message_pump_libevent_unittest.cc 9.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  1. // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "base/message_loop/message_pump_libevent.h"
  5. #include <unistd.h>
  6. #include <memory>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/callback_helpers.h"
  10. #include "base/containers/span.h"
  11. #include "base/files/file_util.h"
  12. #include "base/logging.h"
  13. #include "base/memory/ptr_util.h"
  14. #include "base/memory/raw_ptr.h"
  15. #include "base/message_loop/message_pump_buildflags.h"
  16. #include "base/message_loop/message_pump_type.h"
  17. #include "base/posix/eintr_wrapper.h"
  18. #include "base/run_loop.h"
  19. #include "base/synchronization/waitable_event.h"
  20. #include "base/synchronization/waitable_event_watcher.h"
  21. #include "base/task/single_thread_task_executor.h"
  22. #include "base/task/single_thread_task_runner.h"
  23. #include "base/test/gtest_util.h"
  24. #include "base/test/task_environment.h"
  25. #include "base/threading/sequenced_task_runner_handle.h"
  26. #include "base/threading/thread.h"
  27. #include "base/threading/thread_task_runner_handle.h"
  28. #include "build/build_config.h"
  29. #include "testing/gtest/include/gtest/gtest.h"
  30. #include "third_party/libevent/event.h"
  31. #if BUILDFLAG(ENABLE_MESSAGE_PUMP_EPOLL)
  32. #include "base/message_loop/message_pump_epoll.h"
  33. #endif
  34. namespace base {
  35. enum PumpType {
  36. kLibevent,
  37. kEpoll,
  38. };
  39. class MessagePumpLibeventTest : public testing::Test,
  40. public testing::WithParamInterface<PumpType> {
  41. protected:
  42. MessagePumpLibeventTest()
  43. : task_environment_(std::make_unique<test::SingleThreadTaskEnvironment>(
  44. test::SingleThreadTaskEnvironment::MainThreadType::UI)),
  45. io_thread_("MessagePumpLibeventTestIOThread") {}
  46. ~MessagePumpLibeventTest() override = default;
  47. void SetUp() override {
  48. Thread::Options options(MessagePumpType::IO, 0);
  49. ASSERT_TRUE(io_thread_.StartWithOptions(std::move(options)));
  50. int ret = pipe(pipefds_);
  51. ASSERT_EQ(0, ret);
  52. }
  53. void TearDown() override {
  54. // Some tests watch |pipefds_| from the |io_thread_|. The |io_thread_| must
  55. // thus be joined to ensure those watches are complete before closing the
  56. // pipe.
  57. io_thread_.Stop();
  58. if (IGNORE_EINTR(close(pipefds_[0])) < 0)
  59. PLOG(ERROR) << "close";
  60. if (IGNORE_EINTR(close(pipefds_[1])) < 0)
  61. PLOG(ERROR) << "close";
  62. }
  63. std::unique_ptr<MessagePumpLibevent> CreateMessagePump() {
  64. #if BUILDFLAG(ENABLE_MESSAGE_PUMP_EPOLL)
  65. if (GetParam() == kEpoll) {
  66. return std::make_unique<MessagePumpLibevent>(
  67. MessagePumpLibevent::kUseEpoll);
  68. }
  69. #endif
  70. return std::make_unique<MessagePumpLibevent>();
  71. }
  72. scoped_refptr<SingleThreadTaskRunner> io_runner() const {
  73. return io_thread_.task_runner();
  74. }
  75. void SimulateIOEvent(MessagePumpLibevent* pump,
  76. MessagePumpLibevent::FdWatchController* controller) {
  77. #if BUILDFLAG(ENABLE_MESSAGE_PUMP_EPOLL)
  78. if (GetParam() == kEpoll) {
  79. pump->epoll_pump_->HandleEvent(0, /*can_read=*/true, /*can_write=*/true,
  80. controller);
  81. return;
  82. }
  83. #endif
  84. pump->OnLibeventNotification(0, EV_WRITE | EV_READ, controller);
  85. }
  86. int pipefds_[2];
  87. std::unique_ptr<test::SingleThreadTaskEnvironment> task_environment_;
  88. private:
  89. Thread io_thread_;
  90. };
  91. namespace {
  92. // Concrete implementation of MessagePumpLibevent::FdWatcher that does
  93. // nothing useful.
  94. class StupidWatcher : public MessagePumpLibevent::FdWatcher {
  95. public:
  96. ~StupidWatcher() override = default;
  97. // base:MessagePumpLibevent::FdWatcher interface
  98. void OnFileCanReadWithoutBlocking(int fd) override {}
  99. void OnFileCanWriteWithoutBlocking(int fd) override {}
  100. };
  101. TEST_P(MessagePumpLibeventTest, QuitOutsideOfRun) {
  102. std::unique_ptr<MessagePumpLibevent> pump = CreateMessagePump();
  103. ASSERT_DCHECK_DEATH(pump->Quit());
  104. }
  105. class BaseWatcher : public MessagePumpLibevent::FdWatcher {
  106. public:
  107. explicit BaseWatcher(MessagePumpLibevent::FdWatchController* controller)
  108. : controller_(controller) {
  109. DCHECK(controller_);
  110. }
  111. ~BaseWatcher() override = default;
  112. // base:MessagePumpLibevent::FdWatcher interface
  113. void OnFileCanReadWithoutBlocking(int /* fd */) override { NOTREACHED(); }
  114. void OnFileCanWriteWithoutBlocking(int /* fd */) override { NOTREACHED(); }
  115. protected:
  116. raw_ptr<MessagePumpLibevent::FdWatchController> controller_;
  117. };
  118. class DeleteWatcher : public BaseWatcher {
  119. public:
  120. explicit DeleteWatcher(MessagePumpLibevent::FdWatchController* controller)
  121. : BaseWatcher(controller) {}
  122. ~DeleteWatcher() override { DCHECK(!controller_); }
  123. void OnFileCanWriteWithoutBlocking(int /* fd */) override {
  124. DCHECK(controller_);
  125. delete controller_;
  126. controller_ = nullptr;
  127. }
  128. };
  129. TEST_P(MessagePumpLibeventTest, DeleteWatcher) {
  130. std::unique_ptr<MessagePumpLibevent> pump = CreateMessagePump();
  131. MessagePumpLibevent::FdWatchController* watcher =
  132. new MessagePumpLibevent::FdWatchController(FROM_HERE);
  133. DeleteWatcher delegate(watcher);
  134. pump->WatchFileDescriptor(pipefds_[1], false,
  135. MessagePumpLibevent::WATCH_READ_WRITE, watcher,
  136. &delegate);
  137. SimulateIOEvent(pump.get(), watcher);
  138. }
  139. class StopWatcher : public BaseWatcher {
  140. public:
  141. explicit StopWatcher(MessagePumpLibevent::FdWatchController* controller)
  142. : BaseWatcher(controller) {}
  143. ~StopWatcher() override = default;
  144. void OnFileCanWriteWithoutBlocking(int /* fd */) override {
  145. controller_->StopWatchingFileDescriptor();
  146. }
  147. };
  148. TEST_P(MessagePumpLibeventTest, StopWatcher) {
  149. std::unique_ptr<MessagePumpLibevent> pump = CreateMessagePump();
  150. MessagePumpLibevent::FdWatchController watcher(FROM_HERE);
  151. StopWatcher delegate(&watcher);
  152. pump->WatchFileDescriptor(pipefds_[1], false,
  153. MessagePumpLibevent::WATCH_READ_WRITE, &watcher,
  154. &delegate);
  155. SimulateIOEvent(pump.get(), &watcher);
  156. }
  157. void QuitMessageLoopAndStart(OnceClosure quit_closure) {
  158. std::move(quit_closure).Run();
  159. RunLoop runloop(RunLoop::Type::kNestableTasksAllowed);
  160. ThreadTaskRunnerHandle::Get()->PostTask(FROM_HERE, runloop.QuitClosure());
  161. runloop.Run();
  162. }
  163. class NestedPumpWatcher : public MessagePumpLibevent::FdWatcher {
  164. public:
  165. NestedPumpWatcher() = default;
  166. ~NestedPumpWatcher() override = default;
  167. void OnFileCanReadWithoutBlocking(int /* fd */) override {
  168. RunLoop runloop;
  169. ThreadTaskRunnerHandle::Get()->PostTask(
  170. FROM_HERE, BindOnce(&QuitMessageLoopAndStart, runloop.QuitClosure()));
  171. runloop.Run();
  172. }
  173. void OnFileCanWriteWithoutBlocking(int /* fd */) override {}
  174. };
  175. TEST_P(MessagePumpLibeventTest, NestedPumpWatcher) {
  176. std::unique_ptr<MessagePumpLibevent> pump = CreateMessagePump();
  177. MessagePumpLibevent::FdWatchController watcher(FROM_HERE);
  178. NestedPumpWatcher delegate;
  179. pump->WatchFileDescriptor(pipefds_[1], false, MessagePumpLibevent::WATCH_READ,
  180. &watcher, &delegate);
  181. SimulateIOEvent(pump.get(), &watcher);
  182. }
  183. void FatalClosure() {
  184. FAIL() << "Reached fatal closure.";
  185. }
  186. class QuitWatcher : public BaseWatcher {
  187. public:
  188. QuitWatcher(MessagePumpLibevent::FdWatchController* controller,
  189. base::OnceClosure quit_closure)
  190. : BaseWatcher(controller), quit_closure_(std::move(quit_closure)) {}
  191. void OnFileCanReadWithoutBlocking(int /* fd */) override {
  192. // Post a fatal closure to the MessageLoop before we quit it.
  193. ThreadTaskRunnerHandle::Get()->PostTask(FROM_HERE, BindOnce(&FatalClosure));
  194. if (quit_closure_)
  195. std::move(quit_closure_).Run();
  196. }
  197. private:
  198. base::OnceClosure quit_closure_;
  199. };
  200. void WriteFDWrapper(const int fd,
  201. const char* buf,
  202. int size,
  203. WaitableEvent* event) {
  204. ASSERT_TRUE(WriteFileDescriptor(fd, StringPiece(buf, size)));
  205. }
  206. // Tests that MessagePumpLibevent quits immediately when it is quit from
  207. // libevent's event_base_loop().
  208. TEST_P(MessagePumpLibeventTest, QuitWatcher) {
  209. // Delete the old TaskEnvironment so that we can manage our own one here.
  210. task_environment_.reset();
  211. std::unique_ptr<MessagePumpLibevent> executor_pump = CreateMessagePump();
  212. MessagePumpLibevent* pump = executor_pump.get();
  213. SingleThreadTaskExecutor executor(std::move(executor_pump));
  214. RunLoop run_loop;
  215. MessagePumpLibevent::FdWatchController controller(FROM_HERE);
  216. QuitWatcher delegate(&controller, run_loop.QuitClosure());
  217. WaitableEvent event(WaitableEvent::ResetPolicy::AUTOMATIC,
  218. WaitableEvent::InitialState::NOT_SIGNALED);
  219. std::unique_ptr<WaitableEventWatcher> watcher(new WaitableEventWatcher);
  220. // Tell the pump to watch the pipe.
  221. pump->WatchFileDescriptor(pipefds_[0], false, MessagePumpLibevent::WATCH_READ,
  222. &controller, &delegate);
  223. // Make the IO thread wait for |event| before writing to pipefds[1].
  224. const char buf = 0;
  225. WaitableEventWatcher::EventCallback write_fd_task =
  226. BindOnce(&WriteFDWrapper, pipefds_[1], &buf, 1);
  227. io_runner()->PostTask(
  228. FROM_HERE, BindOnce(IgnoreResult(&WaitableEventWatcher::StartWatching),
  229. Unretained(watcher.get()), &event,
  230. std::move(write_fd_task), io_runner()));
  231. // Queue |event| to signal on |sequence_manager|.
  232. ThreadTaskRunnerHandle::Get()->PostTask(
  233. FROM_HERE, BindOnce(&WaitableEvent::Signal, Unretained(&event)));
  234. // Now run the MessageLoop.
  235. run_loop.Run();
  236. // StartWatching can move |watcher| to IO thread. Release on IO thread.
  237. io_runner()->PostTask(FROM_HERE, BindOnce(&WaitableEventWatcher::StopWatching,
  238. Owned(watcher.release())));
  239. }
  240. #if BUILDFLAG(ENABLE_MESSAGE_PUMP_EPOLL)
  241. #define TEST_PARAM_VALUES kLibevent, kEpoll
  242. #else
  243. #define TEST_PARAM_VALUES kLibevent
  244. #endif
  245. INSTANTIATE_TEST_SUITE_P(,
  246. MessagePumpLibeventTest,
  247. ::testing::Values(TEST_PARAM_VALUES));
  248. } // namespace
  249. } // namespace base