1 /*
2 * Copyright (C) 2015 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "fdevent.h"
18
19 #include <gtest/gtest.h>
20
21 #include <chrono>
22 #include <limits>
23 #include <memory>
24 #include <queue>
25 #include <string>
26 #include <thread>
27 #include <vector>
28
29 #include "adb_io.h"
30 #include "fdevent_test.h"
31
32 using namespace std::chrono_literals;
33
34 class FdHandler {
35 public:
FdHandler(int read_fd,int write_fd,bool use_new_callback)36 FdHandler(int read_fd, int write_fd, bool use_new_callback)
37 : read_fd_(read_fd), write_fd_(write_fd) {
38 if (use_new_callback) {
39 read_fde_ = fdevent_create(read_fd_, FdEventNewCallback, this);
40 write_fde_ = fdevent_create(write_fd_, FdEventNewCallback, this);
41 } else {
42 read_fde_ = fdevent_create(read_fd_, FdEventCallback, this);
43 write_fde_ = fdevent_create(write_fd_, FdEventCallback, this);
44 }
45 fdevent_add(read_fde_, FDE_READ);
46 }
47
~FdHandler()48 ~FdHandler() {
49 fdevent_destroy(read_fde_);
50 fdevent_destroy(write_fde_);
51 }
52
53 private:
FdEventCallback(int fd,unsigned events,void * userdata)54 static void FdEventCallback(int fd, unsigned events, void* userdata) {
55 FdHandler* handler = reinterpret_cast<FdHandler*>(userdata);
56 ASSERT_EQ(0u, (events & ~(FDE_READ | FDE_WRITE))) << "unexpected events: " << events;
57 if (events & FDE_READ) {
58 ASSERT_EQ(fd, handler->read_fd_);
59 char c;
60 ASSERT_EQ(1, adb_read(fd, &c, 1));
61 handler->queue_.push(c);
62 fdevent_add(handler->write_fde_, FDE_WRITE);
63 }
64 if (events & FDE_WRITE) {
65 ASSERT_EQ(fd, handler->write_fd_);
66 ASSERT_FALSE(handler->queue_.empty());
67 char c = handler->queue_.front();
68 handler->queue_.pop();
69 ASSERT_EQ(1, adb_write(fd, &c, 1));
70 if (handler->queue_.empty()) {
71 fdevent_del(handler->write_fde_, FDE_WRITE);
72 }
73 }
74 }
75
FdEventNewCallback(fdevent * fde,unsigned events,void * userdata)76 static void FdEventNewCallback(fdevent* fde, unsigned events, void* userdata) {
77 int fd = fde->fd.get();
78 FdHandler* handler = reinterpret_cast<FdHandler*>(userdata);
79 ASSERT_EQ(0u, (events & ~(FDE_READ | FDE_WRITE))) << "unexpected events: " << events;
80 if (events & FDE_READ) {
81 ASSERT_EQ(fd, handler->read_fd_);
82 char c;
83 ASSERT_EQ(1, adb_read(fd, &c, 1));
84 handler->queue_.push(c);
85 fdevent_add(handler->write_fde_, FDE_WRITE);
86 }
87 if (events & FDE_WRITE) {
88 ASSERT_EQ(fd, handler->write_fd_);
89 ASSERT_FALSE(handler->queue_.empty());
90 char c = handler->queue_.front();
91 handler->queue_.pop();
92 ASSERT_EQ(1, adb_write(fd, &c, 1));
93 if (handler->queue_.empty()) {
94 fdevent_del(handler->write_fde_, FDE_WRITE);
95 }
96 }
97 }
98
99 private:
100 const int read_fd_;
101 const int write_fd_;
102 fdevent* read_fde_;
103 fdevent* write_fde_;
104 std::queue<char> queue_;
105 };
106
107 struct ThreadArg {
108 int first_read_fd;
109 int last_write_fd;
110 size_t middle_pipe_count;
111 };
112
TEST_F(FdeventTest,fdevent_terminate)113 TEST_F(FdeventTest, fdevent_terminate) {
114 PrepareThread();
115 TerminateThread();
116 }
117
TEST_F(FdeventTest,smoke)118 TEST_F(FdeventTest, smoke) {
119 for (bool use_new_callback : {true, false}) {
120 fdevent_reset();
121 const size_t PIPE_COUNT = 10;
122 const size_t MESSAGE_LOOP_COUNT = 100;
123 const std::string MESSAGE = "fdevent_test";
124 int fd_pair1[2];
125 int fd_pair2[2];
126 ASSERT_EQ(0, adb_socketpair(fd_pair1));
127 ASSERT_EQ(0, adb_socketpair(fd_pair2));
128 ThreadArg thread_arg;
129 thread_arg.first_read_fd = fd_pair1[0];
130 thread_arg.last_write_fd = fd_pair2[1];
131 thread_arg.middle_pipe_count = PIPE_COUNT;
132 int writer = fd_pair1[1];
133 int reader = fd_pair2[0];
134
135 PrepareThread();
136
137 std::vector<std::unique_ptr<FdHandler>> fd_handlers;
138 fdevent_run_on_main_thread([&thread_arg, &fd_handlers, use_new_callback]() {
139 std::vector<int> read_fds;
140 std::vector<int> write_fds;
141
142 read_fds.push_back(thread_arg.first_read_fd);
143 for (size_t i = 0; i < thread_arg.middle_pipe_count; ++i) {
144 int fds[2];
145 ASSERT_EQ(0, adb_socketpair(fds));
146 read_fds.push_back(fds[0]);
147 write_fds.push_back(fds[1]);
148 }
149 write_fds.push_back(thread_arg.last_write_fd);
150
151 for (size_t i = 0; i < read_fds.size(); ++i) {
152 fd_handlers.push_back(
153 std::make_unique<FdHandler>(read_fds[i], write_fds[i], use_new_callback));
154 }
155 });
156 WaitForFdeventLoop();
157
158 for (size_t i = 0; i < MESSAGE_LOOP_COUNT; ++i) {
159 std::string read_buffer = MESSAGE;
160 std::string write_buffer(MESSAGE.size(), 'a');
161 ASSERT_TRUE(WriteFdExactly(writer, read_buffer.c_str(), read_buffer.size()));
162 ASSERT_TRUE(ReadFdExactly(reader, &write_buffer[0], write_buffer.size()));
163 ASSERT_EQ(read_buffer, write_buffer);
164 }
165
166 fdevent_run_on_main_thread([&fd_handlers]() { fd_handlers.clear(); });
167 WaitForFdeventLoop();
168
169 TerminateThread();
170 ASSERT_EQ(0, adb_close(writer));
171 ASSERT_EQ(0, adb_close(reader));
172 }
173 }
174
175 struct InvalidFdArg {
176 fdevent* fde;
177 unsigned expected_events;
178 size_t* happened_event_count;
179 };
180
InvalidFdEventCallback(int,unsigned events,void * userdata)181 static void InvalidFdEventCallback(int, unsigned events, void* userdata) {
182 InvalidFdArg* arg = reinterpret_cast<InvalidFdArg*>(userdata);
183 ASSERT_EQ(arg->expected_events, events);
184 fdevent_destroy(arg->fde);
185 if (++*(arg->happened_event_count) == 2) {
186 fdevent_terminate_loop();
187 }
188 }
189
InvalidFdThreadFunc()190 static void InvalidFdThreadFunc() {
191 const int INVALID_READ_FD = std::numeric_limits<int>::max() - 1;
192 size_t happened_event_count = 0;
193 InvalidFdArg read_arg;
194 read_arg.expected_events = FDE_READ | FDE_ERROR;
195 read_arg.happened_event_count = &happened_event_count;
196 read_arg.fde = fdevent_create(INVALID_READ_FD, InvalidFdEventCallback, &read_arg);
197 fdevent_add(read_arg.fde, FDE_READ);
198
199 const int INVALID_WRITE_FD = std::numeric_limits<int>::max();
200 InvalidFdArg write_arg;
201 write_arg.expected_events = FDE_READ | FDE_ERROR;
202 write_arg.happened_event_count = &happened_event_count;
203 write_arg.fde = fdevent_create(INVALID_WRITE_FD, InvalidFdEventCallback, &write_arg);
204 fdevent_add(write_arg.fde, FDE_WRITE);
205 fdevent_loop();
206 }
207
TEST_F(FdeventTest,invalid_fd)208 TEST_F(FdeventTest, invalid_fd) {
209 std::thread thread(InvalidFdThreadFunc);
210 thread.join();
211 }
212
TEST_F(FdeventTest,run_on_main_thread)213 TEST_F(FdeventTest, run_on_main_thread) {
214 std::vector<int> vec;
215
216 PrepareThread();
217
218 // Block the main thread for a long time while we queue our callbacks.
219 fdevent_run_on_main_thread([]() {
220 check_main_thread();
221 std::this_thread::sleep_for(std::chrono::seconds(1));
222 });
223
224 for (int i = 0; i < 1000000; ++i) {
225 fdevent_run_on_main_thread([i, &vec]() {
226 check_main_thread();
227 vec.push_back(i);
228 });
229 }
230
231 TerminateThread();
232
233 ASSERT_EQ(1000000u, vec.size());
234 for (int i = 0; i < 1000000; ++i) {
235 ASSERT_EQ(i, vec[i]);
236 }
237 }
238
make_appender(std::vector<int> * vec,int value)239 static std::function<void()> make_appender(std::vector<int>* vec, int value) {
240 return [vec, value]() {
241 check_main_thread();
242 if (value == 100) {
243 return;
244 }
245
246 vec->push_back(value);
247 fdevent_run_on_main_thread(make_appender(vec, value + 1));
248 };
249 }
250
TEST_F(FdeventTest,run_on_main_thread_reentrant)251 TEST_F(FdeventTest, run_on_main_thread_reentrant) {
252 std::vector<int> vec;
253
254 PrepareThread();
255 fdevent_run_on_main_thread(make_appender(&vec, 0));
256 TerminateThread();
257
258 ASSERT_EQ(100u, vec.size());
259 for (int i = 0; i < 100; ++i) {
260 ASSERT_EQ(i, vec[i]);
261 }
262 }
263
TEST_F(FdeventTest,timeout)264 TEST_F(FdeventTest, timeout) {
265 fdevent_reset();
266 PrepareThread();
267
268 enum class TimeoutEvent {
269 read,
270 timeout,
271 done,
272 };
273
274 struct TimeoutTest {
275 std::vector<std::pair<TimeoutEvent, std::chrono::steady_clock::time_point>> events;
276 fdevent* fde;
277 };
278 TimeoutTest test;
279
280 int fds[2];
281 ASSERT_EQ(0, adb_socketpair(fds));
282 static constexpr auto delta = 100ms;
283 fdevent_run_on_main_thread([&]() {
284 test.fde = fdevent_create(fds[0], [](fdevent* fde, unsigned events, void* arg) {
285 auto test = static_cast<TimeoutTest*>(arg);
286 auto now = std::chrono::steady_clock::now();
287 CHECK((events & FDE_READ) ^ (events & FDE_TIMEOUT));
288 TimeoutEvent event;
289 if ((events & FDE_READ)) {
290 char buf[2];
291 ssize_t rc = adb_read(fde->fd.get(), buf, sizeof(buf));
292 if (rc == 0) {
293 event = TimeoutEvent::done;
294 } else if (rc == 1) {
295 event = TimeoutEvent::read;
296 } else {
297 abort();
298 }
299 } else if ((events & FDE_TIMEOUT)) {
300 event = TimeoutEvent::timeout;
301 } else {
302 abort();
303 }
304
305 CHECK_EQ(fde, test->fde);
306 test->events.emplace_back(event, now);
307
308 if (event == TimeoutEvent::done) {
309 fdevent_destroy(fde);
310 }
311 }, &test);
312 fdevent_add(test.fde, FDE_READ);
313 fdevent_set_timeout(test.fde, delta);
314 });
315
316 ASSERT_EQ(1, adb_write(fds[1], "", 1));
317
318 // Timeout should happen here
319 std::this_thread::sleep_for(delta);
320
321 // and another.
322 std::this_thread::sleep_for(delta);
323
324 // No timeout should happen here.
325 std::this_thread::sleep_for(delta / 2);
326 adb_close(fds[1]);
327
328 TerminateThread();
329
330 ASSERT_EQ(4ULL, test.events.size());
331 ASSERT_EQ(TimeoutEvent::read, test.events[0].first);
332 ASSERT_EQ(TimeoutEvent::timeout, test.events[1].first);
333 ASSERT_EQ(TimeoutEvent::timeout, test.events[2].first);
334 ASSERT_EQ(TimeoutEvent::done, test.events[3].first);
335
336 std::vector<int> time_deltas;
337 for (size_t i = 0; i < test.events.size() - 1; ++i) {
338 auto before = test.events[i].second;
339 auto after = test.events[i + 1].second;
340 auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(after - before);
341 time_deltas.push_back(diff.count());
342 }
343
344 std::vector<int> expected = {
345 delta.count(),
346 delta.count(),
347 delta.count() / 2,
348 };
349
350 std::vector<int> diff;
351 ASSERT_EQ(time_deltas.size(), expected.size());
352 for (size_t i = 0; i < time_deltas.size(); ++i) {
353 diff.push_back(std::abs(time_deltas[i] - expected[i]));
354 }
355
356 ASSERT_LT(diff[0], delta.count() * 0.5);
357 ASSERT_LT(diff[1], delta.count() * 0.5);
358 ASSERT_LT(diff[2], delta.count() * 0.5);
359 }
360