src/mojo/system/raw_channel.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 "mojo/system/raw_channel.h"
   6
   7 #include <string.h>
   8
   9 #include <algorithm>
  10
  11 #include "base/bind.h"
  12 #include "base/location.h"
  13 #include "base/logging.h"
  14 #include "base/message_loop/message_loop.h"
  15 #include "base/stl_util.h"
  16 #include "mojo/system/message_in_transit.h"
  17 #include "mojo/system/transport_data.h"
  18
  19 namespace mojo {
  20 namespace system {
  21
  22 const size_t kReadSize = 4096;
  23
  24 // RawChannel::ReadBuffer ------------------------------------------------------
  25
  26 RawChannel::ReadBuffer::ReadBuffer()
  27     : buffer_(kReadSize),
  28       num_valid_bytes_(0) {
  29 }
  30
  31 RawChannel::ReadBuffer::~ReadBuffer() {
  32 }
  33
  34 void RawChannel::ReadBuffer::GetBuffer(char** addr, size_t* size) {
  35   DCHECK_GE(buffer_.size(), num_valid_bytes_ + kReadSize);
  36   *addr = &buffer_[0] + num_valid_bytes_;
  37   *size = kReadSize;
  38 }
  39
  40 // RawChannel::WriteBuffer -----------------------------------------------------
  41
  42 RawChannel::WriteBuffer::WriteBuffer(size_t serialized_platform_handle_size)
  43     : serialized_platform_handle_size_(serialized_platform_handle_size),
  44       platform_handles_offset_(0),
  45       data_offset_(0) {
  46 }
  47
  48 RawChannel::WriteBuffer::~WriteBuffer() {
  49   STLDeleteElements(&message_queue_);
  50 }
  51
  52 bool RawChannel::WriteBuffer::HavePlatformHandlesToSend() const {
  53   if (message_queue_.empty())
  54     return false;
  55
  56   const TransportData* transport_data =
  57       message_queue_.front()->transport_data();
  58   if (!transport_data)
  59     return false;
  60
  61   const embedder::PlatformHandleVector* all_platform_handles =
  62       transport_data->platform_handles();
  63   if (!all_platform_handles) {
  64     DCHECK_EQ(platform_handles_offset_, 0u);
  65     return false;
  66   }
  67   if (platform_handles_offset_ >= all_platform_handles->size()) {
  68     DCHECK_EQ(platform_handles_offset_, all_platform_handles->size());
  69     return false;
  70   }
  71
  72   return true;
  73 }
  74
  75 void RawChannel::WriteBuffer::GetPlatformHandlesToSend(
  76     size_t* num_platform_handles,
  77     embedder::PlatformHandle** platform_handles,
  78     void** serialization_data) {
  79   DCHECK(HavePlatformHandlesToSend());
  80
  81   TransportData* transport_data = message_queue_.front()->transport_data();
  82   embedder::PlatformHandleVector* all_platform_handles =
  83       transport_data->platform_handles();
  84   *num_platform_handles =
  85       all_platform_handles->size() - platform_handles_offset_;
  86   *platform_handles = &(*all_platform_handles)[platform_handles_offset_];
  87   size_t serialization_data_offset =
  88       transport_data->platform_handle_table_offset();
  89   DCHECK_GT(serialization_data_offset, 0u);
  90   serialization_data_offset +=
  91       platform_handles_offset_ * serialized_platform_handle_size_;
  92   *serialization_data =
  93       static_cast<char*>(transport_data->buffer()) + serialization_data_offset;
  94 }
  95
  96 void RawChannel::WriteBuffer::GetBuffers(std::vector<Buffer>* buffers) const {
  97   buffers->clear();
  98
  99   if (message_queue_.empty())
 100     return;
 101
 102   MessageInTransit* message = message_queue_.front();
 103   DCHECK_LT(data_offset_, message->total_size());
 104   size_t bytes_to_write = message->total_size() - data_offset_;
 105
 106   size_t transport_data_buffer_size = message->transport_data() ?
 107       message->transport_data()->buffer_size() : 0;
 108
 109   if (!transport_data_buffer_size) {
 110     // Only write from the main buffer.
 111     DCHECK_LT(data_offset_, message->main_buffer_size());
 112     DCHECK_LE(bytes_to_write, message->main_buffer_size());
 113     Buffer buffer = {
 114         static_cast<const char*>(message->main_buffer()) + data_offset_,
 115         bytes_to_write};
 116     buffers->push_back(buffer);
 117     return;
 118   }
 119
 120   if (data_offset_ >= message->main_buffer_size()) {
 121     // Only write from the transport data buffer.
 122     DCHECK_LT(data_offset_ - message->main_buffer_size(),
 123               transport_data_buffer_size);
 124     DCHECK_LE(bytes_to_write, transport_data_buffer_size);
 125     Buffer buffer = {
 126         static_cast<const char*>(message->transport_data()->buffer()) +
 127             (data_offset_ - message->main_buffer_size()),
 128         bytes_to_write};
 129     buffers->push_back(buffer);
 130     return;
 131   }
 132
 133   // TODO(vtl): We could actually send out buffers from multiple messages, with
 134   // the "stopping" condition being reaching a message with platform handles
 135   // attached.
 136
 137   // Write from both buffers.
 138   DCHECK_EQ(bytes_to_write, message->main_buffer_size() - data_offset_ +
 139                                 transport_data_buffer_size);
 140   Buffer buffer1 = {
 141     static_cast<const char*>(message->main_buffer()) + data_offset_,
 142     message->main_buffer_size() - data_offset_
 143   };
 144   buffers->push_back(buffer1);
 145   Buffer buffer2 = {
 146     static_cast<const char*>(message->transport_data()->buffer()),
 147     transport_data_buffer_size
 148   };
 149   buffers->push_back(buffer2);
 150 }
 151
 152 // RawChannel ------------------------------------------------------------------
 153
 154 RawChannel::RawChannel()
 155     : message_loop_for_io_(NULL),
 156       delegate_(NULL),
 157       read_stopped_(false),
 158       write_stopped_(false),
 159       weak_ptr_factory_(this) {
 160 }
 161
 162 RawChannel::~RawChannel() {
 163   DCHECK(!read_buffer_);
 164   DCHECK(!write_buffer_);
 165
 166   // No need to take the |write_lock_| here -- if there are still weak pointers
 167   // outstanding, then we're hosed anyway (since we wouldn't be able to
 168   // invalidate them cleanly, since we might not be on the I/O thread).
 169   DCHECK(!weak_ptr_factory_.HasWeakPtrs());
 170 }
 171
 172 bool RawChannel::Init(Delegate* delegate) {
 173   DCHECK(delegate);
 174
 175   DCHECK(!delegate_);
 176   delegate_ = delegate;
 177
 178   CHECK_EQ(base::MessageLoop::current()->type(), base::MessageLoop::TYPE_IO);
 179   DCHECK(!message_loop_for_io_);
 180   message_loop_for_io_ =
 181       static_cast<base::MessageLoopForIO*>(base::MessageLoop::current());
 182
 183   // No need to take the lock. No one should be using us yet.
 184   DCHECK(!read_buffer_);
 185   read_buffer_.reset(new ReadBuffer);
 186   DCHECK(!write_buffer_);
 187   write_buffer_.reset(new WriteBuffer(GetSerializedPlatformHandleSize()));
 188
 189   if (!OnInit()) {
 190     delegate_ = NULL;
 191     message_loop_for_io_ = NULL;
 192     read_buffer_.reset();
 193     write_buffer_.reset();
 194     return false;
 195   }
 196
 197   if (ScheduleRead() != IO_PENDING) {
 198     // This will notify the delegate about the read failure. Although we're on
 199     // the I/O thread, don't call it in the nested context.
 200     message_loop_for_io_->PostTask(
 201         FROM_HERE,
 202         base::Bind(&RawChannel::OnReadCompleted, weak_ptr_factory_.GetWeakPtr(),
 203                    false, 0));
 204   }
 205
 206   // ScheduleRead() failure is treated as a read failure (by notifying the
 207   // delegate), not as an init failure.
 208   return true;
 209 }
 210
 211 void RawChannel::Shutdown() {
 212   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
 213
 214   base::AutoLock locker(write_lock_);
 215
 216   LOG_IF(WARNING, !write_buffer_->message_queue_.empty())
 217       << "Shutting down RawChannel with write buffer nonempty";
 218
 219   // Reset the delegate so that it won't receive further calls.
 220   delegate_ = NULL;
 221   read_stopped_ = true;
 222   write_stopped_ = true;
 223   weak_ptr_factory_.InvalidateWeakPtrs();
 224
 225   OnShutdownNoLock(read_buffer_.Pass(), write_buffer_.Pass());
 226 }
 227
 228 // Reminder: This must be thread-safe.
 229 bool RawChannel::WriteMessage(scoped_ptr<MessageInTransit> message) {
 230   DCHECK(message);
 231
 232   base::AutoLock locker(write_lock_);
 233   if (write_stopped_)
 234     return false;
 235
 236   if (!write_buffer_->message_queue_.empty()) {
 237     EnqueueMessageNoLock(message.Pass());
 238     return true;
 239   }
 240
 241   EnqueueMessageNoLock(message.Pass());
 242   DCHECK_EQ(write_buffer_->data_offset_, 0u);
 243
 244   size_t platform_handles_written = 0;
 245   size_t bytes_written = 0;
 246   IOResult io_result = WriteNoLock(&platform_handles_written, &bytes_written);
 247   if (io_result == IO_PENDING)
 248     return true;
 249
 250   bool result = OnWriteCompletedNoLock(io_result == IO_SUCCEEDED,
 251                                        platform_handles_written,
 252                                        bytes_written);
 253   if (!result) {
 254     // Even if we're on the I/O thread, don't call |OnFatalError()| in the
 255     // nested context.
 256     message_loop_for_io_->PostTask(
 257         FROM_HERE,
 258         base::Bind(&RawChannel::CallOnFatalError,
 259                    weak_ptr_factory_.GetWeakPtr(),
 260                    Delegate::FATAL_ERROR_WRITE));
 261   }
 262
 263   return result;
 264 }
 265
 266 // Reminder: This must be thread-safe.
 267 bool RawChannel::IsWriteBufferEmpty() {
 268   base::AutoLock locker(write_lock_);
 269   return write_buffer_->message_queue_.empty();
 270 }
 271
 272 void RawChannel::OnReadCompleted(bool result, size_t bytes_read) {
 273   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
 274
 275   if (read_stopped_) {
 276     NOTREACHED();
 277     return;
 278   }
 279
 280   IOResult io_result = result ? IO_SUCCEEDED : IO_FAILED;
 281
 282   // Keep reading data in a loop, and dispatch messages if enough data is
 283   // received. Exit the loop if any of the following happens:
 284   //   - one or more messages were dispatched;
 285   //   - the last read failed, was a partial read or would block;
 286   //   - |Shutdown()| was called.
 287   do {
 288     if (io_result != IO_SUCCEEDED) {
 289       read_stopped_ = true;
 290       CallOnFatalError(Delegate::FATAL_ERROR_READ);
 291       return;
 292     }
 293
 294     read_buffer_->num_valid_bytes_ += bytes_read;
 295
 296     // Dispatch all the messages that we can.
 297     bool did_dispatch_message = false;
 298     // Tracks the offset of the first undispatched message in |read_buffer_|.
 299     // Currently, we copy data to ensure that this is zero at the beginning.
 300     size_t read_buffer_start = 0;
 301     size_t remaining_bytes = read_buffer_->num_valid_bytes_;
 302     size_t message_size;
 303     // Note that we rely on short-circuit evaluation here:
 304     //   - |read_buffer_start| may be an invalid index into
 305     //     |read_buffer_->buffer_| if |remaining_bytes| is zero.
 306     //   - |message_size| is only valid if |GetNextMessageSize()| returns true.
 307     // TODO(vtl): Use |message_size| more intelligently (e.g., to request the
 308     // next read).
 309     // TODO(vtl): Validate that |message_size| is sane.
 310     while (remaining_bytes > 0 &&
 311            MessageInTransit::GetNextMessageSize(
 312                &read_buffer_->buffer_[read_buffer_start], remaining_bytes,
 313                &message_size) &&
 314            remaining_bytes >= message_size) {
 315       MessageInTransit::View
 316           message_view(message_size, &read_buffer_->buffer_[read_buffer_start]);
 317       DCHECK_EQ(message_view.total_size(), message_size);
 318
 319       const char* error_message = NULL;
 320       if (!message_view.IsValid(GetSerializedPlatformHandleSize(),
 321                                 &error_message)) {
 322         DCHECK(error_message);
 323         LOG(WARNING) << "Received invalid message: " << error_message;
 324         read_stopped_ = true;
 325         CallOnFatalError(Delegate::FATAL_ERROR_READ);
 326         return;
 327       }
 328
 329       if (message_view.type() == MessageInTransit::kTypeRawChannel) {
 330         if (!OnReadMessageForRawChannel(message_view)) {
 331           read_stopped_ = true;
 332           CallOnFatalError(Delegate::FATAL_ERROR_READ);
 333           return;
 334         }
 335       } else {
 336         embedder::ScopedPlatformHandleVectorPtr platform_handles;
 337         if (message_view.transport_data_buffer()) {
 338           size_t num_platform_handles;
 339           const void* platform_handle_table;
 340           TransportData::GetPlatformHandleTable(
 341               message_view.transport_data_buffer(),
 342               &num_platform_handles,
 343               &platform_handle_table);
 344
 345           if (num_platform_handles > 0) {
 346             platform_handles =
 347                 GetReadPlatformHandles(num_platform_handles,
 348                                        platform_handle_table).Pass();
 349             if (!platform_handles) {
 350               LOG(WARNING) << "Invalid number of platform handles received";
 351               read_stopped_ = true;
 352               CallOnFatalError(Delegate::FATAL_ERROR_READ);
 353               return;
 354             }
 355           }
 356         }
 357
 358         // TODO(vtl): In the case that we aren't expecting any platform handles,
 359         // for the POSIX implementation, we should confirm that none are stored.
 360
 361         // Dispatch the message.
 362         DCHECK(delegate_);
 363         delegate_->OnReadMessage(message_view, platform_handles.Pass());
 364         if (read_stopped_) {
 365           // |Shutdown()| was called in |OnReadMessage()|.
 366           // TODO(vtl): Add test for this case.
 367           return;
 368         }
 369       }
 370
 371       did_dispatch_message = true;
 372
 373       // Update our state.
 374       read_buffer_start += message_size;
 375       remaining_bytes -= message_size;
 376     }
 377
 378     if (read_buffer_start > 0) {
 379       // Move data back to start.
 380       read_buffer_->num_valid_bytes_ = remaining_bytes;
 381       if (read_buffer_->num_valid_bytes_ > 0) {
 382         memmove(&read_buffer_->buffer_[0],
 383                 &read_buffer_->buffer_[read_buffer_start], remaining_bytes);
 384       }
 385       read_buffer_start = 0;
 386     }
 387
 388     if (read_buffer_->buffer_.size() - read_buffer_->num_valid_bytes_ <
 389             kReadSize) {
 390       // Use power-of-2 buffer sizes.
 391       // TODO(vtl): Make sure the buffer doesn't get too large (and enforce the
 392       // maximum message size to whatever extent necessary).
 393       // TODO(vtl): We may often be able to peek at the header and get the real
 394       // required extra space (which may be much bigger than |kReadSize|).
 395       size_t new_size = std::max(read_buffer_->buffer_.size(), kReadSize);
 396       while (new_size < read_buffer_->num_valid_bytes_ + kReadSize)
 397         new_size *= 2;
 398
 399       // TODO(vtl): It's suboptimal to zero out the fresh memory.
 400       read_buffer_->buffer_.resize(new_size, 0);
 401     }
 402
 403     // (1) If we dispatched any messages, stop reading for now (and let the
 404     // message loop do its thing for another round).
 405     // TODO(vtl): Is this the behavior we want? (Alternatives: i. Dispatch only
 406     // a single message. Risks: slower, more complex if we want to avoid lots of
 407     // copying. ii. Keep reading until there's no more data and dispatch all the
 408     // messages we can. Risks: starvation of other users of the message loop.)
 409     // (2) If we didn't max out |kReadSize|, stop reading for now.
 410     bool schedule_for_later = did_dispatch_message || bytes_read < kReadSize;
 411     bytes_read = 0;
 412     io_result = schedule_for_later ? ScheduleRead() : Read(&bytes_read);
 413   } while (io_result != IO_PENDING);
 414 }
 415
 416 void RawChannel::OnWriteCompleted(bool result,
 417                                   size_t platform_handles_written,
 418                                   size_t bytes_written) {
 419   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
 420
 421   bool did_fail = false;
 422   {
 423     base::AutoLock locker(write_lock_);
 424     DCHECK_EQ(write_stopped_, write_buffer_->message_queue_.empty());
 425
 426     if (write_stopped_) {
 427       NOTREACHED();
 428       return;
 429     }
 430
 431     did_fail = !OnWriteCompletedNoLock(result,
 432                                        platform_handles_written,
 433                                        bytes_written);
 434   }
 435
 436   if (did_fail)
 437     CallOnFatalError(Delegate::FATAL_ERROR_WRITE);
 438 }
 439
 440 void RawChannel::EnqueueMessageNoLock(scoped_ptr<MessageInTransit> message) {
 441   write_lock_.AssertAcquired();
 442   write_buffer_->message_queue_.push_back(message.release());
 443 }
 444
 445 bool RawChannel::OnReadMessageForRawChannel(
 446     const MessageInTransit::View& message_view) {
 447   // No non-implementation specific |RawChannel| control messages.
 448   LOG(ERROR) << "Invalid control message (subtype " << message_view.subtype()
 449              << ")";
 450   return false;
 451 }
 452
 453 void RawChannel::CallOnFatalError(Delegate::FatalError fatal_error) {
 454   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
 455   // TODO(vtl): Add a "write_lock_.AssertNotAcquired()"?
 456   if (delegate_)
 457     delegate_->OnFatalError(fatal_error);
 458 }
 459
 460 bool RawChannel::OnWriteCompletedNoLock(bool result,
 461                                         size_t platform_handles_written,
 462                                         size_t bytes_written) {
 463   write_lock_.AssertAcquired();
 464
 465   DCHECK(!write_stopped_);
 466   DCHECK(!write_buffer_->message_queue_.empty());
 467
 468   if (result) {
 469     write_buffer_->platform_handles_offset_ += platform_handles_written;
 470     write_buffer_->data_offset_ += bytes_written;
 471
 472     MessageInTransit* message = write_buffer_->message_queue_.front();
 473     if (write_buffer_->data_offset_ >= message->total_size()) {
 474       // Complete write.
 475       DCHECK_EQ(write_buffer_->data_offset_, message->total_size());
 476       write_buffer_->message_queue_.pop_front();
 477       delete message;
 478       write_buffer_->platform_handles_offset_ = 0;
 479       write_buffer_->data_offset_ = 0;
 480
 481       if (write_buffer_->message_queue_.empty())
 482         return true;
 483     }
 484
 485     // Schedule the next write.
 486     IOResult io_result = ScheduleWriteNoLock();
 487     if (io_result == IO_PENDING)
 488       return true;
 489     DCHECK_EQ(io_result, IO_FAILED);
 490   }
 491
 492   write_stopped_ = true;
 493   STLDeleteElements(&write_buffer_->message_queue_);
 494   write_buffer_->platform_handles_offset_ = 0;
 495   write_buffer_->data_offset_ = 0;
 496   return false;
 497 }
 498
 499 }  // namespace system
 500 }  // namespace mojo