diff options
author | brettw@chromium.org <brettw@chromium.org@0039d316-1c4b-4281-b951-d872f2087c98> | 2013-06-14 22:51:28 +0000 |
---|---|---|
committer | brettw@chromium.org <brettw@chromium.org@0039d316-1c4b-4281-b951-d872f2087c98> | 2013-06-14 22:51:28 +0000 |
commit | e5157497585854be6297aae43040310211e5f3ef (patch) | |
tree | eac9fae1a9f0951df900580a23cf68665369e66e /base/message_pump_libevent.cc | |
parent | 0acd332a7165dfb8f3c9d16d06592695a34b5ff8 (diff) | |
download | chromium_src-e5157497585854be6297aae43040310211e5f3ef.zip chromium_src-e5157497585854be6297aae43040310211e5f3ef.tar.gz chromium_src-e5157497585854be6297aae43040310211e5f3ef.tar.bz2 |
Move message_pump to base/message_loop.
This also fixes some namespace usage inside the message pump files and updates all users of these files to use the new location.
BUG=
R=sky@chromium.org
Review URL: https://codereview.chromium.org/17078005
git-svn-id: svn://svn.chromium.org/chrome/trunk/src@206507 0039d316-1c4b-4281-b951-d872f2087c98
Diffstat (limited to 'base/message_pump_libevent.cc')
-rw-r--r-- | base/message_pump_libevent.cc | 379 |
1 files changed, 0 insertions, 379 deletions
diff --git a/base/message_pump_libevent.cc b/base/message_pump_libevent.cc deleted file mode 100644 index 364f44a..0000000 --- a/base/message_pump_libevent.cc +++ /dev/null @@ -1,379 +0,0 @@ -// 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_pump_libevent.h" - -#include <errno.h> -#include <fcntl.h> -#include <unistd.h> - -#include "base/auto_reset.h" -#include "base/compiler_specific.h" -#include "base/logging.h" -#if defined(OS_MACOSX) -#include "base/mac/scoped_nsautorelease_pool.h" -#endif -#include "base/memory/scoped_ptr.h" -#include "base/observer_list.h" -#include "base/posix/eintr_wrapper.h" -#include "base/time.h" -#include "third_party/libevent/event.h" - -#if defined(OS_MACOSX) -#include "base/mac/scoped_nsautorelease_pool.h" -#endif - -// Lifecycle of struct event -// Libevent uses two main data structures: -// struct event_base (of which there is one per message pump), and -// struct event (of which there is roughly one per socket). -// The socket's struct event is created in -// MessagePumpLibevent::WatchFileDescriptor(), -// is owned by the FileDescriptorWatcher, and is destroyed in -// StopWatchingFileDescriptor(). -// It is moved into and out of lists in struct event_base by -// the libevent functions event_add() and event_del(). -// -// TODO(dkegel): -// At the moment bad things happen if a FileDescriptorWatcher -// is active after its MessagePumpLibevent has been destroyed. -// See MessageLoopTest.FileDescriptorWatcherOutlivesMessageLoop -// Not clear yet whether that situation occurs in practice, -// but if it does, we need to fix it. - -namespace base { - -// Return 0 on success -// Too small a function to bother putting in a library? -static int SetNonBlocking(int fd) { - int flags = fcntl(fd, F_GETFL, 0); - if (flags == -1) - flags = 0; - return fcntl(fd, F_SETFL, flags | O_NONBLOCK); -} - -MessagePumpLibevent::FileDescriptorWatcher::FileDescriptorWatcher() - : event_(NULL), - pump_(NULL), - watcher_(NULL), - weak_factory_(this) { -} - -MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() { - if (event_) { - StopWatchingFileDescriptor(); - } -} - -bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() { - event* e = ReleaseEvent(); - if (e == NULL) - return true; - - // event_del() is a no-op if the event isn't active. - int rv = event_del(e); - delete e; - pump_ = NULL; - watcher_ = NULL; - return (rv == 0); -} - -void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e) { - DCHECK(e); - DCHECK(!event_); - - event_ = e; -} - -event *MessagePumpLibevent::FileDescriptorWatcher::ReleaseEvent() { - struct event *e = event_; - event_ = NULL; - return e; -} - -void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanReadWithoutBlocking( - int fd, MessagePumpLibevent* pump) { - // Since OnFileCanWriteWithoutBlocking() gets called first, it can stop - // watching the file descriptor. - if (!watcher_) - return; - pump->WillProcessIOEvent(); - watcher_->OnFileCanReadWithoutBlocking(fd); - pump->DidProcessIOEvent(); -} - -void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking( - int fd, MessagePumpLibevent* pump) { - DCHECK(watcher_); - pump->WillProcessIOEvent(); - watcher_->OnFileCanWriteWithoutBlocking(fd); - pump->DidProcessIOEvent(); -} - -MessagePumpLibevent::MessagePumpLibevent() - : keep_running_(true), - in_run_(false), - processed_io_events_(false), - event_base_(event_base_new()), - wakeup_pipe_in_(-1), - wakeup_pipe_out_(-1) { - if (!Init()) - NOTREACHED(); -} - -MessagePumpLibevent::~MessagePumpLibevent() { - DCHECK(wakeup_event_); - DCHECK(event_base_); - event_del(wakeup_event_); - delete wakeup_event_; - if (wakeup_pipe_in_ >= 0) { - if (HANDLE_EINTR(close(wakeup_pipe_in_)) < 0) - DPLOG(ERROR) << "close"; - } - if (wakeup_pipe_out_ >= 0) { - if (HANDLE_EINTR(close(wakeup_pipe_out_)) < 0) - DPLOG(ERROR) << "close"; - } - event_base_free(event_base_); -} - -bool MessagePumpLibevent::WatchFileDescriptor(int fd, - bool persistent, - int mode, - FileDescriptorWatcher *controller, - Watcher *delegate) { - DCHECK_GE(fd, 0); - DCHECK(controller); - DCHECK(delegate); - DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE); - // WatchFileDescriptor should be called on the pump thread. It is not - // threadsafe, and your watcher may never be registered. - DCHECK(watch_file_descriptor_caller_checker_.CalledOnValidThread()); - - int event_mask = persistent ? EV_PERSIST : 0; - if (mode & WATCH_READ) { - event_mask |= EV_READ; - } - if (mode & WATCH_WRITE) { - event_mask |= EV_WRITE; - } - - scoped_ptr<event> evt(controller->ReleaseEvent()); - if (evt.get() == NULL) { - // Ownership is transferred to the controller. - evt.reset(new event); - } else { - // Make sure we don't pick up any funky internal libevent masks. - int old_interest_mask = evt.get()->ev_events & - (EV_READ | EV_WRITE | EV_PERSIST); - - // Combine old/new event masks. - event_mask |= old_interest_mask; - - // Must disarm the event before we can reuse it. - event_del(evt.get()); - - // It's illegal to use this function to listen on 2 separate fds with the - // same |controller|. - if (EVENT_FD(evt.get()) != fd) { - NOTREACHED() << "FDs don't match" << EVENT_FD(evt.get()) << "!=" << fd; - return false; - } - } - - // Set current interest mask and message pump for this event. - event_set(evt.get(), fd, event_mask, OnLibeventNotification, controller); - - // Tell libevent which message pump this socket will belong to when we add it. - if (event_base_set(event_base_, evt.get())) { - return false; - } - - // Add this socket to the list of monitored sockets. - if (event_add(evt.get(), NULL)) { - return false; - } - - // Transfer ownership of evt to controller. - controller->Init(evt.release()); - - controller->set_watcher(delegate); - controller->set_pump(this); - - return true; -} - -void MessagePumpLibevent::AddIOObserver(IOObserver *obs) { - io_observers_.AddObserver(obs); -} - -void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) { - io_observers_.RemoveObserver(obs); -} - -// Tell libevent to break out of inner loop. -static void timer_callback(int fd, short events, void *context) -{ - event_base_loopbreak((struct event_base *)context); -} - -// Reentrant! -void MessagePumpLibevent::Run(Delegate* delegate) { - DCHECK(keep_running_) << "Quit must have been called outside of Run!"; - base::AutoReset<bool> auto_reset_in_run(&in_run_, true); - - // event_base_loopexit() + EVLOOP_ONCE is leaky, see http://crbug.com/25641. - // Instead, make our own timer and reuse it on each call to event_base_loop(). - scoped_ptr<event> timer_event(new event); - - for (;;) { -#if defined(OS_MACOSX) - mac::ScopedNSAutoreleasePool autorelease_pool; -#endif - - bool did_work = delegate->DoWork(); - if (!keep_running_) - break; - - event_base_loop(event_base_, EVLOOP_NONBLOCK); - did_work |= processed_io_events_; - processed_io_events_ = false; - if (!keep_running_) - break; - - did_work |= delegate->DoDelayedWork(&delayed_work_time_); - if (!keep_running_) - break; - - if (did_work) - continue; - - did_work = delegate->DoIdleWork(); - if (!keep_running_) - break; - - if (did_work) - continue; - - // EVLOOP_ONCE tells libevent to only block once, - // but to service all pending events when it wakes up. - if (delayed_work_time_.is_null()) { - event_base_loop(event_base_, EVLOOP_ONCE); - } else { - TimeDelta delay = delayed_work_time_ - TimeTicks::Now(); - if (delay > TimeDelta()) { - struct timeval poll_tv; - poll_tv.tv_sec = delay.InSeconds(); - poll_tv.tv_usec = delay.InMicroseconds() % Time::kMicrosecondsPerSecond; - event_set(timer_event.get(), -1, 0, timer_callback, event_base_); - event_base_set(event_base_, timer_event.get()); - event_add(timer_event.get(), &poll_tv); - event_base_loop(event_base_, EVLOOP_ONCE); - event_del(timer_event.get()); - } else { - // It looks like delayed_work_time_ indicates a time in the past, so we - // need to call DoDelayedWork now. - delayed_work_time_ = TimeTicks(); - } - } - } - - keep_running_ = true; -} - -void MessagePumpLibevent::Quit() { - DCHECK(in_run_); - // Tell both libevent and Run that they should break out of their loops. - keep_running_ = false; - ScheduleWork(); -} - -void MessagePumpLibevent::ScheduleWork() { - // Tell libevent (in a threadsafe way) that it should break out of its loop. - char buf = 0; - int nwrite = HANDLE_EINTR(write(wakeup_pipe_in_, &buf, 1)); - DCHECK(nwrite == 1 || errno == EAGAIN) - << "[nwrite:" << nwrite << "] [errno:" << errno << "]"; -} - -void MessagePumpLibevent::ScheduleDelayedWork( - const TimeTicks& delayed_work_time) { - // We know that we can't be blocked on Wait right now since this method can - // only be called on the same thread as Run, so we only need to update our - // record of how long to sleep when we do sleep. - delayed_work_time_ = delayed_work_time; -} - -void MessagePumpLibevent::WillProcessIOEvent() { - FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent()); -} - -void MessagePumpLibevent::DidProcessIOEvent() { - FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent()); -} - -bool MessagePumpLibevent::Init() { - int fds[2]; - if (pipe(fds)) { - DLOG(ERROR) << "pipe() failed, errno: " << errno; - return false; - } - if (SetNonBlocking(fds[0])) { - DLOG(ERROR) << "SetNonBlocking for pipe fd[0] failed, errno: " << errno; - return false; - } - if (SetNonBlocking(fds[1])) { - DLOG(ERROR) << "SetNonBlocking for pipe fd[1] failed, errno: " << errno; - return false; - } - wakeup_pipe_out_ = fds[0]; - wakeup_pipe_in_ = fds[1]; - - wakeup_event_ = new event; - event_set(wakeup_event_, wakeup_pipe_out_, EV_READ | EV_PERSIST, - OnWakeup, this); - event_base_set(event_base_, wakeup_event_); - - if (event_add(wakeup_event_, 0)) - return false; - return true; -} - -// static -void MessagePumpLibevent::OnLibeventNotification(int fd, short flags, - void* context) { - base::WeakPtr<FileDescriptorWatcher> controller = - static_cast<FileDescriptorWatcher*>(context)->weak_factory_.GetWeakPtr(); - DCHECK(controller.get()); - - MessagePumpLibevent* pump = controller->pump(); - pump->processed_io_events_ = true; - - if (flags & EV_WRITE) { - controller->OnFileCanWriteWithoutBlocking(fd, pump); - } - // Check |controller| in case it's been deleted in - // controller->OnFileCanWriteWithoutBlocking(). - if (controller.get() && flags & EV_READ) { - controller->OnFileCanReadWithoutBlocking(fd, pump); - } -} - -// Called if a byte is received on the wakeup pipe. -// static -void MessagePumpLibevent::OnWakeup(int socket, short flags, void* context) { - base::MessagePumpLibevent* that = - static_cast<base::MessagePumpLibevent*>(context); - DCHECK(that->wakeup_pipe_out_ == socket); - - // Remove and discard the wakeup byte. - char buf; - int nread = HANDLE_EINTR(read(socket, &buf, 1)); - DCHECK_EQ(nread, 1); - that->processed_io_events_ = true; - // Tell libevent to break out of inner loop. - event_base_loopbreak(that->event_base_); -} - -} // namespace base |