Mojo: Change data pipe API to take # bytes instead of # elements.

I discovered a whole bunch of bugs in the existing (totally untested)
code. I also noted some refactoring that I want to do urgently (some of
which help with those bugs). I noted them with "TODO" and "FIXME" (the
latter so that it's easy for me to grep) as things to be fixed ASAP.

I wanted to keep this change to (nearly) only the straightforward
conversion of the API. The only bugs fixed are that the checks to
handle all-or-none mode in two-phase write/read were backwards.

R=sky@chromium.org, darin@chromium.org

Review URL: https://codereview.chromium.org/107663010

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@241321 0039d316-1c4b-4281-b951-d872f2087c98

16 files changed, 356 insertions, 308 deletions

diff --git a/mojo/public/system/core.h b/mojo/public/system/core.h
index 097d8fb..af21204 100644
--- a/mojo/public/system/core.h
+++ b/mojo/public/system/core.h
@@ -194,16 +194,22 @@ const MojoReadMessageFlags MOJO_READ_MESSAGE_FLAG_MAY_DISCARD = 1 << 0;
 
 // |MojoCreateDataPipeOptions|: Used to specify creation parameters for a data
 // pipe to |MojoCreateDataPipe()|.
-//   |MojoCreateDataPipeOptionsFlags|: Used to specify different modes of
+//   |size_t struct_size|: Set to the size of the |MojoCreateDataPipeOptions|
+//       struct. (Used to allow for future extensions.)
+//   |MojoCreateDataPipeOptionsFlags flags|: Used to specify different modes of
 //       operation.
 //     |MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE|: No flags; default mode.
 //     |MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD|: May discard data for
 //         whatever reason; best-effort delivery. In particular, if the capacity
 //         is reached, old data may be discard to make room for new data.
-//
-// |element_size * capacity_num_elements| must be less than 2^32 (i.e., it must
-// fit into a 32-bit unsigned integer).
-// TODO(vtl): Finish this.
+//   |uint32_t element_num_bytes|: The size of an element, in bytes. All
+//       transactions and buffers will consist of an integral number of
+//       elements. Must be nonzero.
+//   |uint32_t capacity_num_bytes|: The capacity of the data pipe, in number of
+//       bytes; must be a multiple of |element_num_bytes|. The data pipe will
+//       always be able to queue AT LEAST this much data. Set to zero to opt for
+//       a system-dependent automatically-calculated capacity (which will always
+//       be at least one element).
 typedef uint32_t MojoCreateDataPipeOptionsFlags;
 
 #ifdef __cplusplus
@@ -219,10 +225,10 @@ const MojoCreateDataPipeOptionsFlags
 #endif
 
 struct MojoCreateDataPipeOptions {
-  size_t struct_size;  // Set to the size of this structure.
+  size_t struct_size;
   MojoCreateDataPipeOptionsFlags flags;
-  uint32_t element_size;  // Must be nonzero.
-  uint32_t capacity_num_elements;  // Zero means "default"/automatic.
+  uint32_t element_num_bytes;
+  uint32_t capacity_num_bytes;
 };
 
 // |MojoWriteDataFlags|: Used to specify different modes to |MojoWriteData()|
@@ -431,41 +437,41 @@ MOJO_SYSTEM_EXPORT MojoResult MojoCreateDataPipe(
 MOJO_SYSTEM_EXPORT MojoResult MojoWriteData(
     MojoHandle data_pipe_producer_handle,
     const void* elements,
-    uint32_t* num_elements,
+    uint32_t* num_bytes,
     MojoWriteDataFlags flags);
 
-// TODO(vtl): Note to self: |buffer_num_elements| is an "in-out" parameter:
-// on the "in" side, |*buffer_num_elements| is the number requested; on success,
-// on the "out" side, it's the number available (which may be GREATER or LESS
-// than the number requested; if the "all-or-nothing" flag is set, it's AT LEAST
-// the number requested).
+// TODO(vtl): Note to self: |buffer_num_bytes| is an "in-out" parameter: on the
+// "in" side, |*buffer_num_bytes| is the number requested; on success, on the
+// "out" side, it's the number available (which may be GREATER or LESS than the
+// number requested; if the "all-or-nothing" flag is set, it's AT LEAST the
+// number requested).
 MOJO_SYSTEM_EXPORT MojoResult MojoBeginWriteData(
     MojoHandle data_pipe_producer_handle,
     void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoWriteDataFlags flags);
 
 MOJO_SYSTEM_EXPORT MojoResult MojoEndWriteData(
     MojoHandle data_pipe_producer_handle,
-    uint32_t num_elements_written);
+    uint32_t num_bytes_written);
 
 // TODO(vtl): Note to self: If |MOJO_READ_DATA_FLAG_QUERY| is set, then
 // |elements| must be null (and nothing will be read).
 MOJO_SYSTEM_EXPORT MojoResult MojoReadData(
     MojoHandle data_pipe_consumer_handle,
     void* elements,
-    uint32_t* num_elements,
+    uint32_t* num_bytes,
     MojoReadDataFlags flags);
 
 MOJO_SYSTEM_EXPORT MojoResult MojoBeginReadData(
     MojoHandle data_pipe_consumer_handle,
     const void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoReadDataFlags flags);
 
 MOJO_SYSTEM_EXPORT MojoResult MojoEndReadData(
     MojoHandle data_pipe_consumer_handle,
-    uint32_t num_elements_read);
+    uint32_t num_bytes_read);
 
 #ifdef __cplusplus
 }  // extern "C"
diff --git a/mojo/system/core_impl.cc b/mojo/system/core_impl.cc
index f9260a1..1ed82f4 100644
--- a/mojo/system/core_impl.cc
+++ b/mojo/system/core_impl.cc
@@ -397,70 +397,70 @@ MojoResult CoreImpl::CreateDataPipe(const MojoCreateDataPipeOptions* options,
 
 MojoResult CoreImpl::WriteData(MojoHandle data_pipe_producer_handle,
                                const void* elements,
-                               uint32_t* num_elements,
+                               uint32_t* num_bytes,
                                MojoWriteDataFlags flags) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->WriteData(elements, num_elements, flags);
+  return dispatcher->WriteData(elements, num_bytes, flags);
 }
 
 MojoResult CoreImpl::BeginWriteData(MojoHandle data_pipe_producer_handle,
                                     void** buffer,
-                                    uint32_t* buffer_num_elements,
+                                    uint32_t* buffer_num_bytes,
                                     MojoWriteDataFlags flags) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->BeginWriteData(buffer, buffer_num_elements, flags);
+  return dispatcher->BeginWriteData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult CoreImpl::EndWriteData(MojoHandle data_pipe_producer_handle,
-                                  uint32_t num_elements_written) {
+                                  uint32_t num_bytes_written) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->EndWriteData(num_elements_written);
+  return dispatcher->EndWriteData(num_bytes_written);
 }
 
 MojoResult CoreImpl::ReadData(MojoHandle data_pipe_consumer_handle,
                               void* elements,
-                              uint32_t* num_elements,
+                              uint32_t* num_bytes,
                               MojoReadDataFlags flags) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->ReadData(elements, num_elements, flags);
+  return dispatcher->ReadData(elements, num_bytes, flags);
 }
 
 MojoResult CoreImpl::BeginReadData(MojoHandle data_pipe_consumer_handle,
                                    const void** buffer,
-                                   uint32_t* buffer_num_elements,
+                                   uint32_t* buffer_num_bytes,
                                    MojoReadDataFlags flags) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->BeginReadData(buffer, buffer_num_elements, flags);
+  return dispatcher->BeginReadData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult CoreImpl::EndReadData(MojoHandle data_pipe_consumer_handle,
-                                 uint32_t num_elements_read) {
+                                 uint32_t num_bytes_read) {
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher.get())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return dispatcher->EndReadData(num_elements_read);
+  return dispatcher->EndReadData(num_bytes_read);
 }
 
 CoreImpl::CoreImpl()
