1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/ipc/chromium/src/chrome/common/ipc_channel_win.cc Wed Dec 31 06:09:35 2014 +0100
1.3 @@ -0,0 +1,510 @@
1.4 +// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
1.5 +// Use of this source code is governed by a BSD-style license that can be
1.6 +// found in the LICENSE file.
1.7 +
1.8 +#include "chrome/common/ipc_channel_win.h"
1.9 +
1.10 +#include <windows.h>
1.11 +#include <sstream>
1.12 +
1.13 +#include "base/compiler_specific.h"
1.14 +#include "base/logging.h"
1.15 +#include "base/non_thread_safe.h"
1.16 +#include "base/stats_counters.h"
1.17 +#include "base/win_util.h"
1.18 +#include "chrome/common/ipc_logging.h"
1.19 +#include "chrome/common/ipc_message_utils.h"
1.20 +#include "mozilla/ipc/ProtocolUtils.h"
1.21 +
1.22 +namespace IPC {
1.23 +//------------------------------------------------------------------------------
1.24 +
1.25 +Channel::ChannelImpl::State::State(ChannelImpl* channel) : is_pending(false) {
1.26 + memset(&context.overlapped, 0, sizeof(context.overlapped));
1.27 + context.handler = channel;
1.28 +}
1.29 +
1.30 +Channel::ChannelImpl::State::~State() {
1.31 + COMPILE_ASSERT(!offsetof(Channel::ChannelImpl::State, context),
1.32 + starts_with_io_context);
1.33 +}
1.34 +
1.35 +//------------------------------------------------------------------------------
1.36 +
1.37 +Channel::ChannelImpl::ChannelImpl(const std::wstring& channel_id, Mode mode,
1.38 + Listener* listener)
1.39 + : ALLOW_THIS_IN_INITIALIZER_LIST(input_state_(this)),
1.40 + ALLOW_THIS_IN_INITIALIZER_LIST(output_state_(this)),
1.41 + ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) {
1.42 + Init(mode, listener);
1.43 +
1.44 + if (!CreatePipe(channel_id, mode)) {
1.45 + // The pipe may have been closed already.
1.46 + CHROMIUM_LOG(WARNING) << "Unable to create pipe named \"" << channel_id <<
1.47 + "\" in " << (mode == 0 ? "server" : "client") << " mode.";
1.48 + }
1.49 +}
1.50 +
1.51 +Channel::ChannelImpl::ChannelImpl(const std::wstring& channel_id,
1.52 + HANDLE server_pipe,
1.53 + Mode mode, Listener* listener)
1.54 + : ALLOW_THIS_IN_INITIALIZER_LIST(input_state_(this)),
1.55 + ALLOW_THIS_IN_INITIALIZER_LIST(output_state_(this)),
1.56 + ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) {
1.57 + Init(mode, listener);
1.58 +
1.59 + if (mode == MODE_SERVER) {
1.60 + // Use the existing handle that was dup'd to us
1.61 + pipe_ = server_pipe;
1.62 + EnqueueHelloMessage();
1.63 + } else {
1.64 + // Take the normal init path to connect to the server pipe
1.65 + CreatePipe(channel_id, mode);
1.66 + }
1.67 +}
1.68 +
1.69 +void Channel::ChannelImpl::Init(Mode mode, Listener* listener) {
1.70 + pipe_ = INVALID_HANDLE_VALUE;
1.71 + listener_ = listener;
1.72 + waiting_connect_ = (mode == MODE_SERVER);
1.73 + processing_incoming_ = false;
1.74 + closed_ = false;
1.75 + output_queue_length_ = 0;
1.76 +}
1.77 +
1.78 +void Channel::ChannelImpl::OutputQueuePush(Message* msg)
1.79 +{
1.80 + output_queue_.push(msg);
1.81 + output_queue_length_++;
1.82 +}
1.83 +
1.84 +void Channel::ChannelImpl::OutputQueuePop()
1.85 +{
1.86 + output_queue_.pop();
1.87 + output_queue_length_--;
1.88 +}
1.89 +
1.90 +HANDLE Channel::ChannelImpl::GetServerPipeHandle() const {
1.91 + return pipe_;
1.92 +}
1.93 +
1.94 +void Channel::ChannelImpl::Close() {
1.95 + if (thread_check_.get()) {
1.96 + DCHECK(thread_check_->CalledOnValidThread());
1.97 + }
1.98 +
1.99 + bool waited = false;
1.100 + if (input_state_.is_pending || output_state_.is_pending) {
1.101 + CancelIo(pipe_);
1.102 + waited = true;
1.103 + }
1.104 +
1.105 + // Closing the handle at this point prevents us from issuing more requests
1.106 + // form OnIOCompleted().
1.107 + if (pipe_ != INVALID_HANDLE_VALUE) {
1.108 + CloseHandle(pipe_);
1.109 + pipe_ = INVALID_HANDLE_VALUE;
1.110 + }
1.111 +
1.112 + while (input_state_.is_pending || output_state_.is_pending) {
1.113 + MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this);
1.114 + }
1.115 +
1.116 + while (!output_queue_.empty()) {
1.117 + Message* m = output_queue_.front();
1.118 + OutputQueuePop();
1.119 + delete m;
1.120 + }
1.121 +
1.122 + if (thread_check_.get())
1.123 + thread_check_.reset();
1.124 +
1.125 + closed_ = true;
1.126 +}
1.127 +
1.128 +bool Channel::ChannelImpl::Send(Message* message) {
1.129 + if (thread_check_.get()) {
1.130 + DCHECK(thread_check_->CalledOnValidThread());
1.131 + }
1.132 +#ifdef IPC_MESSAGE_DEBUG_EXTRA
1.133 + DLOG(INFO) << "sending message @" << message << " on channel @" << this
1.134 + << " with type " << message->type()
1.135 + << " (" << output_queue_.size() << " in queue)";
1.136 +#endif
1.137 +
1.138 +#ifdef IPC_MESSAGE_LOG_ENABLED
1.139 + Logging::current()->OnSendMessage(message, L"");
1.140 +#endif
1.141 +
1.142 + if (closed_) {
1.143 + if (mozilla::ipc::LoggingEnabled()) {
1.144 + fprintf(stderr, "Can't send message %s, because this channel is closed.\n",
1.145 + message->name());
1.146 + }
1.147 + delete message;
1.148 + return false;
1.149 + }
1.150 +
1.151 + OutputQueuePush(message);
1.152 + // ensure waiting to write
1.153 + if (!waiting_connect_) {
1.154 + if (!output_state_.is_pending) {
1.155 + if (!ProcessOutgoingMessages(NULL, 0))
1.156 + return false;
1.157 + }
1.158 + }
1.159 +
1.160 + return true;
1.161 +}
1.162 +
1.163 +const std::wstring Channel::ChannelImpl::PipeName(
1.164 + const std::wstring& channel_id) const {
1.165 + std::wostringstream ss;
1.166 + // XXX(darin): get application name from somewhere else
1.167 + ss << L"\\\\.\\pipe\\chrome." << channel_id;
1.168 + return ss.str();
1.169 +}
1.170 +
1.171 +bool Channel::ChannelImpl::CreatePipe(const std::wstring& channel_id,
1.172 + Mode mode) {
1.173 + DCHECK(pipe_ == INVALID_HANDLE_VALUE);
1.174 + const std::wstring pipe_name = PipeName(channel_id);
1.175 + if (mode == MODE_SERVER) {
1.176 + pipe_ = CreateNamedPipeW(pipe_name.c_str(),
1.177 + PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED |
1.178 + FILE_FLAG_FIRST_PIPE_INSTANCE,
1.179 + PIPE_TYPE_BYTE | PIPE_READMODE_BYTE,
1.180 + 1, // number of pipe instances
1.181 + // output buffer size (XXX tune)
1.182 + Channel::kReadBufferSize,
1.183 + // input buffer size (XXX tune)
1.184 + Channel::kReadBufferSize,
1.185 + 5000, // timeout in milliseconds (XXX tune)
1.186 + NULL);
1.187 + } else {
1.188 + pipe_ = CreateFileW(pipe_name.c_str(),
1.189 + GENERIC_READ | GENERIC_WRITE,
1.190 + 0,
1.191 + NULL,
1.192 + OPEN_EXISTING,
1.193 + SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION |
1.194 + FILE_FLAG_OVERLAPPED,
1.195 + NULL);
1.196 + }
1.197 + if (pipe_ == INVALID_HANDLE_VALUE) {
1.198 + // If this process is being closed, the pipe may be gone already.
1.199 + CHROMIUM_LOG(WARNING) << "failed to create pipe: " << GetLastError();
1.200 + return false;
1.201 + }
1.202 +
1.203 + // Create the Hello message to be sent when Connect is called
1.204 + return EnqueueHelloMessage();
1.205 +}
1.206 +
1.207 +bool Channel::ChannelImpl::EnqueueHelloMessage() {
1.208 + scoped_ptr<Message> m(new Message(MSG_ROUTING_NONE,
1.209 + HELLO_MESSAGE_TYPE,
1.210 + IPC::Message::PRIORITY_NORMAL));
1.211 + if (!m->WriteInt(GetCurrentProcessId())) {
1.212 + CloseHandle(pipe_);
1.213 + pipe_ = INVALID_HANDLE_VALUE;
1.214 + return false;
1.215 + }
1.216 +
1.217 + OutputQueuePush(m.release());
1.218 + return true;
1.219 +}
1.220 +
1.221 +bool Channel::ChannelImpl::Connect() {
1.222 + if (!thread_check_.get())
1.223 + thread_check_.reset(new NonThreadSafe());
1.224 +
1.225 + if (pipe_ == INVALID_HANDLE_VALUE)
1.226 + return false;
1.227 +
1.228 + MessageLoopForIO::current()->RegisterIOHandler(pipe_, this);
1.229 +
1.230 + // Check to see if there is a client connected to our pipe...
1.231 + if (waiting_connect_)
1.232 + ProcessConnection();
1.233 +
1.234 + if (!input_state_.is_pending) {
1.235 + // Complete setup asynchronously. By not setting input_state_.is_pending
1.236 + // to true, we indicate to OnIOCompleted that this is the special
1.237 + // initialization signal.
1.238 + MessageLoopForIO::current()->PostTask(FROM_HERE, factory_.NewRunnableMethod(
1.239 + &Channel::ChannelImpl::OnIOCompleted, &input_state_.context, 0, 0));
1.240 + }
1.241 +
1.242 + if (!waiting_connect_)
1.243 + ProcessOutgoingMessages(NULL, 0);
1.244 + return true;
1.245 +}
1.246 +
1.247 +bool Channel::ChannelImpl::ProcessConnection() {
1.248 + DCHECK(thread_check_->CalledOnValidThread());
1.249 + if (input_state_.is_pending)
1.250 + input_state_.is_pending = false;
1.251 +
1.252 + // Do we have a client connected to our pipe?
1.253 + if (INVALID_HANDLE_VALUE == pipe_)
1.254 + return false;
1.255 +
1.256 + BOOL ok = ConnectNamedPipe(pipe_, &input_state_.context.overlapped);
1.257 +
1.258 + DWORD err = GetLastError();
1.259 + if (ok) {
1.260 + // Uhm, the API documentation says that this function should never
1.261 + // return success when used in overlapped mode.
1.262 + NOTREACHED();
1.263 + return false;
1.264 + }
1.265 +
1.266 + switch (err) {
1.267 + case ERROR_IO_PENDING:
1.268 + input_state_.is_pending = true;
1.269 + break;
1.270 + case ERROR_PIPE_CONNECTED:
1.271 + waiting_connect_ = false;
1.272 + break;
1.273 + default:
1.274 + NOTREACHED();
1.275 + return false;
1.276 + }
1.277 +
1.278 + return true;
1.279 +}
1.280 +
1.281 +bool Channel::ChannelImpl::ProcessIncomingMessages(
1.282 + MessageLoopForIO::IOContext* context,
1.283 + DWORD bytes_read) {
1.284 + DCHECK(thread_check_->CalledOnValidThread());
1.285 + if (input_state_.is_pending) {
1.286 + input_state_.is_pending = false;
1.287 + DCHECK(context);
1.288 +
1.289 + if (!context || !bytes_read)
1.290 + return false;
1.291 + } else {
1.292 + // This happens at channel initialization.
1.293 + DCHECK(!bytes_read && context == &input_state_.context);
1.294 + }
1.295 +
1.296 + for (;;) {
1.297 + if (bytes_read == 0) {
1.298 + if (INVALID_HANDLE_VALUE == pipe_)
1.299 + return false;
1.300 +
1.301 + // Read from pipe...
1.302 + BOOL ok = ReadFile(pipe_,
1.303 + input_buf_,
1.304 + Channel::kReadBufferSize,
1.305 + &bytes_read,
1.306 + &input_state_.context.overlapped);
1.307 + if (!ok) {
1.308 + DWORD err = GetLastError();
1.309 + if (err == ERROR_IO_PENDING) {
1.310 + input_state_.is_pending = true;
1.311 + return true;
1.312 + }
1.313 + CHROMIUM_LOG(ERROR) << "pipe error: " << err;
1.314 + return false;
1.315 + }
1.316 + input_state_.is_pending = true;
1.317 + return true;
1.318 + }
1.319 + DCHECK(bytes_read);
1.320 +
1.321 + // Process messages from input buffer.
1.322 +
1.323 + const char* p, *end;
1.324 + if (input_overflow_buf_.empty()) {
1.325 + p = input_buf_;
1.326 + end = p + bytes_read;
1.327 + } else {
1.328 + if (input_overflow_buf_.size() > (kMaximumMessageSize - bytes_read)) {
1.329 + input_overflow_buf_.clear();
1.330 + CHROMIUM_LOG(ERROR) << "IPC message is too big";
1.331 + return false;
1.332 + }
1.333 + input_overflow_buf_.append(input_buf_, bytes_read);
1.334 + p = input_overflow_buf_.data();
1.335 + end = p + input_overflow_buf_.size();
1.336 + }
1.337 +
1.338 + while (p < end) {
1.339 + const char* message_tail = Message::FindNext(p, end);
1.340 + if (message_tail) {
1.341 + int len = static_cast<int>(message_tail - p);
1.342 + const Message m(p, len);
1.343 +#ifdef IPC_MESSAGE_DEBUG_EXTRA
1.344 + DLOG(INFO) << "received message on channel @" << this <<
1.345 + " with type " << m.type();
1.346 +#endif
1.347 + if (m.routing_id() == MSG_ROUTING_NONE &&
1.348 + m.type() == HELLO_MESSAGE_TYPE) {
1.349 + // The Hello message contains only the process id.
1.350 + listener_->OnChannelConnected(MessageIterator(m).NextInt());
1.351 + } else {
1.352 + listener_->OnMessageReceived(m);
1.353 + }
1.354 + p = message_tail;
1.355 + } else {
1.356 + // Last message is partial.
1.357 + break;
1.358 + }
1.359 + }
1.360 + input_overflow_buf_.assign(p, end - p);
1.361 +
1.362 + bytes_read = 0; // Get more data.
1.363 + }
1.364 +
1.365 + return true;
1.366 +}
1.367 +
1.368 +bool Channel::ChannelImpl::ProcessOutgoingMessages(
1.369 + MessageLoopForIO::IOContext* context,
1.370 + DWORD bytes_written) {
1.371 + DCHECK(!waiting_connect_); // Why are we trying to send messages if there's
1.372 + // no connection?
1.373 + DCHECK(thread_check_->CalledOnValidThread());
1.374 +
1.375 + if (output_state_.is_pending) {
1.376 + DCHECK(context);
1.377 + output_state_.is_pending = false;
1.378 + if (!context || bytes_written == 0) {
1.379 + DWORD err = GetLastError();
1.380 + CHROMIUM_LOG(ERROR) << "pipe error: " << err;
1.381 + return false;
1.382 + }
1.383 + // Message was sent.
1.384 + DCHECK(!output_queue_.empty());
1.385 + Message* m = output_queue_.front();
1.386 + OutputQueuePop();
1.387 + delete m;
1.388 + }
1.389 +
1.390 + if (output_queue_.empty())
1.391 + return true;
1.392 +
1.393 + if (INVALID_HANDLE_VALUE == pipe_)
1.394 + return false;
1.395 +
1.396 + // Write to pipe...
1.397 + Message* m = output_queue_.front();
1.398 + BOOL ok = WriteFile(pipe_,
1.399 + m->data(),
1.400 + m->size(),
1.401 + &bytes_written,
1.402 + &output_state_.context.overlapped);
1.403 + if (!ok) {
1.404 + DWORD err = GetLastError();
1.405 + if (err == ERROR_IO_PENDING) {
1.406 + output_state_.is_pending = true;
1.407 +
1.408 +#ifdef IPC_MESSAGE_DEBUG_EXTRA
1.409 + DLOG(INFO) << "sent pending message @" << m << " on channel @" <<
1.410 + this << " with type " << m->type();
1.411 +#endif
1.412 +
1.413 + return true;
1.414 + }
1.415 + CHROMIUM_LOG(ERROR) << "pipe error: " << err;
1.416 + return false;
1.417 + }
1.418 +
1.419 +#ifdef IPC_MESSAGE_DEBUG_EXTRA
1.420 + DLOG(INFO) << "sent message @" << m << " on channel @" << this <<
1.421 + " with type " << m->type();
1.422 +#endif
1.423 +
1.424 + output_state_.is_pending = true;
1.425 + return true;
1.426 +}
1.427 +
1.428 +void Channel::ChannelImpl::OnIOCompleted(MessageLoopForIO::IOContext* context,
1.429 + DWORD bytes_transfered, DWORD error) {
1.430 + bool ok;
1.431 + DCHECK(thread_check_->CalledOnValidThread());
1.432 + if (context == &input_state_.context) {
1.433 + if (waiting_connect_) {
1.434 + if (!ProcessConnection())
1.435 + return;
1.436 + // We may have some messages queued up to send...
1.437 + if (!output_queue_.empty() && !output_state_.is_pending)
1.438 + ProcessOutgoingMessages(NULL, 0);
1.439 + if (input_state_.is_pending)
1.440 + return;
1.441 + // else, fall-through and look for incoming messages...
1.442 + }
1.443 + // we don't support recursion through OnMessageReceived yet!
1.444 + DCHECK(!processing_incoming_);
1.445 + processing_incoming_ = true;
1.446 + ok = ProcessIncomingMessages(context, bytes_transfered);
1.447 + processing_incoming_ = false;
1.448 + } else {
1.449 + DCHECK(context == &output_state_.context);
1.450 + ok = ProcessOutgoingMessages(context, bytes_transfered);
1.451 + }
1.452 + if (!ok && INVALID_HANDLE_VALUE != pipe_) {
1.453 + // We don't want to re-enter Close().
1.454 + Close();
1.455 + listener_->OnChannelError();
1.456 + }
1.457 +}
1.458 +
1.459 +bool Channel::ChannelImpl::Unsound_IsClosed() const
1.460 +{
1.461 + return closed_;
1.462 +}
1.463 +
1.464 +uint32_t Channel::ChannelImpl::Unsound_NumQueuedMessages() const
1.465 +{
1.466 + return output_queue_length_;
1.467 +}
1.468 +
1.469 +//------------------------------------------------------------------------------
1.470 +// Channel's methods simply call through to ChannelImpl.
1.471 +Channel::Channel(const std::wstring& channel_id, Mode mode,
1.472 + Listener* listener)
1.473 + : channel_impl_(new ChannelImpl(channel_id, mode, listener)) {
1.474 +}
1.475 +
1.476 +Channel::Channel(const std::wstring& channel_id, void* server_pipe,
1.477 + Mode mode, Listener* listener)
1.478 + : channel_impl_(new ChannelImpl(channel_id, server_pipe, mode, listener)) {
1.479 +}
1.480 +
1.481 +Channel::~Channel() {
1.482 + delete channel_impl_;
1.483 +}
1.484 +
1.485 +bool Channel::Connect() {
1.486 + return channel_impl_->Connect();
1.487 +}
1.488 +
1.489 +void Channel::Close() {
1.490 + channel_impl_->Close();
1.491 +}
1.492 +
1.493 +void* Channel::GetServerPipeHandle() const {
1.494 + return channel_impl_->GetServerPipeHandle();
1.495 +}
1.496 +
1.497 +Channel::Listener* Channel::set_listener(Listener* listener) {
1.498 + return channel_impl_->set_listener(listener);
1.499 +}
1.500 +
1.501 +bool Channel::Send(Message* message) {
1.502 + return channel_impl_->Send(message);
1.503 +}
1.504 +
1.505 +bool Channel::Unsound_IsClosed() const {
1.506 + return channel_impl_->Unsound_IsClosed();
1.507 +}
1.508 +
1.509 +uint32_t Channel::Unsound_NumQueuedMessages() const {
1.510 + return channel_impl_->Unsound_NumQueuedMessages();
1.511 +}
1.512 +
1.513 +} // namespace IPC
