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 "base/thread_task_runner_handle.h"
  11 #include "ipc/ipc_listener.h"
  12 #include "ipc/ipc_logging.h"
  13 #include "ipc/ipc_message_attachment_set.h"
  14 #include "ipc/ipc_message_macros.h"
  15 #include "ipc/mojo/client_channel.mojom.h"
  16 #include "ipc/mojo/ipc_mojo_bootstrap.h"
  17 #include "ipc/mojo/ipc_mojo_handle_attachment.h"
  18 #include "third_party/mojo/src/mojo/edk/embedder/embedder.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(scoped_refptr<base::TaskRunner> io_runner,
  31                      ChannelHandle channel_handle,
  32                      Channel::Mode mode,
  33                      AttachmentBroker* broker)
  34       : io_runner_(io_runner),
  35         channel_handle_(channel_handle),
  36         mode_(mode),
  37         broker_(broker) {}
  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(io_runner_, channel_handle_, mode_, listener,
  45                                broker_);
  46   }
  47
  48  private:
  49   scoped_refptr<base::TaskRunner> io_runner_;
  50   ChannelHandle channel_handle_;
  51   Channel::Mode mode_;
  52   AttachmentBroker* broker_;
  53 };
  54
  55 //------------------------------------------------------------------------------
  56
  57 class ClientChannelMojo : public ChannelMojo, public ClientChannel {
  58  public:
  59   ClientChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
  60                     const ChannelHandle& handle,
  61                     Listener* listener,
  62                     AttachmentBroker* broker);
  63   ~ClientChannelMojo() override;
  64   // MojoBootstrap::Delegate implementation
  65   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
  66
  67   // ClientChannel implementation
  68   void Init(
  69       mojo::ScopedMessagePipeHandle pipe,
  70       int32_t peer_pid,
  71       const mojo::Callback<void(int32_t)>& callback) override;
  72
  73  private:
  74   void BindPipe(mojo::ScopedMessagePipeHandle handle);
  75   void OnConnectionError();
  76
  77   mojo::Binding<ClientChannel> binding_;
  78   base::WeakPtrFactory<ClientChannelMojo> weak_factory_;
  79
  80   DISALLOW_COPY_AND_ASSIGN(ClientChannelMojo);
  81 };
  82
  83 ClientChannelMojo::ClientChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
  84                                      const ChannelHandle& handle,
  85                                      Listener* listener,
  86                                      AttachmentBroker* broker)
  87     : ChannelMojo(io_runner, handle, Channel::MODE_CLIENT, listener, broker),
  88       binding_(this),
  89       weak_factory_(this) {
  90 }
  91
  92 ClientChannelMojo::~ClientChannelMojo() {
  93 }
  94
  95 void ClientChannelMojo::OnPipeAvailable(
  96     mojo::embedder::ScopedPlatformHandle handle) {
  97   CreateMessagingPipe(handle.Pass(), base::Bind(&ClientChannelMojo::BindPipe,
  98                                                 weak_factory_.GetWeakPtr()));
  99 }
 100
 101 void ClientChannelMojo::Init(
 102     mojo::ScopedMessagePipeHandle pipe,
 103     int32_t peer_pid,
 104     const mojo::Callback<void(int32_t)>& callback) {
 105   InitMessageReader(pipe.Pass(), static_cast<base::ProcessId>(peer_pid));
 106   callback.Run(GetSelfPID());
 107 }
 108
 109 void ClientChannelMojo::BindPipe(mojo::ScopedMessagePipeHandle handle) {
 110   binding_.Bind(handle.Pass());
 111 }
 112
 113 void ClientChannelMojo::OnConnectionError() {
 114   listener()->OnChannelError();
 115 }
 116
 117 //------------------------------------------------------------------------------
 118
 119 class ServerChannelMojo : public ChannelMojo {
 120  public:
 121   ServerChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
 122                     const ChannelHandle& handle,
 123                     Listener* listener,
 124                     AttachmentBroker* broker);
 125   ~ServerChannelMojo() override;
 126
 127   // MojoBootstrap::Delegate implementation
 128   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
 129   // Channel override
 130   void Close() override;
 131
 132  private:
 133   void InitClientChannel(mojo::ScopedMessagePipeHandle peer_handle,
 134                          mojo::ScopedMessagePipeHandle handle);
 135   void OnConnectionError();
 136
 137   // ClientChannelClient implementation
 138   void ClientChannelWasInitialized(int32_t peer_pid);
 139
 140   mojo::InterfacePtr<ClientChannel> client_channel_;
 141   mojo::ScopedMessagePipeHandle message_pipe_;
 142   base::WeakPtrFactory<ServerChannelMojo> weak_factory_;
 143
 144   DISALLOW_COPY_AND_ASSIGN(ServerChannelMojo);
 145 };
 146
 147 ServerChannelMojo::ServerChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
 148                                      const ChannelHandle& handle,
 149                                      Listener* listener,
 150                                      AttachmentBroker* broker)
 151     : ChannelMojo(io_runner, handle, Channel::MODE_SERVER, listener, broker),
 152       weak_factory_(this) {
 153 }
 154
 155 ServerChannelMojo::~ServerChannelMojo() {
 156   Close();
 157 }
 158
 159 void ServerChannelMojo::OnPipeAvailable(
 160     mojo::embedder::ScopedPlatformHandle handle) {
 161   mojo::ScopedMessagePipeHandle peer;
 162   MojoResult create_result =
 163       mojo::CreateMessagePipe(nullptr, &message_pipe_, &peer);
 164   if (create_result != MOJO_RESULT_OK) {
 165     LOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
 166     listener()->OnChannelError();
 167     return;
 168   }
 169   CreateMessagingPipe(
 170       handle.Pass(),
 171       base::Bind(&ServerChannelMojo::InitClientChannel,
 172                  weak_factory_.GetWeakPtr(), base::Passed(&peer)));
 173 }
 174
 175 void ServerChannelMojo::Close() {
 176   client_channel_.reset();
 177   message_pipe_.reset();
 178   ChannelMojo::Close();
 179 }
 180
 181 void ServerChannelMojo::InitClientChannel(
 182     mojo::ScopedMessagePipeHandle peer_handle,
 183     mojo::ScopedMessagePipeHandle handle) {
 184   client_channel_.Bind(
 185       mojo::InterfacePtrInfo<ClientChannel>(handle.Pass(), 0u));
 186   client_channel_.set_connection_error_handler(base::Bind(
 187       &ServerChannelMojo::OnConnectionError, base::Unretained(this)));
 188   client_channel_->Init(
 189       peer_handle.Pass(), static_cast<int32_t>(GetSelfPID()),
 190       base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
 191                  base::Unretained(this)));
 192 }
 193
 194 void ServerChannelMojo::OnConnectionError() {
 195   listener()->OnChannelError();
 196 }
 197
 198 void ServerChannelMojo::ClientChannelWasInitialized(int32_t peer_pid) {
 199   InitMessageReader(message_pipe_.Pass(), peer_pid);
 200 }
 201
 202 #if defined(OS_POSIX) && !defined(OS_NACL)
 203
 204 base::ScopedFD TakeOrDupFile(internal::PlatformFileAttachment* attachment) {
 205   return attachment->Owns() ? base::ScopedFD(attachment->TakePlatformFile())
 206                             : base::ScopedFD(dup(attachment->file()));
 207 }
 208
 209 #endif
 210
 211 }  // namespace
 212
 213 //------------------------------------------------------------------------------
 214
 215 ChannelMojo::ChannelInfoDeleter::ChannelInfoDeleter(
 216     scoped_refptr<base::TaskRunner> io_runner)
 217     : io_runner(io_runner) {
 218 }
 219
 220 ChannelMojo::ChannelInfoDeleter::~ChannelInfoDeleter() {
 221 }
 222
 223 void ChannelMojo::ChannelInfoDeleter::operator()(
 224     mojo::embedder::ChannelInfo* ptr) const {
 225   if (base::ThreadTaskRunnerHandle::Get() == io_runner) {
 226     mojo::embedder::DestroyChannelOnIOThread(ptr);
 227   } else {
 228     io_runner->PostTask(
 229         FROM_HERE, base::Bind(&mojo::embedder::DestroyChannelOnIOThread, ptr));
 230   }
 231 }
 232
 233 //------------------------------------------------------------------------------
 234
 235 // static
 236 bool ChannelMojo::ShouldBeUsed() {
 237   return true;
 238 }
 239
 240 // static
 241 scoped_ptr<ChannelMojo> ChannelMojo::Create(
 242     scoped_refptr<base::TaskRunner> io_runner,
 243     const ChannelHandle& channel_handle,
 244     Mode mode,
 245     Listener* listener,
 246     AttachmentBroker* broker) {
 247   switch (mode) {
 248     case Channel::MODE_CLIENT:
 249       return make_scoped_ptr(
 250           new ClientChannelMojo(io_runner, channel_handle, listener, broker));
 251     case Channel::MODE_SERVER:
 252       return make_scoped_ptr(
 253           new ServerChannelMojo(io_runner, channel_handle, listener, broker));
 254     default:
 255       NOTREACHED();
 256       return nullptr;
 257   }
 258 }
 259
 260 // static
 261 scoped_ptr<ChannelFactory> ChannelMojo::CreateServerFactory(
 262     scoped_refptr<base::TaskRunner> io_runner,
 263     const ChannelHandle& channel_handle,
 264     AttachmentBroker* broker) {
 265   return make_scoped_ptr(new MojoChannelFactory(io_runner, channel_handle,
 266                                                 Channel::MODE_SERVER, broker));
 267 }
 268
 269 // static
 270 scoped_ptr<ChannelFactory> ChannelMojo::CreateClientFactory(
 271     scoped_refptr<base::TaskRunner> io_runner,
 272     const ChannelHandle& channel_handle,
 273     AttachmentBroker* broker) {
 274   return make_scoped_ptr(new MojoChannelFactory(io_runner, channel_handle,
 275                                                 Channel::MODE_CLIENT, broker));
 276 }
 277
 278 ChannelMojo::ChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
 279                          const ChannelHandle& handle,
 280                          Mode mode,
 281                          Listener* listener,
 282                          AttachmentBroker* broker)
 283     : listener_(listener),
 284       peer_pid_(base::kNullProcessId),
 285       io_runner_(io_runner),
 286       channel_info_(nullptr, ChannelInfoDeleter(nullptr)),
 287       waiting_connect_(true),
 288       weak_factory_(this) {
 289   // Create MojoBootstrap after all members are set as it touches
 290   // ChannelMojo from a different thread.
 291   bootstrap_ = MojoBootstrap::Create(handle, mode, this, broker);
 292   if (io_runner == base::MessageLoop::current()->task_runner()) {
 293     InitOnIOThread();
 294   } else {
 295     io_runner->PostTask(FROM_HERE, base::Bind(&ChannelMojo::InitOnIOThread,
 296                                               base::Unretained(this)));
 297   }
 298 }
 299
 300 ChannelMojo::~ChannelMojo() {
 301   Close();
 302 }
 303
 304 void ChannelMojo::InitOnIOThread() {
 305   ipc_support_.reset(
 306       new ScopedIPCSupport(base::MessageLoop::current()->task_runner()));
 307 }
 308
 309 void ChannelMojo::CreateMessagingPipe(
 310     mojo::embedder::ScopedPlatformHandle handle,
 311     const CreateMessagingPipeCallback& callback) {
 312   auto return_callback = base::Bind(&ChannelMojo::OnMessagingPipeCreated,
 313                                     weak_factory_.GetWeakPtr(), callback);
 314   if (base::ThreadTaskRunnerHandle::Get() == io_runner_) {
 315     CreateMessagingPipeOnIOThread(
 316         handle.Pass(), base::ThreadTaskRunnerHandle::Get(), return_callback);
 317   } else {
 318     io_runner_->PostTask(
 319         FROM_HERE,
 320         base::Bind(&ChannelMojo::CreateMessagingPipeOnIOThread,
 321                    base::Passed(&handle), base::ThreadTaskRunnerHandle::Get(),
 322                    return_callback));
 323   }
 324 }
 325
 326 // static
 327 void ChannelMojo::CreateMessagingPipeOnIOThread(
 328     mojo::embedder::ScopedPlatformHandle handle,
 329     scoped_refptr<base::TaskRunner> callback_runner,
 330     const CreateMessagingPipeOnIOThreadCallback& callback) {
 331   mojo::embedder::ChannelInfo* channel_info;
 332   mojo::ScopedMessagePipeHandle pipe =
 333       mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
 334   if (base::ThreadTaskRunnerHandle::Get() == callback_runner) {
 335     callback.Run(pipe.Pass(), channel_info);
 336   } else {
 337     callback_runner->PostTask(
 338         FROM_HERE, base::Bind(callback, base::Passed(&pipe), channel_info));
 339   }
 340 }
 341
 342 void ChannelMojo::OnMessagingPipeCreated(
 343     const CreateMessagingPipeCallback& callback,
 344     mojo::ScopedMessagePipeHandle handle,
 345     mojo::embedder::ChannelInfo* channel_info) {
 346   DCHECK(!channel_info_.get());
 347   channel_info_ = scoped_ptr<mojo::embedder::ChannelInfo, ChannelInfoDeleter>(
 348       channel_info, ChannelInfoDeleter(io_runner_));
 349   callback.Run(handle.Pass());
 350 }
 351
 352 bool ChannelMojo::Connect() {
 353   DCHECK(!message_reader_);
 354   return bootstrap_->Connect();
 355 }
 356
 357 void ChannelMojo::Close() {
 358   scoped_ptr<internal::MessagePipeReader, ReaderDeleter> to_be_deleted;
 359
 360   {
 361     // |message_reader_| has to be cleared inside the lock,
 362     // but the instance has to be deleted outside.
 363     base::AutoLock l(lock_);
 364     to_be_deleted = message_reader_.Pass();
 365     // We might Close() before we Connect().
 366     waiting_connect_ = false;
 367   }
 368
 369   channel_info_.reset();
 370   ipc_support_.reset();
 371   to_be_deleted.reset();
 372 }
 373
 374 void ChannelMojo::OnBootstrapError() {
 375   listener_->OnChannelError();
 376 }
 377
 378 namespace {
 379
 380 // ClosingDeleter calls |CloseWithErrorIfPending| before deleting the
 381 // |MessagePipeReader|.
 382 struct ClosingDeleter {
 383   typedef base::DefaultDeleter<internal::MessagePipeReader> DefaultType;
 384
 385   void operator()(internal::MessagePipeReader* ptr) const {
 386     ptr->CloseWithErrorIfPending();
 387     delete ptr;
 388   }
 389 };
 390
 391 } // namespace
 392
 393 void ChannelMojo::InitMessageReader(mojo::ScopedMessagePipeHandle pipe,
 394                                     int32_t peer_pid) {
 395   scoped_ptr<internal::MessagePipeReader, ClosingDeleter> reader(
 396       new internal::MessagePipeReader(pipe.Pass(), this));
 397
 398   {
 399     base::AutoLock l(lock_);
 400     for (size_t i = 0; i < pending_messages_.size(); ++i) {
 401       bool sent = reader->Send(make_scoped_ptr(pending_messages_[i]));
 402       pending_messages_[i] = nullptr;
 403       if (!sent) {
 404         // OnChannelError() is notified through ClosingDeleter.
 405         pending_messages_.clear();
 406         LOG(ERROR)  << "Failed to flush pending messages";
 407         return;
 408       }
 409     }
 410
 411     // We set |message_reader_| here and won't get any |pending_messages_|
 412     // hereafter. Although we might have some if there is an error, we don't
 413     // care. They cannot be sent anyway.
 414     message_reader_.reset(reader.release());
 415     pending_messages_.clear();
 416     waiting_connect_ = false;
 417   }
 418
 419   set_peer_pid(peer_pid);
 420   listener_->OnChannelConnected(static_cast<int32_t>(GetPeerPID()));
 421   if (message_reader_)
 422     message_reader_->ReadMessagesThenWait();
 423 }
 424
 425 void ChannelMojo::OnPipeClosed(internal::MessagePipeReader* reader) {
 426   Close();
 427 }
 428
 429 void ChannelMojo::OnPipeError(internal::MessagePipeReader* reader) {
 430   listener_->OnChannelError();
 431 }
 432
 433
 434 // Warning: Keep the implementation thread-safe.
 435 bool ChannelMojo::Send(Message* message) {
 436   base::AutoLock l(lock_);
 437   if (!message_reader_) {
 438     pending_messages_.push_back(message);
 439     // Counts as OK before the connection is established, but it's an
 440     // error otherwise.
 441     return waiting_connect_;
 442   }
 443
 444   return message_reader_->Send(make_scoped_ptr(message));
 445 }
 446
 447 bool ChannelMojo::IsSendThreadSafe() const {
 448   return true;
 449 }
 450
 451 base::ProcessId ChannelMojo::GetPeerPID() const {
 452   return peer_pid_;
 453 }
 454
 455 base::ProcessId ChannelMojo::GetSelfPID() const {
 456   return bootstrap_->GetSelfPID();
 457 }
 458
 459 void ChannelMojo::OnMessageReceived(Message& message) {
 460   TRACE_EVENT2("ipc,toplevel", "ChannelMojo::OnMessageReceived",
 461                "class", IPC_MESSAGE_ID_CLASS(message.type()),
 462                "line", IPC_MESSAGE_ID_LINE(message.type()));
 463   listener_->OnMessageReceived(message);
 464   if (message.dispatch_error())
 465     listener_->OnBadMessageReceived(message);
 466 }
 467
 468 #if defined(OS_POSIX) && !defined(OS_NACL)
 469 int ChannelMojo::GetClientFileDescriptor() const {
 470   return bootstrap_->GetClientFileDescriptor();
 471 }
 472
 473 base::ScopedFD ChannelMojo::TakeClientFileDescriptor() {
 474   return bootstrap_->TakeClientFileDescriptor();
 475 }
 476 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 477
 478 // static
 479 MojoResult ChannelMojo::ReadFromMessageAttachmentSet(
 480     Message* message,
 481     std::vector<MojoHandle>* handles) {
 482   // We dup() the handles in IPC::Message to transmit.
 483   // IPC::MessageAttachmentSet has intricate lifecycle semantics
 484   // of FDs, so just to dup()-and-own them is the safest option.
 485   if (message->HasAttachments()) {
 486     MessageAttachmentSet* set = message->attachment_set();
 487     for (unsigned i = 0; i < set->size(); ++i) {
 488       scoped_refptr<MessageAttachment> attachment = set->GetAttachmentAt(i);
 489       switch (attachment->GetType()) {
 490         case MessageAttachment::TYPE_PLATFORM_FILE:
 491 #if defined(OS_POSIX) && !defined(OS_NACL)
 492         {
 493           base::ScopedFD file =
 494               TakeOrDupFile(static_cast<IPC::internal::PlatformFileAttachment*>(
 495                   attachment.get()));
 496           if (!file.is_valid()) {
 497             DPLOG(WARNING) << "Failed to dup FD to transmit.";
 498             set->CommitAll();
 499             return MOJO_RESULT_UNKNOWN;
 500           }
 501
 502           MojoHandle wrapped_handle;
 503           MojoResult wrap_result = CreatePlatformHandleWrapper(
 504               mojo::embedder::ScopedPlatformHandle(
 505                   mojo::embedder::PlatformHandle(file.release())),
 506               &wrapped_handle);
 507           if (MOJO_RESULT_OK != wrap_result) {
 508             LOG(WARNING) << "Pipe failed to wrap handles. Closing: "
 509                          << wrap_result;
 510             set->CommitAll();
 511             return wrap_result;
 512           }
 513
 514           handles->push_back(wrapped_handle);
 515         }
 516 #else
 517           NOTREACHED();
 518 #endif  //  defined(OS_POSIX) && !defined(OS_NACL)
 519         break;
 520         case MessageAttachment::TYPE_MOJO_HANDLE: {
 521           mojo::ScopedHandle handle =
 522               static_cast<IPC::internal::MojoHandleAttachment*>(
 523                   attachment.get())->TakeHandle();
 524           handles->push_back(handle.release().value());
 525         } break;
 526         case MessageAttachment::TYPE_BROKERABLE_ATTACHMENT:
 527           // Brokerable attachments are handled by the AttachmentBroker so
 528           // there's no need to do anything here.
 529           NOTREACHED();
 530           break;
 531       }
 532     }
 533
 534     set->CommitAll();
 535   }
 536
 537   return MOJO_RESULT_OK;
 538 }
 539
 540 // static
 541 MojoResult ChannelMojo::WriteToMessageAttachmentSet(
 542     const std::vector<MojoHandle>& handle_buffer,
 543     Message* message) {
 544   for (size_t i = 0; i < handle_buffer.size(); ++i) {
 545     bool ok = message->attachment_set()->AddAttachment(
 546         new IPC::internal::MojoHandleAttachment(
 547             mojo::MakeScopedHandle(mojo::Handle(handle_buffer[i]))));
 548     DCHECK(ok);
 549     if (!ok) {
 550       LOG(ERROR) << "Failed to add new Mojo handle.";
 551       return MOJO_RESULT_UNKNOWN;
 552     }
 553   }
 554
 555   return MOJO_RESULT_OK;
 556 }
 557
 558 }  // namespace IPC