Mojo: More data pipe implementation.

Implement the dispatcher for the producer side
(DataPipeProducerDispatcher); the consumer side will be similar.

Implement most of the base class for the connecting part (DataPipe).
Subclasses will provide implementations for the various cases (both
local, local producer/remote consumer, remote producer/local consumer).

The setup is similar to that for MessagePipe, but factored a bit
differently since the sides aren't symmetric (hence no gain in factoring
out the endpoint code).

R=sky@chromium.org

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

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

16 files changed, 455 insertions, 24 deletions

diff --git a/mojo/mojo.gyp b/mojo/mojo.gyp
index d778405..51f9050 100644
--- a/mojo/mojo.gyp
+++ b/mojo/mojo.gyp
@@ -78,6 +78,10 @@
         'system/channel.h',
         'system/core_impl.cc',
         'system/core_impl.h',
+        'system/data_pipe.cc',
+        'system/data_pipe.h',
+        'system/data_pipe_producer_dispatcher.cc',
+        'system/data_pipe_producer_dispatcher.h',
         'system/dispatcher.cc',
         'system/dispatcher.h',
         'system/limits.h',
diff --git a/mojo/public/system/core.h b/mojo/public/system/core.h
index dc71ae2..8d4cb02 100644
--- a/mojo/public/system/core.h
+++ b/mojo/public/system/core.h
@@ -406,8 +406,8 @@ MOJO_SYSTEM_EXPORT MojoResult MojoReadMessage(MojoHandle message_pipe_handle,
 
 MOJO_SYSTEM_EXPORT MojoResult MojoCreateDataPipe(
     const struct MojoCreateDataPipeOptions* options,
-    MojoHandle* producer_handle,
-    MojoHandle* consumer_handle);
+    MojoHandle* data_pipe_producer_handle,
+    MojoHandle* data_pipe_consumer_handle);
 
 MOJO_SYSTEM_EXPORT MojoResult MojoWriteData(
     MojoHandle data_pipe_producer_handle,
@@ -415,6 +415,11 @@ MOJO_SYSTEM_EXPORT MojoResult MojoWriteData(
     uint32_t* num_elements,
     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).
 MOJO_SYSTEM_EXPORT MojoResult MojoBeginWriteData(
     MojoHandle data_pipe_producer_handle,
     void** buffer,
diff --git a/mojo/public/system/core_private.cc b/mojo/public/system/core_private.cc
index 702ce48..12f819b 100644
--- a/mojo/public/system/core_private.cc
+++ b/mojo/public/system/core_private.cc
@@ -65,11 +65,12 @@ MojoResult MojoReadMessage(MojoHandle message_pipe_handle,
                              num_handles, flags);
 }
 
-MojoResult MojoCreateDataPipe(const struct MojoCreateDataPipeOptions* options,
-                              MojoHandle* producer_handle,
-                              MojoHandle* consumer_handle) {
+MojoResult MojoCreateDataPipe(const MojoCreateDataPipeOptions* options,
+                              MojoHandle* data_pipe_producer_handle,
+                              MojoHandle* data_pipe_consumer_handle) {
   assert(g_core);
-  return g_core->CreateDataPipe(options, producer_handle, consumer_handle);
+  return g_core->CreateDataPipe(options, data_pipe_producer_handle,
+                                data_pipe_consumer_handle);
 }
 
 MojoResult MojoWriteData(MojoHandle data_pipe_producer_handle,
diff --git a/mojo/public/system/core_private.h b/mojo/public/system/core_private.h
index 1095e98..6df7245 100644
--- a/mojo/public/system/core_private.h
+++ b/mojo/public/system/core_private.h
@@ -41,10 +41,9 @@ class MOJO_SYSTEM_EXPORT CorePrivate {
                                  MojoHandle* handles,
                                  uint32_t* num_handles,
                                  MojoReadMessageFlags flags) = 0;
-  virtual MojoResult CreateDataPipe(
-      const struct MojoCreateDataPipeOptions* options,
-      MojoHandle* producer_handle,
-      MojoHandle* consumer_handle) = 0;
+  virtual MojoResult CreateDataPipe(const MojoCreateDataPipeOptions* options,
+                                    MojoHandle* data_pipe_producer_handle,
+                                    MojoHandle* data_pipe_consumer_handle) = 0;
   virtual MojoResult WriteData(MojoHandle data_pipe_producer_handle,
                                const void* elements,
                                uint32_t* num_elements,
diff --git a/mojo/system/core_impl.cc b/mojo/system/core_impl.cc
index 60b6325..5b26d9f 100644
--- a/mojo/system/core_impl.cc
+++ b/mojo/system/core_impl.cc
@@ -8,6 +8,7 @@
 
 #include "base/logging.h"
 #include "base/time/time.h"
+#include "mojo/system/data_pipe_producer_dispatcher.h"
 #include "mojo/system/dispatcher.h"
 #include "mojo/system/limits.h"
 #include "mojo/system/memory.h"
@@ -349,11 +350,49 @@ MojoResult CoreImpl::ReadMessage(MojoHandle message_pipe_handle,
   return rv;
 }
 
-MojoResult CoreImpl::CreateDataPipe(
-    const struct MojoCreateDataPipeOptions* options,
-    MojoHandle* producer_handle,
-    MojoHandle* consumer_handle) {
-  // TODO(vtl)
+MojoResult CoreImpl::CreateDataPipe(const MojoCreateDataPipeOptions* options,
+                                    MojoHandle* data_pipe_producer_handle,
+                                    MojoHandle* data_pipe_consumer_handle) {
+  if (options && !VerifyUserPointer<void>(options, sizeof(*options)))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  if (!VerifyUserPointer<MojoHandle>(data_pipe_producer_handle, 1))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  if (!VerifyUserPointer<MojoHandle>(data_pipe_consumer_handle, 1))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+/* TODO(vtl): The rest of the code will look something like this:
+  scoped_refptr<LocalDataPipe> data_pipe(new LocalDataPipe());
+  MojoResult result = data_pipe->Init(options);
+  if (result != MOJO_RESULT_OK)
+    return result;
+
+  scoped_refptr<DataPipeProducerDispatcher> producer_dispatcher(
+      new DataPipeProducerDispatcher());
+  scoped_refptr<DataPipeConsumerDispatcher> consumer_dispatcher(
+      new DataPipeConsumerDispatcher());
+
+  MojoHandle producer_handle, consumer_handle;
+  {
+    base::AutoLock locker(handle_table_lock_);
+
+    producer_handle = AddDispatcherNoLock(producer_dispatcher);
+    if (producer_handle == MOJO_HANDLE_INVALID)
+      return MOJO_RESULT_RESOURCE_EXHAUSTED;
+
+    consumer_handle = AddDispatcherNoLock(consumer_dispatcher);
+    if (consumer_handle == MOJO_HANDLE_INVALID) {
+      handle_table_.erase(producer_handle);
+      return MOJO_RESULT_RESOURCE_EXHAUSTED;
+    }
+  }
+
+  producer_dispatcher->Init(data_pipe);
+  consumer_dispatcher->Init(data_pipe);
+
+  *data_pipe_producer_handle = producer_handle;
+  *data_pipe_consumer_handle = consumer_handle;
+  return MOJO_RESULT_OK;
+*/
   NOTIMPLEMENTED();
   return MOJO_RESULT_UNIMPLEMENTED;
 }
@@ -431,8 +470,8 @@ CoreImpl::CoreImpl()
 }
 
 CoreImpl::~CoreImpl() {
-  // This should usually not be reached (the singleton lives forever), except
-  // in tests.
+  // This should usually not be reached (the singleton lives forever), except in
+  // tests.
 }
 
 scoped_refptr<Dispatcher> CoreImpl::GetDispatcher(MojoHandle handle) {
diff --git a/mojo/system/core_impl.h b/mojo/system/core_impl.h
index 51f5cb7..1b47d0a2 100644
--- a/mojo/system/core_impl.h
+++ b/mojo/system/core_impl.h
@@ -57,9 +57,9 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public CorePrivate {
                                  uint32_t* num_handles,
                                  MojoReadMessageFlags flags) OVERRIDE;
   virtual MojoResult CreateDataPipe(
-      const struct MojoCreateDataPipeOptions* options,
-      MojoHandle* producer_handle,
-      MojoHandle* consumer_handle) OVERRIDE;
+      const MojoCreateDataPipeOptions* options,
+      MojoHandle* data_pipe_producer_handle,
+      MojoHandle* data_pipe_consumer_handle) OVERRIDE;
   virtual MojoResult WriteData(MojoHandle data_pipe_producer_handle,
                                const void* elements,
                                uint32_t* num_elements,
diff --git a/mojo/system/data_pipe.cc b/mojo/system/data_pipe.cc
new file mode 100644
index 0000000..3d6aeb5
--- /dev/null
+++ b/mojo/system/data_pipe.cc
@@ -0,0 +1,103 @@
+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "mojo/system/data_pipe.h"
+
+#include <string.h>
+
+#include <algorithm>
+
+#include "base/logging.h"
+#include "mojo/system/memory.h"
+#include "mojo/system/waiter_list.h"
+
+namespace mojo {
+namespace system {
+
+void DataPipe::ProducerCancelAllWaiters() {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+  producer_waiter_list_->CancelAllWaiters();
+}
+
+void DataPipe::ProducerClose() {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+  producer_waiter_list_.reset();
+  ProducerCloseImplNoLock();
+}
+
+MojoResult DataPipe::ProducerWriteData(const void* elements,
+                                       uint32_t* num_elements,
+                                       MojoWriteDataFlags flags) {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+
+  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_);
+
+  rv = ProducerEndWriteDataImplNoLock(num_elements_to_write);
+  if (rv != MOJO_RESULT_OK)
+    return rv;
+
+  *num_elements = num_elements_to_write;
+  return MOJO_RESULT_OK;
+}
+
+MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
+                                            uint32_t* buffer_num_elements,
+                                            MojoWriteDataFlags flags) {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+  return ProducerBeginWriteDataImplNoLock(buffer, buffer_num_elements, flags);
+}
+
+MojoResult DataPipe::ProducerEndWriteData(uint32_t num_elements_written) {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+  return ProducerEndWriteDataImplNoLock(num_elements_written);
+}
+
+MojoResult DataPipe::ProducerAddWaiter(Waiter* waiter,
+                                       MojoWaitFlags flags,
+                                       MojoResult wake_result) {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+
+  if ((flags & ProducerSatisfiedFlagsNoLock()))
+    return MOJO_RESULT_ALREADY_EXISTS;
+  if (!(flags & ProducerSatisfiableFlagsNoLock()))
+    return MOJO_RESULT_FAILED_PRECONDITION;
+
+  producer_waiter_list_->AddWaiter(waiter, flags, wake_result);
+  return MOJO_RESULT_OK;
+}
+
+void DataPipe::ProducerRemoveWaiter(Waiter* waiter) {
+  base::AutoLock locker(lock_);
+  DCHECK(has_local_producer_no_lock());
+  producer_waiter_list_->RemoveWaiter(waiter);
+}
+
+DataPipe::DataPipe(bool has_local_producer, bool has_local_consumer)
+    : producer_waiter_list_(has_local_producer ? new WaiterList() : NULL),
+      consumer_waiter_list_(has_local_consumer ? new WaiterList() : NULL) {
+  DCHECK(has_local_producer || has_local_consumer);
+}
+
+DataPipe::~DataPipe() {
+  DCHECK(!has_local_producer_no_lock());
+  DCHECK(!has_local_consumer_no_lock());
+}
+
+}  // namespace system
+}  // namespace mojo
diff --git a/mojo/system/data_pipe.h b/mojo/system/data_pipe.h
new file mode 100644
index 0000000..44dc592
--- /dev/null
+++ b/mojo/system/data_pipe.h
@@ -0,0 +1,100 @@
+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_SYSTEM_DATA_PIPE_H_
+#define MOJO_SYSTEM_DATA_PIPE_H_
+
+#include "base/basictypes.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/lock.h"
+#include "mojo/public/system/core.h"
+#include "mojo/system/system_impl_export.h"
+
+namespace mojo {
+namespace system {
+
+class Waiter;
+class WaiterList;
+
+// |DataPipe| is a base class for secondary objects implementing data pipes,
+// similar to |MessagePipe| (see the explanatory comment in core_impl.cc). It is
+// typically owned by the dispatcher(s) corresponding to the local endpoints.
+// Its subclasses implement the three cases: local producer and consumer, local
+// producer and remote consumer, and remote producer and local consumer. This
+// class is thread-safe.
+class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
+    public base::RefCountedThreadSafe<DataPipe> {
+ public:
+  // These are called by the producer dispatcher to implement its methods of
+  // corresponding names.
+  void ProducerCancelAllWaiters();
+  void ProducerClose();
+  // This does not validate |elements| or |num_elements| (or |*num_elements|).
+  MojoResult ProducerWriteData(const void* elements,
+                               uint32_t* num_elements,
+                               MojoWriteDataFlags flags);
+  // This does not validate |buffer| or |buffer_num_elements|.
+  MojoResult ProducerBeginWriteData(void** buffer,
+                                    uint32_t* buffer_num_elements,
+                                    MojoWriteDataFlags flags);
+  MojoResult ProducerEndWriteData(uint32_t num_elements_written);
+  MojoResult ProducerAddWaiter(Waiter* waiter,
+                               MojoWaitFlags flags,
+                               MojoResult wake_result);
+  void ProducerRemoveWaiter(Waiter* waiter);
+
+/* TODO(vtl)
+  void ConsumerCancelAllWaiters();
+  void ConsumerClose();
+...
+  MojoResult ConsumerAddWaiter(Waiter* waiter,
+                               MojoWaitFlags flags,
+                               MojoResult wake_result);
+  void ConsumerRemoveWaiter(Waiter* waiter);
+*/
+
+  // Thread-safe and fast (doesn't take the lock).
+  size_t element_size() const { return element_size_; }
+
+ protected:
+  DataPipe(bool has_local_producer, bool has_local_consumer);
+
+  void Init(size_t element_size);
+
+  virtual void ProducerCloseImplNoLock() = 0;
+  virtual MojoResult ProducerBeginWriteDataImplNoLock(
+      void** buffer,
+      uint32_t* buffer_num_elements,
+      MojoWriteDataFlags flags) = 0;
+  virtual MojoResult ProducerEndWriteDataImplNoLock(
+      uint32_t num_elements_written) = 0;
+  virtual MojoWaitFlags ProducerSatisfiedFlagsNoLock() = 0;
+  virtual MojoWaitFlags ProducerSatisfiableFlagsNoLock() = 0;
+
+ private:
+  friend class base::RefCountedThreadSafe<DataPipe>;
+  virtual ~DataPipe();
+
+  bool has_local_producer_no_lock() const {
+    return !!producer_waiter_list_.get();
+  }
+  bool has_local_consumer_no_lock() const {
+    return !!consumer_waiter_list_.get();
+  }
+
+  // Set by |Init()| and never changed afterwards.
+  size_t element_size_;
+
+  base::Lock lock_;  // Protects the following members.
+  scoped_ptr<WaiterList> producer_waiter_list_;
+  scoped_ptr<WaiterList> consumer_waiter_list_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipe);
+};
+
+}  // namespace system
+}  // namespace mojo
+
+#endif  // MOJO_SYSTEM_DATA_PIPE_H_
diff --git a/mojo/system/data_pipe_producer_dispatcher.cc b/mojo/system/data_pipe_producer_dispatcher.cc
new file mode 100644
index 0000000..cb03c31
--- /dev/null
+++ b/mojo/system/data_pipe_producer_dispatcher.cc
@@ -0,0 +1,100 @@
+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "mojo/system/data_pipe_producer_dispatcher.h"
+
+#include "base/logging.h"
+#include "mojo/system/data_pipe.h"
+#include "mojo/system/memory.h"
+
+namespace mojo {
+namespace system {
+
+DataPipeProducerDispatcher::DataPipeProducerDispatcher() {
+}
+
+void DataPipeProducerDispatcher::Init(scoped_refptr<DataPipe> data_pipe) {
+  DCHECK(data_pipe.get());
+  data_pipe_ = data_pipe;
+}
+
+DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
+  // |Close()|/|CloseImplNoLock()| should have taken care of the pipe.
+  DCHECK(!data_pipe_.get());
+}
+
+void DataPipeProducerDispatcher::CancelAllWaitersNoLock() {
+  lock().AssertAcquired();
+  data_pipe_->ProducerCancelAllWaiters();
+}
+
+MojoResult DataPipeProducerDispatcher::CloseImplNoLock() {
+  lock().AssertAcquired();
+  data_pipe_->ProducerClose();
+  data_pipe_ = NULL;
+  return MOJO_RESULT_OK;
+}
+
+MojoResult DataPipeProducerDispatcher::WriteDataImplNoLock(
+    const void* elements,
+    uint32_t* num_elements,
+    MojoWriteDataFlags flags) {
+  lock().AssertAcquired();
+
+  if (!VerifyUserPointer<uint32_t>(num_elements, 1))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  if (!VerifyUserPointerForSize(elements, data_pipe_->element_size(),
+                                *num_elements))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return data_pipe_->ProducerWriteData(elements, num_elements, flags);
+}
+
+MojoResult DataPipeProducerDispatcher::BeginWriteDataImplNoLock(
+    void** buffer,
+    uint32_t* buffer_num_elements,
+    MojoWriteDataFlags flags) {
+  lock().AssertAcquired();
+
+  if (!VerifyUserPointer<void*>(buffer, 1))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  if (!VerifyUserPointer<uint32_t>(buffer_num_elements, 1))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return data_pipe_->ProducerBeginWriteData(buffer, buffer_num_elements, flags);
+}
+
+MojoResult DataPipeProducerDispatcher::EndWriteDataImplNoLock(
+    uint32_t num_elements_written) {
+  lock().AssertAcquired();
+
+  return data_pipe_->ProducerEndWriteData(num_elements_written);
+}
+
+MojoResult DataPipeProducerDispatcher::AddWaiterImplNoLock(
+    Waiter* waiter,
+    MojoWaitFlags flags,
+    MojoResult wake_result) {
+  lock().AssertAcquired();
+  return data_pipe_->ProducerAddWaiter(waiter, flags, wake_result);
+}
+
+void DataPipeProducerDispatcher::RemoveWaiterImplNoLock(Waiter* waiter) {
+  lock().AssertAcquired();
+  data_pipe_->ProducerRemoveWaiter(waiter);
+}
+
+scoped_refptr<Dispatcher>
+DataPipeProducerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock() {
+  lock().AssertAcquired();
+
+  scoped_refptr<DataPipeProducerDispatcher> rv =
+      new DataPipeProducerDispatcher();
+  rv->Init(data_pipe_);
+  data_pipe_ = NULL;
+  return scoped_refptr<Dispatcher>(rv.get());
+}
+
+}  // namespace system
+}  // namespace mojo
diff --git a/mojo/system/data_pipe_producer_dispatcher.h b/mojo/system/data_pipe_producer_dispatcher.h
new file mode 100644
index 0000000..01c8783
--- /dev/null
+++ b/mojo/system/data_pipe_producer_dispatcher.h
@@ -0,0 +1,61 @@
+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_SYSTEM_DATA_PIPE_PRODUCER_DISPATCHER_H_
+#define MOJO_SYSTEM_DATA_PIPE_PRODUCER_DISPATCHER_H_
+
+#include "base/basictypes.h"
+#include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "mojo/system/dispatcher.h"
+#include "mojo/system/system_impl_export.h"
+
+namespace mojo {
+namespace system {
+
+class DataPipe;
+
+// This is the |Dispatcher| implementation for the producer handle for data
+// pipes (created by the Mojo primitive |MojoCreateDataPipe()|). This class is
+// thread-safe.
+class MOJO_SYSTEM_IMPL_EXPORT DataPipeProducerDispatcher : public Dispatcher {
+ public:
+  DataPipeProducerDispatcher();
+
+  // Must be called before any other methods.
+  void Init(scoped_refptr<DataPipe> data_pipe);
+
+ private:
+  friend class base::RefCountedThreadSafe<DataPipeProducerDispatcher>;
+  virtual ~DataPipeProducerDispatcher();
+
+  // |Dispatcher| implementation/overrides:
+  virtual void CancelAllWaitersNoLock() OVERRIDE;
+  virtual MojoResult CloseImplNoLock() OVERRIDE;
+  virtual MojoResult WriteDataImplNoLock(const void* elements,
+                                         uint32_t* num_elements,
+                                         MojoWriteDataFlags flags) OVERRIDE;
+  virtual MojoResult BeginWriteDataImplNoLock(
+      void** buffer,
+      uint32_t* buffer_num_elements,
+      MojoWriteDataFlags flags) OVERRIDE;
+  virtual MojoResult EndWriteDataImplNoLock(
+      uint32_t num_elements_written) OVERRIDE;
+  virtual MojoResult AddWaiterImplNoLock(Waiter* waiter,
+                                         MojoWaitFlags flags,
+                                         MojoResult wake_result) OVERRIDE;
+  virtual void RemoveWaiterImplNoLock(Waiter* waiter) OVERRIDE;
+  virtual scoped_refptr<Dispatcher>
+      CreateEquivalentDispatcherAndCloseImplNoLock() OVERRIDE;
+
+  // Protected by |lock()|:
+  scoped_refptr<DataPipe> data_pipe_;  // This will be null if closed.
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeProducerDispatcher);
+};
+
+}  // namespace system
+}  // namespace mojo
+
+#endif  // MOJO_SYSTEM_DATA_PIPE_PRODUCER_DISPATCHER_H_
diff --git a/mojo/system/memory.cc b/mojo/system/memory.cc
index 633b2be..a10b021 100644
--- a/mojo/system/memory.cc
+++ b/mojo/system/memory.cc
@@ -11,6 +11,8 @@
 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)
