message_pump_glib.cc 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477
  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_glib.h"
  5. #include <fcntl.h>
  6. #include <glib.h>
  7. #include <math.h>
  8. #include "base/logging.h"
  9. #include "base/memory/raw_ptr.h"
  10. #include "base/notreached.h"
  11. #include "base/numerics/safe_conversions.h"
  12. #include "base/posix/eintr_wrapper.h"
  13. #include "base/synchronization/lock.h"
  14. #include "base/threading/platform_thread.h"
  15. namespace base {
  16. namespace {
  17. // Priorities of event sources are important to let everything be processed.
  18. // In particular, GTK event source should have the highest priority (because
  19. // UI events come from it), then Wayland events (the ones coming from the FD
  20. // watcher), and the lowest priority is GLib events (our base message pump).
  21. //
  22. // The g_source API uses ints to denote priorities, and the lower is its value,
  23. // the higher is the priority (i.e., they are ordered backwards).
  24. constexpr int kPriorityWork = G_PRIORITY_DEFAULT_IDLE;
  25. constexpr int kPriorityFdWatch = G_PRIORITY_DEFAULT_IDLE - 10;
  26. // See the explanation above.
  27. static_assert(G_PRIORITY_DEFAULT < kPriorityFdWatch &&
  28. kPriorityFdWatch < kPriorityWork,
  29. "Wrong priorities are set for event sources!");
  30. // Return a timeout suitable for the glib loop according to |next_task_time|, -1
  31. // to block forever, 0 to return right away, or a timeout in milliseconds from
  32. // now.
  33. int GetTimeIntervalMilliseconds(TimeTicks next_task_time) {
  34. if (next_task_time.is_null())
  35. return 0;
  36. else if (next_task_time.is_max())
  37. return -1;
  38. auto timeout_ms =
  39. (next_task_time - TimeTicks::Now()).InMillisecondsRoundedUp();
  40. return timeout_ms < 0 ? 0 : saturated_cast<int>(timeout_ms);
  41. }
  42. bool RunningOnMainThread() {
  43. auto pid = getpid();
  44. auto tid = PlatformThread::CurrentId();
  45. return pid > 0 && tid > 0 && pid == tid;
  46. }
  47. // A brief refresher on GLib:
  48. // GLib sources have four callbacks: Prepare, Check, Dispatch and Finalize.
  49. // On each iteration of the GLib pump, it calls each source's Prepare function.
  50. // This function should return TRUE if it wants GLib to call its Dispatch, and
  51. // FALSE otherwise. It can also set a timeout in this case for the next time
  52. // Prepare should be called again (it may be called sooner).
  53. // After the Prepare calls, GLib does a poll to check for events from the
  54. // system. File descriptors can be attached to the sources. The poll may block
  55. // if none of the Prepare calls returned TRUE. It will block indefinitely, or
  56. // by the minimum time returned by a source in Prepare.
  57. // After the poll, GLib calls Check for each source that returned FALSE
  58. // from Prepare. The return value of Check has the same meaning as for Prepare,
  59. // making Check a second chance to tell GLib we are ready for Dispatch.
  60. // Finally, GLib calls Dispatch for each source that is ready. If Dispatch
  61. // returns FALSE, GLib will destroy the source. Dispatch calls may be recursive
  62. // (i.e., you can call Run from them), but Prepare and Check cannot.
  63. // Finalize is called when the source is destroyed.
  64. // NOTE: It is common for subsystems to want to process pending events while
  65. // doing intensive work, for example the flash plugin. They usually use the
  66. // following pattern (recommended by the GTK docs):
  67. // while (gtk_events_pending()) {
  68. // gtk_main_iteration();
  69. // }
  70. //
  71. // gtk_events_pending just calls g_main_context_pending, which does the
  72. // following:
  73. // - Call prepare on all the sources.
  74. // - Do the poll with a timeout of 0 (not blocking).
  75. // - Call check on all the sources.
  76. // - *Does not* call dispatch on the sources.
  77. // - Return true if any of prepare() or check() returned true.
  78. //
  79. // gtk_main_iteration just calls g_main_context_iteration, which does the whole
  80. // thing, respecting the timeout for the poll (and block, although it is to if
  81. // gtk_events_pending returned true), and call dispatch.
  82. //
  83. // Thus it is important to only return true from prepare or check if we
  84. // actually have events or work to do. We also need to make sure we keep
  85. // internal state consistent so that if prepare/check return true when called
  86. // from gtk_events_pending, they will still return true when called right
  87. // after, from gtk_main_iteration.
  88. //
  89. // For the GLib pump we try to follow the Windows UI pump model:
  90. // - Whenever we receive a wakeup event or the timer for delayed work expires,
  91. // we run DoWork. That part will also run in the other event pumps.
  92. // - We also run DoWork, and possibly DoIdleWork, in the main loop,
  93. // around event handling.
  94. struct WorkSource : public GSource {
  95. raw_ptr<MessagePumpGlib> pump;
  96. };
  97. gboolean WorkSourcePrepare(GSource* source, gint* timeout_ms) {
  98. *timeout_ms = static_cast<WorkSource*>(source)->pump->HandlePrepare();
  99. // We always return FALSE, so that our timeout is honored. If we were
  100. // to return TRUE, the timeout would be considered to be 0 and the poll
  101. // would never block. Once the poll is finished, Check will be called.
  102. return FALSE;
  103. }
  104. gboolean WorkSourceCheck(GSource* source) {
  105. // Only return TRUE if Dispatch should be called.
  106. return static_cast<WorkSource*>(source)->pump->HandleCheck();
  107. }
  108. gboolean WorkSourceDispatch(GSource* source,
  109. GSourceFunc unused_func,
  110. gpointer unused_data) {
  111. static_cast<WorkSource*>(source)->pump->HandleDispatch();
  112. // Always return TRUE so our source stays registered.
  113. return TRUE;
  114. }
  115. // I wish these could be const, but g_source_new wants non-const.
  116. GSourceFuncs WorkSourceFuncs = {WorkSourcePrepare, WorkSourceCheck,
  117. WorkSourceDispatch, nullptr};
  118. struct FdWatchSource : public GSource {
  119. raw_ptr<MessagePumpGlib> pump;
  120. raw_ptr<MessagePumpGlib::FdWatchController> controller;
  121. };
  122. gboolean FdWatchSourcePrepare(GSource* source, gint* timeout_ms) {
  123. *timeout_ms = -1;
  124. return FALSE;
  125. }
  126. gboolean FdWatchSourceCheck(GSource* gsource) {
  127. auto* source = static_cast<FdWatchSource*>(gsource);
  128. return source->pump->HandleFdWatchCheck(source->controller) ? TRUE : FALSE;
  129. }
  130. gboolean FdWatchSourceDispatch(GSource* gsource,
  131. GSourceFunc unused_func,
  132. gpointer unused_data) {
  133. auto* source = static_cast<FdWatchSource*>(gsource);
  134. source->pump->HandleFdWatchDispatch(source->controller);
  135. return TRUE;
  136. }
  137. GSourceFuncs g_fd_watch_source_funcs = {
  138. FdWatchSourcePrepare, FdWatchSourceCheck, FdWatchSourceDispatch, nullptr};
  139. } // namespace
  140. struct MessagePumpGlib::RunState {
  141. raw_ptr<Delegate> delegate;
  142. // Used to flag that the current Run() invocation should return ASAP.
  143. bool should_quit;
  144. // Used to count how many Run() invocations are on the stack.
  145. int run_depth;
  146. // The information of the next task available at this run-level. Stored in
  147. // RunState because different set of tasks can be accessible at various
  148. // run-levels (e.g. non-nestable tasks).
  149. Delegate::NextWorkInfo next_work_info;
  150. };
  151. MessagePumpGlib::MessagePumpGlib()
  152. : state_(nullptr), wakeup_gpollfd_(std::make_unique<GPollFD>()) {
  153. DCHECK(!g_main_context_get_thread_default());
  154. if (RunningOnMainThread()) {
  155. context_ = g_main_context_default();
  156. } else {
  157. owned_context_ = std::unique_ptr<GMainContext, GMainContextDeleter>(
  158. g_main_context_new());
  159. context_ = owned_context_.get();
  160. g_main_context_push_thread_default(context_);
  161. }
  162. // Create our wakeup pipe, which is used to flag when work was scheduled.
  163. int fds[2];
  164. [[maybe_unused]] int ret = pipe(fds);
  165. DCHECK_EQ(ret, 0);
  166. wakeup_pipe_read_ = fds[0];
  167. wakeup_pipe_write_ = fds[1];
  168. wakeup_gpollfd_->fd = wakeup_pipe_read_;
  169. wakeup_gpollfd_->events = G_IO_IN;
  170. work_source_ = std::unique_ptr<GSource, GSourceDeleter>(
  171. g_source_new(&WorkSourceFuncs, sizeof(WorkSource)));
  172. static_cast<WorkSource*>(work_source_.get())->pump = this;
  173. g_source_add_poll(work_source_.get(), wakeup_gpollfd_.get());
  174. g_source_set_priority(work_source_.get(), kPriorityWork);
  175. // This is needed to allow Run calls inside Dispatch.
  176. g_source_set_can_recurse(work_source_.get(), TRUE);
  177. g_source_attach(work_source_.get(), context_);
  178. }
  179. MessagePumpGlib::~MessagePumpGlib() {
  180. work_source_.reset();
  181. close(wakeup_pipe_read_);
  182. close(wakeup_pipe_write_);
  183. context_ = nullptr;
  184. owned_context_.reset();
  185. }
  186. MessagePumpGlib::FdWatchController::FdWatchController(const Location& location)
  187. : FdWatchControllerInterface(location) {}
  188. MessagePumpGlib::FdWatchController::~FdWatchController() {
  189. if (IsInitialized()) {
  190. CHECK(StopWatchingFileDescriptor());
  191. }
  192. if (was_destroyed_) {
  193. DCHECK(!*was_destroyed_);
  194. *was_destroyed_ = true;
  195. }
  196. }
  197. bool MessagePumpGlib::FdWatchController::StopWatchingFileDescriptor() {
  198. if (!IsInitialized())
  199. return false;
  200. g_source_destroy(source_);
  201. g_source_unref(source_);
  202. source_ = nullptr;
  203. watcher_ = nullptr;
  204. return true;
  205. }
  206. bool MessagePumpGlib::FdWatchController::IsInitialized() const {
  207. return !!source_;
  208. }
  209. bool MessagePumpGlib::FdWatchController::InitOrUpdate(int fd,
  210. int mode,
  211. FdWatcher* watcher) {
  212. gushort event_flags = 0;
  213. if (mode & WATCH_READ) {
  214. event_flags |= G_IO_IN;
  215. }
  216. if (mode & WATCH_WRITE) {
  217. event_flags |= G_IO_OUT;
  218. }
  219. if (!IsInitialized()) {
  220. poll_fd_ = std::make_unique<GPollFD>();
  221. poll_fd_->fd = fd;
  222. } else {
  223. if (poll_fd_->fd != fd)
  224. return false;
  225. // Combine old/new event masks.
  226. event_flags |= poll_fd_->events;
  227. // Destroy previous source
  228. bool stopped = StopWatchingFileDescriptor();
  229. DCHECK(stopped);
  230. }
  231. poll_fd_->events = event_flags;
  232. poll_fd_->revents = 0;
  233. source_ = g_source_new(&g_fd_watch_source_funcs, sizeof(FdWatchSource));
  234. DCHECK(source_);
  235. g_source_add_poll(source_, poll_fd_.get());
  236. g_source_set_can_recurse(source_, TRUE);
  237. g_source_set_callback(source_, nullptr, nullptr, nullptr);
  238. g_source_set_priority(source_, kPriorityFdWatch);
  239. watcher_ = watcher;
  240. return true;
  241. }
  242. bool MessagePumpGlib::FdWatchController::Attach(MessagePumpGlib* pump) {
  243. DCHECK(pump);
  244. if (!IsInitialized()) {
  245. return false;
  246. }
  247. auto* source = static_cast<FdWatchSource*>(source_);
  248. source->controller = this;
  249. source->pump = pump;
  250. g_source_attach(source_, pump->context_);
  251. return true;
  252. }
  253. void MessagePumpGlib::FdWatchController::NotifyCanRead() {
  254. if (!watcher_)
  255. return;
  256. DCHECK(poll_fd_);
  257. watcher_->OnFileCanReadWithoutBlocking(poll_fd_->fd);
  258. }
  259. void MessagePumpGlib::FdWatchController::NotifyCanWrite() {
  260. if (!watcher_)
  261. return;
  262. DCHECK(poll_fd_);
  263. watcher_->OnFileCanWriteWithoutBlocking(poll_fd_->fd);
  264. }
  265. bool MessagePumpGlib::WatchFileDescriptor(int fd,
  266. bool persistent,
  267. int mode,
  268. FdWatchController* controller,
  269. FdWatcher* watcher) {
  270. DCHECK_GE(fd, 0);
  271. DCHECK(controller);
  272. DCHECK(watcher);
  273. DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE);
  274. // WatchFileDescriptor should be called on the pump thread. It is not
  275. // threadsafe, so the watcher may never be registered.
  276. DCHECK_CALLED_ON_VALID_THREAD(watch_fd_caller_checker_);
  277. if (!controller->InitOrUpdate(fd, mode, watcher)) {
  278. DPLOG(ERROR) << "FdWatchController init failed (fd=" << fd << ")";
  279. return false;
  280. }
  281. return controller->Attach(this);
  282. }
  283. // Return the timeout we want passed to poll.
  284. int MessagePumpGlib::HandlePrepare() {
  285. // |state_| may be null during tests.
  286. if (!state_)
  287. return 0;
  288. return GetTimeIntervalMilliseconds(state_->next_work_info.delayed_run_time);
  289. }
  290. bool MessagePumpGlib::HandleCheck() {
  291. if (!state_) // state_ may be null during tests.
  292. return false;
  293. // We usually have a single message on the wakeup pipe, since we are only
  294. // signaled when the queue went from empty to non-empty, but there can be
  295. // two messages if a task posted a task, hence we read at most two bytes.
  296. // The glib poll will tell us whether there was data, so this read
  297. // shouldn't block.
  298. if (wakeup_gpollfd_->revents & G_IO_IN) {
  299. char msg[2];
  300. const long num_bytes = HANDLE_EINTR(read(wakeup_pipe_read_, msg, 2));
  301. if (num_bytes < 1) {
  302. NOTREACHED() << "Error reading from the wakeup pipe.";
  303. }
  304. DCHECK((num_bytes == 1 && msg[0] == '!') ||
  305. (num_bytes == 2 && msg[0] == '!' && msg[1] == '!'));
  306. // Since we ate the message, we need to record that we have immediate work,
  307. // because HandleCheck() may be called without HandleDispatch being called
  308. // afterwards.
  309. state_->next_work_info = {TimeTicks()};
  310. return true;
  311. }
  312. // As described in the summary at the top : Check is a second-chance to
  313. // Prepare, verify whether we have work ready again.
  314. if (GetTimeIntervalMilliseconds(state_->next_work_info.delayed_run_time) ==
  315. 0) {
  316. return true;
  317. }
  318. return false;
  319. }
  320. void MessagePumpGlib::HandleDispatch() {
  321. state_->next_work_info = state_->delegate->DoWork();
  322. }
  323. void MessagePumpGlib::Run(Delegate* delegate) {
  324. RunState state;
  325. state.delegate = delegate;
  326. state.should_quit = false;
  327. state.run_depth = state_ ? state_->run_depth + 1 : 1;
  328. RunState* previous_state = state_;
  329. state_ = &state;
  330. // We really only do a single task for each iteration of the loop. If we
  331. // have done something, assume there is likely something more to do. This
  332. // will mean that we don't block on the message pump until there was nothing
  333. // more to do. We also set this to true to make sure not to block on the
  334. // first iteration of the loop, so RunUntilIdle() works correctly.
  335. bool more_work_is_plausible = true;
  336. // We run our own loop instead of using g_main_loop_quit in one of the
  337. // callbacks. This is so we only quit our own loops, and we don't quit
  338. // nested loops run by others. TODO(deanm): Is this what we want?
  339. for (;;) {
  340. // Don't block if we think we have more work to do.
  341. bool block = !more_work_is_plausible;
  342. more_work_is_plausible = g_main_context_iteration(context_, block);
  343. if (state_->should_quit)
  344. break;
  345. state_->next_work_info = state_->delegate->DoWork();
  346. more_work_is_plausible |= state_->next_work_info.is_immediate();
  347. if (state_->should_quit)
  348. break;
  349. if (more_work_is_plausible)
  350. continue;
  351. more_work_is_plausible = state_->delegate->DoIdleWork();
  352. if (state_->should_quit)
  353. break;
  354. }
  355. state_ = previous_state;
  356. }
  357. void MessagePumpGlib::Quit() {
  358. if (state_) {
  359. state_->should_quit = true;
  360. } else {
  361. NOTREACHED() << "Quit called outside Run!";
  362. }
  363. }
  364. void MessagePumpGlib::ScheduleWork() {
  365. // This can be called on any thread, so we don't want to touch any state
  366. // variables as we would then need locks all over. This ensures that if
  367. // we are sleeping in a poll that we will wake up.
  368. char msg = '!';
  369. if (HANDLE_EINTR(write(wakeup_pipe_write_, &msg, 1)) != 1) {
  370. NOTREACHED() << "Could not write to the UI message loop wakeup pipe!";
  371. }
  372. }
  373. void MessagePumpGlib::ScheduleDelayedWork(
  374. const Delegate::NextWorkInfo& next_work_info) {
  375. // We need to wake up the loop in case the poll timeout needs to be
  376. // adjusted. This will cause us to try to do work, but that's OK.
  377. ScheduleWork();
  378. }
  379. bool MessagePumpGlib::HandleFdWatchCheck(FdWatchController* controller) {
  380. DCHECK(controller);
  381. gushort flags = controller->poll_fd_->revents;
  382. return (flags & G_IO_IN) || (flags & G_IO_OUT);
  383. }
  384. void MessagePumpGlib::HandleFdWatchDispatch(FdWatchController* controller) {
  385. DCHECK(controller);
  386. DCHECK(controller->poll_fd_);
  387. gushort flags = controller->poll_fd_->revents;
  388. if ((flags & G_IO_IN) && (flags & G_IO_OUT)) {
  389. // Both callbacks will be called. It is necessary to check that
  390. // |controller| is not destroyed.
  391. bool controller_was_destroyed = false;
  392. controller->was_destroyed_ = &controller_was_destroyed;
  393. controller->NotifyCanWrite();
  394. if (!controller_was_destroyed)
  395. controller->NotifyCanRead();
  396. if (!controller_was_destroyed)
  397. controller->was_destroyed_ = nullptr;
  398. } else if (flags & G_IO_IN) {
  399. controller->NotifyCanRead();
  400. } else if (flags & G_IO_OUT) {
  401. controller->NotifyCanWrite();
  402. }
  403. }
  404. bool MessagePumpGlib::ShouldQuit() const {
  405. CHECK(state_);
  406. return state_->should_quit;
  407. }
  408. } // namespace base