message_pump_libevent refactor:

chrome/common/ipc_channel_posix.cc

 // Copyright (c) 2008 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 "chrome/common/ipc_channel_posix.h"
 
 #include <errno.h>
 #include <fcntl.h>
+#include <stddef.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/stat.h>
-#include <stddef.h>
+#if defined(OS_LINUX)
+#include <linux/un.h>
+#elif defined(OS_MACOSX)
+#include <sys/un.h>

[Mark Mentovai, 2008/12/15 21:43:40]

Did you try to use <sys/un.h> on Linux here?

+#endif
+
 
 #include "base/logging.h"
 #include "base/process_util.h"
 #include "base/scoped_ptr.h"
 #include "base/string_util.h"
 #include "chrome/common/chrome_counters.h"
 #include "chrome/common/ipc_message_utils.h"
 
-#if defined(OS_LINUX)
-#include <linux/un.h>
-#elif defined(OS_MACOSX)
-#include <sys/un.h>
-#endif
-
 namespace IPC {
 
 //------------------------------------------------------------------------------
 // TODO(playmobil): Only use FIFOs for debugging, for real work, use a
 // socketpair.
 namespace {
 
 // The -1 is to take the NULL terminator into account.
 #if defined(OS_LINUX)
 const size_t kMaxPipeNameLength = UNIX_PATH_MAX - 1;
@@ skipping 106 matching lines @@
   *client_socket = fd;
   return true;
 }
 
 }  // namespace
 //------------------------------------------------------------------------------
 
 Channel::ChannelImpl::ChannelImpl(const std::wstring& channel_id, Mode mode,
                                   Listener* listener)
     : mode_(mode),
-    server_listen_connection_event_(new EventHolder()),
-    read_event_(new EventHolder()),
-    write_event_(new EventHolder()),
+    is_blocked_on_write_(false),

[Mark Mentovai, 2008/12/15 21:43:40]

