ipc/mojo/ipc_channel_mojo.cc

   1 // Copyright 2014 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/mojo/ipc_channel_mojo.h"
   6
   7 #include "base/bind.h"
   8 #include "base/bind_helpers.h"
   9 #include "base/lazy_instance.h"
  10 #include "ipc/ipc_listener.h"
  11 #include "mojo/embedder/embedder.h"
  12
  13 #if defined(OS_POSIX) && !defined(OS_NACL)
  14 #include "ipc/file_descriptor_set_posix.h"
  15 #endif
  16
  17 namespace IPC {
  18
  19 namespace {
  20
  21 // IPC::Listener for bootstrap channels.
  22 // It should never receive any message.
  23 class NullListener : public Listener {
  24  public:
  25   virtual bool OnMessageReceived(const Message&) OVERRIDE {
  26     NOTREACHED();
  27     return false;
  28   }
  29
  30   virtual void OnChannelConnected(int32 peer_pid) OVERRIDE {
  31     NOTREACHED();
  32   }
  33
  34   virtual void OnChannelError() OVERRIDE {
  35     NOTREACHED();
  36   }
  37
  38   virtual void OnBadMessageReceived(const Message& message) OVERRIDE {
  39     NOTREACHED();
  40   }
  41 };
  42
  43 base::LazyInstance<NullListener> g_null_listener = LAZY_INSTANCE_INITIALIZER;
  44
  45 class MojoChannelFactory : public ChannelFactory {
  46  public:
  47   MojoChannelFactory(
  48       ChannelHandle channel_handle,
  49       Channel::Mode mode,
  50       scoped_refptr<base::TaskRunner> io_thread_task_runner)
  51       : channel_handle_(channel_handle),
  52         mode_(mode),
  53         io_thread_task_runner_(io_thread_task_runner) {
  54   }
  55
  56   virtual std::string GetName() const OVERRIDE {
  57     return channel_handle_.name;
  58   }
  59
  60   virtual scoped_ptr<Channel> BuildChannel(Listener* listener) OVERRIDE {
  61     return ChannelMojo::Create(
  62         channel_handle_,
  63         mode_,
  64         listener,
  65         io_thread_task_runner_).PassAs<Channel>();
  66   }
  67
  68  private:
  69   ChannelHandle channel_handle_;
  70   Channel::Mode mode_;
  71   scoped_refptr<base::TaskRunner> io_thread_task_runner_;
  72 };
  73
  74 mojo::embedder::PlatformHandle ToPlatformHandle(
  75     const ChannelHandle& handle) {
  76 #if defined(OS_POSIX) && !defined(OS_NACL)
  77   return mojo::embedder::PlatformHandle(handle.socket.fd);
  78 #elif defined(OS_WIN)
  79   return mojo::embedder::PlatformHandle(handle.pipe.handle);
  80 #else
  81 #error "Unsupported Platform!"
  82 #endif
  83 }
  84
  85 //------------------------------------------------------------------------------
  86
  87 // TODO(morrita): This should be built using higher-level Mojo construct
  88 // for clarity and extensibility.
  89 class HelloMessage {
  90  public:
  91   static Pickle CreateRequest(int32 pid) {
  92     Pickle request;
  93     request.WriteString(kHelloRequestMagic);
  94     request.WriteInt(pid);
  95     return request;
  96   }
  97
  98   static bool ReadRequest(Pickle& pickle, int32* pid) {
  99     PickleIterator iter(pickle);
 100     std::string hello;
 101     if (!iter.ReadString(&hello)) {
 102       DLOG(WARNING) << "Failed to Read magic string.";
 103       return false;
 104     }
 105
 106     if (hello != kHelloRequestMagic) {
 107       DLOG(WARNING) << "Magic mismatch:" << hello;
 108       return false;
 109     }
 110
 111     int read_pid;
 112     if (!iter.ReadInt(&read_pid)) {
 113       DLOG(WARNING) << "Failed to Read PID.";
 114       return false;
 115     }
 116
 117     *pid = read_pid;
 118     return true;
 119   }
 120
 121   static Pickle CreateResponse(int32 pid) {
 122     Pickle request;
 123     request.WriteString(kHelloResponseMagic);
 124     request.WriteInt(pid);
 125     return request;
 126   }
 127
 128   static bool ReadResponse(Pickle& pickle, int32* pid) {
 129     PickleIterator iter(pickle);
 130     std::string hello;
 131     if (!iter.ReadString(&hello)) {
 132       DLOG(WARNING) << "Failed to read magic string.";
 133       return false;
 134     }
 135
 136     if (hello != kHelloResponseMagic) {
 137       DLOG(WARNING) << "Magic mismatch:" << hello;
 138       return false;
 139     }
 140
 141     int read_pid;
 142     if (!iter.ReadInt(&read_pid)) {
 143       DLOG(WARNING) << "Failed to read PID.";
 144       return false;
 145     }
 146
 147     *pid = read_pid;
 148     return true;
 149   }
 150
 151  private:
 152   static const char* kHelloRequestMagic;
 153   static const char* kHelloResponseMagic;
 154 };
 155
 156 const char* HelloMessage::kHelloRequestMagic = "MREQ";
 157 const char* HelloMessage::kHelloResponseMagic = "MRES";
 158
 159 } // namespace
 160
 161 //------------------------------------------------------------------------------
 162
 163 // A MessagePipeReader implemenation for IPC::Message communication.
 164 class ChannelMojo::MessageReader : public internal::MessagePipeReader {
 165  public:
 166   MessageReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
 167       : internal::MessagePipeReader(pipe.Pass()),
 168         owner_(owner) {}
 169
 170   bool Send(scoped_ptr<Message> message);
 171   virtual void OnMessageReceived() OVERRIDE;
 172   virtual void OnPipeClosed() OVERRIDE;
 173   virtual void OnPipeError(MojoResult error) OVERRIDE;
 174
 175  private:
 176   ChannelMojo* owner_;
 177 };
 178
 179 void ChannelMojo::MessageReader::OnMessageReceived() {
 180   Message message(data_buffer().empty() ? "" : &data_buffer()[0],
 181                   static_cast<uint32>(data_buffer().size()));
 182
 183   std::vector<MojoHandle> handle_buffer;
 184   TakeHandleBuffer(&handle_buffer);
 185 #if defined(OS_POSIX) && !defined(OS_NACL)
 186   for (size_t i = 0; i < handle_buffer.size(); ++i) {
 187     mojo::embedder::ScopedPlatformHandle platform_handle;
 188     MojoResult unwrap_result = mojo::embedder::PassWrappedPlatformHandle(
 189         handle_buffer[i], &platform_handle);
 190     if (unwrap_result != MOJO_RESULT_OK) {
 191       DLOG(WARNING) << "Pipe failed to covert handles. Closing: "
 192                     << unwrap_result;
 193       CloseWithError(unwrap_result);
 194       return;
 195     }
 196
 197     bool ok = message.file_descriptor_set()->Add(platform_handle.release().fd);
 198     DCHECK(ok);
 199   }
 200 #else
 201   DCHECK(handle_buffer.empty());
 202 #endif
 203
 204   message.TraceMessageEnd();
 205   owner_->OnMessageReceived(message);
 206 }
 207
 208 void ChannelMojo::MessageReader::OnPipeClosed() {
 209   if (!owner_)
 210     return;
 211   owner_->OnPipeClosed(this);
 212   owner_ = NULL;
 213 }
 214
 215 void ChannelMojo::MessageReader::OnPipeError(MojoResult error) {
 216   if (!owner_)
 217     return;
 218   owner_->OnPipeError(this);
 219 }
 220
 221 bool ChannelMojo::MessageReader::Send(scoped_ptr<Message> message) {
 222   DCHECK(IsValid());
 223
 224   message->TraceMessageBegin();
 225   std::vector<MojoHandle> handles;
 226 #if defined(OS_POSIX) && !defined(OS_NACL)
 227   // We dup() the handles in IPC::Message to transmit.
 228   // IPC::FileDescriptorSet has intricate lifecycle semantics
 229   // of FDs, so just to dup()-and-own them is the safest option.
 230   if (message->HasFileDescriptors()) {
 231     FileDescriptorSet* fdset = message->file_descriptor_set();
 232     for (size_t i = 0; i < fdset->size(); ++i) {
 233       int fd_to_send = dup(fdset->GetDescriptorAt(i));
 234       if (-1 == fd_to_send) {
 235         DPLOG(WARNING) << "Failed to dup FD to transmit.";
 236         std::for_each(handles.begin(), handles.end(), &MojoClose);
 237         CloseWithError(MOJO_RESULT_UNKNOWN);
 238         return false;
 239       }
 240
 241       MojoHandle wrapped_handle;
 242       MojoResult wrap_result = CreatePlatformHandleWrapper(
 243           mojo::embedder::ScopedPlatformHandle(
 244               mojo::embedder::PlatformHandle(fd_to_send)),
 245           &wrapped_handle);
 246       if (MOJO_RESULT_OK != wrap_result) {
 247         DLOG(WARNING) << "Pipe failed to wrap handles. Closing: "
 248                       << wrap_result;
 249         std::for_each(handles.begin(), handles.end(), &MojoClose);
 250         CloseWithError(wrap_result);
 251         return false;
 252       }
 253
 254       handles.push_back(wrapped_handle);
 255     }
 256   }
 257 #endif
 258   MojoResult write_result = MojoWriteMessage(
 259       handle(),
 260       message->data(), static_cast<uint32>(message->size()),
 261       handles.empty() ? NULL : &handles[0],
 262       static_cast<uint32>(handles.size()),
 263       MOJO_WRITE_MESSAGE_FLAG_NONE);
 264   if (MOJO_RESULT_OK != write_result) {
 265     std::for_each(handles.begin(), handles.end(), &MojoClose);
 266     CloseWithError(write_result);
 267     return false;
 268   }
 269
 270   return true;
 271 }
 272
 273 //------------------------------------------------------------------------------
 274
 275 // MessagePipeReader implementation for control messages.
 276 // Actual message handling is implemented by sublcasses.
 277 class ChannelMojo::ControlReader : public internal::MessagePipeReader {
 278  public:
 279   ControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
 280       : internal::MessagePipeReader(pipe.Pass()),
 281         owner_(owner) {}
 282
 283   virtual bool Connect() { return true; }
 284   virtual void OnPipeClosed() OVERRIDE;
 285   virtual void OnPipeError(MojoResult error) OVERRIDE;
 286
 287  protected:
 288   ChannelMojo* owner_;
 289 };
 290
 291 void ChannelMojo::ControlReader::OnPipeClosed() {
 292   if (!owner_)
 293     return;
 294   owner_->OnPipeClosed(this);
 295   owner_ = NULL;
 296 }
 297
 298 void ChannelMojo::ControlReader::OnPipeError(MojoResult error) {
 299   if (!owner_)
 300     return;
 301   owner_->OnPipeError(this);
 302 }
 303
 304 //------------------------------------------------------------------------------
 305
 306 // ControlReader for server-side ChannelMojo.
 307 class ChannelMojo::ServerControlReader : public ChannelMojo::ControlReader {
 308  public:
 309   ServerControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
 310       : ControlReader(pipe.Pass(), owner) { }
 311
 312   virtual bool Connect() OVERRIDE;
 313   virtual void OnMessageReceived() OVERRIDE;
 314
 315  private:
 316   MojoResult SendHelloRequest();
 317   MojoResult RespondHelloResponse();
 318
 319   mojo::ScopedMessagePipeHandle message_pipe_;
 320 };
 321
 322 bool ChannelMojo::ServerControlReader::Connect() {
 323   MojoResult result = SendHelloRequest();
 324   if (result != MOJO_RESULT_OK) {
 325     CloseWithError(result);
 326     return false;
 327   }
 328
 329   return true;
 330 }
 331
 332 MojoResult ChannelMojo::ServerControlReader::SendHelloRequest() {
 333   DCHECK(IsValid());
 334   DCHECK(!message_pipe_.is_valid());
 335
 336   mojo::ScopedMessagePipeHandle self;
 337   mojo::ScopedMessagePipeHandle peer;
 338   MojoResult create_result = mojo::CreateMessagePipe(
 339       NULL, &message_pipe_, &peer);
 340   if (MOJO_RESULT_OK != create_result) {
 341     DLOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
 342     return create_result;
 343   }
 344
 345   MojoHandle peer_to_send = peer.get().value();
 346   Pickle request = HelloMessage::CreateRequest(owner_->GetSelfPID());
 347   MojoResult write_result = MojoWriteMessage(
 348       handle(),
 349       request.data(), static_cast<uint32>(request.size()),
 350       &peer_to_send, 1,
 351       MOJO_WRITE_MESSAGE_FLAG_NONE);
 352   if (MOJO_RESULT_OK != write_result) {
 353     DLOG(WARNING) << "Writing Hello request failed: " << create_result;
 354     return write_result;
 355   }
 356
 357   // |peer| is sent and no longer owned by |this|.
 358   (void)peer.release();
 359   return MOJO_RESULT_OK;
 360 }
 361
 362 MojoResult ChannelMojo::ServerControlReader::RespondHelloResponse() {
 363   Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
 364                  static_cast<uint32>(data_buffer().size()));
 365
 366   int32 read_pid = 0;
 367   if (!HelloMessage::ReadResponse(request, &read_pid)) {
 368     DLOG(ERROR) << "Failed to parse Hello response.";
 369     return MOJO_RESULT_UNKNOWN;
 370   }
 371
 372   base::ProcessId pid = static_cast<base::ProcessId>(read_pid);
 373   owner_->set_peer_pid(pid);
 374   owner_->OnConnected(message_pipe_.Pass());
 375   return MOJO_RESULT_OK;
 376 }
 377
 378 void ChannelMojo::ServerControlReader::OnMessageReceived() {
 379   MojoResult result = RespondHelloResponse();
 380   if (result != MOJO_RESULT_OK)
 381     CloseWithError(result);
 382 }
 383
 384 //------------------------------------------------------------------------------
 385
 386 // ControlReader for client-side ChannelMojo.
 387 class ChannelMojo::ClientControlReader : public ChannelMojo::ControlReader {
 388  public:
 389   ClientControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
 390       : ControlReader(pipe.Pass(), owner) {}
 391
 392   virtual void OnMessageReceived() OVERRIDE;
 393
 394  private:
 395   MojoResult RespondHelloRequest(MojoHandle message_channel);
 396 };
 397
 398 MojoResult ChannelMojo::ClientControlReader::RespondHelloRequest(
 399     MojoHandle message_channel) {
 400   DCHECK(IsValid());
 401
 402   mojo::ScopedMessagePipeHandle received_pipe(
 403       (mojo::MessagePipeHandle(message_channel)));
 404
 405   int32 read_request = 0;
 406   Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
 407                  static_cast<uint32>(data_buffer().size()));
 408   if (!HelloMessage::ReadRequest(request, &read_request)) {
 409     DLOG(ERROR) << "Hello request has wrong magic.";
 410     return MOJO_RESULT_UNKNOWN;
 411   }
 412
 413   base::ProcessId pid = read_request;
 414   Pickle response = HelloMessage::CreateResponse(owner_->GetSelfPID());
 415   MojoResult write_result = MojoWriteMessage(
 416       handle(),
 417       response.data(), static_cast<uint32>(response.size()),
 418       NULL, 0,
 419       MOJO_WRITE_MESSAGE_FLAG_NONE);
 420   if (MOJO_RESULT_OK != write_result) {
 421     DLOG(ERROR) << "Writing Hello response failed: " << write_result;
 422     return write_result;
 423   }
 424
 425   owner_->set_peer_pid(pid);
 426   owner_->OnConnected(received_pipe.Pass());
 427   return MOJO_RESULT_OK;
 428 }
 429
 430 void ChannelMojo::ClientControlReader::OnMessageReceived() {
 431   std::vector<MojoHandle> handle_buffer;
 432   TakeHandleBuffer(&handle_buffer);
 433   if (handle_buffer.size() != 1) {
 434     DLOG(ERROR) << "Hello request doesn't contains required handle: "
 435                 << handle_buffer.size();
 436     CloseWithError(MOJO_RESULT_UNKNOWN);
 437     return;
 438   }
 439
 440   MojoResult result = RespondHelloRequest(handle_buffer[0]);
 441   if (result != MOJO_RESULT_OK) {
 442     DLOG(ERROR) << "Failed to respond Hello request. Closing: "
 443                 << result;
 444     CloseWithError(result);
 445   }
 446 }
 447
 448 //------------------------------------------------------------------------------
 449
 450 void ChannelMojo::ChannelInfoDeleter::operator()(
 451     mojo::embedder::ChannelInfo* ptr) const {
 452   mojo::embedder::DestroyChannelOnIOThread(ptr);
 453 }
 454
 455 //------------------------------------------------------------------------------
 456
 457 // static
 458 scoped_ptr<ChannelMojo> ChannelMojo::Create(
 459     scoped_ptr<Channel> bootstrap, Mode mode, Listener* listener,
 460     scoped_refptr<base::TaskRunner> io_thread_task_runner) {
 461   return make_scoped_ptr(new ChannelMojo(
 462       bootstrap.Pass(), mode, listener, io_thread_task_runner));
 463 }
 464
 465 // static
 466 scoped_ptr<ChannelMojo> ChannelMojo::Create(
 467     const ChannelHandle &channel_handle, Mode mode, Listener* listener,
 468     scoped_refptr<base::TaskRunner> io_thread_task_runner) {
 469   return Create(
 470       Channel::Create(channel_handle, mode, g_null_listener.Pointer()),
 471       mode, listener, io_thread_task_runner);
 472 }
 473
 474 // static
 475 scoped_ptr<ChannelFactory> ChannelMojo::CreateFactory(
 476     const ChannelHandle &channel_handle, Mode mode,
 477     scoped_refptr<base::TaskRunner> io_thread_task_runner) {
 478   return make_scoped_ptr(
 479       new MojoChannelFactory(
 480           channel_handle, mode,
 481           io_thread_task_runner)).PassAs<ChannelFactory>();
 482 }
 483
 484 ChannelMojo::ChannelMojo(
 485     scoped_ptr<Channel> bootstrap, Mode mode, Listener* listener,
 486     scoped_refptr<base::TaskRunner> io_thread_task_runner)
 487     : bootstrap_(bootstrap.Pass()),
 488       mode_(mode), listener_(listener),
 489       peer_pid_(base::kNullProcessId),
 490       weak_factory_(this) {
 491   if (base::MessageLoopProxy::current() == io_thread_task_runner.get()) {
 492     InitOnIOThread();
 493   } else {
 494     io_thread_task_runner->PostTask(FROM_HERE,
 495                                     base::Bind(&ChannelMojo::InitOnIOThread,
 496                                                weak_factory_.GetWeakPtr()));
 497   }
 498 }
 499
 500 ChannelMojo::~ChannelMojo() {
 501   Close();
 502 }
 503
 504 void ChannelMojo::InitOnIOThread() {
 505   mojo::embedder::ChannelInfo* channel_info;
 506   mojo::ScopedMessagePipeHandle control_pipe =
 507       mojo::embedder::CreateChannelOnIOThread(
 508           mojo::embedder::ScopedPlatformHandle(
 509               ToPlatformHandle(bootstrap_->TakePipeHandle())),
 510           &channel_info);
 511   channel_info_.reset(channel_info);
 512
 513   switch (mode_) {
 514     case MODE_SERVER:
 515       control_reader_.reset(new ServerControlReader(control_pipe.Pass(), this));
 516       break;
 517     case MODE_CLIENT:
 518       control_reader_.reset(new ClientControlReader(control_pipe.Pass(), this));
 519       break;
 520     default:
 521       NOTREACHED();
 522       break;
 523   }
 524 }
 525
 526 bool ChannelMojo::Connect() {
 527   DCHECK(!message_reader_);
 528   return control_reader_->Connect();
 529 }
 530
 531 void ChannelMojo::Close() {
 532   control_reader_.reset();
 533   message_reader_.reset();
 534   channel_info_.reset();
 535 }
 536
 537 void ChannelMojo::OnConnected(mojo::ScopedMessagePipeHandle pipe) {
 538   message_reader_ = make_scoped_ptr(new MessageReader(pipe.Pass(), this));
 539
 540   for (size_t i = 0; i < pending_messages_.size(); ++i) {
 541     message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
 542     pending_messages_[i] = NULL;
 543   }
 544
 545   pending_messages_.clear();
 546
 547   listener_->OnChannelConnected(GetPeerPID());
 548 }
 549
 550 void ChannelMojo::OnPipeClosed(internal::MessagePipeReader* reader) {
 551   Close();
 552 }
 553
 554 void ChannelMojo::OnPipeError(internal::MessagePipeReader* reader) {
 555   listener_->OnChannelError();
 556 }
 557
 558
 559 bool ChannelMojo::Send(Message* message) {
 560   if (!message_reader_) {
 561     pending_messages_.push_back(message);
 562     return true;
 563   }
 564
 565   return message_reader_->Send(make_scoped_ptr(message));
 566 }
 567
 568 base::ProcessId ChannelMojo::GetPeerPID() const {
 569   return peer_pid_;
 570 }
 571
 572 base::ProcessId ChannelMojo::GetSelfPID() const {
 573   return bootstrap_->GetSelfPID();
 574 }
 575
 576 ChannelHandle ChannelMojo::TakePipeHandle() {
 577   return bootstrap_->TakePipeHandle();
 578 }
 579
 580 void ChannelMojo::OnMessageReceived(Message& message) {
 581   listener_->OnMessageReceived(message);
 582   if (message.dispatch_error())
 583     listener_->OnBadMessageReceived(message);
 584 }
 585
 586 #if defined(OS_POSIX) && !defined(OS_NACL)
 587 int ChannelMojo::GetClientFileDescriptor() const {
 588   return bootstrap_->GetClientFileDescriptor();
 589 }
 590
 591 int ChannelMojo::TakeClientFileDescriptor() {
 592   return bootstrap_->TakeClientFileDescriptor();
 593 }
 594 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 595
 596 }  // namespace IPC