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 "ipc/ipc_logging.h"
  12 #include "ipc/ipc_message_attachment_set.h"
  13 #include "ipc/ipc_message_macros.h"
  14 #include "ipc/mojo/client_channel.mojom.h"
  15 #include "ipc/mojo/ipc_mojo_bootstrap.h"
  16 #include "ipc/mojo/ipc_mojo_handle_attachment.h"
  17 #include "third_party/mojo/src/mojo/edk/embedder/embedder.h"
  18 #include "third_party/mojo/src/mojo/public/cpp/bindings/error_handler.h"
  19
  20 #if defined(OS_POSIX) && !defined(OS_NACL)
  21 #include "ipc/ipc_platform_file_attachment_posix.h"
  22 #endif
  23
  24 namespace IPC {
  25
  26 namespace {
  27
  28 class MojoChannelFactory : public ChannelFactory {
  29  public:
  30   MojoChannelFactory(ChannelMojo::Delegate* delegate,
  31                      scoped_refptr<base::TaskRunner> io_runner,
  32                      ChannelHandle channel_handle,
  33                      Channel::Mode mode)
  34       : delegate_(delegate),
  35         io_runner_(io_runner),
  36         channel_handle_(channel_handle),
  37         mode_(mode) {}
  38
  39   std::string GetName() const override {
  40     return channel_handle_.name;
  41   }
  42
  43   scoped_ptr<Channel> BuildChannel(Listener* listener) override {
  44     return ChannelMojo::Create(delegate_, io_runner_, channel_handle_, mode_,
  45                                listener);
  46   }
  47
  48  private:
  49   ChannelMojo::Delegate* delegate_;
  50   scoped_refptr<base::TaskRunner> io_runner_;
  51   ChannelHandle channel_handle_;
  52   Channel::Mode mode_;
  53 };
  54
  55 //------------------------------------------------------------------------------
  56
  57 class ClientChannelMojo : public ChannelMojo,
  58                           public ClientChannel,
  59                           public mojo::ErrorHandler {
  60  public:
  61   ClientChannelMojo(ChannelMojo::Delegate* delegate,
  62                     scoped_refptr<base::TaskRunner> io_runner,
  63                     const ChannelHandle& handle,
  64                     Listener* listener);
  65   ~ClientChannelMojo() override;
  66   // MojoBootstrap::Delegate implementation
  67   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
  68   // mojo::ErrorHandler implementation
  69   void OnConnectionError() override;
  70   // ClientChannel implementation
  71   void Init(
  72       mojo::ScopedMessagePipeHandle pipe,
  73       int32_t peer_pid,
  74       const mojo::Callback<void(int32_t)>& callback) override;
  75
  76  private:
  77   mojo::Binding<ClientChannel> binding_;
  78
  79   DISALLOW_COPY_AND_ASSIGN(ClientChannelMojo);
  80 };
  81
  82 ClientChannelMojo::ClientChannelMojo(ChannelMojo::Delegate* delegate,
  83                                      scoped_refptr<base::TaskRunner> io_runner,
  84                                      const ChannelHandle& handle,
  85                                      Listener* listener)
  86     : ChannelMojo(delegate, io_runner, handle, Channel::MODE_CLIENT, listener),
  87       binding_(this) {
  88 }
  89
  90 ClientChannelMojo::~ClientChannelMojo() {
  91 }
  92
  93 void ClientChannelMojo::OnPipeAvailable(
  94     mojo::embedder::ScopedPlatformHandle handle) {
  95   binding_.Bind(CreateMessagingPipe(handle.Pass()));
  96 }
  97
  98 void ClientChannelMojo::OnConnectionError() {
  99   listener()->OnChannelError();
 100 }
 101
 102 void ClientChannelMojo::Init(
 103     mojo::ScopedMessagePipeHandle pipe,
 104     int32_t peer_pid,
 105     const mojo::Callback<void(int32_t)>& callback) {
 106   InitMessageReader(pipe.Pass(), static_cast<base::ProcessId>(peer_pid));
 107   callback.Run(GetSelfPID());
 108 }
 109
 110 //------------------------------------------------------------------------------
 111
 112 class ServerChannelMojo : public ChannelMojo, public mojo::ErrorHandler {
 113  public:
 114   ServerChannelMojo(ChannelMojo::Delegate* delegate,
 115                     scoped_refptr<base::TaskRunner> io_runner,
 116                     const ChannelHandle& handle,
 117                     Listener* listener);
 118   ~ServerChannelMojo() override;
 119
 120   // MojoBootstrap::Delegate implementation
 121   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
 122   // mojo::ErrorHandler implementation
 123   void OnConnectionError() override;
 124   // Channel override
 125   void Close() override;
 126
 127  private:
 128   // ClientChannelClient implementation
 129   void ClientChannelWasInitialized(int32_t peer_pid);
 130
 131   mojo::InterfacePtr<ClientChannel> client_channel_;
 132   mojo::ScopedMessagePipeHandle message_pipe_;
 133
 134   DISALLOW_COPY_AND_ASSIGN(ServerChannelMojo);
 135 };
 136
 137 ServerChannelMojo::ServerChannelMojo(ChannelMojo::Delegate* delegate,
 138                                      scoped_refptr<base::TaskRunner> io_runner,
 139                                      const ChannelHandle& handle,
 140                                      Listener* listener)
 141     : ChannelMojo(delegate, io_runner, handle, Channel::MODE_SERVER, listener) {
 142 }
 143
 144 ServerChannelMojo::~ServerChannelMojo() {
 145   Close();
 146 }
 147
 148 void ServerChannelMojo::OnPipeAvailable(
 149     mojo::embedder::ScopedPlatformHandle handle) {
 150   mojo::ScopedMessagePipeHandle peer;
 151   MojoResult create_result =
 152       mojo::CreateMessagePipe(nullptr, &message_pipe_, &peer);
 153   if (create_result != MOJO_RESULT_OK) {
 154     LOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
 155     listener()->OnChannelError();
 156     return;
 157   }
 158
 159   client_channel_.Bind(CreateMessagingPipe(handle.Pass()));
 160   client_channel_.set_error_handler(this);
 161   client_channel_->Init(
 162       peer.Pass(),
 163       static_cast<int32_t>(GetSelfPID()),
 164       base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
 165                  base::Unretained(this)));
 166 }
 167
 168 void ServerChannelMojo::ClientChannelWasInitialized(int32_t peer_pid) {
 169   InitMessageReader(message_pipe_.Pass(), peer_pid);
 170 }
 171
 172 void ServerChannelMojo::OnConnectionError() {
 173   listener()->OnChannelError();
 174 }
 175
 176 void ServerChannelMojo::Close() {
 177   client_channel_.reset();
 178   message_pipe_.reset();
 179   ChannelMojo::Close();
 180 }
 181
 182 #if defined(OS_POSIX) && !defined(OS_NACL)
 183
 184 base::ScopedFD TakeOrDupFile(internal::PlatformFileAttachment* attachment) {
 185   return attachment->Owns() ? base::ScopedFD(attachment->TakePlatformFile())
 186                             : base::ScopedFD(dup(attachment->file()));
 187 }
 188
 189 #endif
 190
 191 } // namespace
 192
 193 //------------------------------------------------------------------------------
 194
 195 void ChannelMojo::ChannelInfoDeleter::operator()(
 196     mojo::embedder::ChannelInfo* ptr) const {
 197   mojo::embedder::DestroyChannelOnIOThread(ptr);
 198 }
 199
 200 //------------------------------------------------------------------------------
 201
 202 // static
 203 bool ChannelMojo::ShouldBeUsed() {
 204   // TODO(morrita): Remove this if it sticks.
 205   return true;
 206 }
 207
 208 // static
 209 scoped_ptr<ChannelMojo> ChannelMojo::Create(
 210     ChannelMojo::Delegate* delegate,
 211     scoped_refptr<base::TaskRunner> io_runner,
 212     const ChannelHandle& channel_handle,
 213     Mode mode,
 214     Listener* listener) {
 215   switch (mode) {
 216     case Channel::MODE_CLIENT:
 217       return make_scoped_ptr(
 218           new ClientChannelMojo(delegate, io_runner, channel_handle, listener));
 219     case Channel::MODE_SERVER:
 220       return make_scoped_ptr(
 221           new ServerChannelMojo(delegate, io_runner, channel_handle, listener));
 222     default:
 223       NOTREACHED();
 224       return nullptr;
 225   }
 226 }
 227
 228 // static
 229 scoped_ptr<ChannelFactory> ChannelMojo::CreateServerFactory(
 230     ChannelMojo::Delegate* delegate,
 231     scoped_refptr<base::TaskRunner> io_runner,
 232     const ChannelHandle& channel_handle) {
 233   return make_scoped_ptr(new MojoChannelFactory(
 234       delegate, io_runner, channel_handle, Channel::MODE_SERVER));
 235 }
 236
 237 // static
 238 scoped_ptr<ChannelFactory> ChannelMojo::CreateClientFactory(
 239     ChannelMojo::Delegate* delegate,
 240     scoped_refptr<base::TaskRunner> io_runner,
 241     const ChannelHandle& channel_handle) {
 242   return make_scoped_ptr(new MojoChannelFactory(
 243       delegate, io_runner, channel_handle, Channel::MODE_CLIENT));
 244 }
 245
 246 ChannelMojo::ChannelMojo(ChannelMojo::Delegate* delegate,
 247                          scoped_refptr<base::TaskRunner> io_runner,
 248                          const ChannelHandle& handle,
 249                          Mode mode,
 250                          Listener* listener)
 251     : mode_(mode),
 252       listener_(listener),
 253       peer_pid_(base::kNullProcessId),
 254       weak_factory_(this) {
 255   // Create MojoBootstrap after all members are set as it touches
 256   // ChannelMojo from a different thread.
 257   bootstrap_ = MojoBootstrap::Create(handle, mode, this);
 258   if (io_runner == base::MessageLoop::current()->message_loop_proxy()) {
 259     InitOnIOThread(delegate);
 260   } else {
 261     io_runner->PostTask(FROM_HERE,
 262                         base::Bind(&ChannelMojo::InitOnIOThread,
 263                                    base::Unretained(this), delegate));
 264   }
 265 }
 266
 267 ChannelMojo::~ChannelMojo() {
 268   Close();
 269 }
 270
 271 void ChannelMojo::InitOnIOThread(ChannelMojo::Delegate* delegate) {
 272   ipc_support_.reset(
 273       new ScopedIPCSupport(base::MessageLoop::current()->task_runner()));
 274   if (!delegate)
 275     return;
 276   delegate_ = delegate->ToWeakPtr();
 277   delegate_->OnChannelCreated(weak_factory_.GetWeakPtr());
 278 }
 279
 280 mojo::ScopedMessagePipeHandle ChannelMojo::CreateMessagingPipe(
 281     mojo::embedder::ScopedPlatformHandle handle) {
 282   DCHECK(!channel_info_.get());
 283   mojo::embedder::ChannelInfo* channel_info;
 284   mojo::ScopedMessagePipeHandle pipe =
 285       mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
 286   channel_info_.reset(channel_info);
 287   return pipe.Pass();
 288 }
 289
 290 bool ChannelMojo::Connect() {
 291   DCHECK(!message_reader_);
 292   return bootstrap_->Connect();
 293 }
 294
 295 void ChannelMojo::Close() {
 296   message_reader_.reset();
 297   channel_info_.reset();
 298   ipc_support_.reset();
 299 }
 300
 301 void ChannelMojo::OnBootstrapError() {
 302   listener_->OnChannelError();
 303 }
 304
 305 void ChannelMojo::InitMessageReader(mojo::ScopedMessagePipeHandle pipe,
 306                                     int32_t peer_pid) {
 307   message_reader_ =
 308       make_scoped_ptr(new internal::MessagePipeReader(pipe.Pass(), this));
 309
 310   for (size_t i = 0; i < pending_messages_.size(); ++i) {
 311     bool sent = message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
 312     pending_messages_[i] = nullptr;
 313     if (!sent) {
 314       pending_messages_.clear();
 315       listener_->OnChannelError();
 316       return;
 317     }
 318   }
 319
 320   pending_messages_.clear();
 321
 322   set_peer_pid(peer_pid);
 323   listener_->OnChannelConnected(static_cast<int32_t>(GetPeerPID()));
 324   if (message_reader_)
 325     message_reader_->ReadMessagesThenWait();
 326 }
 327
 328 void ChannelMojo::OnPipeClosed(internal::MessagePipeReader* reader) {
 329   Close();
 330 }
 331
 332 void ChannelMojo::OnPipeError(internal::MessagePipeReader* reader) {
 333   listener_->OnChannelError();
 334 }
 335
 336
 337 bool ChannelMojo::Send(Message* message) {
 338   if (!message_reader_) {
 339     pending_messages_.push_back(message);
 340     return true;
 341   }
 342
 343   return message_reader_->Send(make_scoped_ptr(message));
 344 }
 345
 346 base::ProcessId ChannelMojo::GetPeerPID() const {
 347   return peer_pid_;
 348 }
 349
 350 base::ProcessId ChannelMojo::GetSelfPID() const {
 351   return bootstrap_->GetSelfPID();
 352 }
 353
 354 void ChannelMojo::OnClientLaunched(base::ProcessHandle handle) {
 355   bootstrap_->OnClientLaunched(handle);
 356 }
 357
 358 void ChannelMojo::OnMessageReceived(Message& message) {
 359   TRACE_EVENT2("ipc,toplevel", "ChannelMojo::OnMessageReceived",
 360                "class", IPC_MESSAGE_ID_CLASS(message.type()),
 361                "line", IPC_MESSAGE_ID_LINE(message.type()));
 362   listener_->OnMessageReceived(message);
 363   if (message.dispatch_error())
 364     listener_->OnBadMessageReceived(message);
 365 }
 366
 367 #if defined(OS_POSIX) && !defined(OS_NACL)
 368 int ChannelMojo::GetClientFileDescriptor() const {
 369   return bootstrap_->GetClientFileDescriptor();
 370 }
 371
 372 base::ScopedFD ChannelMojo::TakeClientFileDescriptor() {
 373   return bootstrap_->TakeClientFileDescriptor();
 374 }
 375 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 376
 377 // static
 378 MojoResult ChannelMojo::ReadFromMessageAttachmentSet(
 379     Message* message,
 380     std::vector<MojoHandle>* handles) {
 381   // We dup() the handles in IPC::Message to transmit.
 382   // IPC::MessageAttachmentSet has intricate lifecycle semantics
 383   // of FDs, so just to dup()-and-own them is the safest option.
 384   if (message->HasAttachments()) {
 385     MessageAttachmentSet* set = message->attachment_set();
 386     for (unsigned i = 0; i < set->size(); ++i) {
 387       scoped_refptr<MessageAttachment> attachment = set->GetAttachmentAt(i);
 388       switch (attachment->GetType()) {
 389         case MessageAttachment::TYPE_PLATFORM_FILE:
 390 #if defined(OS_POSIX) && !defined(OS_NACL)
 391         {
 392           base::ScopedFD file =
 393               TakeOrDupFile(static_cast<IPC::internal::PlatformFileAttachment*>(
 394                   attachment.get()));
 395           if (!file.is_valid()) {
 396             DPLOG(WARNING) << "Failed to dup FD to transmit.";
 397             set->CommitAll();
 398             return MOJO_RESULT_UNKNOWN;
 399           }
 400
 401           MojoHandle wrapped_handle;
 402           MojoResult wrap_result = CreatePlatformHandleWrapper(
 403               mojo::embedder::ScopedPlatformHandle(
 404                   mojo::embedder::PlatformHandle(file.release())),
 405               &wrapped_handle);
 406           if (MOJO_RESULT_OK != wrap_result) {
 407             LOG(WARNING) << "Pipe failed to wrap handles. Closing: "
 408                          << wrap_result;
 409             set->CommitAll();
 410             return wrap_result;
 411           }
 412
 413           handles->push_back(wrapped_handle);
 414         }
 415 #else
 416           NOTREACHED();
 417 #endif  //  defined(OS_POSIX) && !defined(OS_NACL)
 418         break;
 419         case MessageAttachment::TYPE_MOJO_HANDLE: {
 420           mojo::ScopedHandle handle =
 421               static_cast<IPC::internal::MojoHandleAttachment*>(
 422                   attachment.get())->TakeHandle();
 423           handles->push_back(handle.release().value());
 424         } break;
 425       }
 426     }
 427
 428     set->CommitAll();
 429   }
 430
 431   return MOJO_RESULT_OK;
 432 }
 433
 434 // static
 435 MojoResult ChannelMojo::WriteToMessageAttachmentSet(
 436     const std::vector<MojoHandle>& handle_buffer,
 437     Message* message) {
 438   for (size_t i = 0; i < handle_buffer.size(); ++i) {
 439     bool ok = message->attachment_set()->AddAttachment(
 440         new IPC::internal::MojoHandleAttachment(
 441             mojo::MakeScopedHandle(mojo::Handle(handle_buffer[i]))));
 442     DCHECK(ok);
 443     if (!ok) {
 444       LOG(ERROR) << "Failed to add new Mojo handle.";
 445       return MOJO_RESULT_UNKNOWN;
 446     }
 447   }
 448
 449   return MOJO_RESULT_OK;
 450 }
 451
 452 }  // namespace IPC