Add thread checks to the IPC channel....

chrome/common/ipc_channel_win.cc

 // Copyright (c) 2006-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_win.h"
 
 #include <windows.h>
 #include <sstream>
 
 #include "base/compiler_specific.h"
 #include "base/logging.h"
+#include "base/non_thread_safe.h"
 #include "base/win_util.h"
 #include "chrome/common/chrome_counters.h"
 #include "chrome/common/ipc_logging.h"
 #include "chrome/common/ipc_message_utils.h"
 
 namespace IPC {
 //------------------------------------------------------------------------------
 
 Channel::ChannelImpl::State::State(ChannelImpl* channel) : is_pending(false) {
   memset(&context.overlapped, 0, sizeof(context.overlapped));
@@ skipping 17 matching lines @@
       processing_incoming_(false),
       ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) {
   if (!CreatePipe(channel_id, mode)) {
     // The pipe may have been closed already.
     LOG(WARNING) << "Unable to create pipe named \"" << channel_id <<
                     "\" in " << (mode == 0 ? "server" : "client") << " mode.";
   }
 }
 
 void Channel::ChannelImpl::Close() {
+  if (thread_check_.get()) {
+    DCHECK(thread_check_->CalledOnValidThread());
+  }
+
   bool waited = false;
   if (input_state_.is_pending || output_state_.is_pending) {
     CancelIo(pipe_);
     waited = true;
   }
 
   // Closing the handle at this point prevents us from issuing more requests
   // form OnIOCompleted().
   if (pipe_ != INVALID_HANDLE_VALUE) {
     CloseHandle(pipe_);
@@ skipping 11 matching lines @@
   }
 
   while (!output_queue_.empty()) {
     Message* m = output_queue_.front();
     output_queue_.pop();
     delete m;
   }
 }
 
 bool Channel::ChannelImpl::Send(Message* message) {
+  DCHECK(thread_check_->CalledOnValidThread());
   chrome::Counters::ipc_send_counter().Increment();
 #ifdef IPC_MESSAGE_DEBUG_EXTRA
   DLOG(INFO) << "sending message @" << message << " on channel @" << this
              << " with type " << message->type()
              << " (" << output_queue_.size() << " in queue)";
 #endif
 
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging::current()->OnSendMessage(message, L"");
 #endif
@@ skipping 70 matching lines @@
     return false;
   }
 
   output_queue_.push(m.release());
   return true;
 }
 
 bool Channel::ChannelImpl::Connect() {
   DLOG(WARNING) << "Connect called twice";
 
+  if (!thread_check_.get())

[jam, 2009/01/28 03:33:58]

since this is only used in debug builds, it should probably be surrounded by an
ifdef

+    thread_check_.reset(new NonThreadSafe());
+
   if (pipe_ == INVALID_HANDLE_VALUE)
     return false;
 
   MessageLoopForIO::current()->RegisterIOHandler(pipe_, this);
 
   // Check to see if there is a client connected to our pipe...
   if (waiting_connect_)
     ProcessConnection();
 
   if (!input_state_.is_pending) {
     // Complete setup asynchronously. By not setting input_state_.is_pending
     // to true, we indicate to OnIOCompleted that this is the special
     // initialization signal.
     MessageLoopForIO::current()->PostTask(FROM_HERE, factory_.NewRunnableMethod(
         &Channel::ChannelImpl::OnIOCompleted, &input_state_.context, 0, 0));
   }
 
   if (!waiting_connect_)
     ProcessOutgoingMessages(NULL, 0);
   return true;
 }
 
 bool Channel::ChannelImpl::ProcessConnection() {
+  DCHECK(thread_check_->CalledOnValidThread());
   if (input_state_.is_pending)
     input_state_.is_pending = false;
 
   // Do we have a client connected to our pipe?
   if (INVALID_HANDLE_VALUE == pipe_)
     return false;
 
   BOOL ok = ConnectNamedPipe(pipe_, &input_state_.context.overlapped);
 
   DWORD err = GetLastError();
@@ skipping 15 matching lines @@
     NOTREACHED();
     return false;
   }
 
   return true;
 }
 
 bool Channel::ChannelImpl::ProcessIncomingMessages(
     MessageLoopForIO::IOContext* context,
     DWORD bytes_read) {
+  DCHECK(thread_check_->CalledOnValidThread());
   if (input_state_.is_pending) {
     input_state_.is_pending = false;
     DCHECK(context);
 
     if (!context || !bytes_read)
       return false;
   } else {
     // This happens at channel initialization.
     DCHECK(!bytes_read && context == &input_state_.context);
   }
@@ skipping 68 matching lines @@
   }
 
   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) {
     DCHECK(context);
     output_state_.is_pending = false;
     if (!context || bytes_written == 0) {
       DWORD err = GetLastError();
       LOG(ERROR) << "pipe error: " << err;
       return false;
     }
     // Message was sent.
@@ skipping 37 matching lines @@
                 " with type " << m->type();
 #endif
 
   output_state_.is_pending = true;
   return true;
 }
 
 void Channel::ChannelImpl::OnIOCompleted(MessageLoopForIO::IOContext* context,
                             DWORD bytes_transfered, DWORD error) {
   bool ok;
+  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...
@@ skipping 35 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