Refactoring on Windows IPC channel.

ipc/ipc_channel_win.cc

 // 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 "ipc/ipc_channel_win.h"
 
 #include <windows.h>
 
 #include "base/auto_reset.h"
 #include "base/bind.h"
@@ skipping 86 matching lines @@
 // static
 bool Channel::ChannelImpl::IsNamedServerInitialized(
     const std::string& channel_id) {
   if (WaitNamedPipe(PipeName(channel_id).c_str(), 1))
     return true;
   // If ERROR_SEM_TIMEOUT occurred, the pipe exists but is handling another
   // connection.
   return GetLastError() == ERROR_SEM_TIMEOUT;
 }
 
+
+Channel::ChannelImpl::ReadState Channel::ChannelImpl::ReadData(
+    char* buffer,
+    int buffer_len,
+    int* /* bytes_read */) {
+  if (INVALID_HANDLE_VALUE == pipe_)
+    return READ_FAILED;
+
+  DWORD bytes_read = 0;
+  BOOL ok = ReadFile(pipe_, input_buf_, Channel::kReadBufferSize,
+                     &bytes_read, &input_state_.context.overlapped);
+  if (!ok) {
+    DWORD err = GetLastError();
+    if (err == ERROR_IO_PENDING) {
+      input_state_.is_pending = true;
+      return READ_PENDING;
+    }
+    LOG(ERROR) << "pipe error: " << err;
+    return READ_FAILED;
+  }
+
+  // We could return READ_SUCCEEDED here. But the way that this code is
+  // structured, but instead we go back to the message loop. Our completion

[rvargas (doing something else), 2012/03/02 02:22:08]

nit: this comment needs some rewording.

+  // port will be signalled even in the "synchronously completed" state.
+  //
+  // This allows us to potentially process some outgoing messages and
+  // interleave other work on this thread when we're getting hammered with
+  // input messages. Potentially, this could be tuned to be more efficient
+  // with some testing.
+  input_state_.is_pending = true;
+  return READ_PENDING;
+}
+
+bool Channel::ChannelImpl::WillDispatchInputMessage(Message* msg) {
+  return true;
+}
+
+void Channel::ChannelImpl::HandleHelloMessage(const Message& msg) {
+  // The hello message contains one parameter containing the PID.
+  listener_->OnChannelConnected(MessageIterator(msg).NextInt());
+}
+
+bool Channel::ChannelImpl::DidEmptyInputBuffers() {
+  return true;
+}
+
+bool Channel::ChannelImpl::DispatchInputData(const char* input_data,
+                                             int input_data_len) {
+  const char* p;
+  const char* end;
+
+  // Possibly combine with the overflow buffer to make a larger buffer.
+  if (input_overflow_buf_.empty()) {
+    p = input_data;
+    end = input_data + input_data_len;
+  } else {
+    if (input_overflow_buf_.size() >
+        kMaximumMessageSize - input_data_len) {
+      input_overflow_buf_.clear();
+      LOG(ERROR) << "IPC message is too big";
+      return false;
+    }
+    input_overflow_buf_.append(input_data, input_data_len);
+    p = input_overflow_buf_.data();
+    end = p + input_overflow_buf_.size();
+  }
+
+  // Dispatch all complete messages in the data buffer.
+  while (p < end) {
+    const char* message_tail = Message::FindNext(p, end);
+    if (message_tail) {
+      int len = static_cast<int>(message_tail - p);
+      Message m(p, len);
+      if (!WillDispatchInputMessage(&m))
+        return false;
+
+      if (IsHelloMessage(m))
+        HandleHelloMessage(m);
+      else
+        listener_->OnMessageReceived(m);
+      p = message_tail;
+    } else {
+      // Last message is partial.
+      break;
+    }
+  }
+
+  // Save any partial data in the overflow buffer.
+  input_overflow_buf_.assign(p, end - p);
+
+  if (input_overflow_buf_.empty() && !DidEmptyInputBuffers())
+    return false;
+  return true;
+}
+
+bool Channel::ChannelImpl::IsHelloMessage(const Message& m) const {
+  return m.routing_id() == MSG_ROUTING_NONE && m.type() == HELLO_MESSAGE_TYPE;
+}
+
+bool Channel::ChannelImpl::AsyncReadComplete(int bytes_read) {
+  return DispatchInputData(input_buf_, bytes_read);
+}
+
 // static
 const std::wstring Channel::ChannelImpl::PipeName(
     const std::string& channel_id) {
   std::string name("\\\\.\\pipe\\chrome.");
   return ASCIIToWide(name.append(channel_id));
 }
 
 bool Channel::ChannelImpl::CreatePipe(const IPC::ChannelHandle &channel_handle,
                                       Mode mode) {
   DCHECK_EQ(INVALID_HANDLE_VALUE, pipe_);
@@ skipping 133 matching lines @@
     // The pipe is being closed.
     return false;
   default:
     NOTREACHED();
     return false;
   }
 
   return true;
 }
 
