Make DataPipe own an impl.
* Add a DataPipeImpl class (basically consisting of the
DataPipe::...ImplNoLock() methods).
* Make LocalDataPipeImpl subclass that, instead of DataPipe.
* Give DataPipe own such an impl (and call its methods, instead of the
...ImplNoLock() methods).
This is so we'll be able to switch impls.
R=yzshen@chromium.org
Review URL: https://codereview.chromium.org/926553006
diff --git a/mojo/edk/system/BUILD.gn b/mojo/edk/system/BUILD.gn
index c7b4346..06ee017 100644
--- a/mojo/edk/system/BUILD.gn
+++ b/mojo/edk/system/BUILD.gn
@@ -44,6 +44,7 @@
"core.h",
"data_pipe.cc",
"data_pipe.h",
+ "data_pipe_impl.h",
"data_pipe_consumer_dispatcher.cc",
"data_pipe_consumer_dispatcher.h",
"data_pipe_producer_dispatcher.cc",
diff --git a/mojo/edk/system/core.cc b/mojo/edk/system/core.cc
index e82e9f4..80c3d36 100644
--- a/mojo/edk/system/core.cc
+++ b/mojo/edk/system/core.cc
@@ -17,7 +17,6 @@
#include "mojo/edk/system/data_pipe_producer_dispatcher.h"
#include "mojo/edk/system/dispatcher.h"
#include "mojo/edk/system/handle_signals_state.h"
-#include "mojo/edk/system/local_data_pipe_impl.h"
#include "mojo/edk/system/memory.h"
#include "mojo/edk/system/message_pipe.h"
#include "mojo/edk/system/message_pipe_dispatcher.h"
@@ -381,7 +380,7 @@
}
DCHECK_NE(handle_pair.second, MOJO_HANDLE_INVALID);
- scoped_refptr<DataPipe> data_pipe(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> data_pipe(DataPipe::CreateLocal(validated_options));
producer_dispatcher->Init(data_pipe);
consumer_dispatcher->Init(data_pipe);
diff --git a/mojo/edk/system/data_pipe.cc b/mojo/edk/system/data_pipe.cc
index aede492..2431754 100644
--- a/mojo/edk/system/data_pipe.cc
+++ b/mojo/edk/system/data_pipe.cc
@@ -12,6 +12,8 @@
#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/local_data_pipe_impl.h"
#include "mojo/edk/system/memory.h"
#include "mojo/edk/system/options_validation.h"
@@ -83,6 +85,13 @@
return MOJO_RESULT_OK;
}
+// static
+DataPipe* DataPipe::CreateLocal(
+ const MojoCreateDataPipeOptions& validated_options) {
+ return new DataPipe(true, true, validated_options,
+ make_scoped_ptr(new LocalDataPipeImpl()));
+}
+
void DataPipe::ProducerCancelAllAwakables() {
base::AutoLock locker(lock_);
DCHECK(has_local_producer_no_lock());
@@ -116,11 +125,11 @@
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;
@@ -145,8 +154,8 @@
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
@@ -167,25 +176,25 @@
// 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;
@@ -194,7 +203,7 @@
HandleSignalsState DataPipe::ProducerGetHandleSignalsState() {
base::AutoLock locker(lock_);
DCHECK(has_local_producer_no_lock());
- return ProducerGetHandleSignalsStateImplNoLock();
+ return impl_->ProducerGetHandleSignalsState();
}
MojoResult DataPipe::ProducerAddAwakable(Awakable* awakable,
@@ -204,7 +213,7 @@
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;
@@ -226,7 +235,7 @@
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,
@@ -234,7 +243,7 @@
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(
@@ -244,8 +253,8 @@
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 {
@@ -284,11 +293,11 @@
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;
@@ -313,11 +322,11 @@
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;
@@ -331,7 +340,7 @@
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(
@@ -351,8 +360,8 @@
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());
@@ -367,25 +376,25 @@
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;
@@ -394,7 +403,7 @@
HandleSignalsState DataPipe::ConsumerGetHandleSignalsState() {
base::AutoLock locker(lock_);
DCHECK(has_local_consumer_no_lock());
- return ConsumerGetHandleSignalsStateImplNoLock();
+ return impl_->ConsumerGetHandleSignalsState();
}
MojoResult DataPipe::ConsumerAddAwakable(Awakable* awakable,
@@ -404,7 +413,7 @@
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;
@@ -426,7 +435,7 @@
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,
@@ -434,7 +443,7 @@
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(
@@ -444,8 +453,8 @@
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 {
@@ -455,7 +464,8 @@
DataPipe::DataPipe(bool has_local_producer,
bool has_local_consumer,
- const MojoCreateDataPipeOptions& validated_options)
+ 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),
@@ -467,11 +477,16 @@
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) {
+ 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)
}
DataPipe::~DataPipe() {
@@ -491,9 +506,9 @@
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() {
@@ -506,9 +521,9 @@
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(
diff --git a/mojo/edk/system/data_pipe.h b/mojo/edk/system/data_pipe.h
index 585893e..2d6bd12 100644
--- a/mojo/edk/system/data_pipe.h
+++ b/mojo/edk/system/data_pipe.h
@@ -24,6 +24,7 @@
class Awakable;
class AwakableList;
class Channel;
+class DataPipeImpl;
// |DataPipe| is a base class for secondary objects implementing data pipes,
// similar to |MessagePipe| (see the explanatory comment in core.cc). It is
@@ -48,6 +49,14 @@
UserPointer<const MojoCreateDataPipeOptions> in_options,
MojoCreateDataPipeOptions* out_options);
+ // Creates a local (both producer and consumer) data pipe (using
+ // |LocalDataPipeImpl|. |validated_options| should be the output of
+ // |ValidateOptions()|. In particular: |struct_size| is ignored (so
+ // |validated_options| must be the current version of the struct) and
+ // |capacity_num_bytes| must be nonzero.
+ static DataPipe* CreateLocal(
+ const MojoCreateDataPipeOptions& validated_options);
+
// These are called by the producer dispatcher to implement its methods of
// corresponding names.
void ProducerCancelAllAwakables();
@@ -108,75 +117,11 @@
embedder::PlatformHandleVector* platform_handles);
bool ConsumerIsBusy() const;
- protected:
- DataPipe(bool has_local_producer,
- bool has_local_consumer,
- const MojoCreateDataPipeOptions& validated_options);
-
- friend class base::RefCountedThreadSafe<DataPipe>;
- virtual ~DataPipe();
+ // The following are only to be used by |DataPipeImpl| (and its subclasses):
void ProducerCloseNoLock();
void ConsumerCloseNoLock();
- virtual void ProducerCloseImplNoLock() = 0;
- // |num_bytes.Get()| will be a nonzero multiple of |element_num_bytes_|.
- virtual MojoResult ProducerWriteDataImplNoLock(
- UserPointer<const void> elements,
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_write,
- uint32_t min_num_bytes_to_write) = 0;
- virtual MojoResult ProducerBeginWriteDataImplNoLock(
- UserPointer<void*> buffer,
- UserPointer<uint32_t> buffer_num_bytes,
- uint32_t min_num_bytes_to_write) = 0;
- virtual MojoResult ProducerEndWriteDataImplNoLock(
- uint32_t num_bytes_written) = 0;
- // Note: A producer should not be writable during a two-phase write.
- virtual HandleSignalsState ProducerGetHandleSignalsStateImplNoLock()
- const = 0;
- virtual void ProducerStartSerializeImplNoLock(
- Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) = 0;
- virtual bool ProducerEndSerializeImplNoLock(
- Channel* channel,
- void* destination,
- size_t* actual_size,
- embedder::PlatformHandleVector* platform_handles) = 0;
-
- virtual void ConsumerCloseImplNoLock() = 0;
- // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
- virtual MojoResult ConsumerReadDataImplNoLock(UserPointer<void> elements,
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_read,
- uint32_t min_num_bytes_to_read,
- bool peek) = 0;
- virtual MojoResult ConsumerDiscardDataImplNoLock(
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_discard,
- uint32_t min_num_bytes_to_discard) = 0;
- // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
- virtual MojoResult ConsumerQueryDataImplNoLock(
- UserPointer<uint32_t> num_bytes) = 0;
- virtual MojoResult ConsumerBeginReadDataImplNoLock(
- UserPointer<const void*> buffer,
- UserPointer<uint32_t> buffer_num_bytes,
- uint32_t min_num_bytes_to_read) = 0;
- virtual MojoResult ConsumerEndReadDataImplNoLock(uint32_t num_bytes_read) = 0;
- // Note: A consumer should not be writable during a two-phase read.
- virtual HandleSignalsState ConsumerGetHandleSignalsStateImplNoLock()
- const = 0;
- virtual void ConsumerStartSerializeImplNoLock(
- Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) = 0;
- virtual bool ConsumerEndSerializeImplNoLock(
- Channel* channel,
- void* destination,
- size_t* actual_size,
- embedder::PlatformHandleVector* platform_handles) = 0;
-
// Thread-safe and fast (they don't take the lock):
bool may_discard() const { return may_discard_; }
size_t element_num_bytes() const { return element_num_bytes_; }
@@ -218,6 +163,22 @@
}
private:
+ friend class base::RefCountedThreadSafe<DataPipe>;
+
+ // |validated_options| should be the output of |ValidateOptions()|. In
+ // particular: |struct_size| is ignored (so |validated_options| must be the
+ // current version of the struct) and |capacity_num_bytes| must be nonzero.
+ // TODO(vtl): |has_local_producer|/|has_local_consumer| shouldn't really be
+ // arguments here. Instead, they should be determined from the |impl| ... but
+ // the |impl|'s typically figures these out by examining the owner, i.e., the
+ // |DataPipe| object. Probably, this indicates that more stuff should be moved
+ // to |DataPipeImpl|, but for now we'll live with this.
+ DataPipe(bool has_local_producer,
+ bool has_local_consumer,
+ const MojoCreateDataPipeOptions& validated_options,
+ scoped_ptr<DataPipeImpl> impl);
+ virtual ~DataPipe();
+
void AwakeProducerAwakablesForStateChangeNoLock(
const HandleSignalsState& new_producer_state);
void AwakeConsumerAwakablesForStateChangeNoLock(
@@ -246,6 +207,7 @@
// These are nonzero if and only if a two-phase write/read is in progress.
uint32_t producer_two_phase_max_num_bytes_written_;
uint32_t consumer_two_phase_max_num_bytes_read_;
+ scoped_ptr<DataPipeImpl> impl_;
DISALLOW_COPY_AND_ASSIGN(DataPipe);
};
diff --git a/mojo/edk/system/data_pipe_impl.h b/mojo/edk/system/data_pipe_impl.h
new file mode 100644
index 0000000..e1339c0
--- /dev/null
+++ b/mojo/edk/system/data_pipe_impl.h
@@ -0,0 +1,122 @@
+// Copyright 2015 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.
+
+#ifndef MOJO_EDK_SYSTEM_DATA_PIPE_IMPL_H_
+#define MOJO_EDK_SYSTEM_DATA_PIPE_IMPL_H_
+
+#include <stdint.h>
+
+#include "base/macros.h"
+#include "mojo/edk/embedder/platform_handle_vector.h"
+#include "mojo/edk/system/data_pipe.h"
+#include "mojo/edk/system/handle_signals_state.h"
+#include "mojo/edk/system/memory.h"
+#include "mojo/edk/system/system_impl_export.h"
+#include "mojo/public/c/system/types.h"
+
+namespace mojo {
+namespace system {
+
+class Channel;
+
+// Base class/interface for classes that "implement" |DataPipe| for various
+// situations (local versus remote). The methods, other than the constructor,
+// |set_owner()|, and the destructor, are always protected by |DataPipe|'s
+// |lock_|.
+class MOJO_SYSTEM_IMPL_EXPORT DataPipeImpl {
+ public:
+ virtual ~DataPipeImpl() {}
+
+ // This is only called by |DataPipe| during its construction.
+ void set_owner(DataPipe* owner) { owner_ = owner; }
+
+ virtual void ProducerClose() = 0;
+ // |num_bytes.Get()| will be a nonzero multiple of |element_num_bytes()|.
+ virtual MojoResult ProducerWriteData(UserPointer<const void> elements,
+ UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_write,
+ uint32_t min_num_bytes_to_write) = 0;
+ virtual MojoResult ProducerBeginWriteData(
+ UserPointer<void*> buffer,
+ UserPointer<uint32_t> buffer_num_bytes,
+ uint32_t min_num_bytes_to_write) = 0;
+ virtual MojoResult ProducerEndWriteData(uint32_t num_bytes_written) = 0;
+ // Note: A producer should not be writable during a two-phase write.
+ virtual HandleSignalsState ProducerGetHandleSignalsState() const = 0;
+ virtual void ProducerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) = 0;
+ virtual bool ProducerEndSerialize(
+ Channel* channel,
+ void* destination,
+ size_t* actual_size,
+ embedder::PlatformHandleVector* platform_handles) = 0;
+
+ virtual void ConsumerClose() = 0;
+ // |num_bytes.Get()| will be a nonzero multiple of |element_num_bytes()|.
+ virtual MojoResult ConsumerReadData(UserPointer<void> elements,
+ UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_read,
+ uint32_t min_num_bytes_to_read,
+ bool peek) = 0;
+ virtual MojoResult ConsumerDiscardData(UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_discard,
+ uint32_t min_num_bytes_to_discard) = 0;
+ // |num_bytes.Get()| will be a nonzero multiple of |element_num_bytes()|.
+ virtual MojoResult ConsumerQueryData(UserPointer<uint32_t> num_bytes) = 0;
+ virtual MojoResult ConsumerBeginReadData(
+ UserPointer<const void*> buffer,
+ UserPointer<uint32_t> buffer_num_bytes,
+ uint32_t min_num_bytes_to_read) = 0;
+ virtual MojoResult ConsumerEndReadData(uint32_t num_bytes_read) = 0;
+ // Note: A consumer should not be writable during a two-phase read.
+ virtual HandleSignalsState ConsumerGetHandleSignalsState() const = 0;
+ virtual void ConsumerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) = 0;
+ virtual bool ConsumerEndSerialize(
+ Channel* channel,
+ void* destination,
+ size_t* actual_size,
+ embedder::PlatformHandleVector* platform_handles) = 0;
+
+ protected:
+ DataPipeImpl() : owner_() {}
+
+ DataPipe* owner() const { return owner_; }
+
+ bool may_discard() const { return owner_->may_discard(); }
+ size_t element_num_bytes() const { return owner_->element_num_bytes(); }
+ size_t capacity_num_bytes() const { return owner_->capacity_num_bytes(); }
+ bool producer_open() const { return owner_->producer_open_no_lock(); }
+ bool consumer_open() const { return owner_->consumer_open_no_lock(); }
+ uint32_t producer_two_phase_max_num_bytes_written() const {
+ return owner_->producer_two_phase_max_num_bytes_written_no_lock();
+ }
+ uint32_t consumer_two_phase_max_num_bytes_read() const {
+ return owner_->consumer_two_phase_max_num_bytes_read_no_lock();
+ }
+ void set_producer_two_phase_max_num_bytes_written(uint32_t num_bytes) {
+ owner_->set_producer_two_phase_max_num_bytes_written_no_lock(num_bytes);
+ }
+ void set_consumer_two_phase_max_num_bytes_read(uint32_t num_bytes) {
+ owner_->set_consumer_two_phase_max_num_bytes_read_no_lock(num_bytes);
+ }
+ bool producer_in_two_phase_write() const {
+ return owner_->producer_in_two_phase_write_no_lock();
+ }
+ bool consumer_in_two_phase_read() const {
+ return owner_->consumer_in_two_phase_read_no_lock();
+ }
+
+ private:
+ DataPipe* owner_;
+
+ DISALLOW_COPY_AND_ASSIGN(DataPipeImpl);
+};
+
+} // namespace system
+} // namespace mojo
+
+#endif // MOJO_EDK_SYSTEM_DATA_PIPE_IMPL_H_
diff --git a/mojo/edk/system/local_data_pipe_impl.cc b/mojo/edk/system/local_data_pipe_impl.cc
index 01ed107..3ca76d3 100644
--- a/mojo/edk/system/local_data_pipe_impl.cc
+++ b/mojo/edk/system/local_data_pipe_impl.cc
@@ -16,12 +16,13 @@
#include "base/logging.h"
#include "mojo/edk/system/configuration.h"
+#include "mojo/edk/system/data_pipe.h"
namespace mojo {
namespace system {
-LocalDataPipeImpl::LocalDataPipeImpl(const MojoCreateDataPipeOptions& options)
- : DataPipe(true, true, options), start_index_(0), current_num_bytes_(0) {
+LocalDataPipeImpl::LocalDataPipeImpl()
+ : start_index_(0), current_num_bytes_(0) {
// Note: |buffer_| is lazily allocated, since a common case will be that one
// of the handles is immediately passed off to another process.
}
@@ -29,20 +30,20 @@
LocalDataPipeImpl::~LocalDataPipeImpl() {
}
-void LocalDataPipeImpl::ProducerCloseImplNoLock() {
+void LocalDataPipeImpl::ProducerClose() {
// If the consumer is still open and we still have data, we have to keep the
// buffer around. Currently, we won't free it even if it empties later. (We
// could do this -- requiring a check on every read -- but that seems to be
// optimizing for the uncommon case.)
- if (!consumer_open_no_lock() || !current_num_bytes_) {
+ if (!consumer_open() || !current_num_bytes_) {
// Note: There can only be a two-phase *read* (by the consumer) if we still
// have data.
- DCHECK(!consumer_in_two_phase_read_no_lock());
- DestroyBufferNoLock();
+ DCHECK(!consumer_in_two_phase_read());
+ DestroyBuffer();
}
}
-MojoResult LocalDataPipeImpl::ProducerWriteDataImplNoLock(
+MojoResult LocalDataPipeImpl::ProducerWriteData(
UserPointer<const void> elements,
UserPointer<uint32_t> num_bytes,
uint32_t max_num_bytes_to_write,
@@ -50,7 +51,7 @@
DCHECK_EQ(max_num_bytes_to_write % element_num_bytes(), 0u);
DCHECK_EQ(min_num_bytes_to_write % element_num_bytes(), 0u);
DCHECK_GT(max_num_bytes_to_write, 0u);
- DCHECK(consumer_open_no_lock());
+ DCHECK(consumer_open());
size_t num_bytes_to_write = 0;
if (may_discard()) {
@@ -61,8 +62,8 @@
capacity_num_bytes());
if (num_bytes_to_write > capacity_num_bytes() - current_num_bytes_) {
// Discard as much as needed (discard oldest first).
- MarkDataAsConsumedNoLock(num_bytes_to_write -
- (capacity_num_bytes() - current_num_bytes_));
+ MarkDataAsConsumed(num_bytes_to_write -
+ (capacity_num_bytes() - current_num_bytes_));
// No need to wake up write waiters, since we're definitely going to leave
// the buffer full.
}
@@ -81,11 +82,11 @@
// The amount we can write in our first |memcpy()|.
size_t num_bytes_to_write_first =
- std::min(num_bytes_to_write, GetMaxNumBytesToWriteNoLock());
+ std::min(num_bytes_to_write, GetMaxNumBytesToWrite());
// Do the first (and possibly only) |memcpy()|.
size_t first_write_index =
(start_index_ + current_num_bytes_) % capacity_num_bytes();
- EnsureBufferNoLock();
+ EnsureBuffer();
elements.GetArray(buffer_.get() + first_write_index,
num_bytes_to_write_first);
@@ -101,17 +102,17 @@
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ProducerBeginWriteDataImplNoLock(
+MojoResult LocalDataPipeImpl::ProducerBeginWriteData(
UserPointer<void*> buffer,
UserPointer<uint32_t> buffer_num_bytes,
uint32_t min_num_bytes_to_write) {
- DCHECK(consumer_open_no_lock());
+ DCHECK(consumer_open());
// The index we need to start writing at.
size_t write_index =
(start_index_ + current_num_bytes_) % capacity_num_bytes();
- size_t max_num_bytes_to_write = GetMaxNumBytesToWriteNoLock();
+ size_t max_num_bytes_to_write = GetMaxNumBytesToWrite();
if (min_num_bytes_to_write > max_num_bytes_to_write) {
// In "may discard" mode, we can always write from the write index to the
// end of the buffer.
@@ -121,7 +122,7 @@
// We should only reach here if the start index is after the write index!
DCHECK_GE(start_index_, write_index);
DCHECK_GT(min_num_bytes_to_write - max_num_bytes_to_write, 0u);
- MarkDataAsConsumedNoLock(min_num_bytes_to_write - max_num_bytes_to_write);
+ MarkDataAsConsumed(min_num_bytes_to_write - max_num_bytes_to_write);
max_num_bytes_to_write = min_num_bytes_to_write;
} else {
// Don't return "should wait" since you can't wait for a specified amount
@@ -134,30 +135,27 @@
if (max_num_bytes_to_write == 0)
return MOJO_RESULT_SHOULD_WAIT;
- EnsureBufferNoLock();
+ EnsureBuffer();
buffer.Put(buffer_.get() + write_index);
buffer_num_bytes.Put(static_cast<uint32_t>(max_num_bytes_to_write));
- set_producer_two_phase_max_num_bytes_written_no_lock(
+ set_producer_two_phase_max_num_bytes_written(
static_cast<uint32_t>(max_num_bytes_to_write));
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ProducerEndWriteDataImplNoLock(
- uint32_t num_bytes_written) {
- DCHECK_LE(num_bytes_written,
- producer_two_phase_max_num_bytes_written_no_lock());
+MojoResult LocalDataPipeImpl::ProducerEndWriteData(uint32_t num_bytes_written) {
+ DCHECK_LE(num_bytes_written, producer_two_phase_max_num_bytes_written());
current_num_bytes_ += num_bytes_written;
DCHECK_LE(current_num_bytes_, capacity_num_bytes());
- set_producer_two_phase_max_num_bytes_written_no_lock(0);
+ set_producer_two_phase_max_num_bytes_written(0);
return MOJO_RESULT_OK;
}
-HandleSignalsState LocalDataPipeImpl::ProducerGetHandleSignalsStateImplNoLock()
- const {
+HandleSignalsState LocalDataPipeImpl::ProducerGetHandleSignalsState() const {
HandleSignalsState rv;
- if (consumer_open_no_lock()) {
+ if (consumer_open()) {
if ((may_discard() || current_num_bytes_ < capacity_num_bytes()) &&
- !producer_in_two_phase_write_no_lock())
+ !producer_in_two_phase_write())
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
} else {
@@ -167,40 +165,38 @@
return rv;
}
-void LocalDataPipeImpl::ProducerStartSerializeImplNoLock(
- Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) {
+void LocalDataPipeImpl::ProducerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) {
// TODO(vtl): Support serializing producer data pipe handles.
*max_size = 0;
*max_platform_handles = 0;
}
-bool LocalDataPipeImpl::ProducerEndSerializeImplNoLock(
+bool LocalDataPipeImpl::ProducerEndSerialize(
Channel* channel,
void* destination,
size_t* actual_size,
embedder::PlatformHandleVector* platform_handles) {
// TODO(vtl): Support serializing producer data pipe handles.
- ProducerCloseNoLock();
+ owner()->ProducerCloseNoLock();
return false;
}
-void LocalDataPipeImpl::ConsumerCloseImplNoLock() {
+void LocalDataPipeImpl::ConsumerClose() {
// If the producer is around and in a two-phase write, we have to keep the
// buffer around. (We then don't free it until the producer is closed. This
// could be rectified, but again seems like optimizing for the uncommon case.)
- if (!producer_open_no_lock() || !producer_in_two_phase_write_no_lock())
- DestroyBufferNoLock();
+ if (!producer_open() || !producer_in_two_phase_write())
+ DestroyBuffer();
current_num_bytes_ = 0;
}
-MojoResult LocalDataPipeImpl::ConsumerReadDataImplNoLock(
- UserPointer<void> elements,
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_read,
- uint32_t min_num_bytes_to_read,
- bool peek) {
+MojoResult LocalDataPipeImpl::ConsumerReadData(UserPointer<void> elements,
+ UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_read,
+ uint32_t min_num_bytes_to_read,
+ bool peek) {
DCHECK_EQ(max_num_bytes_to_read % element_num_bytes(), 0u);
DCHECK_EQ(min_num_bytes_to_read % element_num_bytes(), 0u);
DCHECK_GT(max_num_bytes_to_read, 0u);
@@ -208,20 +204,20 @@
if (min_num_bytes_to_read > current_num_bytes_) {
// Don't return "should wait" since you can't wait for a specified amount of
// data.
- return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
size_t num_bytes_to_read =
std::min(static_cast<size_t>(max_num_bytes_to_read), current_num_bytes_);
if (num_bytes_to_read == 0) {
- return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
// The amount we can read in our first |memcpy()|.
size_t num_bytes_to_read_first =
- std::min(num_bytes_to_read, GetMaxNumBytesToReadNoLock());
+ std::min(num_bytes_to_read, GetMaxNumBytesToRead());
elements.PutArray(buffer_.get() + start_index_, num_bytes_to_read_first);
if (num_bytes_to_read_first < num_bytes_to_read) {
@@ -231,12 +227,12 @@
}
if (!peek)
- MarkDataAsConsumedNoLock(num_bytes_to_read);
+ MarkDataAsConsumed(num_bytes_to_read);
num_bytes.Put(static_cast<uint32_t>(num_bytes_to_read));
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ConsumerDiscardDataImplNoLock(
+MojoResult LocalDataPipeImpl::ConsumerDiscardData(
UserPointer<uint32_t> num_bytes,
uint32_t max_num_bytes_to_discard,
uint32_t min_num_bytes_to_discard) {
@@ -247,101 +243,98 @@
if (min_num_bytes_to_discard > current_num_bytes_) {
// Don't return "should wait" since you can't wait for a specified amount of
// data.
- return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
// Be consistent with other operations; error if no data available.
if (current_num_bytes_ == 0) {
- return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
size_t num_bytes_to_discard = std::min(
static_cast<size_t>(max_num_bytes_to_discard), current_num_bytes_);
- MarkDataAsConsumedNoLock(num_bytes_to_discard);
+ MarkDataAsConsumed(num_bytes_to_discard);
num_bytes.Put(static_cast<uint32_t>(num_bytes_to_discard));
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ConsumerQueryDataImplNoLock(
+MojoResult LocalDataPipeImpl::ConsumerQueryData(
UserPointer<uint32_t> num_bytes) {
// Note: This cast is safe, since the capacity fits into a |uint32_t|.
num_bytes.Put(static_cast<uint32_t>(current_num_bytes_));
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ConsumerBeginReadDataImplNoLock(
+MojoResult LocalDataPipeImpl::ConsumerBeginReadData(
UserPointer<const void*> buffer,
UserPointer<uint32_t> buffer_num_bytes,
uint32_t min_num_bytes_to_read) {
- size_t max_num_bytes_to_read = GetMaxNumBytesToReadNoLock();
+ size_t max_num_bytes_to_read = GetMaxNumBytesToRead();
if (min_num_bytes_to_read > max_num_bytes_to_read) {
// Don't return "should wait" since you can't wait for a specified amount of
// data.
- return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
// Don't go into a two-phase read if there's no data.
if (max_num_bytes_to_read == 0) {
- return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT
- : MOJO_RESULT_FAILED_PRECONDITION;
+ return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+ : MOJO_RESULT_FAILED_PRECONDITION;
}
buffer.Put(buffer_.get() + start_index_);
buffer_num_bytes.Put(static_cast<uint32_t>(max_num_bytes_to_read));
- set_consumer_two_phase_max_num_bytes_read_no_lock(
+ set_consumer_two_phase_max_num_bytes_read(
static_cast<uint32_t>(max_num_bytes_to_read));
return MOJO_RESULT_OK;
}
-MojoResult LocalDataPipeImpl::ConsumerEndReadDataImplNoLock(
- uint32_t num_bytes_read) {
- DCHECK_LE(num_bytes_read, consumer_two_phase_max_num_bytes_read_no_lock());
+MojoResult LocalDataPipeImpl::ConsumerEndReadData(uint32_t num_bytes_read) {
+ DCHECK_LE(num_bytes_read, consumer_two_phase_max_num_bytes_read());
DCHECK_LE(start_index_ + num_bytes_read, capacity_num_bytes());
- MarkDataAsConsumedNoLock(num_bytes_read);
- set_consumer_two_phase_max_num_bytes_read_no_lock(0);
+ MarkDataAsConsumed(num_bytes_read);
+ set_consumer_two_phase_max_num_bytes_read(0);
return MOJO_RESULT_OK;
}
-HandleSignalsState LocalDataPipeImpl::ConsumerGetHandleSignalsStateImplNoLock()
- const {
+HandleSignalsState LocalDataPipeImpl::ConsumerGetHandleSignalsState() const {
HandleSignalsState rv;
if (current_num_bytes_ > 0) {
- if (!consumer_in_two_phase_read_no_lock())
+ if (!consumer_in_two_phase_read())
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
- } else if (producer_open_no_lock()) {
+ } else if (producer_open()) {
rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
}
- if (!producer_open_no_lock())
+ if (!producer_open())
rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
return rv;
}
-void LocalDataPipeImpl::ConsumerStartSerializeImplNoLock(
- Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) {
+void LocalDataPipeImpl::ConsumerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) {
// TODO(vtl): Support serializing consumer data pipe handles.
*max_size = 0;
*max_platform_handles = 0;
}
-bool LocalDataPipeImpl::ConsumerEndSerializeImplNoLock(
+bool LocalDataPipeImpl::ConsumerEndSerialize(
Channel* channel,
void* destination,
size_t* actual_size,
embedder::PlatformHandleVector* platform_handles) {
// TODO(vtl): Support serializing consumer data pipe handles.
- ConsumerCloseNoLock();
+ owner()->ConsumerCloseNoLock();
return false;
}
-void LocalDataPipeImpl::EnsureBufferNoLock() {
- DCHECK(producer_open_no_lock());
+void LocalDataPipeImpl::EnsureBuffer() {
+ DCHECK(producer_open());
if (buffer_)
return;
buffer_.reset(static_cast<char*>(
@@ -349,7 +342,7 @@
GetConfiguration().data_pipe_buffer_alignment_bytes)));
}
-void LocalDataPipeImpl::DestroyBufferNoLock() {
+void LocalDataPipeImpl::DestroyBuffer() {
#ifndef NDEBUG
// Scribble on the buffer to help detect use-after-frees. (This also helps the
// unit test detect certain bugs without needing ASAN or similar.)
@@ -359,7 +352,7 @@
buffer_.reset();
}
-size_t LocalDataPipeImpl::GetMaxNumBytesToWriteNoLock() {
+size_t LocalDataPipeImpl::GetMaxNumBytesToWrite() {
size_t next_index = start_index_ + current_num_bytes_;
if (next_index >= capacity_num_bytes()) {
next_index %= capacity_num_bytes();
@@ -371,13 +364,13 @@
return capacity_num_bytes() - next_index;
}
-size_t LocalDataPipeImpl::GetMaxNumBytesToReadNoLock() {
+size_t LocalDataPipeImpl::GetMaxNumBytesToRead() {
if (start_index_ + current_num_bytes_ > capacity_num_bytes())
return capacity_num_bytes() - start_index_;
return current_num_bytes_;
}
-void LocalDataPipeImpl::MarkDataAsConsumedNoLock(size_t num_bytes) {
+void LocalDataPipeImpl::MarkDataAsConsumed(size_t num_bytes) {
DCHECK_LE(num_bytes, current_num_bytes_);
start_index_ += num_bytes;
start_index_ %= capacity_num_bytes();
diff --git a/mojo/edk/system/local_data_pipe_impl.h b/mojo/edk/system/local_data_pipe_impl.h
index c241366..14defcb 100644
--- a/mojo/edk/system/local_data_pipe_impl.h
+++ b/mojo/edk/system/local_data_pipe_impl.h
@@ -7,91 +7,79 @@
#include "base/macros.h"
#include "base/memory/aligned_memory.h"
-#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
-#include "mojo/edk/system/data_pipe.h"
+#include "mojo/edk/system/data_pipe_impl.h"
#include "mojo/edk/system/system_impl_export.h"
namespace mojo {
namespace system {
-// |LocalDataPipeImpl| is a subclass that "implements" |DataPipe| for data pipes
-// whose producer and consumer are both local. This class is thread-safe (with
-// protection provided by |DataPipe|'s |lock_|.
-class MOJO_SYSTEM_IMPL_EXPORT LocalDataPipeImpl : public DataPipe {
- public:
- // |validated_options| should be the output of |DataPipe::ValidateOptions()|.
- // In particular: |struct_size| is ignored (so |validated_options| must be the
- // current version of the struct) and |capacity_num_bytes| must be nonzero.
- explicit LocalDataPipeImpl(
- const MojoCreateDataPipeOptions& validated_options);
+class DataPipe;
- private:
- friend class base::RefCountedThreadSafe<LocalDataPipeImpl>;
+// |LocalDataPipeImpl| is a subclass that "implements" |DataPipe| for data pipes
+// whose producer and consumer are both local. See |DataPipeImpl| for more
+// details.
+class MOJO_SYSTEM_IMPL_EXPORT LocalDataPipeImpl : public DataPipeImpl {
+ public:
+ LocalDataPipeImpl();
~LocalDataPipeImpl() override;
- // |DataPipe| implementation:
- void ProducerCloseImplNoLock() override;
- MojoResult ProducerWriteDataImplNoLock(
- UserPointer<const void> elements,
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_write,
- uint32_t min_num_bytes_to_write) override;
- MojoResult ProducerBeginWriteDataImplNoLock(
- UserPointer<void*> buffer,
- UserPointer<uint32_t> buffer_num_bytes,
- uint32_t min_num_bytes_to_write) override;
- MojoResult ProducerEndWriteDataImplNoLock(
- uint32_t num_bytes_written) override;
- HandleSignalsState ProducerGetHandleSignalsStateImplNoLock() const override;
- void ProducerStartSerializeImplNoLock(Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) override;
- bool ProducerEndSerializeImplNoLock(
+ private:
+ // |DataPipeImpl| implementation:
+ void ProducerClose() override;
+ MojoResult ProducerWriteData(UserPointer<const void> elements,
+ UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_write,
+ uint32_t min_num_bytes_to_write) override;
+ MojoResult ProducerBeginWriteData(UserPointer<void*> buffer,
+ UserPointer<uint32_t> buffer_num_bytes,
+ uint32_t min_num_bytes_to_write) override;
+ MojoResult ProducerEndWriteData(uint32_t num_bytes_written) override;
+ HandleSignalsState ProducerGetHandleSignalsState() const override;
+ void ProducerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) override;
+ bool ProducerEndSerialize(
Channel* channel,
void* destination,
size_t* actual_size,
embedder::PlatformHandleVector* platform_handles) override;
- void ConsumerCloseImplNoLock() override;
- MojoResult ConsumerReadDataImplNoLock(UserPointer<void> elements,
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_read,
- uint32_t min_num_bytes_to_read,
- bool peek) override;
- MojoResult ConsumerDiscardDataImplNoLock(
- UserPointer<uint32_t> num_bytes,
- uint32_t max_num_bytes_to_discard,
- uint32_t min_num_bytes_to_discard) override;
- MojoResult ConsumerQueryDataImplNoLock(
- UserPointer<uint32_t> num_bytes) override;
- MojoResult ConsumerBeginReadDataImplNoLock(
- UserPointer<const void*> buffer,
- UserPointer<uint32_t> buffer_num_bytes,
- uint32_t min_num_bytes_to_read) override;
- MojoResult ConsumerEndReadDataImplNoLock(uint32_t num_bytes_read) override;
- HandleSignalsState ConsumerGetHandleSignalsStateImplNoLock() const override;
- void ConsumerStartSerializeImplNoLock(Channel* channel,
- size_t* max_size,
- size_t* max_platform_handles) override;
- bool ConsumerEndSerializeImplNoLock(
+ void ConsumerClose() override;
+ MojoResult ConsumerReadData(UserPointer<void> elements,
+ UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_read,
+ uint32_t min_num_bytes_to_read,
+ bool peek) override;
+ MojoResult ConsumerDiscardData(UserPointer<uint32_t> num_bytes,
+ uint32_t max_num_bytes_to_discard,
+ uint32_t min_num_bytes_to_discard) override;
+ MojoResult ConsumerQueryData(UserPointer<uint32_t> num_bytes) override;
+ MojoResult ConsumerBeginReadData(UserPointer<const void*> buffer,
+ UserPointer<uint32_t> buffer_num_bytes,
+ uint32_t min_num_bytes_to_read) override;
+ MojoResult ConsumerEndReadData(uint32_t num_bytes_read) override;
+ HandleSignalsState ConsumerGetHandleSignalsState() const override;
+ void ConsumerStartSerialize(Channel* channel,
+ size_t* max_size,
+ size_t* max_platform_handles) override;
+ bool ConsumerEndSerialize(
Channel* channel,
void* destination,
size_t* actual_size,
embedder::PlatformHandleVector* platform_handles) override;
- void EnsureBufferNoLock();
- void DestroyBufferNoLock();
+ void EnsureBuffer();
+ void DestroyBuffer();
// Get the maximum (single) write/read size right now (in number of elements);
// result fits in a |uint32_t|.
- size_t GetMaxNumBytesToWriteNoLock();
- size_t GetMaxNumBytesToReadNoLock();
+ size_t GetMaxNumBytesToWrite();
+ size_t GetMaxNumBytesToRead();
// Marks the given number of bytes as consumed/discarded. |num_bytes| must be
// greater than |current_num_bytes_|.
- void MarkDataAsConsumedNoLock(size_t num_bytes);
+ void MarkDataAsConsumed(size_t num_bytes);
- // The members below are protected by |DataPipe|'s |lock_|:
scoped_ptr<char, base::AlignedFreeDeleter> buffer_;
// Circular buffer.
size_t start_index_;
diff --git a/mojo/edk/system/local_data_pipe_impl_unittest.cc b/mojo/edk/system/local_data_pipe_impl_unittest.cc
index 1e66903..3b42e57 100644
--- a/mojo/edk/system/local_data_pipe_impl_unittest.cc
+++ b/mojo/edk/system/local_data_pipe_impl_unittest.cc
@@ -27,7 +27,7 @@
MojoCreateDataPipeOptions default_options = {0};
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
NullUserPointer(), &default_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(default_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(default_options));
dp->ProducerClose();
dp->ConsumerClose();
}
@@ -44,8 +44,7 @@
EXPECT_EQ(MOJO_RESULT_OK,
DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
&validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
dp->ProducerClose();
dp->ConsumerClose();
}
@@ -60,8 +59,7 @@
EXPECT_EQ(MOJO_RESULT_OK,
DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
&validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
dp->ProducerClose();
dp->ConsumerClose();
}
@@ -76,8 +74,7 @@
EXPECT_EQ(MOJO_RESULT_OK,
DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
&validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
dp->ProducerClose();
dp->ConsumerClose();
}
@@ -93,8 +90,7 @@
EXPECT_EQ(MOJO_RESULT_OK,
DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
&validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
dp->ProducerClose();
dp->ConsumerClose();
}
@@ -111,7 +107,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
int32_t elements[10] = {0};
uint32_t num_bytes = 0;
@@ -231,7 +227,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
Waiter waiter;
uint32_t context = 0;
HandleSignalsState hss;
@@ -412,8 +408,7 @@
// Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on producer.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Add a waiter.
waiter.Init();
ASSERT_EQ(MOJO_RESULT_OK,
@@ -437,8 +432,7 @@
// Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on consumer.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Add a waiter.
waiter.Init();
ASSERT_EQ(MOJO_RESULT_OK,
@@ -473,8 +467,7 @@
MakeUserPointer(&options), &validated_options));
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
Waiter waiter;
uint32_t context = 0;
HandleSignalsState hss;
@@ -629,8 +622,7 @@
// Test with two-phase APIs and closing the producer with an active consumer
// waiter.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
Waiter waiter;
uint32_t context = 0;
HandleSignalsState hss;
@@ -732,7 +724,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
Waiter waiter;
HandleSignalsState hss;
@@ -885,7 +877,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
Waiter waiter;
HandleSignalsState hss;
@@ -1017,7 +1009,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
int32_t buffer[100] = {0};
uint32_t num_bytes = 0;
@@ -1210,7 +1202,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Try writing way too much.
uint32_t num_bytes = 20u * sizeof(int32_t);
@@ -1370,7 +1362,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Try writing way too much.
uint32_t num_bytes = 20u * sizeof(int32_t);
@@ -1469,7 +1461,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Try writing way too much (two-phase).
uint32_t num_bytes = 20u * sizeof(int32_t);
@@ -1611,7 +1603,7 @@
// pipe more space.
ASSERT_EQ(100u, validated_options.capacity_num_bytes);
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Write 20 bytes.
uint32_t num_bytes = 20u;
@@ -1694,8 +1686,7 @@
// Close producer first, then consumer.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Write some data, so we'll have something to read.
uint32_t num_bytes = kTestDataSize;
@@ -1751,8 +1742,7 @@
// Close consumer first, then producer.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Write some data, so we'll have something to read.
uint32_t num_bytes = kTestDataSize;
@@ -1808,8 +1798,7 @@
// Test closing the consumer first, then the producer, with an active
// two-phase write.
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Start two-phase write.
void* write_buffer_ptr = nullptr;
@@ -1826,8 +1815,7 @@
// Test closing the producer and then trying to read (with no data).
{
- scoped_refptr<LocalDataPipeImpl> dp(
- new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Write some data, so we'll have something to read.
uint32_t num_bytes = kTestDataSize;
@@ -1890,7 +1878,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// No data.
uint32_t num_bytes = 1000u;
@@ -2016,7 +2004,7 @@
EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
MakeUserPointer(&options), &validated_options));
- scoped_refptr<LocalDataPipeImpl> dp(new LocalDataPipeImpl(validated_options));
+ scoped_refptr<DataPipe> dp(DataPipe::CreateLocal(validated_options));
// Write some elements.
char elements[2] = {'a', 'b'};
