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'};