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/dispatcher.h"
  11 #include "mojo/system/message_in_transit.h"
  12
  13 namespace mojo {
  14 namespace system {
  15
  16 LocalMessagePipeEndpoint::LocalMessagePipeEndpoint()
  17     : is_open_(true),
  18       is_peer_open_(true) {
  19 }
  20
  21 LocalMessagePipeEndpoint::~LocalMessagePipeEndpoint() {
  22   DCHECK(!is_open_);
  23   DCHECK(message_queue_.IsEmpty());  // Should be implied by not being open.
  24 }
  25
  26 MessagePipeEndpoint::Type LocalMessagePipeEndpoint::GetType() const {
  27   return kTypeLocal;
  28 }
  29
  30 void LocalMessagePipeEndpoint::Close() {
  31   DCHECK(is_open_);
  32   is_open_ = false;
  33   message_queue_.Clear();
  34 }
  35
  36 void LocalMessagePipeEndpoint::OnPeerClose() {
  37   DCHECK(is_open_);
  38   DCHECK(is_peer_open_);
  39
  40   MojoWaitFlags old_satisfied_flags = SatisfiedFlags();
  41   MojoWaitFlags old_satisfiable_flags = SatisfiableFlags();
  42   is_peer_open_ = false;
  43   MojoWaitFlags new_satisfied_flags = SatisfiedFlags();
  44   MojoWaitFlags new_satisfiable_flags = SatisfiableFlags();
  45
  46   if (new_satisfied_flags != old_satisfied_flags ||
  47       new_satisfiable_flags != old_satisfiable_flags) {
  48     waiter_list_.AwakeWaitersForStateChange(new_satisfied_flags,
  49                                             new_satisfiable_flags);
  50   }
  51 }
  52
  53 void LocalMessagePipeEndpoint::EnqueueMessage(
  54     scoped_ptr<MessageInTransit> message) {
  55   DCHECK(is_open_);
  56   DCHECK(is_peer_open_);
  57
  58   bool was_empty = message_queue_.IsEmpty();
  59   message_queue_.AddMessage(message.Pass());
  60   if (was_empty) {
  61     waiter_list_.AwakeWaitersForStateChange(SatisfiedFlags(),
  62                                             SatisfiableFlags());
  63   }
  64 }
  65
  66 void LocalMessagePipeEndpoint::CancelAllWaiters() {
  67   DCHECK(is_open_);
  68   waiter_list_.CancelAllWaiters();
  69 }
  70
  71 MojoResult LocalMessagePipeEndpoint::ReadMessage(
  72     void* bytes, uint32_t* num_bytes,
  73     std::vector<scoped_refptr<Dispatcher> >* dispatchers,
  74     uint32_t* num_dispatchers,
  75     MojoReadMessageFlags flags) {
  76   DCHECK(is_open_);
  77   DCHECK(!dispatchers || dispatchers->empty());
  78
  79   const uint32_t max_bytes = num_bytes ? *num_bytes : 0;
  80   const uint32_t max_num_dispatchers = num_dispatchers ? *num_dispatchers : 0;
  81
  82   if (message_queue_.IsEmpty()) {
  83     return is_peer_open_ ? MOJO_RESULT_SHOULD_WAIT :
  84                            MOJO_RESULT_FAILED_PRECONDITION;
  85   }
  86
  87   // TODO(vtl): If |flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD|, we could pop
  88   // and release the lock immediately.
  89   bool enough_space = true;
  90   MessageInTransit* message = message_queue_.PeekMessage();
  91   if (num_bytes)
  92     *num_bytes = message->num_bytes();
  93   if (message->num_bytes() <= max_bytes)
  94     memcpy(bytes, message->bytes(), message->num_bytes());
  95   else
  96     enough_space = false;
  97
  98   if (std::vector<scoped_refptr<Dispatcher> >* queued_dispatchers =
  99           message->dispatchers()) {
 100     if (num_dispatchers)
 101       *num_dispatchers = static_cast<uint32_t>(queued_dispatchers->size());
 102     if (enough_space) {
 103       if (queued_dispatchers->empty()) {
 104         // Nothing to do.
 105       } else if (queued_dispatchers->size() <= max_num_dispatchers) {
 106         DCHECK(dispatchers);
 107         dispatchers->swap(*queued_dispatchers);
 108       } else {
 109         enough_space = false;
 110       }
 111     }
 112   } else {
 113     if (num_dispatchers)
 114       *num_dispatchers = 0;
 115   }
 116
 117   message = NULL;
 118
 119   if (enough_space || (flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD)) {
 120     message_queue_.DiscardMessage();
 121
 122     // Now it's empty, thus no longer readable.
 123     if (message_queue_.IsEmpty()) {
 124       // It's currently not possible to wait for non-readability, but we should
 125       // do the state change anyway.
 126       waiter_list_.AwakeWaitersForStateChange(SatisfiedFlags(),
 127                                               SatisfiableFlags());
 128     }
 129   }
 130
 131   if (!enough_space)
 132     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 133
 134   return MOJO_RESULT_OK;
 135 }
 136
 137 MojoResult LocalMessagePipeEndpoint::AddWaiter(Waiter* waiter,
 138                                                MojoWaitFlags flags,
 139                                                MojoResult wake_result) {
 140   DCHECK(is_open_);
 141
 142   if ((flags & SatisfiedFlags()))
 143     return MOJO_RESULT_ALREADY_EXISTS;
 144   if (!(flags & SatisfiableFlags()))
 145     return MOJO_RESULT_FAILED_PRECONDITION;
 146
 147   waiter_list_.AddWaiter(waiter, flags, wake_result);
 148   return MOJO_RESULT_OK;
 149 }
 150
 151 void LocalMessagePipeEndpoint::RemoveWaiter(Waiter* waiter) {
 152   DCHECK(is_open_);
 153   waiter_list_.RemoveWaiter(waiter);
 154 }
 155
 156 MojoWaitFlags LocalMessagePipeEndpoint::SatisfiedFlags() {
 157   MojoWaitFlags satisfied_flags = 0;
 158   if (!message_queue_.IsEmpty())
 159     satisfied_flags |= MOJO_WAIT_FLAG_READABLE;
 160   if (is_peer_open_)
 161     satisfied_flags |= MOJO_WAIT_FLAG_WRITABLE;
 162   return satisfied_flags;
 163 }
 164
 165 MojoWaitFlags LocalMessagePipeEndpoint::SatisfiableFlags() {
 166   MojoWaitFlags satisfiable_flags = 0;
 167   if (!message_queue_.IsEmpty() || is_peer_open_)
 168     satisfiable_flags |= MOJO_WAIT_FLAG_READABLE;
 169   if (is_peer_open_)
 170     satisfiable_flags |= MOJO_WAIT_FLAG_WRITABLE;
 171   return satisfiable_flags;
 172 }
 173
 174 }  // namespace system
 175 }  // namespace mojo