Home | History | Annotate | Download | only in system 
      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/data_pipe_producer_dispatcher.h"
      6 
      7 #include <stddef.h>
      8 #include <stdint.h>
      9 
     10 #include <utility>
     11 
     12 #include "base/bind.h"
     13 #include "base/logging.h"
     14 #include "base/memory/ref_counted.h"
     15 #include "base/message_loop/message_loop.h"
     16 #include "mojo/edk/embedder/embedder_internal.h"
     17 #include "mojo/edk/embedder/platform_shared_buffer.h"
     18 #include "mojo/edk/system/configuration.h"
     19 #include "mojo/edk/system/core.h"
     20 #include "mojo/edk/system/data_pipe_control_message.h"
     21 #include "mojo/edk/system/node_controller.h"
     22 #include "mojo/edk/system/ports_message.h"
     23 #include "mojo/edk/system/request_context.h"
     24 #include "mojo/public/c/system/data_pipe.h"
     25 
     26 namespace mojo {
     27 namespace edk {
     28 
     29 namespace {
     30 
     31 const uint8_t kFlagPeerClosed = 0x01;
     32 
     33 #pragma pack(push, 1)
     34 
     35 struct SerializedState {
     36   MojoCreateDataPipeOptions options;
     37   uint64_t pipe_id;
     38   uint32_t write_offset;
     39   uint32_t available_capacity;
     40   uint8_t flags;
     41   char padding[7];
     42 };
     43 
     44 static_assert(sizeof(SerializedState) % 8 == 0,
     45               "Invalid SerializedState size.");
     46 
     47 #pragma pack(pop)
     48 
     49 }  // namespace
     50 
     51 // A PortObserver which forwards to a DataPipeProducerDispatcher. This owns a
     52 // reference to the dispatcher to ensure it lives as long as the observed port.
     53 class DataPipeProducerDispatcher::PortObserverThunk
     54     : public NodeController::PortObserver {
     55  public:
     56   explicit PortObserverThunk(
     57       scoped_refptr<DataPipeProducerDispatcher> dispatcher)
     58       : dispatcher_(dispatcher) {}
     59 
     60  private:
     61   ~PortObserverThunk() override {}
     62 
     63   // NodeController::PortObserver:
     64   void OnPortStatusChanged() override { dispatcher_->OnPortStatusChanged(); }
     65 
     66   scoped_refptr<DataPipeProducerDispatcher> dispatcher_;
     67 
     68   DISALLOW_COPY_AND_ASSIGN(PortObserverThunk);
     69 };
     70 
     71 DataPipeProducerDispatcher::DataPipeProducerDispatcher(
     72     NodeController* node_controller,
     73     const ports::PortRef& control_port,
     74     scoped_refptr<PlatformSharedBuffer> shared_ring_buffer,
     75     const MojoCreateDataPipeOptions& options,
     76     bool initialized,
     77     uint64_t pipe_id)
     78     : options_(options),
     79       node_controller_(node_controller),
     80       control_port_(control_port),
     81       pipe_id_(pipe_id),
     82       shared_ring_buffer_(shared_ring_buffer),
     83       available_capacity_(options_.capacity_num_bytes) {
     84   if (initialized) {
     85     base::AutoLock lock(lock_);
     86     InitializeNoLock();
     87   }
     88 }
     89 
     90 Dispatcher::Type DataPipeProducerDispatcher::GetType() const {
     91   return Type::DATA_PIPE_PRODUCER;
     92 }
     93 
     94 MojoResult DataPipeProducerDispatcher::Close() {
     95   base::AutoLock lock(lock_);
     96   DVLOG(1) << "Closing data pipe producer " << pipe_id_;
     97   return CloseNoLock();
     98 }
     99 
    100 MojoResult DataPipeProducerDispatcher::Watch(
    101     MojoHandleSignals signals,
    102     const Watcher::WatchCallback& callback,
    103     uintptr_t context) {
    104   base::AutoLock lock(lock_);
    105 
    106   if (is_closed_ || in_transit_)
    107     return MOJO_RESULT_INVALID_ARGUMENT;
    108 
    109   return awakable_list_.AddWatcher(
    110       signals, callback, context, GetHandleSignalsStateNoLock());
    111 }
    112 
    113 MojoResult DataPipeProducerDispatcher::CancelWatch(uintptr_t context) {
    114   base::AutoLock lock(lock_);
    115 
    116   if (is_closed_ || in_transit_)
    117     return MOJO_RESULT_INVALID_ARGUMENT;
    118 
    119   return awakable_list_.RemoveWatcher(context);
    120 }
    121 
    122 MojoResult DataPipeProducerDispatcher::WriteData(const void* elements,
    123                                                  uint32_t* num_bytes,
    124                                                  MojoWriteDataFlags flags) {
    125   base::AutoLock lock(lock_);
    126   if (!shared_ring_buffer_ || in_transit_)
    127     return MOJO_RESULT_INVALID_ARGUMENT;
    128 
    129   if (in_two_phase_write_)
    130     return MOJO_RESULT_BUSY;
    131 
    132   if (peer_closed_)
    133     return MOJO_RESULT_FAILED_PRECONDITION;
    134 
    135   if (*num_bytes % options_.element_num_bytes != 0)
    136     return MOJO_RESULT_INVALID_ARGUMENT;
    137   if (*num_bytes == 0)
    138     return MOJO_RESULT_OK;  // Nothing to do.
    139 
    140   if ((flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) &&
    141       (*num_bytes > available_capacity_)) {
    142     // Don't return "should wait" since you can't wait for a specified amount of
    143     // data.
    144     return MOJO_RESULT_OUT_OF_RANGE;
    145   }
    146 
    147   DCHECK_LE(available_capacity_, options_.capacity_num_bytes);
    148   uint32_t num_bytes_to_write = std::min(*num_bytes, available_capacity_);
    149   if (num_bytes_to_write == 0)
    150     return MOJO_RESULT_SHOULD_WAIT;
    151 
    152   HandleSignalsState old_state = GetHandleSignalsStateNoLock();
    153 
    154   *num_bytes = num_bytes_to_write;
    155 
    156   CHECK(ring_buffer_mapping_);
    157   uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
    158   CHECK(data);
    159 
    160   const uint8_t* source = static_cast<const uint8_t*>(elements);
    161   CHECK(source);
    162 
    163   DCHECK_LE(write_offset_, options_.capacity_num_bytes);
    164   uint32_t tail_bytes_to_write =
    165       std::min(options_.capacity_num_bytes - write_offset_,
    166                num_bytes_to_write);
    167   uint32_t head_bytes_to_write = num_bytes_to_write - tail_bytes_to_write;
    168 
    169   DCHECK_GT(tail_bytes_to_write, 0u);
    170   memcpy(data + write_offset_, source, tail_bytes_to_write);
    171   if (head_bytes_to_write > 0)
    172     memcpy(data, source + tail_bytes_to_write, head_bytes_to_write);
    173 
    174   DCHECK_LE(num_bytes_to_write, available_capacity_);
    175   available_capacity_ -= num_bytes_to_write;
    176   write_offset_ = (write_offset_ + num_bytes_to_write) %
    177       options_.capacity_num_bytes;
    178 
    179   HandleSignalsState new_state = GetHandleSignalsStateNoLock();
    180   if (!new_state.equals(old_state))
    181     awakable_list_.AwakeForStateChange(new_state);
    182 
    183   base::AutoUnlock unlock(lock_);
    184   NotifyWrite(num_bytes_to_write);
    185 
    186   return MOJO_RESULT_OK;
    187 }
    188 
    189 MojoResult DataPipeProducerDispatcher::BeginWriteData(
    190     void** buffer,
    191     uint32_t* buffer_num_bytes,
    192     MojoWriteDataFlags flags) {
    193   base::AutoLock lock(lock_);
    194   if (!shared_ring_buffer_ || in_transit_)
    195     return MOJO_RESULT_INVALID_ARGUMENT;
    196   if (in_two_phase_write_)
    197     return MOJO_RESULT_BUSY;
    198   if (peer_closed_)
    199     return MOJO_RESULT_FAILED_PRECONDITION;
    200 
    201   if (available_capacity_ == 0) {
    202     return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
    203                         : MOJO_RESULT_SHOULD_WAIT;
    204   }
    205 
    206   in_two_phase_write_ = true;
    207   *buffer_num_bytes = std::min(options_.capacity_num_bytes - write_offset_,
    208                                available_capacity_);
    209   DCHECK_GT(*buffer_num_bytes, 0u);
    210 
    211   CHECK(ring_buffer_mapping_);
    212   uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
    213   *buffer = data + write_offset_;
    214 
    215   return MOJO_RESULT_OK;
    216 }
    217 
    218 MojoResult DataPipeProducerDispatcher::EndWriteData(
    219     uint32_t num_bytes_written) {
    220   base::AutoLock lock(lock_);
    221   if (is_closed_ || in_transit_)
    222     return MOJO_RESULT_INVALID_ARGUMENT;
    223 
    224   if (!in_two_phase_write_)
    225     return MOJO_RESULT_FAILED_PRECONDITION;
    226 
    227   DCHECK(shared_ring_buffer_);
    228   DCHECK(ring_buffer_mapping_);
    229 
    230   // Note: Allow successful completion of the two-phase write even if the other
    231   // side has been closed.
    232   MojoResult rv = MOJO_RESULT_OK;
    233   if (num_bytes_written > available_capacity_ ||
    234       num_bytes_written % options_.element_num_bytes != 0 ||
    235       write_offset_ + num_bytes_written > options_.capacity_num_bytes) {
    236     rv = MOJO_RESULT_INVALID_ARGUMENT;
    237   } else {
    238     DCHECK_LE(num_bytes_written + write_offset_, options_.capacity_num_bytes);
    239     available_capacity_ -= num_bytes_written;
    240     write_offset_ = (write_offset_ + num_bytes_written) %
    241         options_.capacity_num_bytes;
    242 
    243     base::AutoUnlock unlock(lock_);
    244     NotifyWrite(num_bytes_written);
    245   }
    246 
    247   in_two_phase_write_ = false;
    248 
    249   // If we're now writable, we *became* writable (since we weren't writable
    250   // during the two-phase write), so awake producer awakables.
    251   HandleSignalsState new_state = GetHandleSignalsStateNoLock();
    252   if (new_state.satisfies(MOJO_HANDLE_SIGNAL_WRITABLE))
    253     awakable_list_.AwakeForStateChange(new_state);
    254 
    255   return rv;
    256 }
    257 
    258 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsState() const {
    259   base::AutoLock lock(lock_);
    260   return GetHandleSignalsStateNoLock();
    261 }
    262 
    263 MojoResult DataPipeProducerDispatcher::AddAwakable(
    264     Awakable* awakable,
    265     MojoHandleSignals signals,
    266     uintptr_t context,
    267     HandleSignalsState* signals_state) {
    268   base::AutoLock lock(lock_);
    269   if (!shared_ring_buffer_ || in_transit_) {
    270     if (signals_state)
    271       *signals_state = HandleSignalsState();
    272     return MOJO_RESULT_INVALID_ARGUMENT;
    273   }
    274   UpdateSignalsStateNoLock();
    275   HandleSignalsState state = GetHandleSignalsStateNoLock();
    276   if (state.satisfies(signals)) {
    277     if (signals_state)
    278       *signals_state = state;
    279     return MOJO_RESULT_ALREADY_EXISTS;
    280   }
    281   if (!state.can_satisfy(signals)) {
    282     if (signals_state)
    283       *signals_state = state;
    284     return MOJO_RESULT_FAILED_PRECONDITION;
    285   }
    286 
    287   awakable_list_.Add(awakable, signals, context);
    288   return MOJO_RESULT_OK;
    289 }
    290 
    291 void DataPipeProducerDispatcher::RemoveAwakable(
    292     Awakable* awakable,
    293     HandleSignalsState* signals_state) {
    294   base::AutoLock lock(lock_);
    295   if ((!shared_ring_buffer_ || in_transit_) && signals_state)
    296     *signals_state = HandleSignalsState();
    297   else if (signals_state)
    298     *signals_state = GetHandleSignalsStateNoLock();
    299   awakable_list_.Remove(awakable);
    300 }
    301 
    302 void DataPipeProducerDispatcher::StartSerialize(uint32_t* num_bytes,
    303                                                 uint32_t* num_ports,
    304                                                 uint32_t* num_handles) {
    305   base::AutoLock lock(lock_);
    306   DCHECK(in_transit_);
    307   *num_bytes = sizeof(SerializedState);
    308   *num_ports = 1;
    309   *num_handles = 1;
    310 }
    311 
    312 bool DataPipeProducerDispatcher::EndSerialize(
    313     void* destination,
    314     ports::PortName* ports,
    315     PlatformHandle* platform_handles) {
    316   SerializedState* state = static_cast<SerializedState*>(destination);
    317   memcpy(&state->options, &options_, sizeof(MojoCreateDataPipeOptions));
    318   memset(state->padding, 0, sizeof(state->padding));
    319 
    320   base::AutoLock lock(lock_);
    321   DCHECK(in_transit_);
    322   state->pipe_id = pipe_id_;
    323   state->write_offset = write_offset_;
    324   state->available_capacity = available_capacity_;
    325   state->flags = peer_closed_ ? kFlagPeerClosed : 0;
    326 
    327   ports[0] = control_port_.name();
    328 
    329   buffer_handle_for_transit_ = shared_ring_buffer_->DuplicatePlatformHandle();
    330   platform_handles[0] = buffer_handle_for_transit_.get();
    331 
    332   return true;
    333 }
    334 
    335 bool DataPipeProducerDispatcher::BeginTransit() {
    336   base::AutoLock lock(lock_);
    337   if (in_transit_)
    338     return false;
    339   in_transit_ = !in_two_phase_write_;
    340   return in_transit_;
    341 }
    342 
    343 void DataPipeProducerDispatcher::CompleteTransitAndClose() {
    344   node_controller_->SetPortObserver(control_port_, nullptr);
    345 
    346   base::AutoLock lock(lock_);
    347   DCHECK(in_transit_);
    348   transferred_ = true;
    349   in_transit_ = false;
    350   ignore_result(buffer_handle_for_transit_.release());
    351   CloseNoLock();
    352 }
    353 
    354 void DataPipeProducerDispatcher::CancelTransit() {
    355   base::AutoLock lock(lock_);
    356   DCHECK(in_transit_);
    357   in_transit_ = false;
    358   buffer_handle_for_transit_.reset();
    359   awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
    360 }
    361 
    362 // static
    363 scoped_refptr<DataPipeProducerDispatcher>
    364 DataPipeProducerDispatcher::Deserialize(const void* data,
    365                                         size_t num_bytes,
    366                                         const ports::PortName* ports,
    367                                         size_t num_ports,
    368                                         PlatformHandle* handles,
    369                                         size_t num_handles) {
    370   if (num_ports != 1 || num_handles != 1 ||
    371       num_bytes != sizeof(SerializedState)) {
    372     return nullptr;
    373   }
    374 
    375   const SerializedState* state = static_cast<const SerializedState*>(data);
    376 
    377   NodeController* node_controller = internal::g_core->GetNodeController();
    378   ports::PortRef port;
    379   if (node_controller->node()->GetPort(ports[0], &port) != ports::OK)
    380     return nullptr;
    381 
    382   PlatformHandle buffer_handle;
    383   std::swap(buffer_handle, handles[0]);
    384   scoped_refptr<PlatformSharedBuffer> ring_buffer =
    385       PlatformSharedBuffer::CreateFromPlatformHandle(
    386           state->options.capacity_num_bytes,
    387           false /* read_only */,
    388           ScopedPlatformHandle(buffer_handle));
    389   if (!ring_buffer) {
    390     DLOG(ERROR) << "Failed to deserialize shared buffer handle.";
    391     return nullptr;
    392   }
    393 
    394   scoped_refptr<DataPipeProducerDispatcher> dispatcher =
    395       new DataPipeProducerDispatcher(node_controller, port, ring_buffer,
    396                                      state->options, false /* initialized */,
    397                                      state->pipe_id);
    398 
    399   {
    400     base::AutoLock lock(dispatcher->lock_);
    401     dispatcher->write_offset_ = state->write_offset;
    402     dispatcher->available_capacity_ = state->available_capacity;
    403     dispatcher->peer_closed_ = state->flags & kFlagPeerClosed;
    404     dispatcher->InitializeNoLock();
    405     dispatcher->UpdateSignalsStateNoLock();
    406   }
    407 
    408   return dispatcher;
    409 }
    410 
    411 DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
    412   DCHECK(is_closed_ && !in_transit_ && !shared_ring_buffer_ &&
    413          !ring_buffer_mapping_);
    414 }
    415 
    416 void DataPipeProducerDispatcher::InitializeNoLock() {
    417   lock_.AssertAcquired();
    418 
    419   if (shared_ring_buffer_) {
    420     ring_buffer_mapping_ =
    421         shared_ring_buffer_->Map(0, options_.capacity_num_bytes);
    422     if (!ring_buffer_mapping_) {
    423       DLOG(ERROR) << "Failed to map shared buffer.";
    424       shared_ring_buffer_ = nullptr;
    425     }
    426   }
    427 
    428   base::AutoUnlock unlock(lock_);
    429   node_controller_->SetPortObserver(
    430       control_port_,
    431       make_scoped_refptr(new PortObserverThunk(this)));
    432 }
    433 
    434 MojoResult DataPipeProducerDispatcher::CloseNoLock() {
    435   lock_.AssertAcquired();
    436   if (is_closed_ || in_transit_)
    437     return MOJO_RESULT_INVALID_ARGUMENT;
    438   is_closed_ = true;
    439   ring_buffer_mapping_.reset();
    440   shared_ring_buffer_ = nullptr;
    441 
    442   awakable_list_.CancelAll();
    443   if (!transferred_) {
    444     base::AutoUnlock unlock(lock_);
    445     node_controller_->ClosePort(control_port_);
    446   }
    447 
    448   return MOJO_RESULT_OK;
    449 }
    450 
    451 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsStateNoLock()
    452     const {
    453   lock_.AssertAcquired();
    454   HandleSignalsState rv;
    455   if (!peer_closed_) {
    456     if (!in_two_phase_write_ && shared_ring_buffer_ &&
    457         available_capacity_ > 0)
    458       rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
    459     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
    460   } else {
    461     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
    462   }
    463   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
    464   return rv;
    465 }
    466 
    467 void DataPipeProducerDispatcher::NotifyWrite(uint32_t num_bytes) {
    468   DVLOG(1) << "Data pipe producer " << pipe_id_ << " notifying peer: "
    469            << num_bytes << " bytes written. [control_port="
    470            << control_port_.name() << "]";
    471 
    472   SendDataPipeControlMessage(node_controller_, control_port_,
    473                              DataPipeCommand::DATA_WAS_WRITTEN, num_bytes);
    474 }
    475 
    476 void DataPipeProducerDispatcher::OnPortStatusChanged() {
    477   DCHECK(RequestContext::current());
    478 
    479   base::AutoLock lock(lock_);
    480 
    481   // We stop observing the control port as soon it's transferred, but this can
    482   // race with events which are raised right before that happens. This is fine
    483   // to ignore.
    484   if (transferred_)
    485     return;
    486 
    487   DVLOG(1) << "Control port status changed for data pipe producer " << pipe_id_;
    488 
    489   UpdateSignalsStateNoLock();
    490 }
    491 
    492 void DataPipeProducerDispatcher::UpdateSignalsStateNoLock() {
    493   lock_.AssertAcquired();
    494 
    495   bool was_peer_closed = peer_closed_;
    496   size_t previous_capacity = available_capacity_;
    497 
    498   ports::PortStatus port_status;
    499   int rv = node_controller_->node()->GetStatus(control_port_, &port_status);
    500   if (rv != ports::OK || !port_status.receiving_messages) {
    501     DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware of peer closure"
    502              << " [control_port=" << control_port_.name() << "]";
    503     peer_closed_ = true;
    504   } else if (rv == ports::OK && port_status.has_messages && !in_transit_) {
    505     ports::ScopedMessage message;
    506     do {
    507       int rv = node_controller_->node()->GetMessage(
    508           control_port_, &message, nullptr);
    509       if (rv != ports::OK)
    510         peer_closed_ = true;
    511       if (message) {
    512         if (message->num_payload_bytes() < sizeof(DataPipeControlMessage)) {
    513           peer_closed_ = true;
    514           break;
    515         }
    516 
    517         const DataPipeControlMessage* m =
    518             static_cast<const DataPipeControlMessage*>(
    519                 message->payload_bytes());
    520 
    521         if (m->command != DataPipeCommand::DATA_WAS_READ) {
    522           DLOG(ERROR) << "Unexpected message from consumer.";
    523           peer_closed_ = true;
    524           break;
    525         }
    526 
    527         if (static_cast<size_t>(available_capacity_) + m->num_bytes >
    528               options_.capacity_num_bytes) {
    529           DLOG(ERROR) << "Consumer claims to have read too many bytes.";
    530           break;
    531         }
    532 
    533         DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware that "
    534                  << m->num_bytes << " bytes were read. [control_port="
    535                  << control_port_.name() << "]";
    536 
    537         available_capacity_ += m->num_bytes;
    538       }
    539     } while (message);
    540   }
    541 
    542   if (peer_closed_ != was_peer_closed ||
    543       available_capacity_ != previous_capacity) {
    544     awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
    545   }
    546 }
    547 
    548 }  // namespace edk
    549 }  // namespace mojo
    550