ipc/ipc_channel_win.cc

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