diff --git a/mojo/system/core_impl.h b/mojo/system/core_impl.h
index 1b47d0a2..57be33d 100644
--- a/mojo/system/core_impl.h
+++ b/mojo/system/core_impl.h
@@ -62,24 +62,24 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public CorePrivate {
       MojoHandle* data_pipe_consumer_handle) OVERRIDE;
   virtual MojoResult WriteData(MojoHandle data_pipe_producer_handle,
                                const void* elements,
-                               uint32_t* num_elements,
+                               uint32_t* num_bytes,
                                MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult BeginWriteData(MojoHandle data_pipe_producer_handle,
                                     void** buffer,
-                                    uint32_t* buffer_num_elements,
+                                    uint32_t* buffer_num_bytes,
                                     MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult EndWriteData(MojoHandle data_pipe_producer_handle,
-                                  uint32_t num_elements_written) OVERRIDE;
+                                  uint32_t num_bytes_written) OVERRIDE;
   virtual MojoResult ReadData(MojoHandle data_pipe_consumer_handle,
                               void* elements,
-                              uint32_t* num_elements,
+                              uint32_t* num_bytes,
                               MojoReadDataFlags flags) OVERRIDE;
   virtual MojoResult BeginReadData(MojoHandle data_pipe_consumer_handle,
                                    const void** buffer,
-                                   uint32_t* buffer_num_elements,
+                                   uint32_t* buffer_num_bytes,
                                    MojoReadDataFlags flags) OVERRIDE;
   virtual MojoResult EndReadData(MojoHandle data_pipe_consumer_handle,
-                                 uint32_t num_elements_read) OVERRIDE;
+                                 uint32_t num_bytes_read) OVERRIDE;
 
  private:
   friend class test::CoreTestBase;
