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