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
10 * http://www.apache.org/licenses/LICENSE-2.0
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
19 #include <folly/Memory.h>
21 #include <folly/io/async/AsyncTimeout.h>
22 #include <folly/io/async/EventBase.h>
23 #include <folly/io/async/EventHandler.h>
24 #include <folly/io/async/test/SocketPair.h>
25 #include <folly/io/async/test/Util.h>
37 using std::unique_ptr;
42 using std::chrono::milliseconds;
43 using std::chrono::microseconds;
44 using std::chrono::duration_cast;
46 using namespace folly;
48 ///////////////////////////////////////////////////////////////////////////
49 // Tests for read and write events
50 ///////////////////////////////////////////////////////////////////////////
52 enum { BUF_SIZE = 4096 };
54 ssize_t writeToFD(int fd, size_t length) {
55 // write an arbitrary amount of data to the fd
57 memset(buf, 'a', sizeof(buf));
58 ssize_t rc = write(fd, buf, sizeof(buf));
63 size_t writeUntilFull(int fd) {
64 // Write to the fd until EAGAIN is returned
65 size_t bytesWritten = 0;
67 memset(buf, 'a', sizeof(buf));
69 ssize_t rc = write(fd, buf, sizeof(buf));
71 CHECK_EQ(errno, EAGAIN);
80 ssize_t readFromFD(int fd, size_t length) {
81 // write an arbitrary amount of data to the fd
83 return read(fd, buf, sizeof(buf));
86 size_t readUntilEmpty(int fd) {
87 // Read from the fd until EAGAIN is returned
91 int rc = read(fd, buf, sizeof(buf));
93 CHECK(false) << "unexpected EOF";
95 CHECK_EQ(errno, EAGAIN);
104 void checkReadUntilEmpty(int fd, size_t expectedLength) {
105 ASSERT_EQ(readUntilEmpty(fd), expectedLength);
108 struct ScheduledEvent {
114 void perform(int fd) {
115 if (events & EventHandler::READ) {
117 result = readUntilEmpty(fd);
119 result = readFromFD(fd, length);
122 if (events & EventHandler::WRITE) {
124 result = writeUntilFull(fd);
126 result = writeToFD(fd, length);
132 void scheduleEvents(EventBase* eventBase, int fd, ScheduledEvent* events) {
133 for (ScheduledEvent* ev = events; ev->milliseconds > 0; ++ev) {
134 eventBase->tryRunAfterDelay(std::bind(&ScheduledEvent::perform, ev, fd),
139 class TestHandler : public EventHandler {
141 TestHandler(EventBase* eventBase, int fd)
142 : EventHandler(eventBase, fd), fd_(fd) {}
144 void handlerReady(uint16_t events) noexcept override {
145 ssize_t bytesRead = 0;
146 ssize_t bytesWritten = 0;
148 // Read all available data, so EventBase will stop calling us
149 // until new data becomes available
150 bytesRead = readUntilEmpty(fd_);
152 if (events & WRITE) {
153 // Write until the pipe buffer is full, so EventBase will stop calling
154 // us until the other end has read some data
155 bytesWritten = writeUntilFull(fd_);
158 log.emplace_back(events, bytesRead, bytesWritten);
162 EventRecord(uint16_t events, size_t bytesRead, size_t bytesWritten)
165 , bytesRead(bytesRead)
166 , bytesWritten(bytesWritten) {}
171 ssize_t bytesWritten;
174 deque<EventRecord> log;
183 TEST(EventBaseTest, ReadEvent) {
187 // Register for read events
188 TestHandler handler(&eb, sp[0]);
189 handler.registerHandler(EventHandler::READ);
191 // Register timeouts to perform two write events
192 ScheduledEvent events[] = {
193 { 10, EventHandler::WRITE, 2345 },
194 { 160, EventHandler::WRITE, 99 },
197 scheduleEvents(&eb, sp[1], events);
204 // Since we didn't use the EventHandler::PERSIST flag, the handler should
205 // have received the first read, then unregistered itself. Check that only
206 // the first chunk of data was received.
207 ASSERT_EQ(handler.log.size(), 1);
208 ASSERT_EQ(handler.log[0].events, EventHandler::READ);
209 T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
210 milliseconds(events[0].milliseconds), milliseconds(90));
211 ASSERT_EQ(handler.log[0].bytesRead, events[0].length);
212 ASSERT_EQ(handler.log[0].bytesWritten, 0);
213 T_CHECK_TIMEOUT(start, end,
214 milliseconds(events[1].milliseconds), milliseconds(30));
216 // Make sure the second chunk of data is still waiting to be read.
217 size_t bytesRemaining = readUntilEmpty(sp[0]);
218 ASSERT_EQ(bytesRemaining, events[1].length);
222 * Test (READ | PERSIST)
224 TEST(EventBaseTest, ReadPersist) {
228 // Register for read events
229 TestHandler handler(&eb, sp[0]);
230 handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
232 // Register several timeouts to perform writes
233 ScheduledEvent events[] = {
234 { 10, EventHandler::WRITE, 1024 },
235 { 20, EventHandler::WRITE, 2211 },
236 { 30, EventHandler::WRITE, 4096 },
237 { 100, EventHandler::WRITE, 100 },
240 scheduleEvents(&eb, sp[1], events);
242 // Schedule a timeout to unregister the handler after the third write
243 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 85);
250 // The handler should have received the first 3 events,
251 // then been unregistered after that.
252 ASSERT_EQ(handler.log.size(), 3);
253 for (int n = 0; n < 3; ++n) {
254 ASSERT_EQ(handler.log[n].events, EventHandler::READ);
255 T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
256 milliseconds(events[n].milliseconds));
257 ASSERT_EQ(handler.log[n].bytesRead, events[n].length);
258 ASSERT_EQ(handler.log[n].bytesWritten, 0);
260 T_CHECK_TIMEOUT(start, end, milliseconds(events[3].milliseconds));
262 // Make sure the data from the last write is still waiting to be read
263 size_t bytesRemaining = readUntilEmpty(sp[0]);
264 ASSERT_EQ(bytesRemaining, events[3].length);
268 * Test registering for READ when the socket is immediately readable
270 TEST(EventBaseTest, ReadImmediate) {
274 // Write some data to the socket so the other end will
275 // be immediately readable
276 size_t dataLength = 1234;
277 writeToFD(sp[1], dataLength);
279 // Register for read events
280 TestHandler handler(&eb, sp[0]);
281 handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
283 // Register a timeout to perform another write
284 ScheduledEvent events[] = {
285 { 10, EventHandler::WRITE, 2345 },
288 scheduleEvents(&eb, sp[1], events);
290 // Schedule a timeout to unregister the handler
291 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 20);
298 ASSERT_EQ(handler.log.size(), 2);
300 // There should have been 1 event for immediate readability
301 ASSERT_EQ(handler.log[0].events, EventHandler::READ);
302 T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
303 ASSERT_EQ(handler.log[0].bytesRead, dataLength);
304 ASSERT_EQ(handler.log[0].bytesWritten, 0);
306 // There should be another event after the timeout wrote more data
307 ASSERT_EQ(handler.log[1].events, EventHandler::READ);
308 T_CHECK_TIMEOUT(start, handler.log[1].timestamp,
309 milliseconds(events[0].milliseconds));
310 ASSERT_EQ(handler.log[1].bytesRead, events[0].length);
311 ASSERT_EQ(handler.log[1].bytesWritten, 0);
313 T_CHECK_TIMEOUT(start, end, milliseconds(20));
319 TEST(EventBaseTest, WriteEvent) {
323 // Fill up the write buffer before starting
324 size_t initialBytesWritten = writeUntilFull(sp[0]);
326 // Register for write events
327 TestHandler handler(&eb, sp[0]);
328 handler.registerHandler(EventHandler::WRITE);
330 // Register timeouts to perform two reads
331 ScheduledEvent events[] = {
332 { 10, EventHandler::READ, 0 },
333 { 60, EventHandler::READ, 0 },
336 scheduleEvents(&eb, sp[1], events);
343 // Since we didn't use the EventHandler::PERSIST flag, the handler should
344 // have only been able to write once, then unregistered itself.
345 ASSERT_EQ(handler.log.size(), 1);
346 ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
347 T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
348 milliseconds(events[0].milliseconds));
349 ASSERT_EQ(handler.log[0].bytesRead, 0);
350 ASSERT_GT(handler.log[0].bytesWritten, 0);
351 T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
353 ASSERT_EQ(events[0].result, initialBytesWritten);
354 ASSERT_EQ(events[1].result, handler.log[0].bytesWritten);
358 * Test (WRITE | PERSIST)
360 TEST(EventBaseTest, WritePersist) {
364 // Fill up the write buffer before starting
365 size_t initialBytesWritten = writeUntilFull(sp[0]);
367 // Register for write events
368 TestHandler handler(&eb, sp[0]);
369 handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
371 // Register several timeouts to read from the socket at several intervals
372 ScheduledEvent events[] = {
373 { 10, EventHandler::READ, 0 },
374 { 40, EventHandler::READ, 0 },
375 { 70, EventHandler::READ, 0 },
376 { 100, EventHandler::READ, 0 },
379 scheduleEvents(&eb, sp[1], events);
381 // Schedule a timeout to unregister the handler after the third read
382 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 85);
389 // The handler should have received the first 3 events,
390 // then been unregistered after that.
391 ASSERT_EQ(handler.log.size(), 3);
392 ASSERT_EQ(events[0].result, initialBytesWritten);
393 for (int n = 0; n < 3; ++n) {
394 ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
395 T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
396 milliseconds(events[n].milliseconds));
397 ASSERT_EQ(handler.log[n].bytesRead, 0);
398 ASSERT_GT(handler.log[n].bytesWritten, 0);
399 ASSERT_EQ(handler.log[n].bytesWritten, events[n + 1].result);
401 T_CHECK_TIMEOUT(start, end, milliseconds(events[3].milliseconds));
405 * Test registering for WRITE when the socket is immediately writable
407 TEST(EventBaseTest, WriteImmediate) {
411 // Register for write events
412 TestHandler handler(&eb, sp[0]);
413 handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
415 // Register a timeout to perform a read
416 ScheduledEvent events[] = {
417 { 10, EventHandler::READ, 0 },
420 scheduleEvents(&eb, sp[1], events);
422 // Schedule a timeout to unregister the handler
423 int64_t unregisterTimeout = 40;
424 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler),
432 ASSERT_EQ(handler.log.size(), 2);
434 // Since the socket buffer was initially empty,
435 // there should have been 1 event for immediate writability
436 ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
437 T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
438 ASSERT_EQ(handler.log[0].bytesRead, 0);
439 ASSERT_GT(handler.log[0].bytesWritten, 0);
441 // There should be another event after the timeout wrote more data
442 ASSERT_EQ(handler.log[1].events, EventHandler::WRITE);
443 T_CHECK_TIMEOUT(start, handler.log[1].timestamp,
444 milliseconds(events[0].milliseconds));
445 ASSERT_EQ(handler.log[1].bytesRead, 0);
446 ASSERT_GT(handler.log[1].bytesWritten, 0);
448 T_CHECK_TIMEOUT(start, end, milliseconds(unregisterTimeout));
452 * Test (READ | WRITE) when the socket becomes readable first
454 TEST(EventBaseTest, ReadWrite) {
458 // Fill up the write buffer before starting
459 size_t sock0WriteLength = writeUntilFull(sp[0]);
461 // Register for read and write events
462 TestHandler handler(&eb, sp[0]);
463 handler.registerHandler(EventHandler::READ_WRITE);
465 // Register timeouts to perform a write then a read.
466 ScheduledEvent events[] = {
467 { 10, EventHandler::WRITE, 2345 },
468 { 40, EventHandler::READ, 0 },
471 scheduleEvents(&eb, sp[1], events);
478 // Since we didn't use the EventHandler::PERSIST flag, the handler should
479 // have only noticed readability, then unregistered itself. Check that only
480 // one event was logged.
481 ASSERT_EQ(handler.log.size(), 1);
482 ASSERT_EQ(handler.log[0].events, EventHandler::READ);
483 T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
484 milliseconds(events[0].milliseconds));
485 ASSERT_EQ(handler.log[0].bytesRead, events[0].length);
486 ASSERT_EQ(handler.log[0].bytesWritten, 0);
487 ASSERT_EQ(events[1].result, sock0WriteLength);
488 T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
492 * Test (READ | WRITE) when the socket becomes writable first
494 TEST(EventBaseTest, WriteRead) {
498 // Fill up the write buffer before starting
499 size_t sock0WriteLength = writeUntilFull(sp[0]);
501 // Register for read and write events
502 TestHandler handler(&eb, sp[0]);
503 handler.registerHandler(EventHandler::READ_WRITE);
505 // Register timeouts to perform a read then a write.
506 size_t sock1WriteLength = 2345;
507 ScheduledEvent events[] = {
508 { 10, EventHandler::READ, 0 },
509 { 40, EventHandler::WRITE, sock1WriteLength },
512 scheduleEvents(&eb, sp[1], events);
519 // Since we didn't use the EventHandler::PERSIST flag, the handler should
520 // have only noticed writability, then unregistered itself. Check that only
521 // one event was logged.
522 ASSERT_EQ(handler.log.size(), 1);
523 ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
524 T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
525 milliseconds(events[0].milliseconds));
526 ASSERT_EQ(handler.log[0].bytesRead, 0);
527 ASSERT_GT(handler.log[0].bytesWritten, 0);
528 ASSERT_EQ(events[0].result, sock0WriteLength);
529 ASSERT_EQ(events[1].result, sock1WriteLength);
530 T_CHECK_TIMEOUT(start, end, milliseconds(events[1].milliseconds));
532 // Make sure the written data is still waiting to be read.
533 size_t bytesRemaining = readUntilEmpty(sp[0]);
534 ASSERT_EQ(bytesRemaining, events[1].length);
538 * Test (READ | WRITE) when the socket becomes readable and writable
541 TEST(EventBaseTest, ReadWriteSimultaneous) {
545 // Fill up the write buffer before starting
546 size_t sock0WriteLength = writeUntilFull(sp[0]);
548 // Register for read and write events
549 TestHandler handler(&eb, sp[0]);
550 handler.registerHandler(EventHandler::READ_WRITE);
552 // Register a timeout to perform a read and write together
553 ScheduledEvent events[] = {
554 { 10, EventHandler::READ | EventHandler::WRITE, 0 },
557 scheduleEvents(&eb, sp[1], events);
564 // It's not strictly required that the EventBase register us about both
565 // events in the same call. So, it's possible that if the EventBase
566 // implementation changes this test could start failing, and it wouldn't be
567 // considered breaking the API. However for now it's nice to exercise this
569 ASSERT_EQ(handler.log.size(), 1);
570 ASSERT_EQ(handler.log[0].events,
571 EventHandler::READ | EventHandler::WRITE);
572 T_CHECK_TIMEOUT(start, handler.log[0].timestamp,
573 milliseconds(events[0].milliseconds));
574 ASSERT_EQ(handler.log[0].bytesRead, sock0WriteLength);
575 ASSERT_GT(handler.log[0].bytesWritten, 0);
576 T_CHECK_TIMEOUT(start, end, milliseconds(events[0].milliseconds));
580 * Test (READ | WRITE | PERSIST)
582 TEST(EventBaseTest, ReadWritePersist) {
586 // Register for read and write events
587 TestHandler handler(&eb, sp[0]);
588 handler.registerHandler(EventHandler::READ | EventHandler::WRITE |
589 EventHandler::PERSIST);
591 // Register timeouts to perform several reads and writes
592 ScheduledEvent events[] = {
593 { 10, EventHandler::WRITE, 2345 },
594 { 20, EventHandler::READ, 0 },
595 { 35, EventHandler::WRITE, 200 },
596 { 45, EventHandler::WRITE, 15 },
597 { 55, EventHandler::READ, 0 },
598 { 120, EventHandler::WRITE, 2345 },
601 scheduleEvents(&eb, sp[1], events);
603 // Schedule a timeout to unregister the handler
604 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 80);
611 ASSERT_EQ(handler.log.size(), 6);
613 // Since we didn't fill up the write buffer immediately, there should
614 // be an immediate event for writability.
615 ASSERT_EQ(handler.log[0].events, EventHandler::WRITE);
616 T_CHECK_TIMEOUT(start, handler.log[0].timestamp, milliseconds(0));
617 ASSERT_EQ(handler.log[0].bytesRead, 0);
618 ASSERT_GT(handler.log[0].bytesWritten, 0);
620 // Events 1 through 5 should correspond to the scheduled events
621 for (int n = 1; n < 6; ++n) {
622 ScheduledEvent* event = &events[n - 1];
623 T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
624 milliseconds(event->milliseconds));
625 if (event->events == EventHandler::READ) {
626 ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
627 ASSERT_EQ(handler.log[n].bytesRead, 0);
628 ASSERT_GT(handler.log[n].bytesWritten, 0);
630 ASSERT_EQ(handler.log[n].events, EventHandler::READ);
631 ASSERT_EQ(handler.log[n].bytesRead, event->length);
632 ASSERT_EQ(handler.log[n].bytesWritten, 0);
636 // The timeout should have unregistered the handler before the last write.
637 // Make sure that data is still waiting to be read
638 size_t bytesRemaining = readUntilEmpty(sp[0]);
639 ASSERT_EQ(bytesRemaining, events[5].length);
643 class PartialReadHandler : public TestHandler {
645 PartialReadHandler(EventBase* eventBase, int fd, size_t readLength)
646 : TestHandler(eventBase, fd), fd_(fd), readLength_(readLength) {}
648 void handlerReady(uint16_t events) noexcept override {
649 assert(events == EventHandler::READ);
650 ssize_t bytesRead = readFromFD(fd_, readLength_);
651 log.emplace_back(events, bytesRead, 0);
660 * Test reading only part of the available data when a read event is fired.
661 * When PERSIST is used, make sure the handler gets notified again the next
662 * time around the loop.
664 TEST(EventBaseTest, ReadPartial) {
668 // Register for read events
669 size_t readLength = 100;
670 PartialReadHandler handler(&eb, sp[0], readLength);
671 handler.registerHandler(EventHandler::READ | EventHandler::PERSIST);
673 // Register a timeout to perform a single write,
674 // with more data than PartialReadHandler will read at once
675 ScheduledEvent events[] = {
676 { 10, EventHandler::WRITE, (3*readLength) + (readLength / 2) },
679 scheduleEvents(&eb, sp[1], events);
681 // Schedule a timeout to unregister the handler
682 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 30);
689 ASSERT_EQ(handler.log.size(), 4);
691 // The first 3 invocations should read readLength bytes each
692 for (int n = 0; n < 3; ++n) {
693 ASSERT_EQ(handler.log[n].events, EventHandler::READ);
694 T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
695 milliseconds(events[0].milliseconds));
696 ASSERT_EQ(handler.log[n].bytesRead, readLength);
697 ASSERT_EQ(handler.log[n].bytesWritten, 0);
699 // The last read only has readLength/2 bytes
700 ASSERT_EQ(handler.log[3].events, EventHandler::READ);
701 T_CHECK_TIMEOUT(start, handler.log[3].timestamp,
702 milliseconds(events[0].milliseconds));
703 ASSERT_EQ(handler.log[3].bytesRead, readLength / 2);
704 ASSERT_EQ(handler.log[3].bytesWritten, 0);
708 class PartialWriteHandler : public TestHandler {
710 PartialWriteHandler(EventBase* eventBase, int fd, size_t writeLength)
711 : TestHandler(eventBase, fd), fd_(fd), writeLength_(writeLength) {}
713 void handlerReady(uint16_t events) noexcept override {
714 assert(events == EventHandler::WRITE);
715 ssize_t bytesWritten = writeToFD(fd_, writeLength_);
716 log.emplace_back(events, 0, bytesWritten);
725 * Test writing without completely filling up the write buffer when the fd
726 * becomes writable. When PERSIST is used, make sure the handler gets
727 * notified again the next time around the loop.
729 TEST(EventBaseTest, WritePartial) {
733 // Fill up the write buffer before starting
734 size_t initialBytesWritten = writeUntilFull(sp[0]);
736 // Register for write events
737 size_t writeLength = 100;
738 PartialWriteHandler handler(&eb, sp[0], writeLength);
739 handler.registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
741 // Register a timeout to read, so that more data can be written
742 ScheduledEvent events[] = {
743 { 10, EventHandler::READ, 0 },
746 scheduleEvents(&eb, sp[1], events);
748 // Schedule a timeout to unregister the handler
749 eb.tryRunAfterDelay(std::bind(&TestHandler::unregisterHandler, &handler), 30);
756 // Depending on how big the socket buffer is, there will be multiple writes
757 // Only check the first 5
759 ASSERT_GE(handler.log.size(), numChecked);
760 ASSERT_EQ(events[0].result, initialBytesWritten);
762 // The first 3 invocations should read writeLength bytes each
763 for (int n = 0; n < numChecked; ++n) {
764 ASSERT_EQ(handler.log[n].events, EventHandler::WRITE);
765 T_CHECK_TIMEOUT(start, handler.log[n].timestamp,
766 milliseconds(events[0].milliseconds));
767 ASSERT_EQ(handler.log[n].bytesRead, 0);
768 ASSERT_EQ(handler.log[n].bytesWritten, writeLength);
774 * Test destroying a registered EventHandler
776 TEST(EventBaseTest, DestroyHandler) {
777 class DestroyHandler : public AsyncTimeout {
779 DestroyHandler(EventBase* eb, EventHandler* h)
783 void timeoutExpired() noexcept override { delete handler_; }
786 EventHandler* handler_;
792 // Fill up the write buffer before starting
793 size_t initialBytesWritten = writeUntilFull(sp[0]);
795 // Register for write events
796 TestHandler* handler = new TestHandler(&eb, sp[0]);
797 handler->registerHandler(EventHandler::WRITE | EventHandler::PERSIST);
799 // After 10ms, read some data, so that the handler
800 // will be notified that it can write.
801 eb.tryRunAfterDelay(std::bind(checkReadUntilEmpty, sp[1], initialBytesWritten),
804 // Start a timer to destroy the handler after 25ms
805 // This mainly just makes sure the code doesn't break or assert
806 DestroyHandler dh(&eb, handler);
807 dh.scheduleTimeout(25);
813 // Make sure the EventHandler was uninstalled properly when it was
814 // destroyed, and the EventBase loop exited
815 T_CHECK_TIMEOUT(start, end, milliseconds(25));
817 // Make sure that the handler wrote data to the socket
818 // before it was destroyed
819 size_t bytesRemaining = readUntilEmpty(sp[1]);
820 ASSERT_GT(bytesRemaining, 0);
824 ///////////////////////////////////////////////////////////////////////////
825 // Tests for timeout events
826 ///////////////////////////////////////////////////////////////////////////
828 TEST(EventBaseTest, RunAfterDelay) {
831 TimePoint timestamp1(false);
832 TimePoint timestamp2(false);
833 TimePoint timestamp3(false);
834 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp1), 10);
835 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp2), 20);
836 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp3), 40);
842 T_CHECK_TIMEOUT(start, timestamp1, milliseconds(10));
843 T_CHECK_TIMEOUT(start, timestamp2, milliseconds(20));
844 T_CHECK_TIMEOUT(start, timestamp3, milliseconds(40));
845 T_CHECK_TIMEOUT(start, end, milliseconds(40));
849 * Test the behavior of tryRunAfterDelay() when some timeouts are
850 * still scheduled when the EventBase is destroyed.
852 TEST(EventBaseTest, RunAfterDelayDestruction) {
853 TimePoint timestamp1(false);
854 TimePoint timestamp2(false);
855 TimePoint timestamp3(false);
856 TimePoint timestamp4(false);
857 TimePoint start(false);
858 TimePoint end(false);
863 // Run two normal timeouts
864 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp1), 10);
865 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp2), 20);
867 // Schedule a timeout to stop the event loop after 40ms
868 eb.tryRunAfterDelay(std::bind(&EventBase::terminateLoopSoon, &eb), 40);
870 // Schedule 2 timeouts that would fire after the event loop stops
871 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp3), 80);
872 eb.tryRunAfterDelay(std::bind(&TimePoint::reset, ×tamp4), 160);
879 T_CHECK_TIMEOUT(start, timestamp1, milliseconds(10));
880 T_CHECK_TIMEOUT(start, timestamp2, milliseconds(20));
881 T_CHECK_TIMEOUT(start, end, milliseconds(40));
883 ASSERT_TRUE(timestamp3.isUnset());
884 ASSERT_TRUE(timestamp4.isUnset());
886 // Ideally this test should be run under valgrind to ensure that no
890 class TestTimeout : public AsyncTimeout {
892 explicit TestTimeout(EventBase* eventBase)
893 : AsyncTimeout(eventBase)
894 , timestamp(false) {}
896 void timeoutExpired() noexcept override { timestamp.reset(); }
901 TEST(EventBaseTest, BasicTimeouts) {
907 t1.scheduleTimeout(10);
908 t2.scheduleTimeout(20);
909 t3.scheduleTimeout(40);
915 T_CHECK_TIMEOUT(start, t1.timestamp, milliseconds(10));
916 T_CHECK_TIMEOUT(start, t2.timestamp, milliseconds(20));
917 T_CHECK_TIMEOUT(start, t3.timestamp, milliseconds(40));
918 T_CHECK_TIMEOUT(start, end, milliseconds(40));
921 class ReschedulingTimeout : public AsyncTimeout {
923 ReschedulingTimeout(EventBase* evb, const vector<uint32_t>& timeouts)
925 , timeouts_(timeouts)
926 , iterator_(timeouts_.begin()) {}
932 void timeoutExpired() noexcept override {
933 timestamps.emplace_back();
938 if (iterator_ != timeouts_.end()) {
939 uint32_t timeout = *iterator_;
941 scheduleTimeout(timeout);
945 vector<TimePoint> timestamps;
948 vector<uint32_t> timeouts_;
949 vector<uint32_t>::const_iterator iterator_;
953 * Test rescheduling the same timeout multiple times
955 TEST(EventBaseTest, ReuseTimeout) {
958 vector<uint32_t> timeouts;
959 timeouts.push_back(10);
960 timeouts.push_back(30);
961 timeouts.push_back(15);
963 ReschedulingTimeout t(&eb, timeouts);
970 // Use a higher tolerance than usual. We're waiting on 3 timeouts
971 // consecutively. In general, each timeout may go over by a few
972 // milliseconds, and we're tripling this error by witing on 3 timeouts.
973 milliseconds tolerance{6};
975 ASSERT_EQ(timeouts.size(), t.timestamps.size());
977 for (size_t n = 0; n < timeouts.size(); ++n) {
978 total += timeouts[n];
979 T_CHECK_TIMEOUT(start, t.timestamps[n], milliseconds(total), tolerance);
981 T_CHECK_TIMEOUT(start, end, milliseconds(total), tolerance);
985 * Test rescheduling a timeout before it has fired
987 TEST(EventBaseTest, RescheduleTimeout) {
994 t1.scheduleTimeout(15);
995 t2.scheduleTimeout(30);
996 t3.scheduleTimeout(30);
998 auto f = static_cast<bool(AsyncTimeout::*)(uint32_t)>(
999 &AsyncTimeout::scheduleTimeout);
1001 // after 10ms, reschedule t2 to run sooner than originally scheduled
1002 eb.tryRunAfterDelay(std::bind(f, &t2, 10), 10);
1003 // after 10ms, reschedule t3 to run later than originally scheduled
1004 eb.tryRunAfterDelay(std::bind(f, &t3, 40), 10);
1010 T_CHECK_TIMEOUT(start, t1.timestamp, milliseconds(15));
1011 T_CHECK_TIMEOUT(start, t2.timestamp, milliseconds(20));
1012 T_CHECK_TIMEOUT(start, t3.timestamp, milliseconds(50));
1013 T_CHECK_TIMEOUT(start, end, milliseconds(50));
1017 * Test cancelling a timeout
1019 TEST(EventBaseTest, CancelTimeout) {
1022 vector<uint32_t> timeouts;
1023 timeouts.push_back(10);
1024 timeouts.push_back(30);
1025 timeouts.push_back(25);
1027 ReschedulingTimeout t(&eb, timeouts);
1029 eb.tryRunAfterDelay(std::bind(&AsyncTimeout::cancelTimeout, &t), 50);
1035 ASSERT_EQ(t.timestamps.size(), 2);
1036 T_CHECK_TIMEOUT(start, t.timestamps[0], milliseconds(10));
1037 T_CHECK_TIMEOUT(start, t.timestamps[1], milliseconds(40));
1038 T_CHECK_TIMEOUT(start, end, milliseconds(50));
1042 * Test destroying a scheduled timeout object
1044 TEST(EventBaseTest, DestroyTimeout) {
1045 class DestroyTimeout : public AsyncTimeout {
1047 DestroyTimeout(EventBase* eb, AsyncTimeout* t)
1051 void timeoutExpired() noexcept override { delete timeout_; }
1054 AsyncTimeout* timeout_;
1059 TestTimeout* t1 = new TestTimeout(&eb);
1060 t1->scheduleTimeout(30);
1062 DestroyTimeout dt(&eb, t1);
1063 dt.scheduleTimeout(10);
1069 T_CHECK_TIMEOUT(start, end, milliseconds(10));
1073 ///////////////////////////////////////////////////////////////////////////
1074 // Test for runInThreadTestFunc()
1075 ///////////////////////////////////////////////////////////////////////////
1077 struct RunInThreadData {
1078 RunInThreadData(int numThreads, int opsPerThread)
1079 : opsPerThread(opsPerThread)
1080 , opsToGo(numThreads*opsPerThread) {}
1083 deque< pair<int, int> > values;
1089 struct RunInThreadArg {
1090 RunInThreadArg(RunInThreadData* data,
1097 RunInThreadData* data;
1102 void runInThreadTestFunc(RunInThreadArg* arg) {
1103 arg->data->values.emplace_back(arg->thread, arg->value);
1104 RunInThreadData* data = arg->data;
1107 if(--data->opsToGo == 0) {
1108 // Break out of the event base loop if we are the last thread running
1109 data->evb.terminateLoopSoon();
1113 TEST(EventBaseTest, RunInThread) {
1114 uint32_t numThreads = 50;
1115 uint32_t opsPerThread = 100;
1116 RunInThreadData data(numThreads, opsPerThread);
1118 deque<std::thread> threads;
1119 for (uint32_t i = 0; i < numThreads; ++i) {
1120 threads.emplace_back([i, &data] {
1121 for (int n = 0; n < data.opsPerThread; ++n) {
1122 RunInThreadArg* arg = new RunInThreadArg(&data, i, n);
1123 data.evb.runInEventBaseThread(runInThreadTestFunc, arg);
1129 // Add a timeout event to run after 3 seconds.
1130 // Otherwise loop() will return immediately since there are no events to run.
1131 // Once the last thread exits, it will stop the loop(). However, this
1132 // timeout also stops the loop in case there is a bug performing the normal
1134 data.evb.tryRunAfterDelay(std::bind(&EventBase::terminateLoopSoon, &data.evb),
1141 // Verify that the loop exited because all threads finished and requested it
1142 // to stop. This should happen much sooner than the 3 second timeout.
1143 // Assert that it happens in under a second. (This is still tons of extra
1146 auto timeTaken = std::chrono::duration_cast<milliseconds>(
1147 end.getTime() - start.getTime());
1148 ASSERT_LT(timeTaken.count(), 1000);
1149 VLOG(11) << "Time taken: " << timeTaken.count();
1151 // Verify that we have all of the events from every thread
1152 int expectedValues[numThreads];
1153 for (uint32_t n = 0; n < numThreads; ++n) {
1154 expectedValues[n] = 0;
1156 for (deque< pair<int, int> >::const_iterator it = data.values.begin();
1157 it != data.values.end();
1159 int threadID = it->first;
1160 int value = it->second;
1161 ASSERT_EQ(expectedValues[threadID], value);
1162 ++expectedValues[threadID];
1164 for (uint32_t n = 0; n < numThreads; ++n) {
1165 ASSERT_EQ(expectedValues[n], opsPerThread);
1168 // Wait on all of the threads.
1169 for (auto& thread: threads) {
1174 // This test simulates some calls, and verifies that the waiting happens by
1175 // triggering what otherwise would be race conditions, and trying to detect
1176 // whether any of the race conditions happened.
1177 TEST(EventBaseTest, RunInEventBaseThreadAndWait) {
1178 const size_t c = 256;
1179 vector<unique_ptr<atomic<size_t>>> atoms(c);
1180 for (size_t i = 0; i < c; ++i) {
1181 auto& atom = atoms.at(i);
1182 atom = make_unique<atomic<size_t>>(0);
1184 vector<thread> threads(c);
1185 for (size_t i = 0; i < c; ++i) {
1186 auto& atom = *atoms.at(i);
1187 auto& th = threads.at(i);
1188 th = thread([&atom] {
1190 auto ebth = thread([&]{ eb.loopForever(); });
1191 eb.waitUntilRunning();
1192 eb.runInEventBaseThreadAndWait([&] {
1194 atom.compare_exchange_weak(
1195 x, 1, std::memory_order_release, std::memory_order_relaxed);
1198 atom.compare_exchange_weak(
1199 x, 2, std::memory_order_release, std::memory_order_relaxed);
1200 eb.terminateLoopSoon();
1204 for (size_t i = 0; i < c; ++i) {
1205 auto& th = threads.at(i);
1209 for (auto& atom : atoms) sum += *atom;
1213 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadAndWaitCross) {
1215 thread th(&EventBase::loopForever, &eb);
1217 eb.terminateLoopSoon();
1220 auto mutated = false;
1221 eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1224 EXPECT_TRUE(mutated);
1227 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadAndWaitWithin) {
1229 thread th(&EventBase::loopForever, &eb);
1231 eb.terminateLoopSoon();
1234 eb.runInEventBaseThreadAndWait([&] {
1235 auto mutated = false;
1236 eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1239 EXPECT_TRUE(mutated);
1243 TEST(EventBaseTest, RunImmediatelyOrRunInEventBaseThreadNotLooping) {
1245 auto mutated = false;
1246 eb.runImmediatelyOrRunInEventBaseThreadAndWait([&] {
1249 EXPECT_TRUE(mutated);
1252 ///////////////////////////////////////////////////////////////////////////
1253 // Tests for runInLoop()
1254 ///////////////////////////////////////////////////////////////////////////
1256 class CountedLoopCallback : public EventBase::LoopCallback {
1258 CountedLoopCallback(EventBase* eventBase,
1260 std::function<void()> action =
1261 std::function<void()>())
1262 : eventBase_(eventBase)
1264 , action_(action) {}
1266 void runLoopCallback() noexcept override {
1269 eventBase_->runInLoop(this);
1270 } else if (action_) {
1275 unsigned int getCount() const {
1280 EventBase* eventBase_;
1281 unsigned int count_;
1282 std::function<void()> action_;
1285 // Test that EventBase::loop() doesn't exit while there are
1286 // still LoopCallbacks remaining to be invoked.
1287 TEST(EventBaseTest, RepeatedRunInLoop) {
1288 EventBase eventBase;
1290 CountedLoopCallback c(&eventBase, 10);
1291 eventBase.runInLoop(&c);
1292 // The callback shouldn't have run immediately
1293 ASSERT_EQ(c.getCount(), 10);
1296 // loop() should loop until the CountedLoopCallback stops
1297 // re-installing itself.
1298 ASSERT_EQ(c.getCount(), 0);
1301 // Test that EventBase::loop() works as expected without time measurements.
1302 TEST(EventBaseTest, RunInLoopNoTimeMeasurement) {
1303 EventBase eventBase(false);
1305 CountedLoopCallback c(&eventBase, 10);
1306 eventBase.runInLoop(&c);
1307 // The callback shouldn't have run immediately
1308 ASSERT_EQ(c.getCount(), 10);
1311 // loop() should loop until the CountedLoopCallback stops
1312 // re-installing itself.
1313 ASSERT_EQ(c.getCount(), 0);
1316 // Test runInLoop() calls with terminateLoopSoon()
1317 TEST(EventBaseTest, RunInLoopStopLoop) {
1318 EventBase eventBase;
1320 CountedLoopCallback c1(&eventBase, 20);
1321 CountedLoopCallback c2(&eventBase, 10,
1322 std::bind(&EventBase::terminateLoopSoon, &eventBase));
1324 eventBase.runInLoop(&c1);
1325 eventBase.runInLoop(&c2);
1326 ASSERT_EQ(c1.getCount(), 20);
1327 ASSERT_EQ(c2.getCount(), 10);
1329 eventBase.loopForever();
1331 // c2 should have stopped the loop after 10 iterations
1332 ASSERT_EQ(c2.getCount(), 0);
1334 // We allow the EventBase to run the loop callbacks in whatever order it
1335 // chooses. We'll accept c1's count being either 10 (if the loop terminated
1336 // after c1 ran on the 10th iteration) or 11 (if c2 terminated the loop
1339 // (With the current code, c1 will always run 10 times, but we don't consider
1340 // this a hard API requirement.)
1341 ASSERT_GE(c1.getCount(), 10);
1342 ASSERT_LE(c1.getCount(), 11);
1345 TEST(EventBaseTest, TryRunningAfterTerminate) {
1346 EventBase eventBase;
1347 CountedLoopCallback c1(&eventBase, 1,
1348 std::bind(&EventBase::terminateLoopSoon, &eventBase));
1349 eventBase.runInLoop(&c1);
1350 eventBase.loopForever();
1352 eventBase.runInEventBaseThread([&]() {
1359 // Test cancelling runInLoop() callbacks
1360 TEST(EventBaseTest, CancelRunInLoop) {
1361 EventBase eventBase;
1363 CountedLoopCallback c1(&eventBase, 20);
1364 CountedLoopCallback c2(&eventBase, 20);
1365 CountedLoopCallback c3(&eventBase, 20);
1367 std::function<void()> cancelC1Action =
1368 std::bind(&EventBase::LoopCallback::cancelLoopCallback, &c1);
1369 std::function<void()> cancelC2Action =
1370 std::bind(&EventBase::LoopCallback::cancelLoopCallback, &c2);
1372 CountedLoopCallback cancelC1(&eventBase, 10, cancelC1Action);
1373 CountedLoopCallback cancelC2(&eventBase, 10, cancelC2Action);
1375 // Install cancelC1 after c1
1376 eventBase.runInLoop(&c1);
1377 eventBase.runInLoop(&cancelC1);
1379 // Install cancelC2 before c2
1380 eventBase.runInLoop(&cancelC2);
1381 eventBase.runInLoop(&c2);
1384 eventBase.runInLoop(&c3);
1386 ASSERT_EQ(c1.getCount(), 20);
1387 ASSERT_EQ(c2.getCount(), 20);
1388 ASSERT_EQ(c3.getCount(), 20);
1389 ASSERT_EQ(cancelC1.getCount(), 10);
1390 ASSERT_EQ(cancelC2.getCount(), 10);
1395 // cancelC1 and cancelC3 should have both fired after 10 iterations and
1396 // stopped re-installing themselves
1397 ASSERT_EQ(cancelC1.getCount(), 0);
1398 ASSERT_EQ(cancelC2.getCount(), 0);
1399 // c3 should have continued on for the full 20 iterations
1400 ASSERT_EQ(c3.getCount(), 0);
1402 // c1 and c2 should have both been cancelled on the 10th iteration.
1404 // Callbacks are always run in the order they are installed,
1405 // so c1 should have fired 10 times, and been canceled after it ran on the
1406 // 10th iteration. c2 should have only fired 9 times, because cancelC2 will
1407 // have run before it on the 10th iteration, and cancelled it before it
1409 ASSERT_EQ(c1.getCount(), 10);
1410 ASSERT_EQ(c2.getCount(), 11);
1413 class TerminateTestCallback : public EventBase::LoopCallback,
1414 public EventHandler {
1416 TerminateTestCallback(EventBase* eventBase, int fd)
1417 : EventHandler(eventBase, fd),
1418 eventBase_(eventBase),
1419 loopInvocations_(0),
1420 maxLoopInvocations_(0),
1421 eventInvocations_(0),
1422 maxEventInvocations_(0) {}
1424 void reset(uint32_t maxLoopInvocations, uint32_t maxEventInvocations) {
1425 loopInvocations_ = 0;
1426 maxLoopInvocations_ = maxLoopInvocations;
1427 eventInvocations_ = 0;
1428 maxEventInvocations_ = maxEventInvocations;
1430 cancelLoopCallback();
1431 unregisterHandler();
1434 void handlerReady(uint16_t /* events */) noexcept override {
1435 // We didn't register with PERSIST, so we will have been automatically
1436 // unregistered already.
1437 ASSERT_FALSE(isHandlerRegistered());
1439 ++eventInvocations_;
1440 if (eventInvocations_ >= maxEventInvocations_) {
1444 eventBase_->runInLoop(this);
1446 void runLoopCallback() noexcept override {
1448 if (loopInvocations_ >= maxLoopInvocations_) {
1452 registerHandler(READ);
1455 uint32_t getLoopInvocations() const {
1456 return loopInvocations_;
1458 uint32_t getEventInvocations() const {
1459 return eventInvocations_;
1463 EventBase* eventBase_;
1464 uint32_t loopInvocations_;
1465 uint32_t maxLoopInvocations_;
1466 uint32_t eventInvocations_;
1467 uint32_t maxEventInvocations_;
1471 * Test that EventBase::loop() correctly detects when there are no more events
1474 * This uses a single callback, which alternates registering itself as a loop
1475 * callback versus a EventHandler callback. This exercises a regression where
1476 * EventBase::loop() incorrectly exited if there were no more fd handlers
1477 * registered, but a loop callback installed a new fd handler.
1479 TEST(EventBaseTest, LoopTermination) {
1480 EventBase eventBase;
1482 // Open a pipe and close the write end,
1483 // so the read endpoint will be readable
1485 int rc = pipe(pipeFds);
1488 TerminateTestCallback callback(&eventBase, pipeFds[0]);
1490 // Test once where the callback will exit after a loop callback
1491 callback.reset(10, 100);
1492 eventBase.runInLoop(&callback);
1494 ASSERT_EQ(callback.getLoopInvocations(), 10);
1495 ASSERT_EQ(callback.getEventInvocations(), 9);
1497 // Test once where the callback will exit after an fd event callback
1498 callback.reset(100, 7);
1499 eventBase.runInLoop(&callback);
1501 ASSERT_EQ(callback.getLoopInvocations(), 7);
1502 ASSERT_EQ(callback.getEventInvocations(), 7);
1507 ///////////////////////////////////////////////////////////////////////////
1508 // Tests for latency calculations
1509 ///////////////////////////////////////////////////////////////////////////
1511 class IdleTimeTimeoutSeries : public AsyncTimeout {
1515 explicit IdleTimeTimeoutSeries(EventBase *base,
1516 std::deque<std::uint64_t>& timeout) :
1523 ~IdleTimeTimeoutSeries() override {}
1525 void timeoutExpired() noexcept override {
1528 if(timeout_.empty()){
1531 uint64_t sleepTime = timeout_.front();
1532 timeout_.pop_front();
1540 int getTimeouts() const {
1546 std::deque<uint64_t>& timeout_;
1550 * Verify that idle time is correctly accounted for when decaying our loop
1553 * This works by creating a high loop time (via usleep), expecting a latency
1554 * callback with known value, and then scheduling a timeout for later. This
1555 * later timeout is far enough in the future that the idle time should have
1556 * caused the loop time to decay.
1558 TEST(EventBaseTest, IdleTime) {
1559 EventBase eventBase;
1560 eventBase.setLoadAvgMsec(1000);
1561 eventBase.resetLoadAvg(5900.0);
1562 std::deque<uint64_t> timeouts0(4, 8080);
1563 timeouts0.push_front(8000);
1564 timeouts0.push_back(14000);
1565 IdleTimeTimeoutSeries tos0(&eventBase, timeouts0);
1566 std::deque<uint64_t> timeouts(20, 20);
1567 std::unique_ptr<IdleTimeTimeoutSeries> tos;
1568 int64_t testStart = duration_cast<microseconds>(
1569 std::chrono::steady_clock::now().time_since_epoch()).count();
1570 bool hostOverloaded = false;
1572 int latencyCallbacks = 0;
1573 eventBase.setMaxLatency(6000, [&]() {
1576 switch (latencyCallbacks) {
1578 if (tos0.getTimeouts() < 6) {
1579 // This could only happen if the host this test is running
1580 // on is heavily loaded.
1581 int64_t maxLatencyReached = duration_cast<microseconds>(
1582 std::chrono::steady_clock::now().time_since_epoch()).count();
1583 ASSERT_LE(43800, maxLatencyReached - testStart);
1584 hostOverloaded = true;
1587 ASSERT_EQ(6, tos0.getTimeouts());
1588 ASSERT_GE(6100, eventBase.getAvgLoopTime() - 1200);
1589 ASSERT_LE(6100, eventBase.getAvgLoopTime() + 1200);
1590 tos.reset(new IdleTimeTimeoutSeries(&eventBase, timeouts));
1594 FAIL() << "Unexpected latency callback";
1599 // Kick things off with an "immedite" timeout
1600 tos0.scheduleTimeout(1);
1604 if (hostOverloaded) {
1608 ASSERT_EQ(1, latencyCallbacks);
1609 ASSERT_EQ(7, tos0.getTimeouts());
1610 ASSERT_GE(5900, eventBase.getAvgLoopTime() - 1200);
1611 ASSERT_LE(5900, eventBase.getAvgLoopTime() + 1200);
1613 ASSERT_EQ(21, tos->getTimeouts());
1617 * Test that thisLoop functionality works with terminateLoopSoon
1619 TEST(EventBaseTest, ThisLoop) {
1621 bool runInLoop = false;
1622 bool runThisLoop = false;
1625 eb.terminateLoopSoon();
1626 eb.runInLoop([&]() {
1629 eb.runInLoop([&]() {
1636 ASSERT_FALSE(runInLoop);
1637 // Should work with thisLoop
1638 ASSERT_TRUE(runThisLoop);
1641 TEST(EventBaseTest, EventBaseThreadLoop) {
1645 base.runInEventBaseThread([&](){
1650 ASSERT_EQ(true, ran);
1653 TEST(EventBaseTest, EventBaseThreadName) {
1655 base.setName("foo");
1658 #if (__GLIBC__ >= 2) && (__GLIBC_MINOR__ >= 12)
1660 pthread_getname_np(pthread_self(), name, 16);
1661 ASSERT_EQ(0, strcmp("foo", name));
1665 TEST(EventBaseTest, RunBeforeLoop) {
1667 CountedLoopCallback cb(&base, 1, [&](){
1668 base.terminateLoopSoon();
1670 base.runBeforeLoop(&cb);
1672 ASSERT_EQ(cb.getCount(), 0);
1675 TEST(EventBaseTest, RunBeforeLoopWait) {
1677 CountedLoopCallback cb(&base, 1);
1678 base.tryRunAfterDelay([&](){
1679 base.terminateLoopSoon();
1681 base.runBeforeLoop(&cb);
1684 // Check that we only ran once, and did not loop multiple times.
1685 ASSERT_EQ(cb.getCount(), 0);
1688 class PipeHandler : public EventHandler {
1690 PipeHandler(EventBase* eventBase, int fd)
1691 : EventHandler(eventBase, fd) {}
1693 void handlerReady(uint16_t /* events */) noexcept override { abort(); }
1696 TEST(EventBaseTest, StopBeforeLoop) {
1699 // Give the evb something to do.
1701 ASSERT_EQ(0, pipe(p));
1702 PipeHandler handler(&evb, p[0]);
1703 handler.registerHandler(EventHandler::READ);
1705 // It's definitely not running yet
1706 evb.terminateLoopSoon();
1708 // let it run, it should exit quickly.
1709 std::thread t([&] { evb.loop(); });
1712 handler.unregisterHandler();
1719 TEST(EventBaseTest, RunCallbacksOnDestruction) {
1724 base.runInEventBaseThread([&](){