123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477 |
- // Copyright (c) 2012 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- #include "base/message_loop/message_pump_glib.h"
- #include <fcntl.h>
- #include <glib.h>
- #include <math.h>
- #include "base/logging.h"
- #include "base/memory/raw_ptr.h"
- #include "base/notreached.h"
- #include "base/numerics/safe_conversions.h"
- #include "base/posix/eintr_wrapper.h"
- #include "base/synchronization/lock.h"
- #include "base/threading/platform_thread.h"
- namespace base {
- namespace {
- // Priorities of event sources are important to let everything be processed.
- // In particular, GTK event source should have the highest priority (because
- // UI events come from it), then Wayland events (the ones coming from the FD
- // watcher), and the lowest priority is GLib events (our base message pump).
- //
- // The g_source API uses ints to denote priorities, and the lower is its value,
- // the higher is the priority (i.e., they are ordered backwards).
- constexpr int kPriorityWork = G_PRIORITY_DEFAULT_IDLE;
- constexpr int kPriorityFdWatch = G_PRIORITY_DEFAULT_IDLE - 10;
- // See the explanation above.
- static_assert(G_PRIORITY_DEFAULT < kPriorityFdWatch &&
- kPriorityFdWatch < kPriorityWork,
- "Wrong priorities are set for event sources!");
- // Return a timeout suitable for the glib loop according to |next_task_time|, -1
- // to block forever, 0 to return right away, or a timeout in milliseconds from
- // now.
- int GetTimeIntervalMilliseconds(TimeTicks next_task_time) {
- if (next_task_time.is_null())
- return 0;
- else if (next_task_time.is_max())
- return -1;
- auto timeout_ms =
- (next_task_time - TimeTicks::Now()).InMillisecondsRoundedUp();
- return timeout_ms < 0 ? 0 : saturated_cast<int>(timeout_ms);
- }
- bool RunningOnMainThread() {
- auto pid = getpid();
- auto tid = PlatformThread::CurrentId();
- return pid > 0 && tid > 0 && pid == tid;
- }
- // A brief refresher on GLib:
- // GLib sources have four callbacks: Prepare, Check, Dispatch and Finalize.
- // On each iteration of the GLib pump, it calls each source's Prepare function.
- // This function should return TRUE if it wants GLib to call its Dispatch, and
- // FALSE otherwise. It can also set a timeout in this case for the next time
- // Prepare should be called again (it may be called sooner).
- // After the Prepare calls, GLib does a poll to check for events from the
- // system. File descriptors can be attached to the sources. The poll may block
- // if none of the Prepare calls returned TRUE. It will block indefinitely, or
- // by the minimum time returned by a source in Prepare.
- // After the poll, GLib calls Check for each source that returned FALSE
- // from Prepare. The return value of Check has the same meaning as for Prepare,
- // making Check a second chance to tell GLib we are ready for Dispatch.
- // Finally, GLib calls Dispatch for each source that is ready. If Dispatch
- // returns FALSE, GLib will destroy the source. Dispatch calls may be recursive
- // (i.e., you can call Run from them), but Prepare and Check cannot.
- // Finalize is called when the source is destroyed.
- // NOTE: It is common for subsystems to want to process pending events while
- // doing intensive work, for example the flash plugin. They usually use the
- // following pattern (recommended by the GTK docs):
- // while (gtk_events_pending()) {
- // gtk_main_iteration();
- // }
- //
- // gtk_events_pending just calls g_main_context_pending, which does the
- // following:
- // - Call prepare on all the sources.
- // - Do the poll with a timeout of 0 (not blocking).
- // - Call check on all the sources.
- // - *Does not* call dispatch on the sources.
- // - Return true if any of prepare() or check() returned true.
- //
- // gtk_main_iteration just calls g_main_context_iteration, which does the whole
- // thing, respecting the timeout for the poll (and block, although it is to if
- // gtk_events_pending returned true), and call dispatch.
- //
- // Thus it is important to only return true from prepare or check if we
- // actually have events or work to do. We also need to make sure we keep
- // internal state consistent so that if prepare/check return true when called
- // from gtk_events_pending, they will still return true when called right
- // after, from gtk_main_iteration.
- //
- // For the GLib pump we try to follow the Windows UI pump model:
- // - Whenever we receive a wakeup event or the timer for delayed work expires,
- // we run DoWork. That part will also run in the other event pumps.
- // - We also run DoWork, and possibly DoIdleWork, in the main loop,
- // around event handling.
- struct WorkSource : public GSource {
- raw_ptr<MessagePumpGlib> pump;
- };
- gboolean WorkSourcePrepare(GSource* source, gint* timeout_ms) {
- *timeout_ms = static_cast<WorkSource*>(source)->pump->HandlePrepare();
- // We always return FALSE, so that our timeout is honored. If we were
- // to return TRUE, the timeout would be considered to be 0 and the poll
- // would never block. Once the poll is finished, Check will be called.
- return FALSE;
- }
- gboolean WorkSourceCheck(GSource* source) {
- // Only return TRUE if Dispatch should be called.
- return static_cast<WorkSource*>(source)->pump->HandleCheck();
- }
- gboolean WorkSourceDispatch(GSource* source,
- GSourceFunc unused_func,
- gpointer unused_data) {
- static_cast<WorkSource*>(source)->pump->HandleDispatch();
- // Always return TRUE so our source stays registered.
- return TRUE;
- }
- // I wish these could be const, but g_source_new wants non-const.
- GSourceFuncs WorkSourceFuncs = {WorkSourcePrepare, WorkSourceCheck,
- WorkSourceDispatch, nullptr};
- struct FdWatchSource : public GSource {
- raw_ptr<MessagePumpGlib> pump;
- raw_ptr<MessagePumpGlib::FdWatchController> controller;
- };
- gboolean FdWatchSourcePrepare(GSource* source, gint* timeout_ms) {
- *timeout_ms = -1;
- return FALSE;
- }
- gboolean FdWatchSourceCheck(GSource* gsource) {
- auto* source = static_cast<FdWatchSource*>(gsource);
- return source->pump->HandleFdWatchCheck(source->controller) ? TRUE : FALSE;
- }
- gboolean FdWatchSourceDispatch(GSource* gsource,
- GSourceFunc unused_func,
- gpointer unused_data) {
- auto* source = static_cast<FdWatchSource*>(gsource);
- source->pump->HandleFdWatchDispatch(source->controller);
- return TRUE;
- }
- GSourceFuncs g_fd_watch_source_funcs = {
- FdWatchSourcePrepare, FdWatchSourceCheck, FdWatchSourceDispatch, nullptr};
- } // namespace
- struct MessagePumpGlib::RunState {
- raw_ptr<Delegate> delegate;
- // Used to flag that the current Run() invocation should return ASAP.
- bool should_quit;
- // Used to count how many Run() invocations are on the stack.
- int run_depth;
- // The information of the next task available at this run-level. Stored in
- // RunState because different set of tasks can be accessible at various
- // run-levels (e.g. non-nestable tasks).
- Delegate::NextWorkInfo next_work_info;
- };
- MessagePumpGlib::MessagePumpGlib()
- : state_(nullptr), wakeup_gpollfd_(std::make_unique<GPollFD>()) {
- DCHECK(!g_main_context_get_thread_default());
- if (RunningOnMainThread()) {
- context_ = g_main_context_default();
- } else {
- owned_context_ = std::unique_ptr<GMainContext, GMainContextDeleter>(
- g_main_context_new());
- context_ = owned_context_.get();
- g_main_context_push_thread_default(context_);
- }
- // Create our wakeup pipe, which is used to flag when work was scheduled.
- int fds[2];
- [[maybe_unused]] int ret = pipe(fds);
- DCHECK_EQ(ret, 0);
- wakeup_pipe_read_ = fds[0];
- wakeup_pipe_write_ = fds[1];
- wakeup_gpollfd_->fd = wakeup_pipe_read_;
- wakeup_gpollfd_->events = G_IO_IN;
- work_source_ = std::unique_ptr<GSource, GSourceDeleter>(
- g_source_new(&WorkSourceFuncs, sizeof(WorkSource)));
- static_cast<WorkSource*>(work_source_.get())->pump = this;
- g_source_add_poll(work_source_.get(), wakeup_gpollfd_.get());
- g_source_set_priority(work_source_.get(), kPriorityWork);
- // This is needed to allow Run calls inside Dispatch.
- g_source_set_can_recurse(work_source_.get(), TRUE);
- g_source_attach(work_source_.get(), context_);
- }
- MessagePumpGlib::~MessagePumpGlib() {
- work_source_.reset();
- close(wakeup_pipe_read_);
- close(wakeup_pipe_write_);
- context_ = nullptr;
- owned_context_.reset();
- }
- MessagePumpGlib::FdWatchController::FdWatchController(const Location& location)
- : FdWatchControllerInterface(location) {}
- MessagePumpGlib::FdWatchController::~FdWatchController() {
- if (IsInitialized()) {
- CHECK(StopWatchingFileDescriptor());
- }
- if (was_destroyed_) {
- DCHECK(!*was_destroyed_);
- *was_destroyed_ = true;
- }
- }
- bool MessagePumpGlib::FdWatchController::StopWatchingFileDescriptor() {
- if (!IsInitialized())
- return false;
- g_source_destroy(source_);
- g_source_unref(source_);
- source_ = nullptr;
- watcher_ = nullptr;
- return true;
- }
- bool MessagePumpGlib::FdWatchController::IsInitialized() const {
- return !!source_;
- }
- bool MessagePumpGlib::FdWatchController::InitOrUpdate(int fd,
- int mode,
- FdWatcher* watcher) {
- gushort event_flags = 0;
- if (mode & WATCH_READ) {
- event_flags |= G_IO_IN;
- }
- if (mode & WATCH_WRITE) {
- event_flags |= G_IO_OUT;
- }
- if (!IsInitialized()) {
- poll_fd_ = std::make_unique<GPollFD>();
- poll_fd_->fd = fd;
- } else {
- if (poll_fd_->fd != fd)
- return false;
- // Combine old/new event masks.
- event_flags |= poll_fd_->events;
- // Destroy previous source
- bool stopped = StopWatchingFileDescriptor();
- DCHECK(stopped);
- }
- poll_fd_->events = event_flags;
- poll_fd_->revents = 0;
- source_ = g_source_new(&g_fd_watch_source_funcs, sizeof(FdWatchSource));
- DCHECK(source_);
- g_source_add_poll(source_, poll_fd_.get());
- g_source_set_can_recurse(source_, TRUE);
- g_source_set_callback(source_, nullptr, nullptr, nullptr);
- g_source_set_priority(source_, kPriorityFdWatch);
- watcher_ = watcher;
- return true;
- }
- bool MessagePumpGlib::FdWatchController::Attach(MessagePumpGlib* pump) {
- DCHECK(pump);
- if (!IsInitialized()) {
- return false;
- }
- auto* source = static_cast<FdWatchSource*>(source_);
- source->controller = this;
- source->pump = pump;
- g_source_attach(source_, pump->context_);
- return true;
- }
- void MessagePumpGlib::FdWatchController::NotifyCanRead() {
- if (!watcher_)
- return;
- DCHECK(poll_fd_);
- watcher_->OnFileCanReadWithoutBlocking(poll_fd_->fd);
- }
- void MessagePumpGlib::FdWatchController::NotifyCanWrite() {
- if (!watcher_)
- return;
- DCHECK(poll_fd_);
- watcher_->OnFileCanWriteWithoutBlocking(poll_fd_->fd);
- }
- bool MessagePumpGlib::WatchFileDescriptor(int fd,
- bool persistent,
- int mode,
- FdWatchController* controller,
- FdWatcher* watcher) {
- DCHECK_GE(fd, 0);
- DCHECK(controller);
- DCHECK(watcher);
- DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE);
- // WatchFileDescriptor should be called on the pump thread. It is not
- // threadsafe, so the watcher may never be registered.
- DCHECK_CALLED_ON_VALID_THREAD(watch_fd_caller_checker_);
- if (!controller->InitOrUpdate(fd, mode, watcher)) {
- DPLOG(ERROR) << "FdWatchController init failed (fd=" << fd << ")";
- return false;
- }
- return controller->Attach(this);
- }
- // Return the timeout we want passed to poll.
- int MessagePumpGlib::HandlePrepare() {
- // |state_| may be null during tests.
- if (!state_)
- return 0;
- return GetTimeIntervalMilliseconds(state_->next_work_info.delayed_run_time);
- }
- bool MessagePumpGlib::HandleCheck() {
- if (!state_) // state_ may be null during tests.
- return false;
- // We usually have a single message on the wakeup pipe, since we are only
- // signaled when the queue went from empty to non-empty, but there can be
- // two messages if a task posted a task, hence we read at most two bytes.
- // The glib poll will tell us whether there was data, so this read
- // shouldn't block.
- if (wakeup_gpollfd_->revents & G_IO_IN) {
- char msg[2];
- const long num_bytes = HANDLE_EINTR(read(wakeup_pipe_read_, msg, 2));
- if (num_bytes < 1) {
- NOTREACHED() << "Error reading from the wakeup pipe.";
- }
- DCHECK((num_bytes == 1 && msg[0] == '!') ||
- (num_bytes == 2 && msg[0] == '!' && msg[1] == '!'));
- // Since we ate the message, we need to record that we have immediate work,
- // because HandleCheck() may be called without HandleDispatch being called
- // afterwards.
- state_->next_work_info = {TimeTicks()};
- return true;
- }
- // As described in the summary at the top : Check is a second-chance to
- // Prepare, verify whether we have work ready again.
- if (GetTimeIntervalMilliseconds(state_->next_work_info.delayed_run_time) ==
- 0) {
- return true;
- }
- return false;
- }
- void MessagePumpGlib::HandleDispatch() {
- state_->next_work_info = state_->delegate->DoWork();
- }
- void MessagePumpGlib::Run(Delegate* delegate) {
- RunState state;
- state.delegate = delegate;
- state.should_quit = false;
- state.run_depth = state_ ? state_->run_depth + 1 : 1;
- RunState* previous_state = state_;
- state_ = &state;
- // We really only do a single task for each iteration of the loop. If we
- // have done something, assume there is likely something more to do. This
- // will mean that we don't block on the message pump until there was nothing
- // more to do. We also set this to true to make sure not to block on the
- // first iteration of the loop, so RunUntilIdle() works correctly.
- bool more_work_is_plausible = true;
- // We run our own loop instead of using g_main_loop_quit in one of the
- // callbacks. This is so we only quit our own loops, and we don't quit
- // nested loops run by others. TODO(deanm): Is this what we want?
- for (;;) {
- // Don't block if we think we have more work to do.
- bool block = !more_work_is_plausible;
- more_work_is_plausible = g_main_context_iteration(context_, block);
- if (state_->should_quit)
- break;
- state_->next_work_info = state_->delegate->DoWork();
- more_work_is_plausible |= state_->next_work_info.is_immediate();
- if (state_->should_quit)
- break;
- if (more_work_is_plausible)
- continue;
- more_work_is_plausible = state_->delegate->DoIdleWork();
- if (state_->should_quit)
- break;
- }
- state_ = previous_state;
- }
- void MessagePumpGlib::Quit() {
- if (state_) {
- state_->should_quit = true;
- } else {
- NOTREACHED() << "Quit called outside Run!";
- }
- }
- void MessagePumpGlib::ScheduleWork() {
- // This can be called on any thread, so we don't want to touch any state
- // variables as we would then need locks all over. This ensures that if
- // we are sleeping in a poll that we will wake up.
- char msg = '!';
- if (HANDLE_EINTR(write(wakeup_pipe_write_, &msg, 1)) != 1) {
- NOTREACHED() << "Could not write to the UI message loop wakeup pipe!";
- }
- }
- void MessagePumpGlib::ScheduleDelayedWork(
- const Delegate::NextWorkInfo& next_work_info) {
- // We need to wake up the loop in case the poll timeout needs to be
- // adjusted. This will cause us to try to do work, but that's OK.
- ScheduleWork();
- }
- bool MessagePumpGlib::HandleFdWatchCheck(FdWatchController* controller) {
- DCHECK(controller);
- gushort flags = controller->poll_fd_->revents;
- return (flags & G_IO_IN) || (flags & G_IO_OUT);
- }
- void MessagePumpGlib::HandleFdWatchDispatch(FdWatchController* controller) {
- DCHECK(controller);
- DCHECK(controller->poll_fd_);
- gushort flags = controller->poll_fd_->revents;
- if ((flags & G_IO_IN) && (flags & G_IO_OUT)) {
- // Both callbacks will be called. It is necessary to check that
- // |controller| is not destroyed.
- bool controller_was_destroyed = false;
- controller->was_destroyed_ = &controller_was_destroyed;
- controller->NotifyCanWrite();
- if (!controller_was_destroyed)
- controller->NotifyCanRead();
- if (!controller_was_destroyed)
- controller->was_destroyed_ = nullptr;
- } else if (flags & G_IO_IN) {
- controller->NotifyCanRead();
- } else if (flags & G_IO_OUT) {
- controller->NotifyCanWrite();
- }
- }
- bool MessagePumpGlib::ShouldQuit() const {
- CHECK(state_);
- return state_->should_quit;
- }
- } // namespace base
|