Jankometer: Generalize the code more. Add better support for monitoring IO thread.

base/message_pump_libevent.cc

 // Copyright (c) 2009 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 "eintr_wrapper.h"
 #include "base/auto_reset.h"
+#include "base/eintr_wrapper.h"
 #include "base/logging.h"
+#include "base/observer_list.h"
 #include "base/scoped_nsautorelease_pool.h"
 #include "base/scoped_ptr.h"
 #include "base/time.h"
 #if defined(USE_SYSTEM_LIBEVENT)
 #include <event.h>
 #else
 #include "third_party/libevent/event.h"
 #endif
 
 // Lifecycle of struct event
@@ skipping 20 matching lines @@
 // 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()
     : is_persistent_(false),
-      event_(NULL) {
+      event_(NULL),
+      pump_(NULL),
+      watcher_(NULL) {
 }
 
 MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() {
   if (event_) {
     StopWatchingFileDescriptor();
   }
 }
 
 void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e,
                                                       bool is_persistent) {
@@ skipping 11 matching lines @@
 }
 
 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::OnFileCanReadWithoutBlocking(
+    int fd, MessagePumpLibevent* pump) {
+  pump->WillProcessIOEvent();
+  watcher_->OnFileCanReadWithoutBlocking(fd);
+  pump->DidProcessIOEvent();
+}
+
+void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking(
+    int fd, MessagePumpLibevent* pump) {
+  pump->WillProcessIOEvent();
+  watcher_->OnFileCanWriteWithoutBlocking(fd);
+  pump->DidProcessIOEvent();
+}
+
 // Called if a byte is received on the wakeup pipe.
 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);
@@ skipping 37 matching lines @@
     return false;
   return true;
 }
 
 MessagePumpLibevent::~MessagePumpLibevent() {
   DCHECK(wakeup_event_);
   DCHECK(event_base_);
   event_del(wakeup_event_);
   delete wakeup_event_;
   if (wakeup_pipe_in_ >= 0)
-    close(wakeup_pipe_in_);
+    HANDLE_EINTR(close(wakeup_pipe_in_));
   if (wakeup_pipe_out_ >= 0)
-    close(wakeup_pipe_out_);
+    HANDLE_EINTR(close(wakeup_pipe_out_));
   event_base_free(event_base_);
 }
 
 bool MessagePumpLibevent::WatchFileDescriptor(int fd,
                                               bool persistent,
                                               Mode mode,
                                               FileDescriptorWatcher *controller,
                                               Watcher *delegate) {
   DCHECK_GE(fd, 0);
   DCHECK(controller);
@@ skipping 25 matching lines @@
 
     // 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, delegate);
+  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()) != 0) {
     return false;
   }
 
   // Add this socket to the list of monitored sockets.
   if (event_add(evt.get(), NULL) != 0) {
     return false;
   }
 
   // Transfer ownership of evt to controller.
   controller->Init(evt.release(), persistent);
+
+  controller->set_watcher(delegate);
+  controller->set_pump(this);
+
   return true;
 }
 
-
 void MessagePumpLibevent::OnLibeventNotification(int fd, short flags,
                                                  void* context) {
-  Watcher* watcher = static_cast<Watcher*>(context);
+  FileDescriptorWatcher* controller =
+      static_cast<FileDescriptorWatcher*>(context);
+
+  MessagePumpLibevent* pump = controller->pump();
 
   if (flags & EV_WRITE) {
-    watcher->OnFileCanWriteWithoutBlocking(fd);
+    controller->OnFileCanWriteWithoutBlocking(fd, pump);
   }
   if (flags & EV_READ) {
-    watcher->OnFileCanReadWithoutBlocking(fd);
+    controller->OnFileCanReadWithoutBlocking(fd, pump);
   }
 }
 
 // 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!
@@ skipping 67 matching lines @@
       << "[nwrite:" << nwrite << "] [errno:" << errno << "]";
 }
 
 void MessagePumpLibevent::ScheduleDelayedWork(const Time& 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::AddIOObserver(IOObserver *obs) {
+  io_observers_.AddObserver(obs);
+}
+
+void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) {
+  io_observers_.RemoveObserver(obs);
+}
+
+void MessagePumpLibevent::WillProcessIOEvent() {
+  FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent());
+}
+
+void MessagePumpLibevent::DidProcessIOEvent() {
+  FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent());
+}
+
 }  // namespace base