Ensure getVia(eventbase) does not busy wait
[folly.git] / folly / io / async / test / EventBaseTest.cpp
1 /*
2  * Licensed to the Apache Software Foundation (ASF) under one
3  * or more contributor license agreements. See the NOTICE file
4  * distributed with this work for additional information
5  * regarding copyright ownership. The ASF licenses this file
6  * to you under the Apache License, Version 2.0 (the
7  * "License"); you may not use this file except in compliance
8  * with the License. You may obtain a copy of the License at
9  *
10  *   http://www.apache.org/licenses/LICENSE-2.0
11  *
12  * Unless required by applicable law or agreed to in writing,
13  * software distributed under the License is distributed on an
14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15  * KIND, either express or implied. See the License for the
16  * specific language governing permissions and limitations
17  * under the License.
18  */
19
20 #include <folly/Memory.h>
21 #include <folly/ScopeGuard.h>
22
23 #include <folly/io/async/AsyncTimeout.h>
24 #include <folly/io/async/EventBase.h>
25 #include <folly/io/async/EventHandler.h>
26 #include <folly/io/async/test/SocketPair.h>
27 #include <folly/io/async/test/Util.h>
28 #include <folly/portability/Unistd.h>
29
30 #include <folly/futures/Promise.h>
31
32 #include <atomic>
33 #include <iostream>
34 #include <memory>
35 #include <thread>
36
37 using std::atomic;
38 using std::deque;
39 using std::pair;
40 using std::vector;
41 using std::unique_ptr;
42 using std::thread;
43 using std::make_pair;
44 using std::cerr;
45 using std::endl;
46 using std::chrono::milliseconds;
47 using std::chrono::microseconds;
48 using std::chrono::duration_cast;
49
50 using namespace folly;
51
52 ///////////////////////////////////////////////////////////////////////////
53 // Tests for read and write events
54 ///////////////////////////////////////////////////////////////////////////
55
56 enum { BUF_SIZE = 4096 };
57
58 ssize_t writeToFD(int fd, size_t length) {
59   // write an arbitrary amount of data to the fd
60   auto bufv = vector<char>(length);
61   auto buf = bufv.data();
62   memset(buf, 'a', length);
63   ssize_t rc = write(fd, buf, length);
64   CHECK_EQ(rc, length);
65   return rc;
66 }
67
68 size_t writeUntilFull(int fd) {
69   // Write to the fd until EAGAIN is returned
70   size_t bytesWritten = 0;
71   char buf[BUF_SIZE];
72   memset(buf, 'a', sizeof(buf));
73   while (true) {
74     ssize_t rc = write(fd, buf, sizeof(buf));
75     if (rc < 0) {
76       CHECK_EQ(errno, EAGAIN);
77       break;
78     } else {
79       bytesWritten += rc;
80     }
81   }
82   return bytesWritten;
83 }
84
85 ssize_t readFromFD(int fd, size_t length) {
86   // write an arbitrary amount of data to the fd
87   auto buf = vector<char>(length);
88   return read(fd, buf.data(), length);
89 }
90
91 size_t readUntilEmpty(int fd) {
92   // Read from the fd until EAGAIN is returned
93   char buf[BUF_SIZE];
94   size_t bytesRead = 0;
95   while (true) {
96     int rc = read(fd, buf, sizeof(buf));
97     if (rc == 0) {
98       CHECK(false) << "unexpected EOF";
99     } else if (rc < 0) {
100       CHECK_EQ(errno, EAGAIN);
101       break;
102     } else {
103       bytesRead += rc;
104     }
105   }
106   return bytesRead;
107 }
108
109 void checkReadUntilEmpty(int fd, size_t expectedLength) {
110   ASSERT_EQ(readUntilEmpty(fd), expectedLength);
111 }
112
113 struct ScheduledEvent {
114   int milliseconds;
115   uint16_t events;
116   size_t length;
117   ssize_t result;
118
119   void perform(int fd) {
120     if (events & EventHandler::READ) {
121       if (length == 0) {
122         result = readUntilEmpty(fd);
123       } else {
124         result = readFromFD(fd, length);
125       }
126     }
127     if (events & EventHandler::WRITE) {
128       if (length == 0) {
129         result = writeUntilFull(fd);
130       } else {
131         result = writeToFD(fd, length);
132       }
133     }
134   }
135 };
136
137 void scheduleEvents(EventBase* eventBase, int fd, ScheduledEvent* events) {
138   for (ScheduledEvent* ev = events; ev->milliseconds > 0; ++ev) {
139     eventBase->tryRunAfterDelay(std::bind(&ScheduledEvent::perform, ev, fd),
140                              ev->milliseconds);
141   }
142 }
143
144 class TestHandler : public EventHandler {
145  public:
146   TestHandler(EventBase* eventBase, int fd)
147     : EventHandler(eventBase, fd), fd_(fd) {}
148
149   void handlerReady(uint16_t events) noexcept override {
150     ssize_t bytesRead = 0;
151     ssize_t bytesWritten = 0;
152     if (events & READ) {
153       // Read all available data, so EventBase will stop calling us
154       // until new data becomes available
155       bytesRead = readUntilEmpty(fd_);
156     }
157     if (events & WRITE) {
158       // Write until the pipe buffer is full, so EventBase will stop calling
159       // us until the other end has read some data
160       bytesWritten = writeUntilFull(fd_);
161     }
162
163     log.emplace_back(events, bytesRead, bytesWritten);
164   }
165
166   struct EventRecord {
167     EventRecord(uint16_t events, size_t bytesRead, size_t bytesWritten)
168       : events(events)
169       , timestamp()
170       , bytesRead(bytesRead)
171       , bytesWritten(bytesWritten) {}
172
173     uint16_t events;
174     TimePoint timestamp;
175     ssize_t bytesRead;
176     ssize_t bytesWritten;
177   };
178
179   deque<EventRecord> log;
180
181  private:
182   int fd_;
183 };
184
185 /**
186  * Test a READ event
187  */
188 TEST(EventBaseTest, ReadEvent) {
189   EventBase eb;
190   SocketPair sp;
191
192   // Register for read events
193   TestHandler handler(&eb, sp[0]);
194   handler.registerHandler(EventHandler::READ);
195
196   // Register timeouts to perform two write events
197   ScheduledEvent events[] = {
198     { 10, EventHandler::WRITE, 2345, 0 },
199     { 160, EventHandler::WRITE, 99, 0 },
200     { 0, 0, 0, 0 },
201   };
202   scheduleEvents(&eb, sp[1], events);
203
204   // Loop
205   TimePoint start;
206   eb.loop();
207   TimePoint end;
208
209   // Since we didn't use the EventHandler::PERSIST flag, the handler should
210   // have received the first read, then unregistered itself.  Check that only
211   // the first chunk of data was received.
212   ASSERT_EQ(handler.log.size(), 1);
213   ASSERT_EQ(handler.log[0].events, EventHandler::READ);
214   T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
215                   milliseconds(events[0].milliseconds), milliseconds(90));
216   ASSERT_EQ(handler.log[0].bytesRead, events[0].length);
217   ASSERT_EQ(handler.log[0].bytesWritten, 0);
218   T_CHECK_TIMEOUT(start, end,
219                   milliseconds(events[1].milliseconds), milliseconds(30));
220
221   // Make sure the second chunk of data is still waiting to be read.
222   size_t bytesRemaining = readUntilEmpty(sp[0]);
223   ASSERT_EQ(bytesRemaining, events[1].length);
224 }
225
226 /**
227  * Test (READ | PERSIST)
228  */
229 TEST(EventBaseTest, ReadPersist) {
230   EventBase eb;
231   SocketPair sp;
232
233   // Register for read events
234   TestHandler handler(&eb, sp[0]);
235   handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
236
237   // Register several timeouts to perform writes
238   ScheduledEvent events[] = {
239     { 10,  EventHandler::WRITE, 1024, 0 },
240     { 20,  EventHandler::WRITE, 2211, 0 },
241     { 30,  EventHandler::WRITE, 4096, 0 },
242     { 100, EventHandler::WRITE, 100,  0 },
243     { 0, 0, 0, 0 },
244   };
245   scheduleEvents(&eb, sp[1], events);
246
247   // Schedule a timeout to unregister the handler after the third write
248   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 85);
249
250   // Loop
251   TimePoint start;
252   eb.loop();
253   TimePoint end;
254
255   // The handler should have received the first 3 events,
256   // then been unregistered after that.
257   ASSERT_EQ(handler.log.size(), 3);
258   for (int n = 0; n < 3; ++n) {
259     ASSERT_EQ(handler.log[n].events, EventHandler::READ);
260     T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
261                     milliseconds(events[n].milliseconds));
262     ASSERT_EQ(handler.log[n].bytesRead, events[n].length);
263     ASSERT_EQ(handler.log[n].bytesWritten, 0);
264   }
265   T_CHECK_TIMEOUT(start, end, milliseconds(events[3].milliseconds));
266
267   // Make sure the data from the last write is still waiting to be read
268   size_t bytesRemaining = readUntilEmpty(sp[0]);
269   ASSERT_EQ(bytesRemaining, events[3].length);
270 }
271
272 /**
273  * Test registering for READ when the socket is immediately readable
274  */
275 TEST(EventBaseTest, ReadImmediate) {
276   EventBase eb;
277   SocketPair sp;
278
279   // Write some data to the socket so the other end will
280   // be immediately readable
281   size_t dataLength = 1234;
282   writeToFD(sp[1], dataLength);
283
284   // Register for read events
285   TestHandler handler(&eb, sp[0]);
286   handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
287
288   // Register a timeout to perform another write
289   ScheduledEvent events[] = {
290     { 10, EventHandler::WRITE, 2345, 0 },
291     { 0, 0, 0, 0 },
292   };
293   scheduleEvents(&eb, sp[1], events);
294
295   // Schedule a timeout to unregister the handler
296   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 20);
297
298   // Loop
299   TimePoint start;
300   eb.loop();
301   TimePoint end;
302
303   ASSERT_EQ(handler.log.size(), 2);
304
305   // There should have been 1 event for immediate readability
306   ASSERT_EQ(handler.log[0].events, EventHandler::READ);
307   T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
308   ASSERT_EQ(handler.log[0].bytesRead, dataLength);
309   ASSERT_EQ(handler.log[0].bytesWritten, 0);
310
311   // There should be another event after the timeout wrote more data
312   ASSERT_EQ(handler.log[1].events, EventHandler::READ);
313   T_CHECK_TIMEOUT(start, handler.log[1].timestamp,
314                   milliseconds(events[0].milliseconds));
315   ASSERT_EQ(handler.log[1].bytesRead, events[0].length);
316   ASSERT_EQ(handler.log[1].bytesWritten, 0);
317
318   T_CHECK_TIMEOUT(start, end, milliseconds(20));
319 }
320
321 /**
322  * Test a WRITE event
323  */
324 TEST(EventBaseTest, WriteEvent) {
325   EventBase eb;
326   SocketPair sp;
327
328   // Fill up the write buffer before starting
329   size_t initialBytesWritten = writeUntilFull(sp[0]);
330
331   // Register for write events
332   TestHandler handler(&eb, sp[0]);
333   handler.registerHandler(EventHandler::WRITE);
334
335   // Register timeouts to perform two reads
336   ScheduledEvent events[] = {
337     { 10, EventHandler::READ, 0, 0 },
338     { 60, EventHandler::READ, 0, 0 },
339     { 0, 0, 0, 0 },
340   };
341   scheduleEvents(&eb, sp[1], events);
342
343   // Loop
344   TimePoint start;
345   eb.loop();
346   TimePoint end;
347
348   // Since we didn't use the EventHandler::PERSIST flag, the handler should
349   // have only been able to write once, then unregistered itself.
350   ASSERT_EQ(handler.log.size(), 1);
351   ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
352   T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
353                   milliseconds(events[0].milliseconds));
354   ASSERT_EQ(handler.log[0].bytesRead, 0);
355   ASSERT_GT(handler.log[0].bytesWritten, 0);
356   T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
357
358   ASSERT_EQ(events[0].result, initialBytesWritten);
359   ASSERT_EQ(events[1].result, handler.log[0].bytesWritten);
360 }
361
362 /**
363  * Test (WRITE | PERSIST)
364  */
365 TEST(EventBaseTest, WritePersist) {
366   EventBase eb;
367   SocketPair sp;
368
369   // Fill up the write buffer before starting
370   size_t initialBytesWritten = writeUntilFull(sp[0]);
371
372   // Register for write events
373   TestHandler handler(&eb, sp[0]);
374   handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
375
376   // Register several timeouts to read from the socket at several intervals
377   ScheduledEvent events[] = {
378     { 10,  EventHandler::READ, 0, 0 },
379     { 40,  EventHandler::READ, 0, 0 },
380     { 70,  EventHandler::READ, 0, 0 },
381     { 100, EventHandler::READ, 0, 0 },
382     { 0, 0, 0, 0 },
383   };
384   scheduleEvents(&eb, sp[1], events);
385
386   // Schedule a timeout to unregister the handler after the third read
387   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 85);
388
389   // Loop
390   TimePoint start;
391   eb.loop();
392   TimePoint end;
393
394   // The handler should have received the first 3 events,
395   // then been unregistered after that.
396   ASSERT_EQ(handler.log.size(), 3);
397   ASSERT_EQ(events[0].result, initialBytesWritten);
398   for (int n = 0; n < 3; ++n) {
399     ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
400     T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
401                     milliseconds(events[n].milliseconds));
402     ASSERT_EQ(handler.log[n].bytesRead, 0);
403     ASSERT_GT(handler.log[n].bytesWritten, 0);
404     ASSERT_EQ(handler.log[n].bytesWritten, events[n + 1].result);
405   }
406   T_CHECK_TIMEOUT(start, end, milliseconds(events[3].milliseconds));
407 }
408
409 /**
410  * Test registering for WRITE when the socket is immediately writable
411  */
412 TEST(EventBaseTest, WriteImmediate) {
413   EventBase eb;
414   SocketPair sp;
415
416   // Register for write events
417   TestHandler handler(&eb, sp[0]);
418   handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
419
420   // Register a timeout to perform a read
421   ScheduledEvent events[] = {
422     { 10, EventHandler::READ, 0, 0 },
423     { 0, 0, 0, 0 },
424   };
425   scheduleEvents(&eb, sp[1], events);
426
427   // Schedule a timeout to unregister the handler
428   int64_t unregisterTimeout = 40;
429   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler),
430                    unregisterTimeout);
431
432   // Loop
433   TimePoint start;
434   eb.loop();
435   TimePoint end;
436
437   ASSERT_EQ(handler.log.size(), 2);
438
439   // Since the socket buffer was initially empty,
440   // there should have been 1 event for immediate writability
441   ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
442   T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
443   ASSERT_EQ(handler.log[0].bytesRead, 0);
444   ASSERT_GT(handler.log[0].bytesWritten, 0);
445
446   // There should be another event after the timeout wrote more data
447   ASSERT_EQ(handler.log[1].events, EventHandler::WRITE);
448   T_CHECK_TIMEOUT(start, handler.log[1].timestamp,
449                   milliseconds(events[0].milliseconds));
450   ASSERT_EQ(handler.log[1].bytesRead, 0);
451   ASSERT_GT(handler.log[1].bytesWritten, 0);
452
453   T_CHECK_TIMEOUT(start, end, milliseconds(unregisterTimeout));
454 }
455
456 /**
457  * Test (READ | WRITE) when the socket becomes readable first
458  */
459 TEST(EventBaseTest, ReadWrite) {
460   EventBase eb;
461   SocketPair sp;
462
463   // Fill up the write buffer before starting
464   size_t sock0WriteLength = writeUntilFull(sp[0]);
465
466   // Register for read and write events
467   TestHandler handler(&eb, sp[0]);
468   handler.registerHandler(EventHandler::READ_WRITE);
469
470   // Register timeouts to perform a write then a read.
471   ScheduledEvent events[] = {
472     { 10, EventHandler::WRITE, 2345, 0 },
473     { 40, EventHandler::READ, 0, 0 },
474     { 0, 0, 0, 0 },
475   };
476   scheduleEvents(&eb, sp[1], events);
477
478   // Loop
479   TimePoint start;
480   eb.loop();
481   TimePoint end;
482
483   // Since we didn't use the EventHandler::PERSIST flag, the handler should
484   // have only noticed readability, then unregistered itself.  Check that only
485   // one event was logged.
486   ASSERT_EQ(handler.log.size(), 1);
487   ASSERT_EQ(handler.log[0].events, EventHandler::READ);
488   T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
489                   milliseconds(events[0].milliseconds));
490   ASSERT_EQ(handler.log[0].bytesRead, events[0].length);
491   ASSERT_EQ(handler.log[0].bytesWritten, 0);
492   ASSERT_EQ(events[1].result, sock0WriteLength);
493   T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
494 }
495
496 /**
497  * Test (READ | WRITE) when the socket becomes writable first
498  */
499 TEST(EventBaseTest, WriteRead) {
500   EventBase eb;
501   SocketPair sp;
502
503   // Fill up the write buffer before starting
504   size_t sock0WriteLength = writeUntilFull(sp[0]);
505
506   // Register for read and write events
507   TestHandler handler(&eb, sp[0]);
508   handler.registerHandler(EventHandler::READ_WRITE);
509
510   // Register timeouts to perform a read then a write.
511   size_t sock1WriteLength = 2345;
512   ScheduledEvent events[] = {
513     { 10, EventHandler::READ, 0, 0 },
514     { 40, EventHandler::WRITE, sock1WriteLength, 0 },
515     { 0, 0, 0, 0 },
516   };
517   scheduleEvents(&eb, sp[1], events);
518
519   // Loop
520   TimePoint start;
521   eb.loop();
522   TimePoint end;
523
524   // Since we didn't use the EventHandler::PERSIST flag, the handler should
525   // have only noticed writability, then unregistered itself.  Check that only
526   // one event was logged.
527   ASSERT_EQ(handler.log.size(), 1);
528   ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
529   T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
530                   milliseconds(events[0].milliseconds));
531   ASSERT_EQ(handler.log[0].bytesRead, 0);
532   ASSERT_GT(handler.log[0].bytesWritten, 0);
533   ASSERT_EQ(events[0].result, sock0WriteLength);
534   ASSERT_EQ(events[1].result, sock1WriteLength);
535   T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
536
537   // Make sure the written data is still waiting to be read.
538   size_t bytesRemaining = readUntilEmpty(sp[0]);
539   ASSERT_EQ(bytesRemaining, events[1].length);
540 }
541
542 /**
543  * Test (READ | WRITE) when the socket becomes readable and writable
544  * at the same time.
545  */
546 TEST(EventBaseTest, ReadWriteSimultaneous) {
547   EventBase eb;
548   SocketPair sp;
549
550   // Fill up the write buffer before starting
551   size_t sock0WriteLength = writeUntilFull(sp[0]);
552
553   // Register for read and write events
554   TestHandler handler(&eb, sp[0]);
555   handler.registerHandler(EventHandler::READ_WRITE);
556
557   // Register a timeout to perform a read and write together
558   ScheduledEvent events[] = {
559     { 10, EventHandler::READ | EventHandler::WRITE, 0, 0 },
560     { 0, 0, 0, 0 },
561   };
562   scheduleEvents(&eb, sp[1], events);
563
564   // Loop
565   TimePoint start;
566   eb.loop();
567   TimePoint end;
568
569   // It's not strictly required that the EventBase register us about both
570   // events in the same call.  So, it's possible that if the EventBase
571   // implementation changes this test could start failing, and it wouldn't be
572   // considered breaking the API.  However for now it's nice to exercise this
573   // code path.
574   ASSERT_EQ(handler.log.size(), 1);
575   ASSERT_EQ(handler.log[0].events,
576                     EventHandler::READ | EventHandler::WRITE);
577   T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
578                   milliseconds(events[0].milliseconds));
579   ASSERT_EQ(handler.log[0].bytesRead, sock0WriteLength);
580   ASSERT_GT(handler.log[0].bytesWritten, 0);
581   T_CHECK_TIMEOUT(start, end, milliseconds(events[0].milliseconds));
582 }
583
584 /**
585  * Test (READ | WRITE | PERSIST)
586  */
587 TEST(EventBaseTest, ReadWritePersist) {
588   EventBase eb;
589   SocketPair sp;
590
591   // Register for read and write events
592   TestHandler handler(&eb, sp[0]);
593   handler.registerHandler(EventHandler::READ | EventHandler::WRITE |
594                           EventHandler::PERSIST);
595
596   // Register timeouts to perform several reads and writes
597   ScheduledEvent events[] = {
598     { 10, EventHandler::WRITE, 2345, 0 },
599     { 20, EventHandler::READ, 0, 0 },
600     { 35, EventHandler::WRITE, 200, 0 },
601     { 45, EventHandler::WRITE, 15, 0 },
602     { 55, EventHandler::READ, 0, 0 },
603     { 120, EventHandler::WRITE, 2345, 0 },
604     { 0, 0, 0, 0 },
605   };
606   scheduleEvents(&eb, sp[1], events);
607
608   // Schedule a timeout to unregister the handler
609   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 80);
610
611   // Loop
612   TimePoint start;
613   eb.loop();
614   TimePoint end;
615
616   ASSERT_EQ(handler.log.size(), 6);
617
618   // Since we didn't fill up the write buffer immediately, there should
619   // be an immediate event for writability.
620   ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
621   T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
622   ASSERT_EQ(handler.log[0].bytesRead, 0);
623   ASSERT_GT(handler.log[0].bytesWritten, 0);
624
625   // Events 1 through 5 should correspond to the scheduled events
626   for (int n = 1; n < 6; ++n) {
627     ScheduledEvent* event = &events[n - 1];
628     T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
629                     milliseconds(event->milliseconds));
630     if (event->events == EventHandler::READ) {
631       ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
632       ASSERT_EQ(handler.log[n].bytesRead, 0);
633       ASSERT_GT(handler.log[n].bytesWritten, 0);
634     } else {
635       ASSERT_EQ(handler.log[n].events, EventHandler::READ);
636       ASSERT_EQ(handler.log[n].bytesRead, event->length);
637       ASSERT_EQ(handler.log[n].bytesWritten, 0);
638     }
639   }
640
641   // The timeout should have unregistered the handler before the last write.
642   // Make sure that data is still waiting to be read
643   size_t bytesRemaining = readUntilEmpty(sp[0]);
644   ASSERT_EQ(bytesRemaining, events[5].length);
645 }
646
647
648 class PartialReadHandler : public TestHandler {
649  public:
650   PartialReadHandler(EventBase* eventBase, int fd, size_t readLength)
651     : TestHandler(eventBase, fd), fd_(fd), readLength_(readLength) {}
652
653   void handlerReady(uint16_t events) noexcept override {
654     assert(events == EventHandler::READ);
655     ssize_t bytesRead = readFromFD(fd_, readLength_);
656     log.emplace_back(events, bytesRead, 0);
657   }
658
659  private:
660   int fd_;
661   size_t readLength_;
662 };
663
664 /**
665  * Test reading only part of the available data when a read event is fired.
666  * When PERSIST is used, make sure the handler gets notified again the next
667  * time around the loop.
668  */
669 TEST(EventBaseTest, ReadPartial) {
670   EventBase eb;
671   SocketPair sp;
672
673   // Register for read events
674   size_t readLength = 100;
675   PartialReadHandler handler(&eb, sp[0], readLength);
676   handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
677
678   // Register a timeout to perform a single write,
679   // with more data than PartialReadHandler will read at once
680   ScheduledEvent events[] = {
681     { 10, EventHandler::WRITE, (3*readLength) + (readLength / 2), 0 },
682     { 0, 0, 0, 0 },
683   };
684   scheduleEvents(&eb, sp[1], events);
685
686   // Schedule a timeout to unregister the handler
687   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 30);
688
689   // Loop
690   TimePoint start;
691   eb.loop();
692   TimePoint end;
693
694   ASSERT_EQ(handler.log.size(), 4);
695
696   // The first 3 invocations should read readLength bytes each
697   for (int n = 0; n < 3; ++n) {
698     ASSERT_EQ(handler.log[n].events, EventHandler::READ);
699     T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
700                     milliseconds(events[0].milliseconds));
701     ASSERT_EQ(handler.log[n].bytesRead, readLength);
702     ASSERT_EQ(handler.log[n].bytesWritten, 0);
703   }
704   // The last read only has readLength/2 bytes
705   ASSERT_EQ(handler.log[3].events, EventHandler::READ);
706   T_CHECK_TIMEOUT(start, handler.log[3].timestamp,
707                   milliseconds(events[0].milliseconds));
708   ASSERT_EQ(handler.log[3].bytesRead, readLength / 2);
709   ASSERT_EQ(handler.log[3].bytesWritten, 0);
710 }
711
712
713 class PartialWriteHandler : public TestHandler {
714  public:
715   PartialWriteHandler(EventBase* eventBase, int fd, size_t writeLength)
716     : TestHandler(eventBase, fd), fd_(fd), writeLength_(writeLength) {}
717
718   void handlerReady(uint16_t events) noexcept override {
719     assert(events == EventHandler::WRITE);
720     ssize_t bytesWritten = writeToFD(fd_, writeLength_);
721     log.emplace_back(events, 0, bytesWritten);
722   }
723
724  private:
725   int fd_;
726   size_t writeLength_;
727 };
728
729 /**
730  * Test writing without completely filling up the write buffer when the fd
731  * becomes writable.  When PERSIST is used, make sure the handler gets
732  * notified again the next time around the loop.
733  */
734 TEST(EventBaseTest, WritePartial) {
735   EventBase eb;
736   SocketPair sp;
737
738   // Fill up the write buffer before starting
739   size_t initialBytesWritten = writeUntilFull(sp[0]);
740
741   // Register for write events
742   size_t writeLength = 100;
743   PartialWriteHandler handler(&eb, sp[0], writeLength);
744   handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
745
746   // Register a timeout to read, so that more data can be written
747   ScheduledEvent events[] = {
748     { 10, EventHandler::READ, 0, 0 },
749     { 0, 0, 0, 0 },
750   };
751   scheduleEvents(&eb, sp[1], events);
752
753   // Schedule a timeout to unregister the handler
754   eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 30);
755
756   // Loop
757   TimePoint start;
758   eb.loop();
759   TimePoint end;
760
761   // Depending on how big the socket buffer is, there will be multiple writes
762   // Only check the first 5
763   int numChecked = 5;
764   ASSERT_GE(handler.log.size(), numChecked);
765   ASSERT_EQ(events[0].result, initialBytesWritten);
766
767   // The first 3 invocations should read writeLength bytes each
768   for (int n = 0; n < numChecked; ++n) {
769     ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
770     T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
771                     milliseconds(events[0].milliseconds));
772     ASSERT_EQ(handler.log[n].bytesRead, 0);
773     ASSERT_EQ(handler.log[n].bytesWritten, writeLength);
774   }
775 }
776
777
778 /**
779  * Test destroying a registered EventHandler
780  */
781 TEST(EventBaseTest, DestroyHandler) {
782   class DestroyHandler : public AsyncTimeout {
783    public:
784     DestroyHandler(EventBase* eb, EventHandler* h)
785       : AsyncTimeout(eb)
786       , handler_(h) {}
787
788     void timeoutExpired() noexcept override { delete handler_; }
789
790    private:
791     EventHandler* handler_;
792   };
793
794   EventBase eb;
795   SocketPair sp;
796
797   // Fill up the write buffer before starting
798   size_t initialBytesWritten = writeUntilFull(sp[0]);
799
800   // Register for write events
801   TestHandler* handler = new TestHandler(&eb, sp[0]);
802   handler->registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
803
804   // After 10ms, read some data, so that the handler
805   // will be notified that it can write.
806   eb.tryRunAfterDelay(std::bind(checkReadUntilEmpty, sp[1], initialBytesWritten),
807                    10);
808
809   // Start a timer to destroy the handler after 25ms
810   // This mainly just makes sure the code doesn't break or assert
811   DestroyHandler dh(&eb, handler);
812   dh.scheduleTimeout(25);
813
814   TimePoint start;
815   eb.loop();
816   TimePoint end;
817
818   // Make sure the EventHandler was uninstalled properly when it was
819   // destroyed, and the EventBase loop exited
820   T_CHECK_TIMEOUT(start, end, milliseconds(25));
821
822   // Make sure that the handler wrote data to the socket
823   // before it was destroyed
824   size_t bytesRemaining = readUntilEmpty(sp[1]);
825   ASSERT_GT(bytesRemaining, 0);
826 }
827
828
829 ///////////////////////////////////////////////////////////////////////////
830 // Tests for timeout events
831 ///////////////////////////////////////////////////////////////////////////
832
833 TEST(EventBaseTest, RunAfterDelay) {
834   EventBase eb;
835
836   TimePoint timestamp1(false);
837   TimePoint timestamp2(false);
838   TimePoint timestamp3(false);
839   eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp1), 10);
840   eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp2), 20);
841   eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp3), 40);
842
843   TimePoint start;
844   eb.loop();
845   TimePoint end;
846
847   T_CHECK_TIMEOUT(start, timestamp1, milliseconds(10));
848   T_CHECK_TIMEOUT(start, timestamp2, milliseconds(20));
849   T_CHECK_TIMEOUT(start, timestamp3, milliseconds(40));
850   T_CHECK_TIMEOUT(start, end, milliseconds(40));
851 }
852
853 /**
854  * Test the behavior of tryRunAfterDelay() when some timeouts are
855  * still scheduled when the EventBase is destroyed.
856  */
857 TEST(EventBaseTest, RunAfterDelayDestruction) {
858   TimePoint timestamp1(false);
859   TimePoint timestamp2(false);
860   TimePoint timestamp3(false);
861   TimePoint timestamp4(false);
862   TimePoint start(false);
863   TimePoint end(false);
864
865   {
866     EventBase eb;
867
868     // Run two normal timeouts
869     eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp1), 10);
870     eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp2), 20);
871
872     // Schedule a timeout to stop the event loop after 40ms
873     eb.tryRunAfterDelay(std::bind(&EventBase::terminateLoopSoon, &eb), 40);
874
875     // Schedule 2 timeouts that would fire after the event loop stops
876     eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp3), 80);
877     eb.tryRunAfterDelay(std::bind(&TimePoint::reset, &timestamp4), 160);
878
879     start.reset();
880     eb.loop();
881     end.reset();
882   }
883
884   T_CHECK_TIMEOUT(start, timestamp1, milliseconds(10));
885   T_CHECK_TIMEOUT(start, timestamp2, milliseconds(20));
886   T_CHECK_TIMEOUT(start, end, milliseconds(40));
887
888   ASSERT_TRUE(timestamp3.isUnset());
889   ASSERT_TRUE(timestamp4.isUnset());
890
891   // Ideally this test should be run under valgrind to ensure that no
892   // memory is leaked.
893 }
894
895 class TestTimeout : public AsyncTimeout {
896  public:
897   explicit TestTimeout(EventBase* eventBase)
898     : AsyncTimeout(eventBase)
899     , timestamp(false) {}
900
901   void timeoutExpired() noexcept override { timestamp.reset(); }
902
903   TimePoint timestamp;
904 };
905
906 TEST(EventBaseTest, BasicTimeouts) {
907   EventBase eb;
908
909   TestTimeout t1(&eb);
910   TestTimeout t2(&eb);
911   TestTimeout t3(&eb);
912   t1.scheduleTimeout(10);
913   t2.scheduleTimeout(20);
914   t3.scheduleTimeout(40);
915
916   TimePoint start;
917   eb.loop();
918   TimePoint end;
919
920   T_CHECK_TIMEOUT(start, t1.timestamp, milliseconds(10));
921   T_CHECK_TIMEOUT(start, t2.timestamp, milliseconds(20));
922   T_CHECK_TIMEOUT(start, t3.timestamp, milliseconds(40));
923   T_CHECK_TIMEOUT(start, end, milliseconds(40));
924 }
925
926 class ReschedulingTimeout : public AsyncTimeout {
927  public:
928   ReschedulingTimeout(EventBase* evb, const vector<uint32_t>& timeouts)
929     : AsyncTimeout(evb)
930     , timeouts_(timeouts)
931     , iterator_(timeouts_.begin()) {}
932
933   void start() {
934     reschedule();
935   }
936
937   void timeoutExpired() noexcept override {
938     timestamps.emplace_back();
939     reschedule();
940   }
941
942   void reschedule() {
943     if (iterator_ != timeouts_.end()) {
944       uint32_t timeout = *iterator_;
945       ++iterator_;
946       scheduleTimeout(timeout);
947     }
948   }
949
950   vector<TimePoint> timestamps;
951
952  private:
953   vector<uint32_t> timeouts_;
954   vector<uint32_t>::const_iterator iterator_;
955 };
956
957 /**
958  * Test rescheduling the same timeout multiple times
959  */
960 TEST(EventBaseTest, ReuseTimeout) {
961   EventBase eb;
962
963   vector<uint32_t> timeouts;
964   timeouts.push_back(10);
965   timeouts.push_back(30);
966   timeouts.push_back(15);
967
968   ReschedulingTimeout t(&eb, timeouts);
969   t.start();
970
971   TimePoint start;
972   eb.loop();
973   TimePoint end;
974
975   // Use a higher tolerance than usual.  We're waiting on 3 timeouts
976   // consecutively.  In general, each timeout may go over by a few
977   // milliseconds, and we're tripling this error by witing on 3 timeouts.
978   milliseconds tolerance{6};
979
980   ASSERT_EQ(timeouts.size(), t.timestamps.size());
981   uint32_t total = 0;
982   for (size_t n = 0; n < timeouts.size(); ++n) {
983     total += timeouts[n];
984     T_CHECK_TIMEOUT(start, t.timestamps[n], milliseconds(total), tolerance);
985   }
986   T_CHECK_TIMEOUT(start, end, milliseconds(total), tolerance);
987 }
988
989 /**
990  * Test rescheduling a timeout before it has fired
991  */
992 TEST(EventBaseTest, RescheduleTimeout) {
993   EventBase eb;
994
995   TestTimeout t1(&eb);
996   TestTimeout t2(&eb);
997   TestTimeout t3(&eb);
998
999   t1.scheduleTimeout(15);
1000   t2.scheduleTimeout(30);
1001   t3.scheduleTimeout(30);
1002
1003   auto f = static_cast<bool(AsyncTimeout::*)(uint32_t)>(
1004       &AsyncTimeout::scheduleTimeout);
1005
1006   // after 10ms, reschedule t2 to run sooner than originally scheduled
1007   eb.tryRunAfterDelay(std::bind(f, &t2, 10), 10);
1008   // after 10ms, reschedule t3 to run later than originally scheduled
1009   eb.tryRunAfterDelay(std::bind(f, &t3, 40), 10);
1010
1011   TimePoint start;
1012   eb.loop();
1013   TimePoint end;
1014
1015   T_CHECK_TIMEOUT(start, t1.timestamp, milliseconds(15));
1016   T_CHECK_TIMEOUT(start, t2.timestamp, milliseconds(20));
1017   T_CHECK_TIMEOUT(start, t3.timestamp, milliseconds(50));
1018   T_CHECK_TIMEOUT(start, end, milliseconds(50));
1019 }
1020
1021 /**
1022  * Test cancelling a timeout
1023  */
1024 TEST(EventBaseTest, CancelTimeout) {
1025   EventBase eb;
1026
1027   vector<uint32_t> timeouts;
1028   timeouts.push_back(10);
1029   timeouts.push_back(30);
1030   timeouts.push_back(25);
1031
1032   ReschedulingTimeout t(&eb, timeouts);
1033   t.start();
1034   eb.tryRunAfterDelay(std::bind(&AsyncTimeout::cancelTimeout, &t), 50);
1035
1036   TimePoint start;
1037   eb.loop();
1038   TimePoint end;
1039
1040   ASSERT_EQ(t.timestamps.size(), 2);
1041   T_CHECK_TIMEOUT(start, t.timestamps[0], milliseconds(10));
1042   T_CHECK_TIMEOUT(start, t.timestamps[1], milliseconds(40));
1043   T_CHECK_TIMEOUT(start, end, milliseconds(50));
1044 }
1045
1046 /**
1047  * Test destroying a scheduled timeout object
1048  */
1049 TEST(EventBaseTest, DestroyTimeout) {
1050   class DestroyTimeout : public AsyncTimeout {
1051    public:
1052     DestroyTimeout(EventBase* eb, AsyncTimeout* t)
1053       : AsyncTimeout(eb)
1054       , timeout_(t) {}
1055
1056     void timeoutExpired() noexcept override { delete timeout_; }
1057
1058    private:
1059     AsyncTimeout* timeout_;
1060   };
1061
1062   EventBase eb;
1063
1064   TestTimeout* t1 = new TestTimeout(&eb);
1065   t1->scheduleTimeout(30);
1066
1067   DestroyTimeout dt(&eb, t1);
1068   dt.scheduleTimeout(10);
1069
1070   TimePoint start;
1071   eb.loop();
1072   TimePoint end;
1073
1074   T_CHECK_TIMEOUT(start, end, milliseconds(10));
1075 }
1076
1077
1078 ///////////////////////////////////////////////////////////////////////////
1079 // Test for runInThreadTestFunc()
1080 ///////////////////////////////////////////////////////////////////////////
1081
1082 struct RunInThreadData {
1083   RunInThreadData(int numThreads, int opsPerThread)
1084     : opsPerThread(opsPerThread)
1085     , opsToGo(numThreads*opsPerThread) {}
1086
1087   EventBase evb;
1088   deque< pair<int, int> > values;
1089
1090   int opsPerThread;
1091   int opsToGo;
1092 };
1093
1094 struct RunInThreadArg {
1095   RunInThreadArg(RunInThreadData* data,
1096                  int threadId,
1097                  int value)
1098     : data(data)
1099     , thread(threadId)
1100     , value(value) {}
1101
1102   RunInThreadData* data;
1103   int thread;
1104   int value;
1105 };
1106
1107 void runInThreadTestFunc(RunInThreadArg* arg) {
1108   arg->data->values.emplace_back(arg->thread, arg->value);
1109   RunInThreadData* data = arg->data;
1110   delete arg;
1111
1112   if(--data->opsToGo == 0) {
1113     // Break out of the event base loop if we are the last thread running
1114     data->evb.terminateLoopSoon();
1115   }
1116 }
1117
1118 TEST(EventBaseTest, RunInThread) {
1119   constexpr uint32_t numThreads = 50;
1120   constexpr uint32_t opsPerThread = 100;
1121   RunInThreadData data(numThreads, opsPerThread);
1122
1123   deque<std::thread> threads;
1124   for (uint32_t i = 0; i < numThreads; ++i) {
1125     threads.emplace_back([i, &data] {
1126         for (int n = 0; n < data.opsPerThread; ++n) {
1127           RunInThreadArg* arg = new RunInThreadArg(&data, i, n);
1128           data.evb.runInEventBaseThread(runInThreadTestFunc, arg);
1129           usleep(10);
1130         }
1131       });
1132   }
1133
1134   // Add a timeout event to run after 3 seconds.
1135   // Otherwise loop() will return immediately since there are no events to run.
1136   // Once the last thread exits, it will stop the loop().  However, this
1137   // timeout also stops the loop in case there is a bug performing the normal
1138   // stop.
1139   data.evb.tryRunAfterDelay(std::bind(&EventBase::terminateLoopSoon, &data.evb),
1140                          3000);
1141
1142   TimePoint start;
1143   data.evb.loop();
1144   TimePoint end;
1145
1146   // Verify that the loop exited because all threads finished and requested it
1147   // to stop.  This should happen much sooner than the 3 second timeout.
1148   // Assert that it happens in under a second.  (This is still tons of extra
1149   // padding.)
1150
1151   auto timeTaken = std::chrono::duration_cast<milliseconds>(
1152     end.getTime() - start.getTime());
1153   ASSERT_LT(timeTaken.count(), 1000);
1154   VLOG(11) << "Time taken: " << timeTaken.count();
1155
1156   // Verify that we have all of the events from every thread
1157   int expectedValues[numThreads];
1158   for (uint32_t n = 0; n < numThreads; ++n) {
1159     expectedValues[n] = 0;
1160   }
1161   for (deque< pair<int, int> >::const_iterator it = data.values.begin();
1162        it != data.values.end();
1163        ++it) {
1164     int threadID = it->first;
1165     int value = it->second;
1166     ASSERT_EQ(expectedValues[threadID], value);
1167     ++expectedValues[threadID];
1168   }
1169   for (uint32_t n = 0; n < numThreads; ++n) {
1170     ASSERT_EQ(expectedValues[n], opsPerThread);
1171   }
1172
1173   // Wait on all of the threads.
1174   for (auto& thread: threads) {
1175     thread.join();
1176   }
1177 }
1178
1179 //  This test simulates some calls, and verifies that the waiting happens by
1180 //  triggering what otherwise would be race conditions, and trying to detect
1181 //  whether any of the race conditions happened.
1182 TEST(EventBaseTest, RunInEventBaseThreadAndWait) {
1183   const size_t c = 256;
1184   vector<unique_ptr<atomic<size_t>>> atoms(c);
1185   for (size_t i = 0; i < c; ++i) {
1186     auto& atom = atoms.at(i);
1187     atom = make_unique<atomic<size_t>>(0);
1188   }
1189   vector<thread> threads(c);
1190   for (size_t i = 0; i < c; ++i) {
1191     auto& atom = *atoms.at(i);
1192     auto& th = threads.at(i);
1193     th = thread([&atom] {
1194         EventBase eb;
1195         auto ebth = thread([&]{ eb.loopForever(); });
1196         eb.waitUntilRunning();
1197         eb.runInEventBaseThreadAndWait([&] {
1198           size_t x = 0;
1199           atom.compare_exchange_weak(
1200               x, 1, std::memory_order_release, std::memory_order_relaxed);
1201         });
1202         size_t x = 0;
1203         atom.compare_exchange_weak(
1204             x, 2, std::memory_order_release, std::memory_order_relaxed);
1205         eb.terminateLoopSoon();
1206         ebth.join();
1207     });
1208   }
1209   for (size_t i = 0; i < c; ++i) {
1210     auto& th = threads.at(i);
1211     th.join();
1212   }
1213   size_t sum = 0;
1214   for (auto& atom : atoms) sum += *atom;
1215   EXPECT_EQ(c, sum);
1216 }
1217
1218 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadAndWaitCross) {
1219   EventBase eb;
1220   thread th(&EventBase::loopForever, &eb);
1221   SCOPE_EXIT {
1222     eb.terminateLoopSoon();
1223     th.join();
1224   };
1225   auto mutated = false;
1226   eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1227       mutated = true;
1228   });
1229   EXPECT_TRUE(mutated);
1230 }
1231
1232 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadAndWaitWithin) {
1233   EventBase eb;
1234   thread th(&EventBase::loopForever, &eb);
1235   SCOPE_EXIT {
1236     eb.terminateLoopSoon();
1237     th.join();
1238   };
1239   eb.runInEventBaseThreadAndWait([&] {
1240       auto mutated = false;
1241       eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1242           mutated = true;
1243       });
1244       EXPECT_TRUE(mutated);
1245   });
1246 }
1247
1248 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadNotLooping) {
1249   EventBase eb;
1250   auto mutated = false;
1251   eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1252       mutated = true;
1253     });
1254   EXPECT_TRUE(mutated);
1255 }
1256
1257 ///////////////////////////////////////////////////////////////////////////
1258 // Tests for runInLoop()
1259 ///////////////////////////////////////////////////////////////////////////
1260
1261 class CountedLoopCallback : public EventBase::LoopCallback {
1262  public:
1263   CountedLoopCallback(EventBase* eventBase,
1264                       unsigned int count,
1265                       std::function<void()> action =
1266                         std::function<void()>())
1267     : eventBase_(eventBase)
1268     , count_(count)
1269     , action_(action) {}
1270
1271   void runLoopCallback() noexcept override {
1272     --count_;
1273     if (count_ > 0) {
1274       eventBase_->runInLoop(this);
1275     } else if (action_) {
1276       action_();
1277     }
1278   }
1279
1280   unsigned int getCount() const {
1281     return count_;
1282   }
1283
1284  private:
1285   EventBase* eventBase_;
1286   unsigned int count_;
1287   std::function<void()> action_;
1288 };
1289
1290 // Test that EventBase::loop() doesn't exit while there are
1291 // still LoopCallbacks remaining to be invoked.
1292 TEST(EventBaseTest, RepeatedRunInLoop) {
1293   EventBase eventBase;
1294
1295   CountedLoopCallback c(&eventBase, 10);
1296   eventBase.runInLoop(&c);
1297   // The callback shouldn't have run immediately
1298   ASSERT_EQ(c.getCount(), 10);
1299   eventBase.loop();
1300
1301   // loop() should loop until the CountedLoopCallback stops
1302   // re-installing itself.
1303   ASSERT_EQ(c.getCount(), 0);
1304 }
1305
1306 // Test that EventBase::loop() works as expected without time measurements.
1307 TEST(EventBaseTest, RunInLoopNoTimeMeasurement) {
1308   EventBase eventBase(false);
1309
1310   CountedLoopCallback c(&eventBase, 10);
1311   eventBase.runInLoop(&c);
1312   // The callback shouldn't have run immediately
1313   ASSERT_EQ(c.getCount(), 10);
1314   eventBase.loop();
1315
1316   // loop() should loop until the CountedLoopCallback stops
1317   // re-installing itself.
1318   ASSERT_EQ(c.getCount(), 0);
1319 }
1320
1321 // Test runInLoop() calls with terminateLoopSoon()
1322 TEST(EventBaseTest, RunInLoopStopLoop) {
1323   EventBase eventBase;
1324
1325   CountedLoopCallback c1(&eventBase, 20);
1326   CountedLoopCallback c2(&eventBase, 10,
1327                          std::bind(&EventBase::terminateLoopSoon, &eventBase));
1328
1329   eventBase.runInLoop(&c1);
1330   eventBase.runInLoop(&c2);
1331   ASSERT_EQ(c1.getCount(), 20);
1332   ASSERT_EQ(c2.getCount(), 10);
1333
1334   eventBase.loopForever();
1335
1336   // c2 should have stopped the loop after 10 iterations
1337   ASSERT_EQ(c2.getCount(), 0);
1338
1339   // We allow the EventBase to run the loop callbacks in whatever order it
1340   // chooses.  We'll accept c1's count being either 10 (if the loop terminated
1341   // after c1 ran on the 10th iteration) or 11 (if c2 terminated the loop
1342   // before c1 ran).
1343   //
1344   // (With the current code, c1 will always run 10 times, but we don't consider
1345   // this a hard API requirement.)
1346   ASSERT_GE(c1.getCount(), 10);
1347   ASSERT_LE(c1.getCount(), 11);
1348 }
1349
1350 TEST(EventBaseTest, TryRunningAfterTerminate) {
1351   EventBase eventBase;
1352   CountedLoopCallback c1(&eventBase, 1,
1353                          std::bind(&EventBase::terminateLoopSoon, &eventBase));
1354   eventBase.runInLoop(&c1);
1355   eventBase.loopForever();
1356   bool ran = false;
1357   eventBase.runInEventBaseThread([&]() {
1358     ran = true;
1359   });
1360
1361   ASSERT_FALSE(ran);
1362 }
1363
1364 // Test cancelling runInLoop() callbacks
1365 TEST(EventBaseTest, CancelRunInLoop) {
1366   EventBase eventBase;
1367
1368   CountedLoopCallback c1(&eventBase, 20);
1369   CountedLoopCallback c2(&eventBase, 20);
1370   CountedLoopCallback c3(&eventBase, 20);
1371
1372   std::function<void()> cancelC1Action =
1373     std::bind(&EventBase::LoopCallback::cancelLoopCallback, &c1);
1374   std::function<void()> cancelC2Action =
1375     std::bind(&EventBase::LoopCallback::cancelLoopCallback, &c2);
1376
1377   CountedLoopCallback cancelC1(&eventBase, 10, cancelC1Action);
1378   CountedLoopCallback cancelC2(&eventBase, 10, cancelC2Action);
1379
1380   // Install cancelC1 after c1
1381   eventBase.runInLoop(&c1);
1382   eventBase.runInLoop(&cancelC1);
1383
1384   // Install cancelC2 before c2
1385   eventBase.runInLoop(&cancelC2);
1386   eventBase.runInLoop(&c2);
1387
1388   // Install c3
1389   eventBase.runInLoop(&c3);
1390
1391   ASSERT_EQ(c1.getCount(), 20);
1392   ASSERT_EQ(c2.getCount(), 20);
1393   ASSERT_EQ(c3.getCount(), 20);
1394   ASSERT_EQ(cancelC1.getCount(), 10);
1395   ASSERT_EQ(cancelC2.getCount(), 10);
1396
1397   // Run the loop
1398   eventBase.loop();
1399
1400   // cancelC1 and cancelC3 should have both fired after 10 iterations and
1401   // stopped re-installing themselves
1402   ASSERT_EQ(cancelC1.getCount(), 0);
1403   ASSERT_EQ(cancelC2.getCount(), 0);
1404   // c3 should have continued on for the full 20 iterations
1405   ASSERT_EQ(c3.getCount(), 0);
1406
1407   // c1 and c2 should have both been cancelled on the 10th iteration.
1408   //
1409   // Callbacks are always run in the order they are installed,
1410   // so c1 should have fired 10 times, and been canceled after it ran on the
1411   // 10th iteration.  c2 should have only fired 9 times, because cancelC2 will
1412   // have run before it on the 10th iteration, and cancelled it before it
1413   // fired.
1414   ASSERT_EQ(c1.getCount(), 10);
1415   ASSERT_EQ(c2.getCount(), 11);
1416 }
1417
1418 class TerminateTestCallback : public EventBase::LoopCallback,
1419                               public EventHandler {
1420  public:
1421   TerminateTestCallback(EventBase* eventBase, int fd)
1422     : EventHandler(eventBase, fd),
1423       eventBase_(eventBase),
1424       loopInvocations_(0),
1425       maxLoopInvocations_(0),
1426       eventInvocations_(0),
1427       maxEventInvocations_(0) {}
1428
1429   void reset(uint32_t maxLoopInvocations, uint32_t maxEventInvocations) {
1430     loopInvocations_ = 0;
1431     maxLoopInvocations_ = maxLoopInvocations;
1432     eventInvocations_ = 0;
1433     maxEventInvocations_ = maxEventInvocations;
1434
1435     cancelLoopCallback();
1436     unregisterHandler();
1437   }
1438
1439   void handlerReady(uint16_t /* events */) noexcept override {
1440     // We didn't register with PERSIST, so we will have been automatically
1441     // unregistered already.
1442     ASSERT_FALSE(isHandlerRegistered());
1443
1444     ++eventInvocations_;
1445     if (eventInvocations_ >= maxEventInvocations_) {
1446       return;
1447     }
1448
1449     eventBase_->runInLoop(this);
1450   }
1451   void runLoopCallback() noexcept override {
1452     ++loopInvocations_;
1453     if (loopInvocations_ >= maxLoopInvocations_) {
1454       return;
1455     }
1456
1457     registerHandler(READ);
1458   }
1459
1460   uint32_t getLoopInvocations() const {
1461     return loopInvocations_;
1462   }
1463   uint32_t getEventInvocations() const {
1464     return eventInvocations_;
1465   }
1466
1467  private:
1468   EventBase* eventBase_;
1469   uint32_t loopInvocations_;
1470   uint32_t maxLoopInvocations_;
1471   uint32_t eventInvocations_;
1472   uint32_t maxEventInvocations_;
1473 };
1474
1475 /**
1476  * Test that EventBase::loop() correctly detects when there are no more events
1477  * left to run.
1478  *
1479  * This uses a single callback, which alternates registering itself as a loop
1480  * callback versus a EventHandler callback.  This exercises a regression where
1481  * EventBase::loop() incorrectly exited if there were no more fd handlers
1482  * registered, but a loop callback installed a new fd handler.
1483  */
1484 TEST(EventBaseTest, LoopTermination) {
1485   EventBase eventBase;
1486
1487   // Open a pipe and close the write end,
1488   // so the read endpoint will be readable
1489   int pipeFds[2];
1490   int rc = pipe(pipeFds);
1491   ASSERT_EQ(rc, 0);
1492   close(pipeFds[1]);
1493   TerminateTestCallback callback(&eventBase, pipeFds[0]);
1494
1495   // Test once where the callback will exit after a loop callback
1496   callback.reset(10, 100);
1497   eventBase.runInLoop(&callback);
1498   eventBase.loop();
1499   ASSERT_EQ(callback.getLoopInvocations(), 10);
1500   ASSERT_EQ(callback.getEventInvocations(), 9);
1501
1502   // Test once where the callback will exit after an fd event callback
1503   callback.reset(100, 7);
1504   eventBase.runInLoop(&callback);
1505   eventBase.loop();
1506   ASSERT_EQ(callback.getLoopInvocations(), 7);
1507   ASSERT_EQ(callback.getEventInvocations(), 7);
1508
1509   close(pipeFds[0]);
1510 }
1511
1512 ///////////////////////////////////////////////////////////////////////////
1513 // Tests for latency calculations
1514 ///////////////////////////////////////////////////////////////////////////
1515
1516 class IdleTimeTimeoutSeries : public AsyncTimeout {
1517
1518  public:
1519
1520   explicit IdleTimeTimeoutSeries(EventBase *base,
1521                                  std::deque<std::uint64_t>& timeout) :
1522     AsyncTimeout(base),
1523     timeouts_(0),
1524     timeout_(timeout) {
1525       scheduleTimeout(1);
1526     }
1527
1528     ~IdleTimeTimeoutSeries() override {}
1529
1530     void timeoutExpired() noexcept override {
1531     ++timeouts_;
1532
1533     if(timeout_.empty()){
1534       cancelTimeout();
1535     } else {
1536       uint64_t sleepTime = timeout_.front();
1537       timeout_.pop_front();
1538       if (sleepTime) {
1539         usleep(sleepTime);
1540       }
1541       scheduleTimeout(1);
1542     }
1543   }
1544
1545   int getTimeouts() const {
1546     return timeouts_;
1547   }
1548
1549  private:
1550   int timeouts_;
1551   std::deque<uint64_t>& timeout_;
1552 };
1553
1554 /**
1555  * Verify that idle time is correctly accounted for when decaying our loop
1556  * time.
1557  *
1558  * This works by creating a high loop time (via usleep), expecting a latency
1559  * callback with known value, and then scheduling a timeout for later. This
1560  * later timeout is far enough in the future that the idle time should have
1561  * caused the loop time to decay.
1562  */
1563 TEST(EventBaseTest, IdleTime) {
1564   EventBase eventBase;
1565   eventBase.setLoadAvgMsec(1000);
1566   eventBase.resetLoadAvg(5900.0);
1567   std::deque<uint64_t> timeouts0(4, 8080);
1568   timeouts0.push_front(8000);
1569   timeouts0.push_back(14000);
1570   IdleTimeTimeoutSeries tos0(&eventBase, timeouts0);
1571   std::deque<uint64_t> timeouts(20, 20);
1572   std::unique_ptr<IdleTimeTimeoutSeries> tos;
1573   int64_t testStart = duration_cast<microseconds>(
1574     std::chrono::steady_clock::now().time_since_epoch()).count();
1575   bool hostOverloaded = false;
1576
1577   int latencyCallbacks = 0;
1578   eventBase.setMaxLatency(6000, [&]() {
1579     ++latencyCallbacks;
1580
1581     switch (latencyCallbacks) {
1582     case 1:
1583       if (tos0.getTimeouts() < 6) {
1584         // This could only happen if the host this test is running
1585         // on is heavily loaded.
1586         int64_t maxLatencyReached = duration_cast<microseconds>(
1587             std::chrono::steady_clock::now().time_since_epoch()).count();
1588         ASSERT_LE(43800, maxLatencyReached - testStart);
1589         hostOverloaded = true;
1590         break;
1591       }
1592       ASSERT_EQ(6, tos0.getTimeouts());
1593       ASSERT_GE(6100, eventBase.getAvgLoopTime() - 1200);
1594       ASSERT_LE(6100, eventBase.getAvgLoopTime() + 1200);
1595       tos.reset(new IdleTimeTimeoutSeries(&eventBase, timeouts));
1596       break;
1597
1598     default:
1599       FAIL() << "Unexpected latency callback";
1600       break;
1601     }
1602   });
1603
1604   // Kick things off with an "immedite" timeout
1605   tos0.scheduleTimeout(1);
1606
1607   eventBase.loop();
1608
1609   if (hostOverloaded) {
1610     return;
1611   }
1612
1613   ASSERT_EQ(1, latencyCallbacks);
1614   ASSERT_EQ(7, tos0.getTimeouts());
1615   ASSERT_GE(5900, eventBase.getAvgLoopTime() - 1200);
1616   ASSERT_LE(5900, eventBase.getAvgLoopTime() + 1200);
1617   ASSERT_TRUE(!!tos);
1618   ASSERT_EQ(21, tos->getTimeouts());
1619 }
1620
1621 /**
1622  * Test that thisLoop functionality works with terminateLoopSoon
1623  */
1624 TEST(EventBaseTest, ThisLoop) {
1625   EventBase eb;
1626   bool runInLoop = false;
1627   bool runThisLoop = false;
1628
1629   eb.runInLoop([&](){
1630       eb.terminateLoopSoon();
1631       eb.runInLoop([&]() {
1632           runInLoop = true;
1633         });
1634       eb.runInLoop([&]() {
1635           runThisLoop = true;
1636         }, true);
1637     }, true);
1638   eb.loopForever();
1639
1640   // Should not work
1641   ASSERT_FALSE(runInLoop);
1642   // Should work with thisLoop
1643   ASSERT_TRUE(runThisLoop);
1644 }
1645
1646 TEST(EventBaseTest, EventBaseThreadLoop) {
1647   EventBase base;
1648   bool ran = false;
1649
1650   base.runInEventBaseThread([&](){
1651     ran = true;
1652   });
1653   base.loop();
1654
1655   ASSERT_EQ(true, ran);
1656 }
1657
1658 TEST(EventBaseTest, EventBaseThreadName) {
1659   EventBase base;
1660   base.setName("foo");
1661   base.loop();
1662
1663 #if (__GLIBC__ >= 2) && (__GLIBC_MINOR__ >= 12)
1664   char name[16];
1665   pthread_getname_np(pthread_self(), name, 16);
1666   ASSERT_EQ(0, strcmp("foo", name));
1667 #endif
1668 }
1669
1670 TEST(EventBaseTest, RunBeforeLoop) {
1671   EventBase base;
1672   CountedLoopCallback cb(&base, 1, [&](){
1673     base.terminateLoopSoon();
1674   });
1675   base.runBeforeLoop(&cb);
1676   base.loopForever();
1677   ASSERT_EQ(cb.getCount(), 0);
1678 }
1679
1680 TEST(EventBaseTest, RunBeforeLoopWait) {
1681   EventBase base;
1682   CountedLoopCallback cb(&base, 1);
1683   base.tryRunAfterDelay([&](){
1684       base.terminateLoopSoon();
1685     }, 500);
1686   base.runBeforeLoop(&cb);
1687   base.loopForever();
1688
1689   // Check that we only ran once, and did not loop multiple times.
1690   ASSERT_EQ(cb.getCount(), 0);
1691 }
1692
1693 class PipeHandler : public EventHandler {
1694 public:
1695   PipeHandler(EventBase* eventBase, int fd)
1696     : EventHandler(eventBase, fd) {}
1697
1698   void handlerReady(uint16_t /* events */) noexcept override { abort(); }
1699 };
1700
1701 TEST(EventBaseTest, StopBeforeLoop) {
1702   EventBase evb;
1703
1704   // Give the evb something to do.
1705   int p[2];
1706   ASSERT_EQ(0, pipe(p));
1707   PipeHandler handler(&evb, p[0]);
1708   handler.registerHandler(EventHandler::READ);
1709
1710   // It's definitely not running yet
1711   evb.terminateLoopSoon();
1712
1713   // let it run, it should exit quickly.
1714   std::thread t([&] { evb.loop(); });
1715   t.join();
1716
1717   handler.unregisterHandler();
1718   close(p[0]);
1719   close(p[1]);
1720
1721   SUCCEED();
1722 }
1723
1724 TEST(EventBaseTest, RunCallbacksOnDestruction) {
1725   bool ran = false;
1726
1727   {
1728     EventBase base;
1729     base.runInEventBaseThread([&](){
1730       ran = true;
1731     });
1732   }
1733
1734   ASSERT_TRUE(ran);
1735 }
1736
1737 TEST(EventBaseTest, LoopKeepAlive) {
1738   EventBase evb;
1739
1740   bool done = false;
1741   std::thread t([&, loopKeepAlive = evb.loopKeepAlive() ]() mutable {
1742     /* sleep override */ std::this_thread::sleep_for(
1743         std::chrono::milliseconds(100));
1744     evb.runInEventBaseThread(
1745         [&done, loopKeepAlive = std::move(loopKeepAlive) ] { done = true; });
1746   });
1747
1748   evb.loop();
1749
1750   ASSERT_TRUE(done);
1751
1752   t.join();
1753 }
1754
1755 TEST(EventBaseTest, LoopKeepAliveInLoop) {
1756   EventBase evb;
1757
1758   bool done = false;
1759   std::thread t;
1760
1761   evb.runInEventBaseThread([&] {
1762     t = std::thread([&, loopKeepAlive = evb.loopKeepAlive() ]() mutable {
1763       /* sleep override */ std::this_thread::sleep_for(
1764           std::chrono::milliseconds(100));
1765       evb.runInEventBaseThread(
1766           [&done, loopKeepAlive = std::move(loopKeepAlive) ] { done = true; });
1767     });
1768   });
1769
1770   evb.loop();
1771
1772   ASSERT_TRUE(done);
1773
1774   t.join();
1775 }
1776
1777 TEST(EventBaseTest, LoopKeepAliveWithLoopForever) {
1778   std::unique_ptr<EventBase> evb = folly::make_unique<EventBase>();
1779
1780   bool done = false;
1781
1782   std::thread evThread([&] {
1783     evb->loopForever();
1784     evb.reset();
1785     done = true;
1786   });
1787
1788   {
1789     auto* ev = evb.get();
1790     EventBase::LoopKeepAlive keepAlive;
1791     ev->runInEventBaseThreadAndWait(
1792         [&ev, &keepAlive] { keepAlive = ev->loopKeepAlive(); });
1793     ASSERT_FALSE(done) << "Loop finished before we asked it to";
1794     ev->terminateLoopSoon();
1795     /* sleep override */
1796     std::this_thread::sleep_for(std::chrono::milliseconds(30));
1797     ASSERT_FALSE(done) << "Loop terminated early";
1798     ev->runInEventBaseThread([&ev, keepAlive = std::move(keepAlive) ]{});
1799   }
1800
1801   evThread.join();
1802   ASSERT_TRUE(done);
1803 }
1804
1805 TEST(EventBaseTest, LoopKeepAliveShutdown) {
1806   auto evb = folly::make_unique<EventBase>();
1807
1808   bool done = false;
1809
1810   std::thread t([
1811     &done,
1812     loopKeepAlive = evb->loopKeepAlive(),
1813     evbPtr = evb.get()
1814   ]() mutable {
1815     /* sleep override */ std::this_thread::sleep_for(
1816         std::chrono::milliseconds(100));
1817     evbPtr->runInEventBaseThread(
1818         [&done, loopKeepAlive = std::move(loopKeepAlive) ] { done = true; });
1819   });
1820
1821   evb.reset();
1822
1823   ASSERT_TRUE(done);
1824
1825   t.join();
1826 }
1827
1828 TEST(EventBaseTest, DrivableExecutorTest) {
1829   folly::Promise<bool> p;
1830   auto f = p.getFuture();
1831   EventBase base;
1832   bool finished = false;
1833
1834   std::thread t([&] {
1835     /* sleep override */
1836     std::this_thread::sleep_for(std::chrono::microseconds(10));
1837     finished = true;
1838     base.runInEventBaseThread([&]() { p.setValue(true); });
1839   });
1840
1841   // Ensure drive does not busy wait
1842   base.drive(); // TODO: fix notification queue init() extra wakeup
1843   base.drive();
1844   EXPECT_TRUE(finished);
1845
1846   folly::Promise<bool> p2;
1847   auto f2 = p2.getFuture();
1848   // Ensure waitVia gets woken up properly, even from
1849   // a separate thread.
1850   base.runAfterDelay([&]() { p2.setValue(true); }, 10);
1851   f2.waitVia(&base);
1852   EXPECT_TRUE(f2.isReady());
1853
1854   t.join();
1855 }