@@ -22,11 +24,25 @@ 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);
 template MOJO_SYSTEM_IMPL_EXPORT bool VerifyUserPointerForSize<4>(
     const void*, size_t);
+template MOJO_SYSTEM_IMPL_EXPORT bool VerifyUserPointerForSize<8>(
+    const void*, size_t);
 
 }  // namespace system
 }  // namespace mojo
diff --git a/mojo/system/memory.h b/mojo/system/memory.h
index 6483e03..9a25ef9 100644
--- a/mojo/system/memory.h
+++ b/mojo/system/memory.h
@@ -19,6 +19,12 @@ 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|,
diff --git a/mojo/system/message_pipe.cc b/mojo/system/message_pipe.cc
index 10cf9cd..4083cd8 100644
--- a/mojo/system/message_pipe.cc
+++ b/mojo/system/message_pipe.cc
@@ -5,7 +5,6 @@
 #include "mojo/system/message_pipe.h"
 
 #include "base/logging.h"
-#include "base/stl_util.h"
 #include "mojo/system/channel.h"
 #include "mojo/system/dispatcher.h"
 #include "mojo/system/local_message_pipe_endpoint.h"
diff --git a/mojo/system/message_pipe.h b/mojo/system/message_pipe.h
index 716f580..111d690 100644
--- a/mojo/system/message_pipe.h
+++ b/mojo/system/message_pipe.h
@@ -8,7 +8,6 @@
 #include <vector>
 
 #include "base/basictypes.h"
