ipc/ipc_channel_win.cc

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