src/mojo/system/local_message_pipe_endpoint.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/system/local_message_pipe_endpoint.h"
   6
   7 #include <string.h>
   8
   9 #include "base/logging.h"
  10 #include "mojo/system/message_in_transit.h"
  11
  12 namespace mojo {
  13 namespace system {
  14
  15 LocalMessagePipeEndpoint::LocalMessagePipeEndpoint()
  16     : is_open_(true),
  17       is_peer_open_(true) {
  18 }
  19
  20 LocalMessagePipeEndpoint::~LocalMessagePipeEndpoint() {
  21   DCHECK(!is_open_);
  22 }
  23
  24 void LocalMessagePipeEndpoint::OnPeerClose() {
  25   DCHECK(is_open_);
  26   DCHECK(is_peer_open_);
  27
  28   MojoWaitFlags old_satisfied_flags = SatisfiedFlags();
  29   MojoWaitFlags old_satisfiable_flags = SatisfiableFlags();
  30   is_peer_open_ = false;
  31   MojoWaitFlags new_satisfied_flags = SatisfiedFlags();
  32   MojoWaitFlags new_satisfiable_flags = SatisfiableFlags();
  33
  34   if (new_satisfied_flags != old_satisfied_flags ||
  35       new_satisfiable_flags != old_satisfiable_flags) {
  36     waiter_list_.AwakeWaitersForStateChange(new_satisfied_flags,
  37                                             new_satisfiable_flags);
  38   }
  39 }
  40
  41 MojoResult LocalMessagePipeEndpoint::EnqueueMessage(
  42     const void* bytes, uint32_t num_bytes,
  43     const MojoHandle* handles, uint32_t num_handles,
  44     MojoWriteMessageFlags /*flags*/) {
  45   DCHECK(is_open_);
  46   DCHECK(is_peer_open_);
  47
  48   bool was_empty = message_queue_.empty();
  49
  50   // TODO(vtl): Eventually (with C++11), this should be an |emplace_back()|.
  51   message_queue_.push_back(MessageInTransit::Create(bytes, num_bytes));
  52   // TODO(vtl): Support sending handles.
  53
  54   if (was_empty) {
  55     waiter_list_.AwakeWaitersForStateChange(SatisfiedFlags(),
  56                                             SatisfiableFlags());
  57   }
  58
  59   return MOJO_RESULT_OK;
  60 }
  61
  62 void LocalMessagePipeEndpoint::CancelAllWaiters() {
  63   DCHECK(is_open_);
  64   waiter_list_.CancelAllWaiters();
  65 }
  66
  67 void LocalMessagePipeEndpoint::Close() {
  68   DCHECK(is_open_);
  69   is_open_ = false;
  70   for (std::deque<MessageInTransit*>::iterator it = message_queue_.begin();
  71        it != message_queue_.end();
  72        ++it) {
  73     (*it)->Destroy();
  74   }
  75   message_queue_.clear();
  76 }
  77
  78 MojoResult LocalMessagePipeEndpoint::ReadMessage(
  79     void* bytes, uint32_t* num_bytes,
  80     MojoHandle* handles, uint32_t* num_handles,
  81     MojoReadMessageFlags flags) {
  82   DCHECK(is_open_);
  83
  84   const uint32_t max_bytes = num_bytes ? *num_bytes : 0;
  85   // TODO(vtl): We'll need this later:
  86   //  const uint32_t max_handles = num_handles ? *num_handles : 0;
  87
  88   if (message_queue_.empty())
  89     return MOJO_RESULT_NOT_FOUND;
  90
  91   // TODO(vtl): If |flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD|, we could pop
  92   // and release the lock immediately.
  93   bool not_enough_space = false;
  94   MessageInTransit* const message = message_queue_.front();
  95   if (num_bytes)
  96     *num_bytes = message->data_size();
  97   if (message->data_size() <= max_bytes)
  98     memcpy(bytes, message->data(), message->data_size());
  99   else
 100     not_enough_space = true;
 101
 102   // TODO(vtl): Support receiving handles.
 103   if (num_handles)
 104     *num_handles = 0;
 105
 106   if (!not_enough_space || (flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD)) {
 107     message_queue_.pop_front();
 108     message->Destroy();
 109
 110     // Now it's empty, thus no longer readable.
 111     if (message_queue_.empty()) {
 112       // It's currently not possible to wait for non-readability, but we should
 113       // do the state change anyway.
 114       waiter_list_.AwakeWaitersForStateChange(SatisfiedFlags(),
 115                                               SatisfiableFlags());
 116     }
 117   }
 118
 119   if (not_enough_space)
 120     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 121
 122   return MOJO_RESULT_OK;
 123 }
 124
 125 MojoResult LocalMessagePipeEndpoint::AddWaiter(Waiter* waiter,
 126                                                MojoWaitFlags flags,
 127                                                MojoResult wake_result) {
 128   DCHECK(is_open_);
 129
 130   if ((flags & SatisfiedFlags()))
 131     return MOJO_RESULT_ALREADY_EXISTS;
 132   if (!(flags & SatisfiableFlags()))
 133     return MOJO_RESULT_FAILED_PRECONDITION;
 134
 135   waiter_list_.AddWaiter(waiter, flags, wake_result);
 136   return MOJO_RESULT_OK;
 137 }
 138
 139 void LocalMessagePipeEndpoint::RemoveWaiter(Waiter* waiter) {
 140   DCHECK(is_open_);
 141   waiter_list_.RemoveWaiter(waiter);
 142 }
 143
 144 MojoWaitFlags LocalMessagePipeEndpoint::SatisfiedFlags() {
 145   MojoWaitFlags satisfied_flags = 0;
 146   if (!message_queue_.empty())
 147     satisfied_flags |= MOJO_WAIT_FLAG_READABLE;
 148   if (is_peer_open_)
 149     satisfied_flags |= MOJO_WAIT_FLAG_WRITABLE;
 150   return satisfied_flags;
 151 }
 152
 153 MojoWaitFlags LocalMessagePipeEndpoint::SatisfiableFlags() {
 154   MojoWaitFlags satisfiable_flags = 0;
 155   if (!message_queue_.empty() || is_peer_open_)
 156     satisfiable_flags |= MOJO_WAIT_FLAG_READABLE;
 157   if (is_peer_open_)
 158     satisfiable_flags |= MOJO_WAIT_FLAG_WRITABLE;
 159   return satisfiable_flags;
 160 }
 161
 162 }  // namespace system
 163 }  // namespace mojo
 164