Make DataPipe own an impl.

mojo/edk/system/data_pipe.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/edk/system/data_pipe.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <limits>
 
 #include "base/logging.h"
 #include "mojo/edk/system/awakable_list.h"
 #include "mojo/edk/system/configuration.h"
+#include "mojo/edk/system/data_pipe_impl.h"
 #include "mojo/edk/system/memory.h"
 #include "mojo/edk/system/options_validation.h"
 
 namespace mojo {
 namespace system {
 
+DataPipe::DataPipe(bool has_local_producer,
+                   bool has_local_consumer,
+                   const MojoCreateDataPipeOptions& validated_options,
+                   scoped_ptr<DataPipeImpl> impl)
+    : may_discard_((validated_options.flags &
+                    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD)),
+      element_num_bytes_(validated_options.element_num_bytes),
+      capacity_num_bytes_(validated_options.capacity_num_bytes),
+      producer_open_(true),
+      consumer_open_(true),
+      producer_awakable_list_(has_local_producer ? new AwakableList()
+                                                 : nullptr),
+      consumer_awakable_list_(has_local_consumer ? new AwakableList()
+                                                 : nullptr),
+      producer_two_phase_max_num_bytes_written_(0),
+      consumer_two_phase_max_num_bytes_read_(0),
+      impl_(impl.Pass()) {
+  impl_->set_owner(this);
+
+#if !defined(NDEBUG) || defined(DCHECK_ALWAYS_ON)
+  // Check that the passed in options actually are validated.
+  MojoCreateDataPipeOptions unused = {0};
+  DCHECK_EQ(ValidateCreateOptions(MakeUserPointer(&validated_options), &unused),
+            MOJO_RESULT_OK);
+#endif  // !defined(NDEBUG) || defined(DCHECK_ALWAYS_ON)
+}
+
 // static
 MojoCreateDataPipeOptions DataPipe::GetDefaultCreateOptions() {
   MojoCreateDataPipeOptions result = {
       static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions)),
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,
       1u,
       static_cast<uint32_t>(
           GetConfiguration().default_data_pipe_capacity_bytes)};
   return result;
 }