-bool Channel::ChannelImpl::ProcessIncomingMessages(
-    MessageLoopForIO::IOContext* context,
-    DWORD bytes_read) {
-  DCHECK(thread_check_->CalledOnValidThread());
-  if (input_state_.is_pending) {

[brettw, 2012/03/01 21:24:16]

This check moved to OnIOCompleted.

-    input_state_.is_pending = false;
-    DCHECK(context);
-
-    if (!context || !bytes_read)
+bool Channel::ChannelImpl::ProcessIncomingMessages() {
+  while (true) {
+    int bytes_read = 0;
+    ReadState read_state = ReadData(input_buf_, Channel::kReadBufferSize,
+                                    &bytes_read);
+    if (read_state == READ_FAILED)
       return false;
-  } else {
-    // This happens at channel initialization.
-    DCHECK(!bytes_read && context == &input_state_.context);
+    if (read_state == READ_PENDING)
+      return true;
+    DCHECK(bytes_read > 0);
+    if (!DispatchInputData(input_buf_, bytes_read))
+      return false;
   }
-
-  for (;;) {
-    if (bytes_read == 0) {

[brettw, 2012/03/01 21:24:16]

This code moved to ReadData

-      if (INVALID_HANDLE_VALUE == pipe_)
-        return false;
-
-      // Read from pipe...
-      BOOL ok = ReadFile(pipe_,
-                         input_buf_,
-                         Channel::kReadBufferSize,
-                         &bytes_read,
-                         &input_state_.context.overlapped);
-      if (!ok) {
-        DWORD err = GetLastError();
-        if (err == ERROR_IO_PENDING) {
-          input_state_.is_pending = true;
-          return true;
-        }
-        LOG(ERROR) << "pipe error: " << err;
-        return false;
-      }
-      input_state_.is_pending = true;
-      return true;
-    }
-    DCHECK(bytes_read);
-
-    // Process messages from input buffer.
-
-    const char* p, *end;

[brettw, 2012/03/01 21:24:16]

The rest of this loop moved to DispatchInputData

-    if (input_overflow_buf_.empty()) {
-      p = input_buf_;
-      end = p + bytes_read;
-    } else {
-      if (input_overflow_buf_.size() > (kMaximumMessageSize - bytes_read)) {
-        input_overflow_buf_.clear();
-        LOG(ERROR) << "IPC message is too big";
-        return false;
-      }
-      input_overflow_buf_.append(input_buf_, bytes_read);
-      p = input_overflow_buf_.data();
-      end = p + input_overflow_buf_.size();
-    }
-
-    while (p < end) {
-      const char* message_tail = Message::FindNext(p, end);
-      if (message_tail) {
-        int len = static_cast<int>(message_tail - p);
-        const Message m(p, len);
-        DVLOG(2) << "received message on channel @" << this
-                 << " with type " << m.type();
-        if (m.routing_id() == MSG_ROUTING_NONE &&
-            m.type() == HELLO_MESSAGE_TYPE) {
-          // The Hello message contains only the process id.
-          listener_->OnChannelConnected(MessageIterator(m).NextInt());
-        } else {
-          listener_->OnMessageReceived(m);
-        }
-        p = message_tail;
-      } else {
-        // Last message is partial.
-        break;
-      }
-    }
-    input_overflow_buf_.assign(p, end - p);
-
-    bytes_read = 0;  // Get more data.
-  }
-
-  return true;
 }
 
 bool Channel::ChannelImpl::ProcessOutgoingMessages(
     MessageLoopForIO::IOContext* context,
     DWORD bytes_written) {
   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
                               // no connection?
   DCHECK(thread_check_->CalledOnValidThread());
 
   if (output_state_.is_pending) {
@@ skipping 40 matching lines @@
   }
 
   DVLOG(2) << "sent message @" << m << " on channel @" << this
            << " with type " << m->type();
 
   output_state_.is_pending = true;
   return true;
 }
 
 void Channel::ChannelImpl::OnIOCompleted(MessageLoopForIO::IOContext* context,
-                            DWORD bytes_transfered, DWORD error) {
-  bool ok;
+                                         DWORD bytes_transfered,
+                                         DWORD error) {
+  bool ok = true;
   DCHECK(thread_check_->CalledOnValidThread());
   if (context == &input_state_.context) {
     if (waiting_connect_) {
       if (!ProcessConnection())
         return;
       // We may have some messages queued up to send...
       if (!output_queue_.empty() && !output_state_.is_pending)
         ProcessOutgoingMessages(NULL, 0);
       if (input_state_.is_pending)
         return;
       // else, fall-through and look for incoming messages...
     }
-    // we don't support recursion through OnMessageReceived yet!
+
+    // We don't support recursion through OnMessageReceived yet!
     DCHECK(!processing_incoming_);
     AutoReset<bool> auto_reset_processing_incoming(&processing_incoming_, true);
-    ok = ProcessIncomingMessages(context, bytes_transfered);
+
+    // Process the new data.
+    if (input_state_.is_pending) {
+      // This is the normal case for everything except the initialization step.
+      input_state_.is_pending = false;

[rvargas (doing something else), 2012/03/02 02:22:08]

It may be better to move this block to AsyncReadComplete.

Maybe even the whole if/else... even if the else is not part of a Read... but it
can be an early return there, and it's only slightly out of place.

[brettw, 2012/03/02 05:31:18]

I was going to make AsyncReadComplete cross-platform, and since this is
platform-specific, it needs to go here.

[rvargas (doing something else), 2012/03/02 18:44:17]

OK.

Does that mean that AsyncReadComplete is going to be part of the interface that
this class implements? The description on the header was not clear about how one
would use this method, but I assumed that it was going to be private to the
windows version (so it didn't matter that much). I guess when the rest of the
code appear it would be clear how to use it :)

+      if (!bytes_transfered)
+        ok = false;
+      else
+        ok = AsyncReadComplete(bytes_transfered);

[brettw, 2012/03/01 21:24:16]

ProcessIncomingMessages doesn't need any parameters now that processing async
data is a two-step process.

+    } else {
+      DCHECK(!bytes_transfered);
+    }
+
+    // Request more data.
+    if (ok)
+      ok = ProcessIncomingMessages();
   } else {
     DCHECK(context == &output_state_.context);
     ok = ProcessOutgoingMessages(context, bytes_transfered);
   }
   if (!ok && INVALID_HANDLE_VALUE != pipe_) {
     // We don't want to re-enter Close().
     Close();
     listener_->OnChannelError();
   }
 }
@@ skipping 24 matching lines @@
 bool Channel::Send(Message* message) {
   return channel_impl_->Send(message);
 }
 
 // static
 bool Channel::IsNamedServerInitialized(const std::string& channel_id) {
   return ChannelImpl::IsNamedServerInitialized(channel_id);
 }
 
 }  // namespace IPC