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 = 512;
122 const size_t MESSAGE_LOOP_COUNT = 10;
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
TEST_F(FdeventTest,run_on_main_thread)175 TEST_F(FdeventTest, run_on_main_thread) {
176 std::vector<int> vec;
177
178 PrepareThread();
179
180 // Block the main thread for a long time while we queue our callbacks.
181 fdevent_run_on_main_thread([]() {
182 check_main_thread();
183 std::this_thread::sleep_for(std::chrono::seconds(1));
184 });
185
186 for (int i = 0; i < 1000000; ++i) {
187 fdevent_run_on_main_thread([i, &vec]() {
188 check_main_thread();
189 vec.push_back(i);
190 });
191 }
192
193 TerminateThread();
194
195 ASSERT_EQ(1000000u, vec.size());
196 for (int i = 0; i < 1000000; ++i) {
197 ASSERT_EQ(i, vec[i]);
198 }
199 }
200
make_appender(std::vector<int> * vec,int value)201 static std::function<void()> make_appender(std::vector<int>* vec, int value) {
202 return [vec, value]() {
203 check_main_thread();
204 if (value == 100) {
205 return;
206 }
207
208 vec->push_back(value);
209 fdevent_run_on_main_thread(make_appender(vec, value + 1));
210 };
211 }
212
TEST_F(FdeventTest,run_on_main_thread_reentrant)213 TEST_F(FdeventTest, run_on_main_thread_reentrant) {
214 std::vector<int> vec;
215
216 PrepareThread();
217 fdevent_run_on_main_thread(make_appender(&vec, 0));
218 TerminateThread();
219
220 ASSERT_EQ(100u, vec.size());
221 for (int i = 0; i < 100; ++i) {
222 ASSERT_EQ(i, vec[i]);
223 }
224 }
225
TEST_F(FdeventTest,timeout)226 TEST_F(FdeventTest, timeout) {
227 fdevent_reset();
228 PrepareThread();
229
230 enum class TimeoutEvent {
231 read,
232 timeout,
233 done,
234 };
235
236 struct TimeoutTest {
237 std::vector<std::pair<TimeoutEvent, std::chrono::steady_clock::time_point>> events;
238 fdevent* fde;
239 };
240 TimeoutTest test;
241
242 int fds[2];
243 ASSERT_EQ(0, adb_socketpair(fds));
244 static constexpr auto delta = 100ms;
245 fdevent_run_on_main_thread([&]() {
246 test.fde = fdevent_create(fds[0], [](fdevent* fde, unsigned events, void* arg) {
247 auto test = static_cast<TimeoutTest*>(arg);
248 auto now = std::chrono::steady_clock::now();
249 CHECK((events & FDE_READ) ^ (events & FDE_TIMEOUT));
250 TimeoutEvent event;
251 if ((events & FDE_READ)) {
252 char buf[2];
253 ssize_t rc = adb_read(fde->fd.get(), buf, sizeof(buf));
254 if (rc == 0) {
255 event = TimeoutEvent::done;
256 } else if (rc == 1) {
257 event = TimeoutEvent::read;
258 } else {
259 abort();
260 }
261 } else if ((events & FDE_TIMEOUT)) {
262 event = TimeoutEvent::timeout;
263 } else {
264 abort();
265 }
266
267 CHECK_EQ(fde, test->fde);
268 test->events.emplace_back(event, now);
269
270 if (event == TimeoutEvent::done) {
271 fdevent_destroy(fde);
272 }
273 }, &test);
274 fdevent_add(test.fde, FDE_READ);
275 fdevent_set_timeout(test.fde, delta);
276 });
277
278 ASSERT_EQ(1, adb_write(fds[1], "", 1));
279
280 // Timeout should happen here
281 std::this_thread::sleep_for(delta);
282
283 // and another.
284 std::this_thread::sleep_for(delta);
285
286 // No timeout should happen here.
287 std::this_thread::sleep_for(delta / 2);
288 adb_close(fds[1]);
289
290 TerminateThread();
291
292 ASSERT_EQ(4ULL, test.events.size());
293 ASSERT_EQ(TimeoutEvent::read, test.events[0].first);
294 ASSERT_EQ(TimeoutEvent::timeout, test.events[1].first);
295 ASSERT_EQ(TimeoutEvent::timeout, test.events[2].first);
296 ASSERT_EQ(TimeoutEvent::done, test.events[3].first);
297
298 std::vector<int> time_deltas;
299 for (size_t i = 0; i < test.events.size() - 1; ++i) {
300 auto before = test.events[i].second;
301 auto after = test.events[i + 1].second;
302 auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(after - before);
303 time_deltas.push_back(diff.count());
304 }
305
306 std::vector<int> expected = {
307 delta.count(),
308 delta.count(),
309 delta.count() / 2,
310 };
311
312 std::vector<int> diff;
313 ASSERT_EQ(time_deltas.size(), expected.size());
314 for (size_t i = 0; i < time_deltas.size(); ++i) {
315 diff.push_back(std::abs(time_deltas[i] - expected[i]));
316 }
317
318 ASSERT_LT(diff[0], delta.count() * 0.5);
319 ASSERT_LT(diff[1], delta.count() * 0.5);
320 ASSERT_LT(diff[2], delta.count() * 0.5);
321 }
322