@@ skipping 78 matching lines @@
   uint32_t max_num_bytes_to_write = num_bytes.Get();
   if (max_num_bytes_to_write % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (max_num_bytes_to_write == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   uint32_t min_num_bytes_to_write = all_or_none ? max_num_bytes_to_write : 0;
 
   HandleSignalsState old_consumer_state =
-      ConsumerGetHandleSignalsStateImplNoLock();
-  MojoResult rv = ProducerWriteDataImplNoLock(
+      impl_->ConsumerGetHandleSignalsState();
+  MojoResult rv = impl_->ProducerWriteData(
       elements, num_bytes, max_num_bytes_to_write, min_num_bytes_to_write);
   HandleSignalsState new_consumer_state =
-      ConsumerGetHandleSignalsStateImplNoLock();
+      impl_->ConsumerGetHandleSignalsState();
   if (!new_consumer_state.equals(old_consumer_state))
     AwakeConsumerAwakablesForStateChangeNoLock(new_consumer_state);
   return rv;
 }
 
 MojoResult DataPipe::ProducerBeginWriteData(
     UserPointer<void*> buffer,
     UserPointer<uint32_t> buffer_num_bytes,
     bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   if (producer_in_two_phase_write_no_lock())
     return MOJO_RESULT_BUSY;
   if (!consumer_open_no_lock())
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   uint32_t min_num_bytes_to_write = 0;
   if (all_or_none) {
     min_num_bytes_to_write = buffer_num_bytes.Get();
     if (min_num_bytes_to_write % element_num_bytes_ != 0)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  MojoResult rv = ProducerBeginWriteDataImplNoLock(buffer, buffer_num_bytes,
-                                                   min_num_bytes_to_write);
+  MojoResult rv = impl_->ProducerBeginWriteData(buffer, buffer_num_bytes,
+                                                min_num_bytes_to_write);
   if (rv != MOJO_RESULT_OK)
     return rv;
   // Note: No need to awake producer awakables, even though we're going from
   // writable to non-writable (since you can't wait on non-writability).
   // Similarly, though this may have discarded data (in "may discard" mode),
   // making it non-readable, there's still no need to awake consumer awakables.
   DCHECK(producer_in_two_phase_write_no_lock());
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipe::ProducerEndWriteData(uint32_t num_bytes_written) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   if (!producer_in_two_phase_write_no_lock())
     return MOJO_RESULT_FAILED_PRECONDITION;
   // Note: Allow successful completion of the two-phase write even if the
   // consumer has been closed.
 
   HandleSignalsState old_consumer_state =
-      ConsumerGetHandleSignalsStateImplNoLock();
+      impl_->ConsumerGetHandleSignalsState();
   MojoResult rv;
   if (num_bytes_written > producer_two_phase_max_num_bytes_written_ ||
       num_bytes_written % element_num_bytes_ != 0) {
     rv = MOJO_RESULT_INVALID_ARGUMENT;
     producer_two_phase_max_num_bytes_written_ = 0;
   } else {
-    rv = ProducerEndWriteDataImplNoLock(num_bytes_written);
+    rv = impl_->ProducerEndWriteData(num_bytes_written);
   }
   // Two-phase write ended even on failure.
   DCHECK(!producer_in_two_phase_write_no_lock());
   // If we're now writable, we *became* writable (since we weren't writable
   // during the two-phase write), so awake producer awakables.
   HandleSignalsState new_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
+      impl_->ProducerGetHandleSignalsState();
   if (new_producer_state.satisfies(MOJO_HANDLE_SIGNAL_WRITABLE))
     AwakeProducerAwakablesForStateChangeNoLock(new_producer_state);
   HandleSignalsState new_consumer_state =
-      ConsumerGetHandleSignalsStateImplNoLock();
+      impl_->ConsumerGetHandleSignalsState();
   if (!new_consumer_state.equals(old_consumer_state))
     AwakeConsumerAwakablesForStateChangeNoLock(new_consumer_state);
   return rv;
 }
 
 HandleSignalsState DataPipe::ProducerGetHandleSignalsState() {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
-  return ProducerGetHandleSignalsStateImplNoLock();
+  return impl_->ProducerGetHandleSignalsState();
 }
 
 MojoResult DataPipe::ProducerAddAwakable(Awakable* awakable,
                                          MojoHandleSignals signals,
                                          uint32_t context,
                                          HandleSignalsState* signals_state) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
-  HandleSignalsState producer_state = ProducerGetHandleSignalsStateImplNoLock();
+  HandleSignalsState producer_state = impl_->ProducerGetHandleSignalsState();
   if (producer_state.satisfies(signals)) {
     if (signals_state)
       *signals_state = producer_state;
     return MOJO_RESULT_ALREADY_EXISTS;
   }
   if (!producer_state.can_satisfy(signals)) {
     if (signals_state)
       *signals_state = producer_state;
     return MOJO_RESULT_FAILED_PRECONDITION;
   }
 
   producer_awakable_list_->Add(awakable, signals, context);
   return MOJO_RESULT_OK;
 }
 
 void DataPipe::ProducerRemoveAwakable(Awakable* awakable,
                                       HandleSignalsState* signals_state) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
   producer_awakable_list_->Remove(awakable);
   if (signals_state)
-    *signals_state = ProducerGetHandleSignalsStateImplNoLock();
+    *signals_state = impl_->ProducerGetHandleSignalsState();
 }
 
 void DataPipe::ProducerStartSerialize(Channel* channel,
                                       size_t* max_size,
                                       size_t* max_platform_handles) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
-  ProducerStartSerializeImplNoLock(channel, max_size, max_platform_handles);
+  impl_->ProducerStartSerialize(channel, max_size, max_platform_handles);
 }
 
 bool DataPipe::ProducerEndSerialize(
     Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
-  return ProducerEndSerializeImplNoLock(channel, destination, actual_size,
-                                        platform_handles);
+  return impl_->ProducerEndSerialize(channel, destination, actual_size,
+                                     platform_handles);
 }
 
 bool DataPipe::ProducerIsBusy() const {
   base::AutoLock locker(lock_);
   return producer_in_two_phase_write_no_lock();
 }
 
 void DataPipe::ConsumerCancelAllAwakables() {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
@@ skipping 18 matching lines @@
   uint32_t max_num_bytes_to_read = num_bytes.Get();
   if (max_num_bytes_to_read % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (max_num_bytes_to_read == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   uint32_t min_num_bytes_to_read = all_or_none ? max_num_bytes_to_read : 0;
 
   HandleSignalsState old_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
-  MojoResult rv = ConsumerReadDataImplNoLock(
+      impl_->ProducerGetHandleSignalsState();
+  MojoResult rv = impl_->ConsumerReadData(
       elements, num_bytes, max_num_bytes_to_read, min_num_bytes_to_read, peek);
   HandleSignalsState new_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
+      impl_->ProducerGetHandleSignalsState();
   if (!new_producer_state.equals(old_producer_state))
     AwakeProducerAwakablesForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 MojoResult DataPipe::ConsumerDiscardData(UserPointer<uint32_t> num_bytes,
                                          bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
   uint32_t max_num_bytes_to_discard = num_bytes.Get();
   if (max_num_bytes_to_discard % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (max_num_bytes_to_discard == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   uint32_t min_num_bytes_to_discard =
       all_or_none ? max_num_bytes_to_discard : 0;
 
   HandleSignalsState old_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
-  MojoResult rv = ConsumerDiscardDataImplNoLock(
+      impl_->ProducerGetHandleSignalsState();
+  MojoResult rv = impl_->ConsumerDiscardData(
       num_bytes, max_num_bytes_to_discard, min_num_bytes_to_discard);
   HandleSignalsState new_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
+      impl_->ProducerGetHandleSignalsState();
   if (!new_producer_state.equals(old_producer_state))
     AwakeProducerAwakablesForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 MojoResult DataPipe::ConsumerQueryData(UserPointer<uint32_t> num_bytes) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
   // Note: Don't need to validate |*num_bytes| for query.
-  return ConsumerQueryDataImplNoLock(num_bytes);
+  return impl_->ConsumerQueryData(num_bytes);
 }
 
 MojoResult DataPipe::ConsumerBeginReadData(
     UserPointer<const void*> buffer,
     UserPointer<uint32_t> buffer_num_bytes,
     bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
   uint32_t min_num_bytes_to_read = 0;
   if (all_or_none) {
     min_num_bytes_to_read = buffer_num_bytes.Get();
     if (min_num_bytes_to_read % element_num_bytes_ != 0)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  MojoResult rv = ConsumerBeginReadDataImplNoLock(buffer, buffer_num_bytes,
-                                                  min_num_bytes_to_read);
+  MojoResult rv = impl_->ConsumerBeginReadData(buffer, buffer_num_bytes,
+                                               min_num_bytes_to_read);
   if (rv != MOJO_RESULT_OK)
     return rv;
   DCHECK(consumer_in_two_phase_read_no_lock());
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipe::ConsumerEndReadData(uint32_t num_bytes_read) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (!consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   HandleSignalsState old_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
+      impl_->ProducerGetHandleSignalsState();
   MojoResult rv;
   if (num_bytes_read > consumer_two_phase_max_num_bytes_read_ ||
       num_bytes_read % element_num_bytes_ != 0) {
     rv = MOJO_RESULT_INVALID_ARGUMENT;
     consumer_two_phase_max_num_bytes_read_ = 0;
   } else {
-    rv = ConsumerEndReadDataImplNoLock(num_bytes_read);
+    rv = impl_->ConsumerEndReadData(num_bytes_read);
   }
   // Two-phase read ended even on failure.
   DCHECK(!consumer_in_two_phase_read_no_lock());
   // If we're now readable, we *became* readable (since we weren't readable
   // during the two-phase read), so awake consumer awakables.
   HandleSignalsState new_consumer_state =
-      ConsumerGetHandleSignalsStateImplNoLock();
+      impl_->ConsumerGetHandleSignalsState();
   if (new_consumer_state.satisfies(MOJO_HANDLE_SIGNAL_READABLE))
     AwakeConsumerAwakablesForStateChangeNoLock(new_consumer_state);
   HandleSignalsState new_producer_state =
-      ProducerGetHandleSignalsStateImplNoLock();
+      impl_->ProducerGetHandleSignalsState();
   if (!new_producer_state.equals(old_producer_state))
     AwakeProducerAwakablesForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 HandleSignalsState DataPipe::ConsumerGetHandleSignalsState() {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
-  return ConsumerGetHandleSignalsStateImplNoLock();
+  return impl_->ConsumerGetHandleSignalsState();
 }
 
 MojoResult DataPipe::ConsumerAddAwakable(Awakable* awakable,
                                          MojoHandleSignals signals,
                                          uint32_t context,
                                          HandleSignalsState* signals_state) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
-  HandleSignalsState consumer_state = ConsumerGetHandleSignalsStateImplNoLock();
+  HandleSignalsState consumer_state = impl_->ConsumerGetHandleSignalsState();
   if (consumer_state.satisfies(signals)) {
     if (signals_state)
       *signals_state = consumer_state;
     return MOJO_RESULT_ALREADY_EXISTS;
   }
   if (!consumer_state.can_satisfy(signals)) {
     if (signals_state)
       *signals_state = consumer_state;
     return MOJO_RESULT_FAILED_PRECONDITION;
   }
 
   consumer_awakable_list_->Add(awakable, signals, context);
   return MOJO_RESULT_OK;
 }
 
 void DataPipe::ConsumerRemoveAwakable(Awakable* awakable,
                                       HandleSignalsState* signals_state) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
   consumer_awakable_list_->Remove(awakable);
   if (signals_state)
-    *signals_state = ConsumerGetHandleSignalsStateImplNoLock();
+    *signals_state = impl_->ConsumerGetHandleSignalsState();
 }
 
 void DataPipe::ConsumerStartSerialize(Channel* channel,
                                       size_t* max_size,
                                       size_t* max_platform_handles) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
-  ConsumerStartSerializeImplNoLock(channel, max_size, max_platform_handles);
+  impl_->ConsumerStartSerialize(channel, max_size, max_platform_handles);
 }
 
 bool DataPipe::ConsumerEndSerialize(
     Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
-  return ConsumerEndSerializeImplNoLock(channel, destination, actual_size,
-                                        platform_handles);
+  return impl_->ConsumerEndSerialize(channel, destination, actual_size,
+                                     platform_handles);
 }
 
 bool DataPipe::ConsumerIsBusy() const {
   base::AutoLock locker(lock_);
   return consumer_in_two_phase_read_no_lock();
 }
 
-DataPipe::DataPipe(bool has_local_producer,
-                   bool has_local_consumer,
-                   const MojoCreateDataPipeOptions& validated_options)
-    : may_discard_((validated_options.flags &
-                    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD)),
-      element_num_bytes_(validated_options.element_num_bytes),
-      capacity_num_bytes_(validated_options.capacity_num_bytes),
-      producer_open_(true),
-      consumer_open_(true),
-      producer_awakable_list_(has_local_producer ? new AwakableList()
-                                                 : nullptr),
-      consumer_awakable_list_(has_local_consumer ? new AwakableList()
-                                                 : nullptr),
-      producer_two_phase_max_num_bytes_written_(0),
-      consumer_two_phase_max_num_bytes_read_(0) {
-  // Check that the passed in options actually are validated.
-  MojoCreateDataPipeOptions unused = {0};
-  DCHECK_EQ(ValidateCreateOptions(MakeUserPointer(&validated_options), &unused),
-            MOJO_RESULT_OK);
-}
-
 DataPipe::~DataPipe() {
   DCHECK(!producer_open_);
   DCHECK(!consumer_open_);
   DCHECK(!producer_awakable_list_);
   DCHECK(!consumer_awakable_list_);
 }
 
 void DataPipe::ProducerCloseNoLock() {
   lock_.AssertAcquired();
   DCHECK(producer_open_);
   producer_open_ = false;
   DCHECK(has_local_producer_no_lock());
   producer_awakable_list_.reset();
   // Not a bug, except possibly in "user" code.
   DVLOG_IF(2, producer_in_two_phase_write_no_lock())
       << "Producer closed with active two-phase write";
   producer_two_phase_max_num_bytes_written_ = 0;
-  ProducerCloseImplNoLock();
+  impl_->ProducerClose();
   AwakeConsumerAwakablesForStateChangeNoLock(
-      ConsumerGetHandleSignalsStateImplNoLock());
+      impl_->ConsumerGetHandleSignalsState());
 }
 
 void DataPipe::ConsumerCloseNoLock() {
   lock_.AssertAcquired();
   DCHECK(consumer_open_);
   consumer_open_ = false;
   DCHECK(has_local_consumer_no_lock());
   consumer_awakable_list_.reset();
   // Not a bug, except possibly in "user" code.
   DVLOG_IF(2, consumer_in_two_phase_read_no_lock())
       << "Consumer closed with active two-phase read";
   consumer_two_phase_max_num_bytes_read_ = 0;
-  ConsumerCloseImplNoLock();
+  impl_->ConsumerClose();
   AwakeProducerAwakablesForStateChangeNoLock(
-      ProducerGetHandleSignalsStateImplNoLock());
+      impl_->ProducerGetHandleSignalsState());
 }
 
 void DataPipe::AwakeProducerAwakablesForStateChangeNoLock(
     const HandleSignalsState& new_producer_state) {
   lock_.AssertAcquired();
   if (!has_local_producer_no_lock())
     return;
   producer_awakable_list_->AwakeForStateChange(new_producer_state);
 }
 
 void DataPipe::AwakeConsumerAwakablesForStateChangeNoLock(
     const HandleSignalsState& new_consumer_state) {
   lock_.AssertAcquired();
   if (!has_local_consumer_no_lock())
     return;
   consumer_awakable_list_->AwakeForStateChange(new_consumer_state);
 }
 
 }  // namespace system
 }  // namespace mojo