mojo/edk/system/channel.cc

   1 // Copyright 2013 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 "mojo/edk/system/channel.h"
   6
   7 #include <algorithm>
   8
   9 #include "base/bind.h"
  10 #include "base/logging.h"
  11 #include "base/macros.h"
  12 #include "base/strings/stringprintf.h"
  13 #include "mojo/edk/embedder/platform_handle_vector.h"
  14 #include "mojo/edk/system/transport_data.h"
  15
  16 namespace mojo {
  17 namespace system {
  18
  19 namespace {
  20
  21 struct SerializedEndpoint {
  22   // This is the endpoint ID on the receiving side, and should be a "remote ID".
  23   // (The receiving side should already have had an endpoint attached and been
  24   // run via the |Channel|s. This endpoint will have both IDs assigned, so this
  25   // ID is only needed to associate that endpoint with a particular dispatcher.)
  26   ChannelEndpointId receiver_endpoint_id;
  27 };
  28
  29 }  // namespace
  30
  31 Channel::Channel(embedder::PlatformSupport* platform_support)
  32     : platform_support_(platform_support),
  33       is_running_(false),
  34       is_shutting_down_(false),
  35       channel_manager_(nullptr) {
  36 }
  37
  38 bool Channel::Init(scoped_ptr<RawChannel> raw_channel) {
  39   DCHECK(creation_thread_checker_.CalledOnValidThread());
  40   DCHECK(raw_channel);
  41
  42   // No need to take |lock_|, since this must be called before this object
  43   // becomes thread-safe.
  44   DCHECK(!is_running_);
  45   raw_channel_ = raw_channel.Pass();
  46
  47   if (!raw_channel_->Init(this)) {
  48     raw_channel_.reset();
  49     return false;
  50   }
  51
  52   is_running_ = true;
  53   return true;
  54 }
  55
  56 void Channel::SetChannelManager(ChannelManager* channel_manager) {
  57   DCHECK(channel_manager);
  58
  59   base::AutoLock locker(lock_);
  60   DCHECK(!is_shutting_down_);
  61   DCHECK(!channel_manager_);
  62   channel_manager_ = channel_manager;
  63 }
  64
  65 void Channel::Shutdown() {
  66   DCHECK(creation_thread_checker_.CalledOnValidThread());
  67
  68   IdToEndpointMap to_destroy;
  69   {
  70     base::AutoLock locker(lock_);
  71     if (!is_running_)
  72       return;
  73
  74     // Note: Don't reset |raw_channel_|, in case we're being called from within
  75     // |OnReadMessage()| or |OnError()|.
  76     raw_channel_->Shutdown();
  77     is_running_ = false;
  78
  79     // We need to deal with it outside the lock.
  80     std::swap(to_destroy, local_id_to_endpoint_map_);
  81   }
  82
  83   size_t num_live = 0;
  84   size_t num_zombies = 0;
  85   for (IdToEndpointMap::iterator it = to_destroy.begin();
  86        it != to_destroy.end(); ++it) {
  87     if (it->second) {
  88       num_live++;
  89       it->second->DetachFromChannel();
  90     } else {
  91       num_zombies++;
  92     }
  93   }
  94   DVLOG_IF(2, num_live || num_zombies) << "Shut down Channel with " << num_live
  95                                        << " live endpoints and " << num_zombies
  96                                        << " zombies";
  97 }
  98
  99 void Channel::WillShutdownSoon() {
 100   base::AutoLock locker(lock_);
 101   is_shutting_down_ = true;
 102   channel_manager_ = nullptr;
 103 }
 104
 105 void Channel::SetBootstrapEndpoint(scoped_refptr<ChannelEndpoint> endpoint) {
 106   DCHECK(endpoint);
 107
 108   // Used for both local and remote IDs.
 109   ChannelEndpointId bootstrap_id = ChannelEndpointId::GetBootstrap();
 110
 111   {
 112     base::AutoLock locker(lock_);
 113
 114     DLOG_IF(WARNING, is_shutting_down_)
 115         << "SetBootstrapEndpoint() while shutting down";
 116
 117     // Bootstrap endpoint should be the first.
 118     DCHECK(local_id_to_endpoint_map_.empty());
 119
 120     local_id_to_endpoint_map_[bootstrap_id] = endpoint;
 121   }
 122
 123   endpoint->AttachAndRun(this, bootstrap_id, bootstrap_id);
 124 }
 125
 126 bool Channel::WriteMessage(scoped_ptr<MessageInTransit> message) {
 127   base::AutoLock locker(lock_);
 128   if (!is_running_) {
 129     // TODO(vtl): I think this is probably not an error condition, but I should
 130     // think about it (and the shutdown sequence) more carefully.
 131     LOG(WARNING) << "WriteMessage() after shutdown";
 132     return false;
 133   }
 134
 135   DLOG_IF(WARNING, is_shutting_down_) << "WriteMessage() while shutting down";
 136   return raw_channel_->WriteMessage(message.Pass());
 137 }
 138
 139 bool Channel::IsWriteBufferEmpty() {
 140   base::AutoLock locker(lock_);
 141   if (!is_running_)
 142     return true;
 143   return raw_channel_->IsWriteBufferEmpty();
 144 }
 145
 146 void Channel::DetachEndpoint(ChannelEndpoint* endpoint,
 147                              ChannelEndpointId local_id,
 148                              ChannelEndpointId remote_id) {
 149   DCHECK(endpoint);
 150   DCHECK(local_id.is_valid());
 151
 152   if (!remote_id.is_valid())
 153     return;  // Nothing to do.
 154
 155   {
 156     base::AutoLock locker_(lock_);
 157     if (!is_running_)
 158       return;
 159
 160     IdToEndpointMap::iterator it = local_id_to_endpoint_map_.find(local_id);
 161     // We detach immediately if we receive a remove message, so it's possible
 162     // that the local ID is no longer in |local_id_to_endpoint_map_|, or even
 163     // that it's since been reused for another endpoint. In both cases, there's
 164     // nothing more to do.
 165     if (it == local_id_to_endpoint_map_.end() || it->second.get() != endpoint)
 166       return;
 167
 168     DCHECK(it->second);
 169     it->second = nullptr;
 170
 171     // Send a remove message outside the lock.
 172   }
 173
 174   if (!SendControlMessage(MessageInTransit::kSubtypeChannelRemoveEndpoint,
 175                           local_id, remote_id)) {
 176     HandleLocalError(base::StringPrintf(
 177         "Failed to send message to remove remote endpoint (local ID %u, remote "
 178         "ID %u)",
 179         static_cast<unsigned>(local_id.value()),
 180         static_cast<unsigned>(remote_id.value())));
 181   }
 182 }
 183
 184 size_t Channel::GetSerializedEndpointSize() const {
 185   return sizeof(SerializedEndpoint);
 186 }
 187
 188 void Channel::SerializeEndpoint(scoped_refptr<ChannelEndpoint> endpoint,
 189                                 void* destination) {
 190   SerializedEndpoint* s = static_cast<SerializedEndpoint*>(destination);
 191   s->receiver_endpoint_id = AttachAndRunEndpoint(endpoint);
 192   DVLOG(2) << "Serializing endpoint (remote ID = " << s->receiver_endpoint_id
 193            << ")";
 194 }
 195
 196 scoped_refptr<IncomingEndpoint> Channel::DeserializeEndpoint(
 197     const void* source) {
 198   const SerializedEndpoint* s = static_cast<const SerializedEndpoint*>(source);
 199   ChannelEndpointId local_id = s->receiver_endpoint_id;
 200   // No need to check the validity of |local_id| -- if it's not valid, it simply
 201   // won't be in |incoming_endpoints_|.
 202   DVLOG_IF(2, !local_id.is_valid() || !local_id.is_remote())
 203       << "Attempt to get incoming endpoint for invalid ID " << local_id;
 204
 205   base::AutoLock locker(lock_);
 206
 207   auto it = incoming_endpoints_.find(local_id);
 208   if (it == incoming_endpoints_.end()) {
 209     LOG(ERROR) << "Failed to deserialize endpoint (ID = " << local_id << ")";
 210     return nullptr;
 211   }
 212
 213   DVLOG(2) << "Deserializing endpoint (new local ID = " << local_id << ")";
 214
 215   scoped_refptr<IncomingEndpoint> rv;
 216   rv.swap(it->second);
 217   incoming_endpoints_.erase(it);
 218   return rv;
 219 }
 220
 221 size_t Channel::GetSerializedPlatformHandleSize() const {
 222   return raw_channel_->GetSerializedPlatformHandleSize();
 223 }
 224
 225 Channel::~Channel() {
 226   // The channel should have been shut down first.
 227   DCHECK(!is_running_);
 228 }
 229
 230 void Channel::OnReadMessage(
 231     const MessageInTransit::View& message_view,
 232     embedder::ScopedPlatformHandleVectorPtr platform_handles) {
 233   DCHECK(creation_thread_checker_.CalledOnValidThread());
 234
 235   switch (message_view.type()) {
 236     case MessageInTransit::kTypeEndpoint:
 237       OnReadMessageForEndpoint(message_view, platform_handles.Pass());
 238       break;
 239     case MessageInTransit::kTypeChannel:
 240       OnReadMessageForChannel(message_view, platform_handles.Pass());
 241       break;
 242     default:
 243       HandleRemoteError(
 244           base::StringPrintf("Received message of invalid type %u",
 245                              static_cast<unsigned>(message_view.type())));
 246       break;
 247   }
 248 }
 249
 250 void Channel::OnError(Error error) {
 251   DCHECK(creation_thread_checker_.CalledOnValidThread());
 252
 253   switch (error) {
 254     case ERROR_READ_SHUTDOWN:
 255       // The other side was cleanly closed, so this isn't actually an error.
 256       DVLOG(1) << "RawChannel read error (shutdown)";
 257       break;
 258     case ERROR_READ_BROKEN: {
 259       base::AutoLock locker(lock_);
 260       LOG_IF(ERROR, !is_shutting_down_)
 261           << "RawChannel read error (connection broken)";
 262       break;
 263     }
 264     case ERROR_READ_BAD_MESSAGE:
 265       // Receiving a bad message means either a bug, data corruption, or
 266       // malicious attack (probably due to some other bug).
 267       LOG(ERROR) << "RawChannel read error (received bad message)";
 268       break;
 269     case ERROR_READ_UNKNOWN:
 270       LOG(ERROR) << "RawChannel read error (unknown)";
 271       break;
 272     case ERROR_WRITE:
 273       // Write errors are slightly notable: they probably shouldn't happen under
 274       // normal operation (but maybe the other side crashed).
 275       LOG(WARNING) << "RawChannel write error";
 276       break;
 277   }
 278   Shutdown();
 279 }
 280
 281 void Channel::OnReadMessageForEndpoint(
 282     const MessageInTransit::View& message_view,
 283     embedder::ScopedPlatformHandleVectorPtr platform_handles) {
 284   DCHECK(creation_thread_checker_.CalledOnValidThread());
 285   DCHECK(message_view.type() == MessageInTransit::kTypeEndpoint);
 286
 287   ChannelEndpointId local_id = message_view.destination_id();
 288   if (!local_id.is_valid()) {
 289     HandleRemoteError("Received message with no destination ID");
 290     return;
 291   }
 292
 293   scoped_refptr<ChannelEndpoint> endpoint;
 294   {
 295     base::AutoLock locker(lock_);
 296
 297     // Since we own |raw_channel_|, and this method and |Shutdown()| should only
 298     // be called from the creation thread, |raw_channel_| should never be null
 299     // here.
 300     DCHECK(is_running_);
 301
 302     IdToEndpointMap::const_iterator it =
 303         local_id_to_endpoint_map_.find(local_id);
 304     if (it != local_id_to_endpoint_map_.end()) {
 305       // Ignore messages for zombie endpoints (not an error).
 306       if (!it->second) {
 307         DVLOG(2) << "Ignoring downstream message for zombie endpoint (local ID "
 308                     "= " << local_id
 309                  << ", remote ID = " << message_view.source_id() << ")";
 310         return;
 311       }
 312
 313       endpoint = it->second;
 314     }
 315   }
 316   if (!endpoint) {
 317     HandleRemoteError(base::StringPrintf(
 318         "Received a message for nonexistent local destination ID %u",
 319         static_cast<unsigned>(local_id.value())));
 320     // This is strongly indicative of some problem. However, it's not a fatal
 321     // error, since it may indicate a buggy (or hostile) remote process. Don't
 322     // die even for Debug builds, since handling this properly needs to be
 323     // tested (TODO(vtl)).
 324     DLOG(ERROR) << "This should not happen under normal operation.";
 325     return;
 326   }
 327
 328   scoped_ptr<MessageInTransit> message(new MessageInTransit(message_view));
 329   if (message_view.transport_data_buffer_size() > 0) {
 330     DCHECK(message_view.transport_data_buffer());
 331     message->SetDispatchers(TransportData::DeserializeDispatchers(
 332         message_view.transport_data_buffer(),
 333         message_view.transport_data_buffer_size(), platform_handles.Pass(),
 334         this));
 335   }
 336
 337   endpoint->OnReadMessage(message.Pass());
 338 }
 339
 340 void Channel::OnReadMessageForChannel(
 341     const MessageInTransit::View& message_view,
 342     embedder::ScopedPlatformHandleVectorPtr platform_handles) {
 343   DCHECK(creation_thread_checker_.CalledOnValidThread());
 344   DCHECK_EQ(message_view.type(), MessageInTransit::kTypeChannel);
 345
 346   // Currently, no channel messages take platform handles.
 347   if (platform_handles) {
 348     HandleRemoteError(
 349         "Received invalid channel message (has platform handles)");
 350     NOTREACHED();
 351     return;
 352   }
 353
 354   switch (message_view.subtype()) {
 355     case MessageInTransit::kSubtypeChannelAttachAndRunEndpoint:
 356       DVLOG(2) << "Handling channel message to attach and run endpoint (local "
 357                   "ID " << message_view.destination_id() << ", remote ID "
 358                << message_view.source_id() << ")";
 359       if (!OnAttachAndRunEndpoint(message_view.destination_id(),
 360                                   message_view.source_id())) {
 361         HandleRemoteError(
 362             "Received invalid channel message to attach and run endpoint");
 363       }
 364       break;
 365     case MessageInTransit::kSubtypeChannelRemoveEndpoint:
 366       DVLOG(2) << "Handling channel message to remove endpoint (local ID "
 367                << message_view.destination_id() << ", remote ID "
 368                << message_view.source_id() << ")";
 369       if (!OnRemoveEndpoint(message_view.destination_id(),
 370                             message_view.source_id())) {
 371         HandleRemoteError(
 372             "Received invalid channel message to remove endpoint");
 373       }
 374       break;
 375     case MessageInTransit::kSubtypeChannelRemoveEndpointAck:
 376       DVLOG(2) << "Handling channel message to ack remove endpoint (local ID "
 377                << message_view.destination_id() << ", remote ID "
 378                << message_view.source_id() << ")";
 379       if (!OnRemoveEndpointAck(message_view.destination_id())) {
 380         HandleRemoteError(
 381             "Received invalid channel message to ack remove endpoint");
 382       }
 383       break;
 384     default:
 385       HandleRemoteError("Received invalid channel message");
 386       NOTREACHED();
 387       break;
 388   }
 389 }
 390
 391 bool Channel::OnAttachAndRunEndpoint(ChannelEndpointId local_id,
 392                                      ChannelEndpointId remote_id) {
 393   // We should only get this for remotely-created local endpoints, so our local
 394   // ID should be "remote".
 395   if (!local_id.is_valid() || !local_id.is_remote()) {
 396     DVLOG(2) << "Received attach and run endpoint with invalid local ID";
 397     return false;
 398   }
 399
 400   // Conversely, the remote end should be "local".
 401   if (!remote_id.is_valid() || remote_id.is_remote()) {
 402     DVLOG(2) << "Received attach and run endpoint with invalid remote ID";
 403     return false;
 404   }
 405
 406   // Create/initialize an |IncomingEndpoint| and thus an endpoint (outside the
 407   // lock).
 408   scoped_refptr<IncomingEndpoint> incoming_endpoint(new IncomingEndpoint());
 409   scoped_refptr<ChannelEndpoint> endpoint = incoming_endpoint->Init();
 410
 411   bool success = true;
 412   {
 413     base::AutoLock locker(lock_);
 414
 415     if (local_id_to_endpoint_map_.find(local_id) ==
 416         local_id_to_endpoint_map_.end()) {
 417       DCHECK(incoming_endpoints_.find(local_id) == incoming_endpoints_.end());
 418
 419       // TODO(vtl): Use emplace when we move to C++11 unordered_maps. (It'll
 420       // avoid some refcount churn.)
 421       local_id_to_endpoint_map_[local_id] = endpoint;
 422       incoming_endpoints_[local_id] = incoming_endpoint;
 423     } else {
 424       // We need to call |Close()| outside the lock.
 425       success = false;
 426     }
 427   }
 428   if (!success) {
 429     DVLOG(2) << "Received attach and run endpoint for existing local ID";
 430     incoming_endpoint->Close();
 431     return false;
 432   }
 433
 434   endpoint->AttachAndRun(this, local_id, remote_id);
 435   return true;
 436 }
 437
 438 bool Channel::OnRemoveEndpoint(ChannelEndpointId local_id,
 439                                ChannelEndpointId remote_id) {
 440   DCHECK(creation_thread_checker_.CalledOnValidThread());
 441
 442   scoped_refptr<ChannelEndpoint> endpoint;
 443   {
 444     base::AutoLock locker(lock_);
 445
 446     IdToEndpointMap::iterator it = local_id_to_endpoint_map_.find(local_id);
 447     if (it == local_id_to_endpoint_map_.end()) {
 448       DVLOG(2) << "Remove endpoint error: not found";
 449       return false;
 450     }
 451
 452     if (!it->second) {
 453       // Remove messages "crossed"; we have to wait for the ack.
 454       return true;
 455     }
 456
 457     endpoint = it->second;
 458     local_id_to_endpoint_map_.erase(it);
 459     // Detach and send the remove ack message outside the lock.
 460   }
 461
 462   endpoint->DetachFromChannel();
 463
 464   if (!SendControlMessage(MessageInTransit::kSubtypeChannelRemoveEndpointAck,
 465                           local_id, remote_id)) {
 466     HandleLocalError(base::StringPrintf(
 467         "Failed to send message to ack remove remote endpoint (local ID %u, "
 468         "remote ID %u)",
 469         static_cast<unsigned>(local_id.value()),
 470         static_cast<unsigned>(remote_id.value())));
 471   }
 472
 473   return true;
 474 }
 475
 476 bool Channel::OnRemoveEndpointAck(ChannelEndpointId local_id) {
 477   DCHECK(creation_thread_checker_.CalledOnValidThread());
 478
 479   base::AutoLock locker(lock_);
 480
 481   IdToEndpointMap::iterator it = local_id_to_endpoint_map_.find(local_id);
 482   if (it == local_id_to_endpoint_map_.end()) {
 483     DVLOG(2) << "Remove endpoint ack error: not found";
 484     return false;
 485   }
 486
 487   if (it->second) {
 488     DVLOG(2) << "Remove endpoint ack error: wrong state";
 489     return false;
 490   }
 491
 492   local_id_to_endpoint_map_.erase(it);
 493   return true;
 494 }
 495
 496 void Channel::HandleRemoteError(const base::StringPiece& error_message) {
 497   // TODO(vtl): Is this how we really want to handle this? Probably we want to
 498   // terminate the connection, since it's spewing invalid stuff.
 499   LOG(WARNING) << error_message;
 500 }
 501
 502 void Channel::HandleLocalError(const base::StringPiece& error_message) {
 503   // TODO(vtl): Is this how we really want to handle this?
 504   // Sometimes we'll want to propagate the error back to the message pipe
 505   // (endpoint), and notify it that the remote is (effectively) closed.
 506   // Sometimes we'll want to kill the channel (and notify all the endpoints that
 507   // their remotes are dead.
 508   LOG(WARNING) << error_message;
 509 }
 510
 511 // Note: |endpoint| being a |scoped_refptr| makes this function safe, since it
 512 // keeps the endpoint alive even after the lock is released. Otherwise, there's
 513 // the temptation to simply pass the result of |new ChannelEndpoint(...)|
 514 // directly to this function, which wouldn't be sufficient for safety.
 515 ChannelEndpointId Channel::AttachAndRunEndpoint(
 516     scoped_refptr<ChannelEndpoint> endpoint) {
 517   DCHECK(endpoint);
 518
 519   ChannelEndpointId local_id;
 520   ChannelEndpointId remote_id;
 521   {
 522     base::AutoLock locker(lock_);
 523
 524     DLOG_IF(WARNING, is_shutting_down_)
 525         << "AttachAndRunEndpoint() while shutting down";
 526
 527     do {
 528       local_id = local_id_generator_.GetNext();
 529     } while (local_id_to_endpoint_map_.find(local_id) !=
 530              local_id_to_endpoint_map_.end());
 531
 532     // TODO(vtl): We also need to check for collisions of remote IDs here.
 533     remote_id = remote_id_generator_.GetNext();
 534
 535     local_id_to_endpoint_map_[local_id] = endpoint;
 536   }
 537
 538   if (!SendControlMessage(MessageInTransit::kSubtypeChannelAttachAndRunEndpoint,
 539                           local_id, remote_id)) {
 540     HandleLocalError(base::StringPrintf(
 541         "Failed to send message to run remote endpoint (local ID %u, remote ID "
 542         "%u)",
 543         static_cast<unsigned>(local_id.value()),
 544         static_cast<unsigned>(remote_id.value())));
 545     // TODO(vtl): Should we continue on to |AttachAndRun()|?
 546   }
 547
 548   endpoint->AttachAndRun(this, local_id, remote_id);
 549   return remote_id;
 550 }
 551
 552 bool Channel::SendControlMessage(MessageInTransit::Subtype subtype,
 553                                  ChannelEndpointId local_id,
 554                                  ChannelEndpointId remote_id) {
 555   DVLOG(2) << "Sending channel control message: subtype " << subtype
 556            << ", local ID " << local_id << ", remote ID " << remote_id;
 557   scoped_ptr<MessageInTransit> message(new MessageInTransit(
 558       MessageInTransit::kTypeChannel, subtype, 0, nullptr));
 559   message->set_source_id(local_id);
 560   message->set_destination_id(remote_id);
 561   return WriteMessage(message.Pass());
 562 }
 563
 564 }  // namespace system
 565 }  // namespace mojo