-#include "base/callback.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/synchronization/lock.h"
diff --git a/mojo/system/message_pipe_dispatcher.cc b/mojo/system/message_pipe_dispatcher.cc
index 0c60f48..0642440 100644
--- a/mojo/system/message_pipe_dispatcher.cc
+++ b/mojo/system/message_pipe_dispatcher.cc
@@ -101,7 +101,7 @@ scoped_refptr<Dispatcher>
 MessagePipeDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock() {
   lock().AssertAcquired();
 
-  scoped_refptr<MessagePipeDispatcher> rv = new MessagePipeDispatcher;
+  scoped_refptr<MessagePipeDispatcher> rv = new MessagePipeDispatcher();
   rv->Init(message_pipe_, port_);
   message_pipe_ = NULL;
   port_ = kInvalidPort;
diff --git a/mojo/system/message_pipe_dispatcher.h b/mojo/system/message_pipe_dispatcher.h
index 3f00b03..8101613 100644
--- a/mojo/system/message_pipe_dispatcher.h
+++ b/mojo/system/message_pipe_dispatcher.h
@@ -45,7 +45,6 @@ class MOJO_SYSTEM_IMPL_EXPORT MessagePipeDispatcher : public Dispatcher {
                                          MojoWaitFlags flags,
                                          MojoResult wake_result) OVERRIDE;
   virtual void RemoveWaiterImplNoLock(Waiter* waiter) OVERRIDE;
-
   virtual scoped_refptr<Dispatcher>
       CreateEquivalentDispatcherAndCloseImplNoLock() OVERRIDE;