diff --git a/mojo/system/core_test_base.cc b/mojo/system/core_test_base.cc
index 5edcb0f..e702321 100644
--- a/mojo/system/core_test_base.cc
+++ b/mojo/system/core_test_base.cc
@@ -79,7 +79,7 @@ class MockDispatcher : public Dispatcher {
 
   virtual MojoResult WriteDataImplNoLock(
       const void* /*elements*/,
-      uint32_t* /*num_elements*/,
+      uint32_t* /*num_bytes*/,
       MojoWriteDataFlags /*flags*/) OVERRIDE {
     info_->IncrementWriteDataCallCount();
     lock().AssertAcquired();
@@ -88,7 +88,7 @@ class MockDispatcher : public Dispatcher {
 
   virtual MojoResult BeginWriteDataImplNoLock(
       void** /*buffer*/,
-      uint32_t* /*buffer_num_elements*/,
+      uint32_t* /*buffer_num_bytes*/,
       MojoWriteDataFlags /*flags*/) OVERRIDE {
     info_->IncrementBeginWriteDataCallCount();
     lock().AssertAcquired();
@@ -96,14 +96,14 @@ class MockDispatcher : public Dispatcher {
   }
 
   virtual MojoResult EndWriteDataImplNoLock(
-      uint32_t /*num_elements_written*/) OVERRIDE {
+      uint32_t /*num_bytes_written*/) OVERRIDE {
     info_->IncrementEndWriteDataCallCount();
     lock().AssertAcquired();
     return MOJO_RESULT_UNIMPLEMENTED;
   }
 
   virtual MojoResult ReadDataImplNoLock(void* /*elements*/,
-                                        uint32_t* /*num_elements*/,
+                                        uint32_t* /*num_bytes*/,
                                         MojoReadDataFlags /*flags*/) OVERRIDE {
     info_->IncrementReadDataCallCount();
     lock().AssertAcquired();
@@ -112,7 +112,7 @@ class MockDispatcher : public Dispatcher {
 
   virtual MojoResult BeginReadDataImplNoLock(
       const void** /*buffer*/,
-      uint32_t* /*buffer_num_elements*/,
+      uint32_t* /*buffer_num_bytes*/,
       MojoReadDataFlags /*flags*/) OVERRIDE {
     info_->IncrementBeginReadDataCallCount();
     lock().AssertAcquired();
@@ -120,7 +120,7 @@ class MockDispatcher : public Dispatcher {
   }
 
   virtual MojoResult EndReadDataImplNoLock(
-      uint32_t /*num_elements_read*/) OVERRIDE {
+      uint32_t /*num_bytes_read*/) OVERRIDE {
     info_->IncrementEndReadDataCallCount();
     lock().AssertAcquired();
     return MOJO_RESULT_UNIMPLEMENTED;
diff --git a/mojo/system/data_pipe.cc b/mojo/system/data_pipe.cc
index e1594ea..360ec63 100644
--- a/mojo/system/data_pipe.cc
+++ b/mojo/system/data_pipe.cc
@@ -27,11 +27,12 @@ void DataPipe::ProducerClose() {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
   producer_waiter_list_.reset();
+  // TODO(vtl): FIXME -- "cancel" any two-phase write (do we need to do this?)
   ProducerCloseImplNoLock();
 }
 
 MojoResult DataPipe::ProducerWriteData(const void* elements,
-                                       uint32_t* num_elements,
+                                       uint32_t* num_bytes,
                                        MojoWriteDataFlags flags) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
@@ -39,29 +40,18 @@ MojoResult DataPipe::ProducerWriteData(const void* elements,
   if (producer_in_two_phase_write_)
     return MOJO_RESULT_BUSY;
 
-  // TODO(vtl): This implementation may write less than requested, even if room
-  // is available. Fix this. (Probably make a subclass-specific impl.)
-  void* buffer = NULL;
-  uint32_t buffer_num_elements = *num_elements;
-  MojoResult rv = ProducerBeginWriteDataImplNoLock(&buffer,
-                                                   &buffer_num_elements,
-                                                   flags);
-  if (rv != MOJO_RESULT_OK)
-    return rv;
-
-  uint32_t num_elements_to_write = std::min(*num_elements, buffer_num_elements);
-  memcpy(buffer, elements, num_elements_to_write * element_size_);
+  // Returning "busy" takes priority over "invalid argument".
+  if (*num_bytes % element_num_bytes_ != 0)
+    return MOJO_RESULT_INVALID_ARGUMENT;
 
-  rv = ProducerEndWriteDataImplNoLock(num_elements_to_write);
-  if (rv != MOJO_RESULT_OK)
-    return rv;
+  if (*num_bytes == 0)
+    return MOJO_RESULT_OK;  // Nothing to do.
 
-  *num_elements = num_elements_to_write;
-  return MOJO_RESULT_OK;
+  return ProducerWriteDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
-                                            uint32_t* buffer_num_elements,
+                                            uint32_t* buffer_num_bytes,
                                             MojoWriteDataFlags flags) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
@@ -70,7 +60,7 @@ MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
     return MOJO_RESULT_BUSY;
 
   MojoResult rv = ProducerBeginWriteDataImplNoLock(buffer,
-                                                   buffer_num_elements,
+                                                   buffer_num_bytes,
                                                    flags);
   if (rv != MOJO_RESULT_OK)
     return rv;
@@ -79,14 +69,14 @@ MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
   return MOJO_RESULT_OK;
 }
 
-MojoResult DataPipe::ProducerEndWriteData(uint32_t num_elements_written) {
+MojoResult DataPipe::ProducerEndWriteData(uint32_t num_bytes_written) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   if (!producer_in_two_phase_write_)
     return MOJO_RESULT_FAILED_PRECONDITION;
 
-  MojoResult rv = ProducerEndWriteDataImplNoLock(num_elements_written);
+  MojoResult rv = ProducerEndWriteDataImplNoLock(num_bytes_written);
   producer_in_two_phase_write_ = false;  // End two-phase write even on failure.
   return rv;
 }
@@ -122,11 +112,12 @@ void DataPipe::ConsumerClose() {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
   consumer_waiter_list_.reset();
+  // TODO(vtl): FIXME -- "cancel" any two-phase read (do we need to do this?)
   ConsumerCloseImplNoLock();
 }
 
 MojoResult DataPipe::ConsumerReadData(void* elements,
-                                      uint32_t* num_elements,
+                                      uint32_t* num_bytes,
                                       MojoReadDataFlags flags) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
@@ -134,36 +125,26 @@ MojoResult DataPipe::ConsumerReadData(void* elements,
   if (consumer_in_two_phase_read_)
     return MOJO_RESULT_BUSY;
 
-  if ((flags & MOJO_READ_DATA_FLAG_DISCARD)) {
-    return ConsumerDiscardDataNoLock(num_elements,
-                                     (flags & MOJO_READ_DATA_FLAG_ALL_OR_NONE));
-  }
+  // Note: Don't need to validate |*num_bytes| for query.
   if ((flags & MOJO_READ_DATA_FLAG_QUERY))
-    return ConsumerQueryDataNoLock(num_elements);
-
-  // TODO(vtl): This implementation may write less than requested, even if room
-  // is available. Fix this. (Probably make a subclass-specific impl.)
-  const void* buffer = NULL;
-  uint32_t buffer_num_elements = 0;
-  MojoResult rv = ConsumerBeginReadDataImplNoLock(&buffer,
-                                                  &buffer_num_elements,
-                                                  flags);
-  if (rv != MOJO_RESULT_OK)
-    return rv;
+    return ConsumerQueryDataNoLock(num_bytes);
 
-  uint32_t num_elements_to_read = std::min(*num_elements, buffer_num_elements);
-  memcpy(elements, buffer, num_elements_to_read * element_size_);
+  if (*num_bytes % element_num_bytes_ != 0)
+    return MOJO_RESULT_INVALID_ARGUMENT;
 
-  rv = ConsumerEndReadDataImplNoLock(num_elements_to_read);
-  if (rv != MOJO_RESULT_OK)
-    return rv;
+  if (*num_bytes == 0)
+    return MOJO_RESULT_OK;  // Nothing to do (for read or for discard).
 
-  *num_elements = num_elements_to_read;
-  return MOJO_RESULT_OK;
+  if ((flags & MOJO_READ_DATA_FLAG_DISCARD)) {
+    return ConsumerDiscardDataNoLock(num_bytes,
+                                     (flags & MOJO_READ_DATA_FLAG_ALL_OR_NONE));
+  }
+
+  return ConsumerReadDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult DataPipe::ConsumerBeginReadData(const void** buffer,
-                                           uint32_t* buffer_num_elements,
+                                           uint32_t* buffer_num_bytes,
                                            MojoReadDataFlags flags) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
@@ -172,7 +153,7 @@ MojoResult DataPipe::ConsumerBeginReadData(const void** buffer,
     return MOJO_RESULT_BUSY;
 
   MojoResult rv = ConsumerBeginReadDataImplNoLock(buffer,
-                                                  buffer_num_elements,
+                                                  buffer_num_bytes,
                                                   flags);
   if (rv != MOJO_RESULT_OK)
     return rv;
@@ -181,14 +162,14 @@ MojoResult DataPipe::ConsumerBeginReadData(const void** buffer,
   return MOJO_RESULT_OK;
 }
 
-MojoResult DataPipe::ConsumerEndReadData(uint32_t num_elements_read) {
+MojoResult DataPipe::ConsumerEndReadData(uint32_t num_bytes_read) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (!consumer_in_two_phase_read_)
     return MOJO_RESULT_FAILED_PRECONDITION;
 
-  MojoResult rv = ConsumerEndReadDataImplNoLock(num_elements_read);
+  MojoResult rv = ConsumerEndReadDataImplNoLock(num_bytes_read);
   consumer_in_two_phase_read_ = false;  // End two-phase read even on failure.
   return rv;
 }
@@ -215,7 +196,7 @@ void DataPipe::ConsumerRemoveWaiter(Waiter* waiter) {
 }
 
 DataPipe::DataPipe(bool has_local_producer, bool has_local_consumer)
-    : element_size_(0),
+    : element_num_bytes_(0),
       producer_waiter_list_(has_local_producer ? new WaiterList() : NULL),
       consumer_waiter_list_(has_local_consumer ? new WaiterList() : NULL),
       producer_in_two_phase_write_(false),
@@ -229,28 +210,26 @@ DataPipe::~DataPipe() {
 }
 
 MojoResult DataPipe::Init(bool may_discard,
-                          size_t element_size,
-                          size_t capacity_num_elements) {
+                          size_t element_num_bytes,
+                          size_t capacity_num_bytes) {
   // No need to lock: This method is not thread-safe.
 
-  if (element_size == 0)
+  if (element_num_bytes == 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
-  if (!capacity_num_elements) {
-    // Set the capacity to the default (rounded down by element size, but always
-    // at least one element).
-    capacity_num_elements =
-        std::max(static_cast<size_t>(1),
-                 kDefaultDataPipeCapacityBytes / element_size);
+  if (capacity_num_bytes == 0) {
+    // Round the default capacity down to a multiple of the element size.
+    capacity_num_bytes = kDefaultDataPipeCapacityBytes -
+        (kDefaultDataPipeCapacityBytes % element_num_bytes);
+    // But make sure that the capacity is still at least one.
+    if (capacity_num_bytes == 0)
+      capacity_num_bytes = element_num_bytes;
   }
-  if (capacity_num_elements >
-          std::numeric_limits<uint32_t>::max() / element_size)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-  if (capacity_num_elements * element_size > kMaxDataPipeCapacityBytes)
+  if (capacity_num_bytes > kMaxDataPipeCapacityBytes)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 
   may_discard_ = may_discard;
-  element_size_ = element_size;
-  capacity_num_elements_ = capacity_num_elements;
+  element_num_bytes_ = element_num_bytes;
+  capacity_num_bytes_ = capacity_num_bytes;
   return MOJO_RESULT_OK;
 }
 
diff --git a/mojo/system/data_pipe.h b/mojo/system/data_pipe.h
index 3138435..f4b160b 100644
--- a/mojo/system/data_pipe.h
+++ b/mojo/system/data_pipe.h
@@ -31,15 +31,16 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
   // corresponding names.
   void ProducerCancelAllWaiters();
   void ProducerClose();
-  // This does not validate its arguments.
+  // This does not validate its arguments, except to check that |*num_bytes| is
+  // a multiple of |element_num_bytes_|.
   MojoResult ProducerWriteData(const void* elements,
-                               uint32_t* num_elements,
+                               uint32_t* num_bytes,
                                MojoWriteDataFlags flags);
   // This does not validate its arguments.
   MojoResult ProducerBeginWriteData(void** buffer,
-                                    uint32_t* buffer_num_elements,
+                                    uint32_t* buffer_num_bytes,
                                     MojoWriteDataFlags flags);
-  MojoResult ProducerEndWriteData(uint32_t num_elements_written);
+  MojoResult ProducerEndWriteData(uint32_t num_bytes_written);
   MojoResult ProducerAddWaiter(Waiter* waiter,
                                MojoWaitFlags flags,
                                MojoResult wake_result);
@@ -49,25 +50,21 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
   // corresponding names.
   void ConsumerCancelAllWaiters();
   void ConsumerClose();
-  // This does not validate its arguments.
+  // This does not validate its arguments, except to check that |*num_bytes| is
+  // a multiple of |element_num_bytes_|.
   MojoResult ConsumerReadData(void* elements,
-                              uint32_t* num_elements,
+                              uint32_t* num_bytes,
                               MojoReadDataFlags flags);
   // This does not validate its arguments.
   MojoResult ConsumerBeginReadData(const void** buffer,
-                                   uint32_t* buffer_num_elements,
+                                   uint32_t* buffer_num_bytes,
                                    MojoReadDataFlags flags);
-  MojoResult ConsumerEndReadData(uint32_t num_elements_read);
+  MojoResult ConsumerEndReadData(uint32_t num_bytes_read);
   MojoResult ConsumerAddWaiter(Waiter* waiter,
                                MojoWaitFlags flags,
                                MojoResult wake_result);
   void ConsumerRemoveWaiter(Waiter* waiter);
 
-  // Thread-safe and fast (they don't take the lock):
-  bool may_discard() const { return may_discard_; }
-  size_t element_size() const { return element_size_; }
-  size_t capacity_num_elements() const { return capacity_num_elements_; }
-
  protected:
   DataPipe(bool has_local_producer, bool has_local_consumer);
 
@@ -77,35 +74,49 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
   // Not thread-safe; must be called before any other methods are called. This
   // object is only usable on success.
   MojoResult Init(bool may_discard,
-                  size_t element_size,
-                  size_t capacity_num_elements);
+                  size_t element_num_bytes,
+                  size_t capacity_num_bytes);
 
   void AwakeProducerWaitersForStateChangeNoLock();
   void AwakeConsumerWaitersForStateChangeNoLock();
 
   virtual void ProducerCloseImplNoLock() = 0;
+  // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
+  virtual MojoResult ProducerWriteDataImplNoLock(const void* elements,
+                                                 uint32_t* num_bytes,
+                                                 MojoWriteDataFlags flags) = 0;
   virtual MojoResult ProducerBeginWriteDataImplNoLock(
       void** buffer,
-      uint32_t* buffer_num_elements,
+      uint32_t* buffer_num_bytes,
       MojoWriteDataFlags flags) = 0;
   virtual MojoResult ProducerEndWriteDataImplNoLock(
-      uint32_t num_elements_written) = 0;
+      uint32_t num_bytes_written) = 0;
   virtual MojoWaitFlags ProducerSatisfiedFlagsNoLock() = 0;
   virtual MojoWaitFlags ProducerSatisfiableFlagsNoLock() = 0;
 
   virtual void ConsumerCloseImplNoLock() = 0;
-  virtual MojoResult ConsumerDiscardDataNoLock(uint32_t* num_elements,
+  // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
+  virtual MojoResult ConsumerReadDataImplNoLock(void* elements,
+                                                uint32_t* num_bytes,
+                                                MojoReadDataFlags flags) = 0;
+  virtual MojoResult ConsumerDiscardDataNoLock(uint32_t* num_bytes,
                                                bool all_or_none) = 0;
-  virtual MojoResult ConsumerQueryDataNoLock(uint32_t* num_elements) = 0;
+  // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
+  virtual MojoResult ConsumerQueryDataNoLock(uint32_t* num_bytes) = 0;
   virtual MojoResult ConsumerBeginReadDataImplNoLock(
       const void** buffer,
-      uint32_t* buffer_num_elements,
+      uint32_t* buffer_num_bytes,
       MojoReadDataFlags flags) = 0;
-  virtual MojoResult ConsumerEndReadDataImplNoLock(
-      uint32_t num_elements_read) = 0;
+  virtual MojoResult ConsumerEndReadDataImplNoLock(uint32_t num_bytes_read) = 0;
   virtual MojoWaitFlags ConsumerSatisfiedFlagsNoLock() = 0;
   virtual MojoWaitFlags ConsumerSatisfiableFlagsNoLock() = 0;
 
+  // Thread-safe and fast (they don't take the lock):
+  // TODO(vtl): FIXME -- "may discard" not respected
+  bool may_discard() const { return may_discard_; }
+  size_t element_num_bytes() const { return element_num_bytes_; }
+  size_t capacity_num_bytes() const { return capacity_num_bytes_; }
+
  private:
   bool has_local_producer_no_lock() const {
     return !!producer_waiter_list_.get();
@@ -116,8 +127,8 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
 
   // Set by |Init()| and never changed afterwards:
   bool may_discard_;
-  size_t element_size_;
-  size_t capacity_num_elements_;
+  size_t element_num_bytes_;
+  size_t capacity_num_bytes_;
 
   base::Lock lock_;  // Protects the following members.
   scoped_ptr<WaiterList> producer_waiter_list_;
diff --git a/mojo/system/data_pipe_consumer_dispatcher.cc b/mojo/system/data_pipe_consumer_dispatcher.cc
index f0fc451..07878f9 100644
--- a/mojo/system/data_pipe_consumer_dispatcher.cc
+++ b/mojo/system/data_pipe_consumer_dispatcher.cc
@@ -38,11 +38,11 @@ MojoResult DataPipeConsumerDispatcher::CloseImplNoLock() {
 
 MojoResult DataPipeConsumerDispatcher::ReadDataImplNoLock(
     void* elements,
-    uint32_t* num_elements,
+    uint32_t* num_bytes,
     MojoReadDataFlags flags) {
   lock().AssertAcquired();
 
-  if (!VerifyUserPointer<uint32_t>(num_elements, 1))
+  if (!VerifyUserPointer<uint32_t>(num_bytes, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
   // These flags are mutally exclusive.
   if ((flags & MOJO_READ_DATA_FLAG_DISCARD) &&
@@ -54,37 +54,36 @@ MojoResult DataPipeConsumerDispatcher::ReadDataImplNoLock(
     elements = NULL;  // Null it out for safety.
   } else {
     // Only verify |elements| if we're neither discarding nor querying.
-    if (!VerifyUserPointerForSize(elements, data_pipe_->element_size(),
-                                  *num_elements))
+    if (!VerifyUserPointer<void>(elements, *num_bytes))
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  return data_pipe_->ConsumerReadData(elements, num_elements, flags);
+  return data_pipe_->ConsumerReadData(elements, num_bytes, flags);
 }
 
 MojoResult DataPipeConsumerDispatcher::BeginReadDataImplNoLock(
     const void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoReadDataFlags flags) {
   lock().AssertAcquired();
 
   if (!VerifyUserPointer<const void*>(buffer, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
-  if (!VerifyUserPointer<uint32_t>(buffer_num_elements, 1))
+  if (!VerifyUserPointer<uint32_t>(buffer_num_bytes, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
   // These flags may not be used in two-phase mode.
   if ((flags & MOJO_READ_DATA_FLAG_DISCARD) ||
       (flags & MOJO_READ_DATA_FLAG_QUERY))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return data_pipe_->ConsumerBeginReadData(buffer, buffer_num_elements, flags);
+  return data_pipe_->ConsumerBeginReadData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult DataPipeConsumerDispatcher::EndReadDataImplNoLock(
-    uint32_t num_elements_read) {
+    uint32_t num_bytes_read) {
   lock().AssertAcquired();
 
-  return data_pipe_->ConsumerEndReadData(num_elements_read);
+  return data_pipe_->ConsumerEndReadData(num_bytes_read);
 }
 
 MojoResult DataPipeConsumerDispatcher::AddWaiterImplNoLock(
diff --git a/mojo/system/data_pipe_consumer_dispatcher.h b/mojo/system/data_pipe_consumer_dispatcher.h
index cd24db06..e079db6 100644
--- a/mojo/system/data_pipe_consumer_dispatcher.h
+++ b/mojo/system/data_pipe_consumer_dispatcher.h
@@ -34,12 +34,12 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipeConsumerDispatcher : public Dispatcher {
   virtual void CancelAllWaitersNoLock() OVERRIDE;
   virtual MojoResult CloseImplNoLock() OVERRIDE;
   virtual MojoResult ReadDataImplNoLock(void* elements,
-                                        uint32_t* num_elements,
+                                        uint32_t* num_bytes,
                                         MojoReadDataFlags flags) OVERRIDE;
   virtual MojoResult BeginReadDataImplNoLock(const void** buffer,
-                                             uint32_t* buffer_num_elements,
+                                             uint32_t* buffer_num_bytes,
                                              MojoReadDataFlags flags) OVERRIDE;
-  virtual MojoResult EndReadDataImplNoLock(uint32_t num_elements_read) OVERRIDE;
+  virtual MojoResult EndReadDataImplNoLock(uint32_t num_bytes_read) OVERRIDE;
   virtual MojoResult AddWaiterImplNoLock(Waiter* waiter,
                                          MojoWaitFlags flags,
                                          MojoResult wake_result) OVERRIDE;
diff --git a/mojo/system/data_pipe_producer_dispatcher.cc b/mojo/system/data_pipe_producer_dispatcher.cc
index cb03c31..fce93aa 100644
--- a/mojo/system/data_pipe_producer_dispatcher.cc
+++ b/mojo/system/data_pipe_producer_dispatcher.cc
@@ -38,38 +38,37 @@ MojoResult DataPipeProducerDispatcher::CloseImplNoLock() {
 
 MojoResult DataPipeProducerDispatcher::WriteDataImplNoLock(
     const void* elements,
-    uint32_t* num_elements,
+    uint32_t* num_bytes,
     MojoWriteDataFlags flags) {
   lock().AssertAcquired();
 
-  if (!VerifyUserPointer<uint32_t>(num_elements, 1))
+  if (!VerifyUserPointer<uint32_t>(num_bytes, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
-  if (!VerifyUserPointerForSize(elements, data_pipe_->element_size(),
-                                *num_elements))
+  if (!VerifyUserPointer<void>(elements, *num_bytes))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return data_pipe_->ProducerWriteData(elements, num_elements, flags);
+  return data_pipe_->ProducerWriteData(elements, num_bytes, flags);
 }
 
 MojoResult DataPipeProducerDispatcher::BeginWriteDataImplNoLock(
     void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoWriteDataFlags flags) {
   lock().AssertAcquired();
 
   if (!VerifyUserPointer<void*>(buffer, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
-  if (!VerifyUserPointer<uint32_t>(buffer_num_elements, 1))
+  if (!VerifyUserPointer<uint32_t>(buffer_num_bytes, 1))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return data_pipe_->ProducerBeginWriteData(buffer, buffer_num_elements, flags);
+  return data_pipe_->ProducerBeginWriteData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult DataPipeProducerDispatcher::EndWriteDataImplNoLock(
-    uint32_t num_elements_written) {
+    uint32_t num_bytes_written) {
   lock().AssertAcquired();
 
-  return data_pipe_->ProducerEndWriteData(num_elements_written);
+  return data_pipe_->ProducerEndWriteData(num_bytes_written);
 }
 
 MojoResult DataPipeProducerDispatcher::AddWaiterImplNoLock(
diff --git a/mojo/system/data_pipe_producer_dispatcher.h b/mojo/system/data_pipe_producer_dispatcher.h
index 01c8783..0d6916a 100644
--- a/mojo/system/data_pipe_producer_dispatcher.h
+++ b/mojo/system/data_pipe_producer_dispatcher.h
@@ -34,14 +34,14 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipeProducerDispatcher : public Dispatcher {
   virtual void CancelAllWaitersNoLock() OVERRIDE;
   virtual MojoResult CloseImplNoLock() OVERRIDE;
   virtual MojoResult WriteDataImplNoLock(const void* elements,
-                                         uint32_t* num_elements,
+                                         uint32_t* num_bytes,
                                          MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult BeginWriteDataImplNoLock(
       void** buffer,
-      uint32_t* buffer_num_elements,
+      uint32_t* buffer_num_bytes,
       MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult EndWriteDataImplNoLock(
-      uint32_t num_elements_written) OVERRIDE;
+      uint32_t num_bytes_written) OVERRIDE;
   virtual MojoResult AddWaiterImplNoLock(Waiter* waiter,
                                          MojoWaitFlags flags,
                                          MojoResult wake_result) OVERRIDE;
diff --git a/mojo/system/dispatcher.cc b/mojo/system/dispatcher.cc
index 702df3a..bca9d9a 100644
--- a/mojo/system/dispatcher.cc
+++ b/mojo/system/dispatcher.cc
@@ -52,59 +52,59 @@ MojoResult Dispatcher::ReadMessage(
 }
 
 MojoResult Dispatcher::WriteData(const void* elements,
-                                 uint32_t* num_elements,
+                                 uint32_t* num_bytes,
                                  MojoWriteDataFlags flags) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return WriteDataImplNoLock(elements, num_elements, flags);
+  return WriteDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult Dispatcher::BeginWriteData(void** buffer,
-                                      uint32_t* buffer_num_elements,
+                                      uint32_t* buffer_num_bytes,
                                       MojoWriteDataFlags flags) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return BeginWriteDataImplNoLock(buffer, buffer_num_elements, flags);
+  return BeginWriteDataImplNoLock(buffer, buffer_num_bytes, flags);
 }
 
-MojoResult Dispatcher::EndWriteData(uint32_t num_elements_written) {
+MojoResult Dispatcher::EndWriteData(uint32_t num_bytes_written) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return EndWriteDataImplNoLock(num_elements_written);
+  return EndWriteDataImplNoLock(num_bytes_written);
 }
 
 MojoResult Dispatcher::ReadData(void* elements,
-                                uint32_t* num_elements,
+                                uint32_t* num_bytes,
                                 MojoReadDataFlags flags) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return ReadDataImplNoLock(elements, num_elements, flags);
+  return ReadDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult Dispatcher::BeginReadData(const void** buffer,
-                                     uint32_t* buffer_num_elements,
+                                     uint32_t* buffer_num_bytes,
                                      MojoReadDataFlags flags) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return BeginReadDataImplNoLock(buffer, buffer_num_elements, flags);
+  return BeginReadDataImplNoLock(buffer, buffer_num_bytes, flags);
 }
 
-MojoResult Dispatcher::EndReadData(uint32_t num_elements_read) {
+MojoResult Dispatcher::EndReadData(uint32_t num_bytes_read) {
   base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return EndReadDataImplNoLock(num_elements_read);
+  return EndReadDataImplNoLock(num_bytes_read);
 }
 
 MojoResult Dispatcher::AddWaiter(Waiter* waiter,
@@ -184,7 +184,7 @@ MojoResult Dispatcher::ReadMessageImplNoLock(
 }
 
 MojoResult Dispatcher::WriteDataImplNoLock(const void* /*elements*/,
-                                           uint32_t* /*num_elements*/,
+                                           uint32_t* /*num_bytes*/,
                                            MojoWriteDataFlags /*flags*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
@@ -192,18 +192,16 @@ MojoResult Dispatcher::WriteDataImplNoLock(const void* /*elements*/,
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
-MojoResult Dispatcher::BeginWriteDataImplNoLock(
-    void** /*buffer*/,
-    uint32_t* /*buffer_num_elements*/,
-    MojoWriteDataFlags /*flags*/) {
+MojoResult Dispatcher::BeginWriteDataImplNoLock(void** /*buffer*/,
+                                                uint32_t* /*buffer_num_bytes*/,
+                                                MojoWriteDataFlags /*flags*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
-MojoResult Dispatcher::EndWriteDataImplNoLock(
-    uint32_t /*num_elements_written*/) {
+MojoResult Dispatcher::EndWriteDataImplNoLock(uint32_t /*num_bytes_written*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
@@ -211,7 +209,7 @@ MojoResult Dispatcher::EndWriteDataImplNoLock(
 }
 
 MojoResult Dispatcher::ReadDataImplNoLock(void* /*elements*/,
-                                          uint32_t* /*num_elements*/,
+                                          uint32_t* /*num_bytes*/,
                                           MojoReadDataFlags /*flags*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
@@ -219,17 +217,16 @@ MojoResult Dispatcher::ReadDataImplNoLock(void* /*elements*/,
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
-MojoResult Dispatcher::BeginReadDataImplNoLock(
-    const void** /*buffer*/,
-    uint32_t* /*buffer_num_elements*/,
-    MojoReadDataFlags /*flags*/) {
+MojoResult Dispatcher::BeginReadDataImplNoLock(const void** /*buffer*/,
+                                               uint32_t* /*buffer_num_bytes*/,
+                                               MojoReadDataFlags /*flags*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
-MojoResult Dispatcher::EndReadDataImplNoLock(uint32_t /*num_elements_read*/) {
+MojoResult Dispatcher::EndReadDataImplNoLock(uint32_t /*num_bytes_read*/) {
   lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
diff --git a/mojo/system/dispatcher.h b/mojo/system/dispatcher.h
index ec89401..c081770 100644
--- a/mojo/system/dispatcher.h
+++ b/mojo/system/dispatcher.h
@@ -53,19 +53,19 @@ class MOJO_SYSTEM_IMPL_EXPORT Dispatcher :
       uint32_t* num_dispatchers,
       MojoReadMessageFlags flags);
   MojoResult WriteData(const void* elements,
-                       uint32_t* num_elements,
+                       uint32_t* elements_num_bytes,
                        MojoWriteDataFlags flags);
   MojoResult BeginWriteData(void** buffer,
-                            uint32_t* buffer_num_elements,
+                            uint32_t* buffer_num_bytes,
                             MojoWriteDataFlags flags);
-  MojoResult EndWriteData(uint32_t num_elements_written);
+  MojoResult EndWriteData(uint32_t num_bytes_written);
   MojoResult ReadData(void* elements,
-                      uint32_t* num_elements,
+                      uint32_t* num_bytes,
                       MojoReadDataFlags flags);
   MojoResult BeginReadData(const void** buffer,
-                           uint32_t* buffer_num_elements,
+                           uint32_t* buffer_num_bytes,
                            MojoReadDataFlags flags);
-  MojoResult EndReadData(uint32_t num_elements_read);
+  MojoResult EndReadData(uint32_t num_bytes_read);
 
   // Adds a waiter to this dispatcher. The waiter will be woken up when this
   // object changes state to satisfy |flags| with result |wake_result| (which
@@ -119,19 +119,19 @@ class MOJO_SYSTEM_IMPL_EXPORT Dispatcher :
       uint32_t* num_dispatchers,
       MojoReadMessageFlags flags);
   virtual MojoResult WriteDataImplNoLock(const void* elements,
-                                         uint32_t* num_elements,
+                                         uint32_t* num_bytes,
                                          MojoWriteDataFlags flags);
   virtual MojoResult BeginWriteDataImplNoLock(void** buffer,
-                                              uint32_t* buffer_num_elements,
+                                              uint32_t* buffer_num_bytes,
                                               MojoWriteDataFlags flags);
-  virtual MojoResult EndWriteDataImplNoLock(uint32_t num_elements_written);
+  virtual MojoResult EndWriteDataImplNoLock(uint32_t num_bytes_written);
   virtual MojoResult ReadDataImplNoLock(void* elements,
-                                        uint32_t* num_elements,
+                                        uint32_t* num_bytes,
                                         MojoReadDataFlags flags);
   virtual MojoResult BeginReadDataImplNoLock(const void** buffer,
-                                             uint32_t* buffer_num_elements,
+                                             uint32_t* buffer_num_bytes,
                                              MojoReadDataFlags flags);
-  virtual MojoResult EndReadDataImplNoLock(uint32_t num_elements_read);
+  virtual MojoResult EndReadDataImplNoLock(uint32_t num_bytes_read);
   virtual MojoResult AddWaiterImplNoLock(Waiter* waiter,
                                          MojoWaitFlags flags,
                                          MojoResult wake_result);
diff --git a/mojo/system/local_data_pipe.cc b/mojo/system/local_data_pipe.cc
index a81d421..c1bf4bc 100644
--- a/mojo/system/local_data_pipe.cc
+++ b/mojo/system/local_data_pipe.cc
@@ -3,10 +3,10 @@
 // found in the LICENSE file.
 
 // TODO(vtl): I currently potentially overflow in doing index calculations.
-// E.g., |buffer_first_element_index_| and |buffer_current_num_elements_| fit
-// into a |uint32_t|, but their sum may not. This is bad and poses a security
-// risk. (We're currently saved by the limit on capacity -- the maximum size of
-// the buffer, checked in |DataPipe::Init()|, is currently sufficiently small.
+// E.g., |start_index_| and |current_num_bytes_| fit into a |uint32_t|, but
+// their sum may not. This is bad and poses a security risk. (We're currently
+// saved by the limit on capacity -- the maximum size of the buffer, checked in
+// |DataPipe::Init()|, is currently sufficiently small.
 
 #include "mojo/system/local_data_pipe.h"
 
@@ -22,17 +22,17 @@ LocalDataPipe::LocalDataPipe()
     : DataPipe(true, true),
       producer_open_(true),
       consumer_open_(true),
-      buffer_first_element_index_(0),
-      buffer_current_num_elements_(0),
-      two_phase_max_elements_written_(0),
-      two_phase_max_elements_read_(0) {
+      start_index_(0),
+      current_num_bytes_(0),
+      two_phase_max_num_bytes_written_(0),
+      two_phase_max_num_bytes_read_(0) {
 }
 
 MojoResult LocalDataPipe::Init(const MojoCreateDataPipeOptions* options) {
   static const MojoCreateDataPipeOptions kDefaultOptions = {
-    sizeof(MojoCreateDataPipeOptions),  // |struct_size|.
-    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
-    1u,  // |element_size|.
+    sizeof(MojoCreateDataPipeOptions),
+    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,
+    1u,
     static_cast<uint32_t>(kDefaultDataPipeCapacityBytes)
   };
   if (!options)
@@ -45,8 +45,8 @@ MojoResult LocalDataPipe::Init(const MojoCreateDataPipeOptions* options) {
   // handles is immediately passed off to another process.
   return DataPipe::Init(
       (options->flags & MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD),
-      static_cast<size_t>(options->element_size),
-      static_cast<size_t>(options->capacity_num_elements));
+      static_cast<size_t>(options->element_num_bytes),
+      static_cast<size_t>(options->capacity_num_bytes));
 }
 
 LocalDataPipe::~LocalDataPipe() {
@@ -57,50 +57,82 @@ LocalDataPipe::~LocalDataPipe() {
 void LocalDataPipe::ProducerCloseImplNoLock() {
   DCHECK(producer_open_);
   producer_open_ = false;
-  if (!buffer_current_num_elements_) {
+  if (!current_num_bytes_)
     buffer_.reset();
-    buffer_current_num_elements_ = 0;
-  }
   AwakeConsumerWaitersForStateChangeNoLock();
 }
 
+MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(
+    const void* elements,
+    uint32_t* num_bytes,
+    MojoWriteDataFlags flags) {
+  DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
+  DCHECK_GT(*num_bytes, 0u);
+
+  // TODO(vtl): This implementation may write less than requested, even if room
+  // is available. Fix this.
+  void* buffer = NULL;
+  uint32_t buffer_num_bytes = *num_bytes;
+  MojoResult rv = ProducerBeginWriteDataImplNoLock(&buffer,
+                                                   &buffer_num_bytes,
+                                                   flags);
+  if (rv != MOJO_RESULT_OK)
+    return rv;
+  DCHECK_EQ(buffer_num_bytes % element_num_bytes(), 0u);
+
+  uint32_t num_bytes_to_write = std::min(*num_bytes, buffer_num_bytes);
+  memcpy(buffer, elements, num_bytes_to_write);
+
+  rv = ProducerEndWriteDataImplNoLock(num_bytes_to_write);
+  if (rv != MOJO_RESULT_OK)
+    return rv;
+
+  *num_bytes = num_bytes_to_write;
+  return MOJO_RESULT_OK;
+}
+
 MojoResult LocalDataPipe::ProducerBeginWriteDataImplNoLock(
     void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoWriteDataFlags flags) {
-  size_t max_elements_to_write = GetMaxElementsToWriteNoLock();
+  size_t max_num_bytes_to_write = GetMaxNumBytesToWriteNoLock();
   // TODO(vtl): Consider this return value.
   if ((flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) &&
-      *buffer_num_elements < max_elements_to_write)
+      *buffer_num_bytes > max_num_bytes_to_write)
     return MOJO_RESULT_OUT_OF_RANGE;
 
-  size_t next_index = (buffer_first_element_index_ +
-                       buffer_current_num_elements_) % capacity_num_elements();
+  // Don't go into a two-phase write if there's no room.
+  // TODO(vtl): Change this to "should wait" when we have that error code.
+  if (max_num_bytes_to_write == 0)
+    return MOJO_RESULT_NOT_FOUND;
+
+  size_t write_index =
+      (start_index_ + current_num_bytes_) % capacity_num_bytes();
   EnsureBufferNoLock();
-  *buffer = buffer_.get() + next_index * element_size();
-  *buffer_num_elements = static_cast<uint32_t>(max_elements_to_write);
-  two_phase_max_elements_written_ =
-      static_cast<uint32_t>(max_elements_to_write);
+  *buffer = buffer_.get() + write_index;
+  *buffer_num_bytes = static_cast<uint32_t>(max_num_bytes_to_write);
+  two_phase_max_num_bytes_written_ =
+      static_cast<uint32_t>(max_num_bytes_to_write);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ProducerEndWriteDataImplNoLock(
-    uint32_t num_elements_written) {
-  if (num_elements_written > two_phase_max_elements_written_) {
+    uint32_t num_bytes_written) {
+  if (num_bytes_written > two_phase_max_num_bytes_written_) {
     // Note: The two-phase write ends here even on failure.
-    two_phase_max_elements_written_ = 0;  // For safety.
+    two_phase_max_num_bytes_written_ = 0;  // For safety.
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  buffer_current_num_elements_ += num_elements_written;
-  DCHECK_LE(buffer_current_num_elements_, capacity_num_elements());
-  two_phase_max_elements_written_ = 0;  // For safety.
+  current_num_bytes_ += num_bytes_written;
+  DCHECK_LE(current_num_bytes_, capacity_num_bytes());
+  two_phase_max_num_bytes_written_ = 0;  // For safety.
   return MOJO_RESULT_OK;
 }
 
 MojoWaitFlags LocalDataPipe::ProducerSatisfiedFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
-  if (consumer_open_ && buffer_current_num_elements_ < capacity_num_elements())
+  if (consumer_open_ && current_num_bytes_ < capacity_num_bytes())
     rv |= MOJO_WAIT_FLAG_WRITABLE;
   return rv;
 }
@@ -115,81 +147,118 @@ MojoWaitFlags LocalDataPipe::ProducerSatisfiableFlagsNoLock() {
 void LocalDataPipe::ConsumerCloseImplNoLock() {
   DCHECK(consumer_open_);
   consumer_open_ = false;
+  // TODO(vtl): FIXME -- broken if two-phase write ongoing
   buffer_.reset();
-  buffer_current_num_elements_ = 0;
+  current_num_bytes_ = 0;
   AwakeProducerWaitersForStateChangeNoLock();
 }
 
-MojoResult LocalDataPipe::ConsumerDiscardDataNoLock(uint32_t* num_elements,
+MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
+                                                     uint32_t* num_bytes,
+                                                     MojoReadDataFlags flags) {
+  DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
+  DCHECK_GT(*num_bytes, 0u);
+  // These cases are handled by more specific methods.
+  DCHECK(!(flags & MOJO_READ_DATA_FLAG_DISCARD));
+  DCHECK(!(flags & MOJO_READ_DATA_FLAG_QUERY));
+
+  // TODO(vtl): This implementation may write less than requested, even if room
+  // is available. Fix this.
+  const void* buffer = NULL;
+  uint32_t buffer_num_bytes = 0;
+  MojoResult rv = ConsumerBeginReadDataImplNoLock(&buffer,
+                                                  &buffer_num_bytes,
+                                                  MOJO_READ_DATA_FLAG_NONE);
+  if (rv != MOJO_RESULT_OK)
+    return rv;
+  DCHECK_EQ(buffer_num_bytes % element_num_bytes(), 0u);
+
+  uint32_t num_bytes_to_read = std::min(*num_bytes, buffer_num_bytes);
+  memcpy(elements, buffer, num_bytes_to_read);
+
+  rv = ConsumerEndReadDataImplNoLock(num_bytes_to_read);
+  if (rv != MOJO_RESULT_OK)
+    return rv;
+
+  *num_bytes = num_bytes_to_read;
+  return MOJO_RESULT_OK;
+}
+
+MojoResult LocalDataPipe::ConsumerDiscardDataNoLock(uint32_t* num_bytes,
                                                     bool all_or_none) {
+  DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
+  DCHECK_GT(*num_bytes, 0u);
+
   // TODO(vtl): Think about the error code in this case.
-  if (all_or_none && *num_elements > buffer_current_num_elements_)
+  if (all_or_none && *num_bytes > current_num_bytes_)
     return MOJO_RESULT_OUT_OF_RANGE;
 
-  size_t num_elements_to_discard =
-      std::min(static_cast<size_t>(*num_elements),
-               buffer_current_num_elements_);
-  buffer_first_element_index_ =
-      (buffer_first_element_index_ + num_elements_to_discard) %
-          capacity_num_elements();
-  buffer_current_num_elements_ -= num_elements_to_discard;
+  size_t num_bytes_to_discard =
+      std::min(static_cast<size_t>(*num_bytes), current_num_bytes_);
+  start_index_ = (start_index_ + num_bytes_to_discard) % capacity_num_bytes();
+  current_num_bytes_ -= num_bytes_to_discard;
 
   AwakeProducerWaitersForStateChangeNoLock();
   AwakeConsumerWaitersForStateChangeNoLock();
 
-  *num_elements = static_cast<uint32_t>(num_elements_to_discard);
+  *num_bytes = static_cast<uint32_t>(num_bytes_to_discard);
   return MOJO_RESULT_OK;
 }
 
-MojoResult LocalDataPipe::ConsumerQueryDataNoLock(uint32_t* num_elements) {
+MojoResult LocalDataPipe::ConsumerQueryDataNoLock(uint32_t* num_bytes) {
   // Note: This cast is safe, since the capacity fits into a |uint32_t|.
-  *num_elements = static_cast<uint32_t>(buffer_current_num_elements_);
+  *num_bytes = static_cast<uint32_t>(current_num_bytes_);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ConsumerBeginReadDataImplNoLock(
     const void** buffer,
-    uint32_t* buffer_num_elements,
+    uint32_t* buffer_num_bytes,
     MojoReadDataFlags flags) {
-  size_t max_elements_to_read = GetMaxElementsToReadNoLock();
+  size_t max_num_bytes_to_read = GetMaxNumBytesToReadNoLock();
   // TODO(vtl): Consider this return value.
   if ((flags & MOJO_READ_DATA_FLAG_ALL_OR_NONE) &&
-      *buffer_num_elements < max_elements_to_read)
+      *buffer_num_bytes > max_num_bytes_to_read)
     return MOJO_RESULT_OUT_OF_RANGE;
-  // Note: This works even if |buffer_| is null.
-  *buffer = buffer_.get() + buffer_first_element_index_ * element_size();
-  *buffer_num_elements = static_cast<uint32_t>(max_elements_to_read);
-  two_phase_max_elements_read_ = static_cast<uint32_t>(max_elements_to_read);
+
+  // Don't go into a two-phase read if there's no data.
+  // TODO(vtl): Change this to "should wait" when we have that error code.
+  if (max_num_bytes_to_read == 0)
+    return MOJO_RESULT_NOT_FOUND;
+
+  *buffer = buffer_.get() + start_index_;
+  *buffer_num_bytes = static_cast<uint32_t>(max_num_bytes_to_read);
+  two_phase_max_num_bytes_read_ = static_cast<uint32_t>(max_num_bytes_to_read);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ConsumerEndReadDataImplNoLock(
-    uint32_t num_elements_read) {
-  if (num_elements_read > two_phase_max_elements_read_) {
+    uint32_t num_bytes_read) {
+  if (num_bytes_read > two_phase_max_num_bytes_read_) {
     // Note: The two-phase read ends here even on failure.
-    two_phase_max_elements_read_ = 0;  // For safety.
+    two_phase_max_num_bytes_read_ = 0;  // For safety.
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  buffer_first_element_index_ += num_elements_read;
-  DCHECK_LE(buffer_first_element_index_, capacity_num_elements());
-  buffer_first_element_index_ %= capacity_num_elements();
-  DCHECK_LE(num_elements_read, buffer_current_num_elements_);
-  buffer_current_num_elements_ -= num_elements_read;
-  two_phase_max_elements_read_ = 0;  // For safety.
+  start_index_ += num_bytes_read;
+  DCHECK_LE(start_index_, capacity_num_bytes());
+  start_index_ %= capacity_num_bytes();
+  DCHECK_LE(num_bytes_read, current_num_bytes_);
+  current_num_bytes_ -= num_bytes_read;
+  two_phase_max_num_bytes_read_ = 0;  // For safety.
   return MOJO_RESULT_OK;
 }
 
 MojoWaitFlags LocalDataPipe::ConsumerSatisfiedFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
-  if (buffer_current_num_elements_ > 0)
+  if (current_num_bytes_ > 0)
     rv |= MOJO_WAIT_FLAG_READABLE;
   return rv;
 }
 
 MojoWaitFlags LocalDataPipe::ConsumerSatisfiableFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
-  if (buffer_current_num_elements_ > 0 || producer_open_)
+  if (current_num_bytes_ > 0 || producer_open_)
     rv |= MOJO_WAIT_FLAG_READABLE;
   return rv;
 }
@@ -199,27 +268,25 @@ void LocalDataPipe::EnsureBufferNoLock() {
   if (buffer_.get())
     return;
   buffer_.reset(static_cast<char*>(
-      base::AlignedAlloc(static_cast<size_t>(capacity_num_elements()) *
-                             element_size(),
-                         kDataPipeBufferAlignmentBytes)));
-}
-
-size_t LocalDataPipe::GetMaxElementsToWriteNoLock() {
-  size_t next_index = buffer_first_element_index_ +
-                      buffer_current_num_elements_;
-  if (next_index >= capacity_num_elements()) {
-    next_index %= capacity_num_elements();
-    DCHECK_GE(buffer_first_element_index_, next_index);
-    return buffer_first_element_index_ - next_index;
+      base::AlignedAlloc(capacity_num_bytes(), kDataPipeBufferAlignmentBytes)));
+}
+
+size_t LocalDataPipe::GetMaxNumBytesToWriteNoLock() {
+  size_t next_index = start_index_ + current_num_bytes_;
+  if (next_index >= capacity_num_bytes()) {
+    next_index %= capacity_num_bytes();
+    DCHECK_GE(start_index_, next_index);
+    DCHECK_EQ(start_index_ - next_index,
+              capacity_num_bytes() - current_num_bytes_);
+    return start_index_ - next_index;
   }
-  return capacity_num_elements() - next_index;
+  return capacity_num_bytes() - next_index;
 }
 
-size_t LocalDataPipe::GetMaxElementsToReadNoLock() {
-  if (buffer_first_element_index_ + buffer_current_num_elements_ >
-          capacity_num_elements())
-    return capacity_num_elements() - buffer_first_element_index_;
-  return buffer_current_num_elements_;
+size_t LocalDataPipe::GetMaxNumBytesToReadNoLock() {
+  if (start_index_ + current_num_bytes_ > capacity_num_bytes())
+    return capacity_num_bytes() - start_index_;
+  return current_num_bytes_;
 }
 
 }  // namespace system
diff --git a/mojo/system/local_data_pipe.h b/mojo/system/local_data_pipe.h
index 0c9e4b6..5603873 100644
--- a/mojo/system/local_data_pipe.h
+++ b/mojo/system/local_data_pipe.h
@@ -30,24 +30,32 @@ class MOJO_SYSTEM_IMPL_EXPORT LocalDataPipe : public DataPipe {
 
   // |DataPipe| implementation:
   virtual void ProducerCloseImplNoLock() OVERRIDE;
+  virtual MojoResult ProducerWriteDataImplNoLock(
+      const void* elements,
+      uint32_t* num_bytes,
+      MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult ProducerBeginWriteDataImplNoLock(
       void** buffer,
-      uint32_t* buffer_num_elements,
+      uint32_t* buffer_num_bytes,
       MojoWriteDataFlags flags) OVERRIDE;
   virtual MojoResult ProducerEndWriteDataImplNoLock(
-      uint32_t num_elements_written) OVERRIDE;
+      uint32_t num_bytes_written) OVERRIDE;
   virtual MojoWaitFlags ProducerSatisfiedFlagsNoLock() OVERRIDE;
   virtual MojoWaitFlags ProducerSatisfiableFlagsNoLock() OVERRIDE;
   virtual void ConsumerCloseImplNoLock() OVERRIDE;
-  virtual MojoResult ConsumerDiscardDataNoLock(uint32_t* num_elements,
+  virtual MojoResult ConsumerReadDataImplNoLock(
+      void* elements,
+      uint32_t* num_bytes,
+      MojoReadDataFlags flags) OVERRIDE;
+  virtual MojoResult ConsumerDiscardDataNoLock(uint32_t* num_bytes,
                                                bool all_or_none) OVERRIDE;
-  virtual MojoResult ConsumerQueryDataNoLock(uint32_t* num_elements) OVERRIDE;
+  virtual MojoResult ConsumerQueryDataNoLock(uint32_t* num_bytes) OVERRIDE;
   virtual MojoResult ConsumerBeginReadDataImplNoLock(
       const void** buffer,
-      uint32_t* buffer_num_elements,
+      uint32_t* buffer_num_bytes,
       MojoReadDataFlags flags) OVERRIDE;
   virtual MojoResult ConsumerEndReadDataImplNoLock(
-      uint32_t num_elements_read) OVERRIDE;
+      uint32_t num_bytes_read) OVERRIDE;
   virtual MojoWaitFlags ConsumerSatisfiedFlagsNoLock() OVERRIDE;
   virtual MojoWaitFlags ConsumerSatisfiableFlagsNoLock() OVERRIDE;
 
@@ -55,20 +63,22 @@ class MOJO_SYSTEM_IMPL_EXPORT LocalDataPipe : public DataPipe {
 
   // Get the maximum (single) write/read size right now (in number of elements);
   // result fits in a |uint32_t|.
-  size_t GetMaxElementsToWriteNoLock();
-  size_t GetMaxElementsToReadNoLock();
+  size_t GetMaxNumBytesToWriteNoLock();
+  size_t GetMaxNumBytesToReadNoLock();
 
   // The members below are protected by |DataPipe|'s |lock_|:
+  // TODO(vtl): FIXME -- move this to DataPipe?
   bool producer_open_;
   bool consumer_open_;
 
   scoped_ptr_malloc<char, base::ScopedPtrAlignedFree> buffer_;
   // Circular buffer.
-  size_t buffer_first_element_index_;
-  size_t buffer_current_num_elements_;
+  size_t start_index_;
+  size_t current_num_bytes_;
 
-  uint32_t two_phase_max_elements_written_;
-  uint32_t two_phase_max_elements_read_;
+  // TODO(vtl): FIXME -- merge this with bool in superclass
+  uint32_t two_phase_max_num_bytes_written_;
+  uint32_t two_phase_max_num_bytes_read_;
 
   DISALLOW_COPY_AND_ASSIGN(LocalDataPipe);
 };
diff --git a/mojo/system/memory.cc b/mojo/system/memory.cc
index a10b021..dd8b3d2 100644
--- a/mojo/system/memory.cc
+++ b/mojo/system/memory.cc
@@ -11,8 +11,6 @@
 namespace mojo {
 namespace system {
 
-// TODO(vtl): Can I make this use the non-templatized function without a
-// performance hit?
 template <size_t size>
 bool VerifyUserPointerForSize(const void* pointer, size_t count) {
   if (count > std::numeric_limits<size_t>::max() / size)
@@ -24,18 +22,6 @@ bool VerifyUserPointerForSize(const void* pointer, size_t count) {
   return count == 0 || !!pointer;
 }
 
-bool MOJO_SYSTEM_IMPL_EXPORT VerifyUserPointerForSize(const void* pointer,
-                                                      size_t size,
-                                                      size_t count) {
-  if (count > std::numeric_limits<size_t>::max() / size)
-    return false;
-
-  // TODO(vtl): If running in kernel mode, do a full verification. For now, just
-  // check that it's non-null if |size| is nonzero. (A faster user mode
-  // implementation is also possible if this check is skipped.)
-  return count == 0 || !!pointer;
-}
-
 // Explicitly instantiate the sizes we need. Add instantiations as needed.
 template MOJO_SYSTEM_IMPL_EXPORT bool VerifyUserPointerForSize<1>(
     const void*, size_t);
diff --git a/mojo/system/memory.h b/mojo/system/memory.h
index 9a25ef9..6483e03 100644
--- a/mojo/system/memory.h
+++ b/mojo/system/memory.h
@@ -19,12 +19,6 @@ template <size_t size>
 bool MOJO_SYSTEM_IMPL_EXPORT VerifyUserPointerForSize(const void* pointer,
                                                       size_t count);
 
-// A non-templatized version of the above, for when the element size isn't
-// fixed.
-bool MOJO_SYSTEM_IMPL_EXPORT VerifyUserPointerForSize(const void* pointer,
-                                                      size_t size,
-                                                      size_t count);
-
 // Verify that |count * sizeof(T)| bytes can be read from the user |pointer|
 // insofar as possible/necessary (note: this is done carefully since |count *
 // sizeof(T)| may overflow a |size_t|. |count| may be zero. If |T| is |void|,