Indentation messed up through here.

     message_send_bytes_written_(0),
     server_listen_pipe_(-1),
     pipe_(-1),
     listener_(listener),
     waiting_connect_(true),
     processing_incoming_(false),
     factory_(this) {
   if (!CreatePipe(channel_id, mode)) {
     // The pipe may have been closed already.
     LOG(WARNING) << "Unable to create pipe named \"" << channel_id <<
@@ skipping 41 matching lines @@
 
   output_queue_.push(msg.release());
   return true;
 }
 
 bool Channel::ChannelImpl::Connect() {
   if (mode_ == MODE_SERVER) {
     if (server_listen_pipe_ == -1) {
       return false;
     }
-    event *ev = &(server_listen_connection_event_->event);
-    MessageLoopForIO::current()->WatchFileHandle(server_listen_pipe_,
-                                                 EV_READ | EV_PERSIST,
-                                                 ev,
-                                                 this);
-    server_listen_connection_event_->is_active = true;
+    MessageLoopForIO::current()->WatchFileDescriptor(
+        server_listen_pipe_,
+        true,
+        MessageLoopForIO::WATCH_READ,
+        &server_listen_connection_watcher_,
+        this);
   } else {
     if (pipe_ == -1) {
       return false;
     }
-    MessageLoopForIO::current()->WatchFileHandle(pipe_,
-                                                 EV_READ | EV_PERSIST,
-                                                 &(read_event_->event),
-                                                 this);
-    read_event_->is_active = true;
+    MessageLoopForIO::current()->WatchFileDescriptor(
+        pipe_,
+        true,
+        MessageLoopForIO::WATCH_READ,
+        &read_watcher_,
+        this);
     waiting_connect_ = false;
   }
 
   if (!waiting_connect_)
     return ProcessOutgoingMessages();
   return true;
 }
 
 bool Channel::ChannelImpl::ProcessIncomingMessages() {
   ssize_t bytes_read = 0;
@@ skipping 68 matching lines @@
 
     bytes_read = 0;  // Get more data.
   }
 
   return true;
 }
 
 bool Channel::ChannelImpl::ProcessOutgoingMessages() {
   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
                               // no connection?
+  is_blocked_on_write_ = false;
 
   if (output_queue_.empty())
     return true;
 
   if (pipe_ == -1)
     return false;
 
-  // If libevent was monitoring the socket for us (we blocked when trying to
-  // write a message last time), then delete the underlying libevent structure.
-  if (write_event_->is_active) {
-    // TODO(playmobil): This calls event_del(), but we can probably
-    // do with just calling event_add here.
-    MessageLoopForIO::current()->UnwatchFileHandle(&(write_event_->event));
-    write_event_->is_active = false;
-  }
-
   // Write out all the messages we can till the write blocks or there are no
   // more outgoing messages.
   while (!output_queue_.empty()) {
     Message* msg = output_queue_.front();
 
     size_t amt_to_write = msg->size() - message_send_bytes_written_;
     const char *out_bytes = reinterpret_cast<const char*>(msg->data()) +
         message_send_bytes_written_;
     ssize_t bytes_written = -1;
     do {
       bytes_written = write(pipe_, out_bytes, amt_to_write);
     } while (bytes_written == -1 && errno == EINTR);
 
     if (bytes_written < 0) {
       LOG(ERROR) << "pipe error: " << strerror(errno);
       return false;
     }
 
     if (static_cast<size_t>(bytes_written) != amt_to_write) {
       message_send_bytes_written_ += bytes_written;
 
       // Tell libevent to call us back once things are unblocked.
-      MessageLoopForIO::current()->WatchFileHandle(server_listen_pipe_,
-                                                   EV_WRITE,
-                                                   &(write_event_->event),
-                                                   this);
-      write_event_->is_active = true;
-
+      is_blocked_on_write_ = true;
+      MessageLoopForIO::current()->WatchFileDescriptor(
+          pipe_,
+          false,  // One shot
+          MessageLoopForIO::WATCH_WRITE,
+          &write_watcher_,
+          this);
     } else {
       message_send_bytes_written_ = 0;
 
       // Message sent OK!
 #ifdef IPC_MESSAGE_DEBUG_EXTRA
       DLOG(INFO) << "sent message @" << msg << " on channel @" << this <<
                     " with type " << msg->type();
 #endif
       output_queue_.pop();
       delete msg;
@@ skipping 10 matching lines @@
              << " (" << output_queue_.size() << " in queue)";
 #endif
 
 // TODO(playmobil): implement
 //  #ifdef IPC_MESSAGE_LOG_ENABLED
 //    Logging::current()->OnSendMessage(message, L"");
 //  #endif
 
   output_queue_.push(message);
   if (!waiting_connect_) {
-    if (!write_event_->is_active) {
+    if (!is_blocked_on_write_) {
       if (!ProcessOutgoingMessages())
         return false;
     }
   }
 
   return true;
 }
 
 // Called by libevent when we can read from th pipe without blocking.
-void Channel::ChannelImpl::OnFileReadReady(int fd) {
+void Channel::ChannelImpl::OnFileCanReadWithoutBlocking(int fd) {
   bool send_server_hello_msg = false;
   if (waiting_connect_ && mode_ == MODE_SERVER) {
     if (!ServerAcceptFifoConnection(server_listen_pipe_, &pipe_)) {
       Close();
     }
 
     // No need to watch the listening socket any longer since only one client
     // can connect.  So unregister with libevent.
-    event *ev = &(server_listen_connection_event_->event);
-    MessageLoopForIO::current()->UnwatchFileHandle(ev);
-    server_listen_connection_event_->is_active = false;
+    server_listen_connection_watcher_.StopWatchingFileDescriptor();
 
     // Start watching our end of the socket.
-    MessageLoopForIO::current()->WatchFileHandle(pipe_,
-                                                 EV_READ | EV_PERSIST,
-                                                 &(read_event_->event),
-                                                 this);
-    read_event_->is_active = true;
+    MessageLoopForIO::current()->WatchFileDescriptor(
+        pipe_,
+        true,
+        MessageLoopForIO::WATCH_READ,
+        &read_watcher_,
+        this);
+
     waiting_connect_ = false;
     send_server_hello_msg = true;
   }
 
   if (!waiting_connect_ && fd == pipe_) {
     if (!ProcessIncomingMessages()) {
       Close();
       listener_->OnChannelError();
     }
   }
 
   // If we're a server and handshaking, then we want to make sure that we
   // only send our handshake message after we've processed the client's.
   // This gives us a chance to kill the client if the incoming handshake
   // is invalid.
   if (send_server_hello_msg) {
+    // This should be our first write so there' sno chance we can block here...
+    DCHECK(is_blocked_on_write_ == false);
     ProcessOutgoingMessages();
   }
 }
 
 // Called by libevent when we can write to the pipe without blocking.
-void Channel::ChannelImpl::OnFileWriteReady(int fd) {
+void Channel::ChannelImpl::OnFileCanWriteWithoutBlocking(int fd) {
   if (!ProcessOutgoingMessages()) {
     Close();
     listener_->OnChannelError();
   }
 }
 
 void Channel::ChannelImpl::Close() {
   // Close can be called multiple time, so we need to make sure we're
   // idempotent.
 
   // Unregister libevent for the listening socket and close it.
-  if (server_listen_connection_event_ &&
-      server_listen_connection_event_->is_active) {
-    MessageLoopForIO::current()->UnwatchFileHandle(
-          &(server_listen_connection_event_->event));
-  }
+  server_listen_connection_watcher_.StopWatchingFileDescriptor();
 
   if (server_listen_pipe_ != -1) {
     close(server_listen_pipe_);
     server_listen_pipe_ = -1;
   }
 
   // Unregister libevent for the FIFO and close it.
-  if (read_event_ && read_event_->is_active) {
-    MessageLoopForIO::current()->UnwatchFileHandle(&(read_event_->event));
-  }
-  if (write_event_ && write_event_->is_active) {
-    MessageLoopForIO::current()->UnwatchFileHandle(&(write_event_->event));
-  }
+  read_watcher_.StopWatchingFileDescriptor();
+  write_watcher_.StopWatchingFileDescriptor();
   if (pipe_ != -1) {
     close(pipe_);
     pipe_ = -1;
   }
 
-  delete server_listen_connection_event_;
-  server_listen_connection_event_ = NULL;
-  delete read_event_;
-  read_event_ = NULL;
-  delete write_event_;
-  write_event_ = NULL;
-
   // Unlink the FIFO
   unlink(pipe_name_.c_str());
 
   while (!output_queue_.empty()) {
     Message* m = output_queue_.front();
     output_queue_.pop();
     delete m;
   }
 }
 
@@ skipping 18 matching lines @@
 
 void Channel::set_listener(Listener* listener) {
   channel_impl_->set_listener(listener);
 }
 
 bool Channel::Send(Message* message) {
   return channel_impl_->Send(message);
 }
 
 }  // namespace IPC