ipc/ipc_channel_win.cc

   1 // Copyright (c) 2011 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "ipc/ipc_channel_win.h"
   6
   7 #include <windows.h>
   8
   9 #include "base/auto_reset.h"
  10 #include "base/compiler_specific.h"
  11 #include "base/logging.h"
  12 #include "base/threading/non_thread_safe.h"
  13 #include "base/utf_string_conversions.h"
  14 #include "base/win/scoped_handle.h"
  15 #include "ipc/ipc_logging.h"
  16 #include "ipc/ipc_message_utils.h"
  17
  18 namespace IPC {
  19
  20 Channel::ChannelImpl::State::State(ChannelImpl* channel) : is_pending(false) {
  21   memset(&context.overlapped, 0, sizeof(context.overlapped));
  22   context.handler = channel;
  23 }
  24
  25 Channel::ChannelImpl::State::~State() {
  26   COMPILE_ASSERT(!offsetof(Channel::ChannelImpl::State, context),
  27                  starts_with_io_context);
  28 }
  29
  30 Channel::ChannelImpl::ChannelImpl(const IPC::ChannelHandle &channel_handle,
  31                                   Mode mode, Listener* listener)
  32     : ALLOW_THIS_IN_INITIALIZER_LIST(input_state_(this)),
  33       ALLOW_THIS_IN_INITIALIZER_LIST(output_state_(this)),
  34       pipe_(INVALID_HANDLE_VALUE),
  35       listener_(listener),
  36       waiting_connect_(mode & MODE_SERVER_FLAG),
  37       processing_incoming_(false),
  38       ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) {
  39   CreatePipe(channel_handle, mode);
  40 }
  41
  42 Channel::ChannelImpl::~ChannelImpl() {
  43   Close();
  44 }
  45
  46 void Channel::ChannelImpl::Close() {
  47   if (thread_check_.get()) {
  48     DCHECK(thread_check_->CalledOnValidThread());
  49   }
  50
  51   if (input_state_.is_pending || output_state_.is_pending)
  52     CancelIo(pipe_);
  53
  54   // Closing the handle at this point prevents us from issuing more requests
  55   // form OnIOCompleted().
  56   if (pipe_ != INVALID_HANDLE_VALUE) {
  57     CloseHandle(pipe_);
  58     pipe_ = INVALID_HANDLE_VALUE;
  59   }
  60
  61   // Make sure all IO has completed.
  62   base::Time start = base::Time::Now();
  63   while (input_state_.is_pending || output_state_.is_pending) {
  64     MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this);
  65   }
  66
  67   while (!output_queue_.empty()) {
  68     Message* m = output_queue_.front();
  69     output_queue_.pop();
  70     delete m;
  71   }
  72 }
  73
  74 bool Channel::ChannelImpl::Send(Message* message) {
  75   DCHECK(thread_check_->CalledOnValidThread());
  76   DVLOG(2) << "sending message @" << message << " on channel @" << this
  77            << " with type " << message->type()
  78            << " (" << output_queue_.size() << " in queue)";
  79
  80 #ifdef IPC_MESSAGE_LOG_ENABLED
  81   Logging::GetInstance()->OnSendMessage(message, "");
  82 #endif
  83
  84   output_queue_.push(message);
  85   // ensure waiting to write
  86   if (!waiting_connect_) {
  87     if (!output_state_.is_pending) {
  88       if (!ProcessOutgoingMessages(NULL, 0))
  89         return false;
  90     }
  91   }
  92
  93   return true;
  94 }
  95
  96 const std::wstring Channel::ChannelImpl::PipeName(
  97     const std::string& channel_id) const {
  98   std::string name("\\\\.\\pipe\\chrome.");
  99   return ASCIIToWide(name.append(channel_id));
 100 }
 101
 102 bool Channel::ChannelImpl::CreatePipe(const IPC::ChannelHandle &channel_handle,
 103                                       Mode mode) {
 104   DCHECK_EQ(INVALID_HANDLE_VALUE, pipe_);
 105   const std::wstring pipe_name = PipeName(channel_handle.name);
 106   if (mode & MODE_SERVER_FLAG) {
 107     pipe_ = CreateNamedPipeW(pipe_name.c_str(),
 108                              PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED |
 109                               FILE_FLAG_FIRST_PIPE_INSTANCE,
 110                              PIPE_TYPE_BYTE | PIPE_READMODE_BYTE,
 111                              1,
 112                              Channel::kReadBufferSize,
 113                              Channel::kReadBufferSize,
 114                              5000,
 115                              NULL);
 116   } else if (mode & MODE_CLIENT_FLAG) {
 117     pipe_ = CreateFileW(pipe_name.c_str(),
 118                         GENERIC_READ | GENERIC_WRITE,
 119                         0,
 120                         NULL,
 121                         OPEN_EXISTING,
 122                         SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION |
 123                             FILE_FLAG_OVERLAPPED,
 124                         NULL);
 125   } else {
 126     NOTREACHED();
 127   }
 128   if (pipe_ == INVALID_HANDLE_VALUE) {
 129     // If this process is being closed, the pipe may be gone already.
 130     LOG(WARNING) << "Unable to create pipe \"" << pipe_name <<
 131                     "\" in " << (mode == 0 ? "server" : "client")
 132                     << " mode. Error :" << GetLastError();
 133     return false;
 134   }
 135
 136   // Create the Hello message to be sent when Connect is called
 137   scoped_ptr<Message> m(new Message(MSG_ROUTING_NONE,
 138                                     HELLO_MESSAGE_TYPE,
 139                                     IPC::Message::PRIORITY_NORMAL));
 140   if (!m->WriteInt(GetCurrentProcessId())) {
 141     CloseHandle(pipe_);
 142     pipe_ = INVALID_HANDLE_VALUE;
 143     return false;
 144   }
 145
 146   output_queue_.push(m.release());
 147   return true;
 148 }
 149
 150 bool Channel::ChannelImpl::Connect() {
 151   DLOG_IF(WARNING, thread_check_.get()) << "Connect called more than once";
 152
 153   if (!thread_check_.get())
 154     thread_check_.reset(new base::NonThreadSafe());
 155
 156   if (pipe_ == INVALID_HANDLE_VALUE)
 157     return false;
 158
 159   MessageLoopForIO::current()->RegisterIOHandler(pipe_, this);
 160
 161   // Check to see if there is a client connected to our pipe...
 162   if (waiting_connect_)
 163     ProcessConnection();
 164
 165   if (!input_state_.is_pending) {
 166     // Complete setup asynchronously. By not setting input_state_.is_pending
 167     // to true, we indicate to OnIOCompleted that this is the special
 168     // initialization signal.
 169     MessageLoopForIO::current()->PostTask(FROM_HERE, factory_.NewRunnableMethod(
 170         &Channel::ChannelImpl::OnIOCompleted, &input_state_.context, 0, 0));
 171   }
 172
 173   if (!waiting_connect_)
 174     ProcessOutgoingMessages(NULL, 0);
 175   return true;
 176 }
 177
 178 bool Channel::ChannelImpl::ProcessConnection() {
 179   DCHECK(thread_check_->CalledOnValidThread());
 180   if (input_state_.is_pending)
 181     input_state_.is_pending = false;
 182
 183   // Do we have a client connected to our pipe?
 184   if (INVALID_HANDLE_VALUE == pipe_)
 185     return false;
 186
 187   BOOL ok = ConnectNamedPipe(pipe_, &input_state_.context.overlapped);
 188
 189   DWORD err = GetLastError();
 190   if (ok) {
 191     // Uhm, the API documentation says that this function should never
 192     // return success when used in overlapped mode.
 193     NOTREACHED();
 194     return false;
 195   }
 196
 197   switch (err) {
 198   case ERROR_IO_PENDING:
 199     input_state_.is_pending = true;
 200     break;
 201   case ERROR_PIPE_CONNECTED:
 202     waiting_connect_ = false;
 203     break;
 204   case ERROR_NO_DATA:
 205     // The pipe is being closed.
 206     return false;
 207   default:
 208     NOTREACHED();
 209     return false;
 210   }
 211
 212   return true;
 213 }
 214
 215 bool Channel::ChannelImpl::ProcessIncomingMessages(
 216     MessageLoopForIO::IOContext* context,
 217     DWORD bytes_read) {
 218   DCHECK(thread_check_->CalledOnValidThread());
 219   if (input_state_.is_pending) {
 220     input_state_.is_pending = false;
 221     DCHECK(context);
 222
 223     if (!context || !bytes_read)
 224       return false;
 225   } else {
 226     // This happens at channel initialization.
 227     DCHECK(!bytes_read && context == &input_state_.context);
 228   }
 229
 230   for (;;) {
 231     if (bytes_read == 0) {
 232       if (INVALID_HANDLE_VALUE == pipe_)
 233         return false;
 234
 235       // Read from pipe...
 236       BOOL ok = ReadFile(pipe_,
 237                          input_buf_,
 238                          Channel::kReadBufferSize,
 239                          &bytes_read,
 240                          &input_state_.context.overlapped);
 241       if (!ok) {
 242         DWORD err = GetLastError();
 243         if (err == ERROR_IO_PENDING) {
 244           input_state_.is_pending = true;
 245           return true;
 246         }
 247         LOG(ERROR) << "pipe error: " << err;
 248         return false;
 249       }
 250       input_state_.is_pending = true;
 251       return true;
 252     }
 253     DCHECK(bytes_read);
 254
 255     // Process messages from input buffer.
 256
 257     const char* p, *end;
 258     if (input_overflow_buf_.empty()) {
 259       p = input_buf_;
 260       end = p + bytes_read;
 261     } else {
 262       if (input_overflow_buf_.size() > (kMaximumMessageSize - bytes_read)) {
 263         input_overflow_buf_.clear();
 264         LOG(ERROR) << "IPC message is too big";
 265         return false;
 266       }
 267       input_overflow_buf_.append(input_buf_, bytes_read);
 268       p = input_overflow_buf_.data();
 269       end = p + input_overflow_buf_.size();
 270     }
 271
 272     while (p < end) {
 273       const char* message_tail = Message::FindNext(p, end);
 274       if (message_tail) {
 275         int len = static_cast<int>(message_tail - p);
 276         const Message m(p, len);
 277         DVLOG(2) << "received message on channel @" << this
 278                  << " with type " << m.type();
 279         if (m.routing_id() == MSG_ROUTING_NONE &&
 280             m.type() == HELLO_MESSAGE_TYPE) {
 281           // The Hello message contains only the process id.
 282           listener_->OnChannelConnected(MessageIterator(m).NextInt());
 283         } else {
 284           listener_->OnMessageReceived(m);
 285         }
 286         p = message_tail;
 287       } else {
 288         // Last message is partial.
 289         break;
 290       }
 291     }
 292     input_overflow_buf_.assign(p, end - p);
 293
 294     bytes_read = 0;  // Get more data.
 295   }
 296
 297   return true;
 298 }
 299
 300 bool Channel::ChannelImpl::ProcessOutgoingMessages(
 301     MessageLoopForIO::IOContext* context,
 302     DWORD bytes_written) {
 303   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
 304                               // no connection?
 305   DCHECK(thread_check_->CalledOnValidThread());
 306
 307   if (output_state_.is_pending) {
 308     DCHECK(context);
 309     output_state_.is_pending = false;
 310     if (!context || bytes_written == 0) {
 311       DWORD err = GetLastError();
 312       LOG(ERROR) << "pipe error: " << err;
 313       return false;
 314     }
 315     // Message was sent.
 316     DCHECK(!output_queue_.empty());
 317     Message* m = output_queue_.front();
 318     output_queue_.pop();
 319     delete m;
 320   }
 321
 322   if (output_queue_.empty())
 323     return true;
 324
 325   if (INVALID_HANDLE_VALUE == pipe_)
 326     return false;
 327
 328   // Write to pipe...
 329   Message* m = output_queue_.front();
 330   DCHECK(m->size() <= INT_MAX);
 331   BOOL ok = WriteFile(pipe_,
 332                       m->data(),
 333                       static_cast<int>(m->size()),
 334                       &bytes_written,
 335                       &output_state_.context.overlapped);
 336   if (!ok) {
 337     DWORD err = GetLastError();
 338     if (err == ERROR_IO_PENDING) {
 339       output_state_.is_pending = true;
 340
 341       DVLOG(2) << "sent pending message @" << m << " on channel @" << this
 342                << " with type " << m->type();
 343
 344       return true;
 345     }
 346     LOG(ERROR) << "pipe error: " << err;
 347     return false;
 348   }
 349
 350   DVLOG(2) << "sent message @" << m << " on channel @" << this
 351            << " with type " << m->type();
 352
 353   output_state_.is_pending = true;
 354   return true;
 355 }
 356
 357 void Channel::ChannelImpl::OnIOCompleted(MessageLoopForIO::IOContext* context,
 358                             DWORD bytes_transfered, DWORD error) {
 359   bool ok;
 360   DCHECK(thread_check_->CalledOnValidThread());
 361   if (context == &input_state_.context) {
 362     if (waiting_connect_) {
 363       if (!ProcessConnection())
 364         return;
 365       // We may have some messages queued up to send...
 366       if (!output_queue_.empty() && !output_state_.is_pending)
 367         ProcessOutgoingMessages(NULL, 0);
 368       if (input_state_.is_pending)
 369         return;
 370       // else, fall-through and look for incoming messages...
 371     }
 372     // we don't support recursion through OnMessageReceived yet!
 373     DCHECK(!processing_incoming_);
 374     AutoReset<bool> auto_reset_processing_incoming(&processing_incoming_, true);
 375     ok = ProcessIncomingMessages(context, bytes_transfered);
 376   } else {
 377     DCHECK(context == &output_state_.context);
 378     ok = ProcessOutgoingMessages(context, bytes_transfered);
 379   }
 380   if (!ok && INVALID_HANDLE_VALUE != pipe_) {
 381     // We don't want to re-enter Close().
 382     Close();
 383     listener_->OnChannelError();
 384   }
 385 }
 386
 387 //------------------------------------------------------------------------------
 388 // Channel's methods simply call through to ChannelImpl.
 389 Channel::Channel(const IPC::ChannelHandle &channel_handle, Mode mode,
 390                  Listener* listener)
 391     : channel_impl_(new ChannelImpl(channel_handle, mode, listener)) {
 392 }
 393
 394 Channel::~Channel() {
 395   delete channel_impl_;
 396 }
 397
 398 bool Channel::Connect() {
 399   return channel_impl_->Connect();
 400 }
 401
 402 void Channel::Close() {
 403   channel_impl_->Close();
 404 }
 405
 406 void Channel::set_listener(Listener* listener) {
 407   channel_impl_->set_listener(listener);
 408 }
 409
 410 bool Channel::Send(Message* message) {
 411   return channel_impl_->Send(message);
 412 }
 413
 414 }  // namespace IPC