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