Mojo: Move the handle table details out of CoreImpl into its own class.

(This will make it much easier to improve the handle table, e.g., by caching
handle look-ups.)

R=darin@chromium.org

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

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

8 files changed, 451 insertions, 217 deletions

diff --git a/mojo/embedder/test_embedder.cc b/mojo/embedder/test_embedder.cc
index 912a7ac..855d1da 100644
--- a/mojo/embedder/test_embedder.cc
+++ b/mojo/embedder/test_embedder.cc
@@ -7,6 +7,7 @@
 #include "base/logging.h"
 #include "base/macros.h"
 #include "mojo/system/core_impl.h"
+#include "mojo/system/handle_table.h"
 
 namespace mojo {
 
@@ -15,12 +16,15 @@ namespace internal {
 
 bool ShutdownCheckNoLeaks(CoreImpl* core_impl) {
   // No point in taking the lock.
-  if (core_impl->handle_table_.empty())
+  const HandleTable::HandleToEntryMap& handle_to_entry_map =
+      core_impl->handle_table_.handle_to_entry_map_;
+
+  if (handle_to_entry_map.empty())
     return true;
 
-  for (CoreImpl::HandleTableMap::const_iterator it =
-           core_impl->handle_table_.begin();
-       it != core_impl->handle_table_.end();
+  for (HandleTable::HandleToEntryMap::const_iterator it =
+           handle_to_entry_map.begin();
+       it != handle_to_entry_map.end();
        ++it) {
     LOG(ERROR) << "Mojo embedder shutdown: Leaking handle " << (*it).first;
   }
diff --git a/mojo/mojo.gyp b/mojo/mojo.gyp
index e2457fd..4c24e47 100644
--- a/mojo/mojo.gyp
+++ b/mojo/mojo.gyp
@@ -126,6 +126,8 @@
         'system/data_pipe_producer_dispatcher.h',
         'system/dispatcher.cc',
         'system/dispatcher.h',
+        'system/handle_table.cc',
+        'system/handle_table.h',
         'system/local_data_pipe.cc',
         'system/local_data_pipe.h',
         'system/local_message_pipe_endpoint.cc',
diff --git a/mojo/system/core_impl.cc b/mojo/system/core_impl.cc
index ea1ebda..c3fa10d 100644
--- a/mojo/system/core_impl.cc
+++ b/mojo/system/core_impl.cc
@@ -87,8 +87,7 @@ CoreImpl::HandleTableEntry::~HandleTableEntry() {
   DCHECK(!busy);
 }
 
-CoreImpl::CoreImpl()
-    : next_handle_(MOJO_HANDLE_INVALID + 1) {
+CoreImpl::CoreImpl() {
 }
 
 CoreImpl::~CoreImpl() {
@@ -99,7 +98,7 @@ CoreImpl::~CoreImpl() {
 MojoHandle CoreImpl::AddDispatcher(
     const scoped_refptr<Dispatcher>& dispatcher) {
   base::AutoLock locker(handle_table_lock_);
-  return AddDispatcherNoLock(dispatcher);
+  return handle_table_.AddDispatcher(dispatcher);
 }
 
 MojoTimeTicks CoreImpl::GetTimeTicksNow() {
@@ -113,13 +112,10 @@ MojoResult CoreImpl::Close(MojoHandle handle) {
   scoped_refptr<Dispatcher> dispatcher;
   {
     base::AutoLock locker(handle_table_lock_);
-    HandleTableMap::iterator it = handle_table_.find(handle);
-    if (it == handle_table_.end())
-      return MOJO_RESULT_INVALID_ARGUMENT;
-    if (it->second.busy)
-      return MOJO_RESULT_BUSY;
-    dispatcher = it->second.dispatcher;
-    handle_table_.erase(it);
+    MojoResult result = handle_table_.GetAndRemoveDispatcher(handle,
+                                                             &dispatcher);
+    if (result != MOJO_RESULT_OK)
+      return result;
   }
 
   // The dispatcher doesn't have a say in being closed, but gets notified of it.
@@ -160,31 +156,25 @@ MojoResult CoreImpl::CreateMessagePipe(MojoHandle* message_pipe_handle0,
   scoped_refptr<MessagePipeDispatcher> dispatcher0(new MessagePipeDispatcher());
   scoped_refptr<MessagePipeDispatcher> dispatcher1(new MessagePipeDispatcher());
 
-  MojoHandle h0, h1;
+  std::pair<MojoHandle, MojoHandle> handle_pair;
   {
     base::AutoLock locker(handle_table_lock_);
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close the dispatcher
-    // (outside the table lock).
-    h0 = AddDispatcherNoLock(dispatcher0);
-    if (h0 == MOJO_HANDLE_INVALID)
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close both dispatchers
-    // (outside the table lock).
-    h1 = AddDispatcherNoLock(dispatcher1);
-    if (h1 == MOJO_HANDLE_INVALID) {
-      handle_table_.erase(h0);
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
-    }
+    handle_pair = handle_table_.AddDispatcherPair(dispatcher0, dispatcher1);
+  }
+  if (handle_pair.first == MOJO_HANDLE_INVALID) {
+    DCHECK_EQ(handle_pair.second, MOJO_HANDLE_INVALID);
+    LOG(ERROR) << "Handle table full";
+    dispatcher0->Close();
+    dispatcher1->Close();
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   scoped_refptr<MessagePipe> message_pipe(new MessagePipe());
   dispatcher0->Init(message_pipe, 0);
   dispatcher1->Init(message_pipe, 1);
 
-  *message_pipe_handle0 = h0;
-  *message_pipe_handle1 = h1;
+  *message_pipe_handle0 = handle_pair.first;
+  *message_pipe_handle1 = handle_pair.second;
   return MOJO_RESULT_OK;
 }
 
@@ -234,71 +224,10 @@ MojoResult CoreImpl::WriteMessage(MojoHandle message_pipe_handle,
   // handles from the handle table.
   {
     base::AutoLock locker(handle_table_lock_);
-
-    std::vector<HandleTableEntry*> entries(num_handles);
-
-    // First verify all the handles and get their dispatchers.
-    uint32_t i;
-    MojoResult error_result = MOJO_RESULT_INTERNAL;
-    for (i = 0; i < num_handles; i++) {
-      // Sending your own handle is not allowed (and, for consistency, returns
-      // "busy").
-      if (handles[i] == message_pipe_handle) {
-        error_result = MOJO_RESULT_BUSY;
-        break;
-      }
-
-      HandleTableMap::iterator it = handle_table_.find(handles[i]);
-      if (it == handle_table_.end()) {
-        error_result = MOJO_RESULT_INVALID_ARGUMENT;
-        break;
-      }
-
-      entries[i] = &it->second;
-      if (entries[i]->busy) {
-        error_result = MOJO_RESULT_BUSY;
-        break;
-      }
-      // Note: By marking the handle as busy here, we're also preventing the
-      // same handle from being sent multiple times in the same message.
-      entries[i]->busy = true;
-
-      // Try to start the transport.
-      DispatcherTransport transport =
-          Dispatcher::CoreImplAccess::TryStartTransport(
-              entries[i]->dispatcher.get());
-      if (!transport.is_valid()) {
-        // Unset the busy flag (since it won't be unset below).
-        entries[i]->busy = false;
-        error_result = MOJO_RESULT_BUSY;
-        break;
-      }
-
-      // Check if the dispatcher is busy (e.g., in a two-phase read/write).
-      // (Note that this must be done after the dispatcher's lock is acquired.)
-      if (transport.IsBusy()) {
-        // Unset the busy flag and end the transport (since it won't be done
-        // below).
-        entries[i]->busy = false;
-        transport.End();
-        error_result = MOJO_RESULT_BUSY;
-        break;
-      }
-
-      // Hang on to the transport (which we'll need to end the transport).
-      transports[i] = transport;
-    }
-    if (i < num_handles) {
-      DCHECK_NE(error_result, MOJO_RESULT_INTERNAL);
-
-      // Unset the busy flags and release the locks.
-      for (uint32_t j = 0; j < i; j++) {
-        DCHECK(entries[j]->busy);
-        entries[j]->busy = false;
-        transports[j].End();
-      }
-      return error_result;
-    }
+    MojoResult result = handle_table_.MarkBusyAndStartTransport(
+        message_pipe_handle, handles, num_handles, &transports);
+    if (result != MOJO_RESULT_OK)
+      return result;
   }
 
   MojoResult rv = dispatcher->WriteMessage(bytes, num_bytes, &transports,
@@ -309,28 +238,12 @@ MojoResult CoreImpl::WriteMessage(MojoHandle message_pipe_handle,
   for (uint32_t i = 0; i < num_handles; i++)
     transports[i].End();
 
-  if (rv == MOJO_RESULT_OK) {
-    base::AutoLock locker(handle_table_lock_);
-
-    // Succeeded, so the handles should be removed from the handle table. (The
-    // transferring to new dispatchers/closing must have already been done.)
-    for (uint32_t i = 0; i < num_handles; i++) {
-      HandleTableMap::iterator it = handle_table_.find(handles[i]);
-      DCHECK(it != handle_table_.end());
-      DCHECK(it->second.busy);
-      it->second.busy = false;  // For the sake of a |DCHECK()|.
-      handle_table_.erase(it);
-    }
-  } else {
+  {
     base::AutoLock locker(handle_table_lock_);
-
-    // Failed, so the handles should go back to their normal state.
-    for (uint32_t i = 0; i < num_handles; i++) {
-      HandleTableMap::iterator it = handle_table_.find(handles[i]);
-      DCHECK(it != handle_table_.end());
-      DCHECK(it->second.busy);
-      it->second.busy = false;
-    }
+    if (rv == MOJO_RESULT_OK)
+      handle_table_.RemoveBusyHandles(handles, num_handles);
+    else
+      handle_table_.RestoreBusyHandles(handles, num_handles);
   }
 
   return rv;
@@ -366,17 +279,19 @@ MojoResult CoreImpl::ReadMessage(MojoHandle message_pipe_handle,
     DCHECK(num_handles);
     DCHECK_LE(dispatchers.size(), static_cast<size_t>(*num_handles));
 
-    base::AutoLock locker(handle_table_lock_);
-
-    for (size_t i = 0; i < dispatchers.size(); i++) {
-      // TODO(vtl): What should we do if we hit the maximum handle table size
-      // here? Currently, we'll just fill in those handles with
-      // |MOJO_HANDLE_INVALID| (and return success anyway).
-      // TODO(vtl): crbug.com/345911: On failure, we should close the dispatcher
-      // (outside the table lock).
-      handles[i] = AddDispatcherNoLock(dispatchers[i]);
-      LOG_IF(ERROR, handles[i] == MOJO_HANDLE_INVALID)
-          << "Failed to add dispatcher (" << dispatchers[i].get() << ")";
+    bool success;
+    {
+      base::AutoLock locker(handle_table_lock_);
+      success = handle_table_.AddDispatcherVector(dispatchers, handles);
+    }
+    if (!success) {
+      LOG(ERROR) << "Received message with " << dispatchers.size()
+                 << " handles, but handle table full";
+      // Close dispatchers (outside the lock).
+      for (size_t i = 0; i < dispatchers.size(); i++) {
+        if (dispatchers[i])
+          dispatchers[i]->Close();
+      }
     }
   }
 
@@ -409,31 +324,27 @@ MojoResult CoreImpl::CreateDataPipe(const MojoCreateDataPipeOptions* options,
   scoped_refptr<DataPipeConsumerDispatcher> consumer_dispatcher(
       new DataPipeConsumerDispatcher());
 
-  MojoHandle producer_handle, consumer_handle;
+  std::pair<MojoHandle, MojoHandle> handle_pair;
   {
     base::AutoLock locker(handle_table_lock_);
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close the dispatcher
-    // (outside the table lock).
-    producer_handle = AddDispatcherNoLock(producer_dispatcher);
-    if (producer_handle == MOJO_HANDLE_INVALID)
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close both dispatchers
-    // (outside the table lock).
-    consumer_handle = AddDispatcherNoLock(consumer_dispatcher);
-    if (consumer_handle == MOJO_HANDLE_INVALID) {
-      handle_table_.erase(producer_handle);
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
-    }
+    handle_pair = handle_table_.AddDispatcherPair(producer_dispatcher,
+                                                  consumer_dispatcher);
+  }
+  if (handle_pair.first == MOJO_HANDLE_INVALID) {
+    DCHECK_EQ(handle_pair.second, MOJO_HANDLE_INVALID);
+    LOG(ERROR) << "Handle table full";
+    producer_dispatcher->Close();
+    consumer_dispatcher->Close();
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
+  DCHECK_NE(handle_pair.second, MOJO_HANDLE_INVALID);
 
   scoped_refptr<DataPipe> data_pipe(new LocalDataPipe(validated_options));
   producer_dispatcher->Init(data_pipe);
   consumer_dispatcher->Init(data_pipe);
 
-  *data_pipe_producer_handle = producer_handle;
-  *data_pipe_consumer_handle = consumer_handle;
+  *data_pipe_producer_handle = handle_pair.first;
+  *data_pipe_consumer_handle = handle_pair.second;
   return MOJO_RESULT_OK;
 }
 
@@ -534,15 +445,11 @@ MojoResult CoreImpl::CreateSharedBuffer(
     return result;
   }
 
-  MojoHandle h;
-  {
-    base::AutoLock locker(handle_table_lock_);
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close the dispatcher
-    // (outside the table lock).
-    h = AddDispatcherNoLock(dispatcher);
-    if (h == MOJO_HANDLE_INVALID)
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
+  MojoHandle h = AddDispatcher(dispatcher);
+  if (h == MOJO_HANDLE_INVALID) {
+    LOG(ERROR) << "Handle table full";
+    dispatcher->Close();
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   *shared_buffer_handle = h;
@@ -567,15 +474,11 @@ MojoResult CoreImpl::DuplicateBufferHandle(
   if (result != MOJO_RESULT_OK)
     return result;
 
-  MojoHandle new_handle;
-  {
-    base::AutoLock locker(handle_table_lock_);
-
-    // TODO(vtl): crbug.com/345911: On failure, we should close the dispatcher
-    // (outside the table lock).
-    new_handle = AddDispatcherNoLock(new_dispatcher);
-    if (new_handle == MOJO_HANDLE_INVALID)
-      return MOJO_RESULT_RESOURCE_EXHAUSTED;
+  MojoHandle new_handle = AddDispatcher(new_dispatcher);
+  if (new_handle == MOJO_HANDLE_INVALID) {
+    LOG(ERROR) << "Handle table full";
+    dispatcher->Close();
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   *new_buffer_handle = new_handle;
@@ -620,37 +523,7 @@ scoped_refptr<Dispatcher> CoreImpl::GetDispatcher(MojoHandle handle) {
     return NULL;
 
   base::AutoLock locker(handle_table_lock_);
-  HandleTableMap::iterator it = handle_table_.find(handle);
-  if (it == handle_table_.end())
-    return NULL;
-
-  return it->second.dispatcher;
-}
-
-MojoHandle CoreImpl::AddDispatcherNoLock(
-    const scoped_refptr<Dispatcher>& dispatcher) {
-  handle_table_lock_.AssertAcquired();
-  DCHECK_NE(next_handle_, MOJO_HANDLE_INVALID);
-
-  if (!dispatcher.get() || handle_table_.size() >= kMaxHandleTableSize)
-    return MOJO_HANDLE_INVALID;
-
-  // TODO(vtl): Maybe we want to do something different/smarter. (Or maybe try
-  // assigning randomly?)
-  while (handle_table_.find(next_handle_) != handle_table_.end()) {
-    next_handle_++;
-    if (next_handle_ == MOJO_HANDLE_INVALID)
-      next_handle_++;
-  }
-
-  MojoHandle new_handle = next_handle_;
-  handle_table_[new_handle] = HandleTableEntry(dispatcher);
-
-  next_handle_++;
-  if (next_handle_ == MOJO_HANDLE_INVALID)
-    next_handle_++;
-
-  return new_handle;
+  return handle_table_.GetDispatcher(handle);
 }
 
 // Note: We allow |handles| to repeat the same handle multiple times, since
diff --git a/mojo/system/core_impl.h b/mojo/system/core_impl.h
index b82f8d2..63b4b08 100644
--- a/mojo/system/core_impl.h
+++ b/mojo/system/core_impl.h
@@ -7,24 +7,17 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
-#include "base/containers/hash_tables.h"
 #include "base/memory/ref_counted.h"
 #include "base/synchronization/lock.h"
 #include "mojo/public/system/core_private.h"
+#include "mojo/system/handle_table.h"
 #include "mojo/system/system_impl_export.h"
 
 namespace mojo {
 namespace system {
 
-class CoreImpl;
 class Dispatcher;
 
-// Test-only function (defined/used in embedder/test_embedder.cc). Declared here
-// so it can be friended.
-namespace internal {
-bool ShutdownCheckNoLeaks(CoreImpl*);
-}
-
 // |CoreImpl| is a singleton object that implements the Mojo system calls. All
 // public methods are thread-safe.
 class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
@@ -131,11 +124,6 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
   // invalid.
   scoped_refptr<Dispatcher> GetDispatcher(MojoHandle handle);
 
-  // Assigns a new handle for the given dispatcher; returns
-  // |MOJO_HANDLE_INVALID| on failure (due to hitting resource limits) or if
-  // |dispatcher| is null. Must be called under |handle_table_lock_|.
-  MojoHandle AddDispatcherNoLock(const scoped_refptr<Dispatcher>& dispatcher);
-
   // Internal implementation of |Wait()| and |WaitMany()|; doesn't do basic
   // validation of arguments.
   MojoResult WaitManyInternal(const MojoHandle* handles,
@@ -147,9 +135,8 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
 
   // TODO(vtl): |handle_table_lock_| should be a reader-writer lock (if only we
   // had them).
-  base::Lock handle_table_lock_;  // Protects the immediately-following members.
-  HandleTableMap handle_table_;
-  MojoHandle next_handle_;  // Invariant: never |MOJO_HANDLE_INVALID|.
+  base::Lock handle_table_lock_;  // Protects |handle_table_|.
+  HandleTable handle_table_;
 
   // ---------------------------------------------------------------------------
 
diff --git a/mojo/system/dispatcher.cc b/mojo/system/dispatcher.cc
index de702aa..47b7e40 100644
--- a/mojo/system/dispatcher.cc
+++ b/mojo/system/dispatcher.cc
@@ -16,7 +16,7 @@ namespace test {
 // TODO(vtl): Maybe this should be defined in a test-only file instead.
 DispatcherTransport DispatcherTryStartTransport(
     Dispatcher* dispatcher) {
-  return Dispatcher::CoreImplAccess::TryStartTransport(dispatcher);
+  return Dispatcher::HandleTableAccess::TryStartTransport(dispatcher);
 }
 
 }  // namespace test
@@ -24,7 +24,7 @@ DispatcherTransport DispatcherTryStartTransport(
 // Dispatcher ------------------------------------------------------------------
 
 // static
-DispatcherTransport Dispatcher::CoreImplAccess::TryStartTransport(
+DispatcherTransport Dispatcher::HandleTableAccess::TryStartTransport(
     Dispatcher* dispatcher) {
   DCHECK(dispatcher);
 
diff --git a/mojo/system/dispatcher.h b/mojo/system/dispatcher.h
index c6ef752..75d9d42 100644
--- a/mojo/system/dispatcher.h
+++ b/mojo/system/dispatcher.h
@@ -27,6 +27,7 @@ class Channel;
 class CoreImpl;
 class Dispatcher;
 class DispatcherTransport;
+class HandleTable;
 class LocalMessagePipeEndpoint;
 class MessageInTransit;
 class ProxyMessagePipeEndpoint;
@@ -126,16 +127,17 @@ class MOJO_SYSTEM_IMPL_EXPORT Dispatcher :
   void RemoveWaiter(Waiter* waiter);
 
   // A dispatcher must be put into a special state in order to be sent across a
-  // message pipe. Outside of tests, only |CoreImplAccess| is allowed to do
+  // message pipe. Outside of tests, only |HandleTableAccess| is allowed to do
   // this, since there are requirements on the handle table (see below).
   //
   // In this special state, only a restricted set of operations is allowed.
   // These are the ones available as |DispatcherTransport| methods. Other
   // |Dispatcher| methods must not be called until |DispatcherTransport::End()|
   // has been called.
-  class CoreImplAccess {
+  class HandleTableAccess {
    private:
     friend class CoreImpl;
+    friend class HandleTable;
     // Tests also need this, to avoid needing |CoreImpl|.
     friend DispatcherTransport test::DispatcherTryStartTransport(Dispatcher*);
 
@@ -300,8 +302,8 @@ class MOJO_SYSTEM_IMPL_EXPORT Dispatcher :
 };
 
 // Wrapper around a |Dispatcher| pointer, while it's being processed to be
-// passed in a message pipe. See the comment about |Dispatcher::CoreImplAccess|
-// for more details.
+// passed in a message pipe. See the comment about
+// |Dispatcher::HandleTableAccess| for more details.
 //
 // Note: This class is deliberately "thin" -- no more expensive than a
 // |Dispatcher*|.
@@ -324,7 +326,7 @@ class MOJO_SYSTEM_IMPL_EXPORT DispatcherTransport {
   Dispatcher* dispatcher() { return dispatcher_; }
 
  private:
-  friend class Dispatcher::CoreImplAccess;
+  friend class Dispatcher::HandleTableAccess;
 
   explicit DispatcherTransport(Dispatcher* dispatcher)
       : dispatcher_(dispatcher) {}
diff --git a/mojo/system/handle_table.cc b/mojo/system/handle_table.cc
new file mode 100644
index 0000000..fc03e2e
--- /dev/null
+++ b/mojo/system/handle_table.cc
@@ -0,0 +1,240 @@
+// Copyright 2014 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/handle_table.h"
+
+#include "base/basictypes.h"
+#include "base/logging.h"
+#include "mojo/system/constants.h"
+#include "mojo/system/dispatcher.h"
+
+namespace mojo {
+namespace system {
+
+HandleTable::Entry::Entry()
+    : busy(false) {
+}
+
+HandleTable::Entry::Entry(const scoped_refptr<Dispatcher>& dispatcher)
+    : dispatcher(dispatcher),
+      busy(false) {
+}
+
+HandleTable::Entry::~Entry() {
+  DCHECK(!busy);
+}
+
+HandleTable::HandleTable()
+    : next_handle_(MOJO_HANDLE_INVALID + 1) {
+}
+
+HandleTable::~HandleTable() {
+  // This should usually not be reached (the only instance should be owned by
+  // the singleton |CoreImpl|, which lives forever), except in tests.
+}
+
+Dispatcher* HandleTable::GetDispatcher(MojoHandle handle) {
+  DCHECK_NE(handle, MOJO_HANDLE_INVALID);
+
+  HandleToEntryMap::iterator it = handle_to_entry_map_.find(handle);
+  if (it == handle_to_entry_map_.end())
+    return NULL;
+  return it->second.dispatcher;
+}
+
+MojoResult HandleTable::GetAndRemoveDispatcher(
+    MojoHandle handle,
+    scoped_refptr<Dispatcher>* dispatcher) {
+  DCHECK_NE(handle, MOJO_HANDLE_INVALID);
+  DCHECK(dispatcher);
+
+  HandleToEntryMap::iterator it = handle_to_entry_map_.find(handle);
+  if (it == handle_to_entry_map_.end())
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  if (it->second.busy)
+    return MOJO_RESULT_BUSY;
+  *dispatcher = it->second.dispatcher;
+  handle_to_entry_map_.erase(it);
+
+  return MOJO_RESULT_OK;
+}
+
+MojoHandle HandleTable::AddDispatcher(
+    const scoped_refptr<Dispatcher>& dispatcher) {
+  if (handle_to_entry_map_.size() >= kMaxHandleTableSize)
+    return MOJO_HANDLE_INVALID;
+  return AddDispatcherNoSizeCheck(dispatcher);
+}
+
+std::pair<MojoHandle, MojoHandle> HandleTable::AddDispatcherPair(
+    const scoped_refptr<Dispatcher>& dispatcher0,
+    const scoped_refptr<Dispatcher>& dispatcher1) {
+  if (handle_to_entry_map_.size() + 1 >= kMaxHandleTableSize)
+    return std::make_pair(MOJO_HANDLE_INVALID, MOJO_HANDLE_INVALID);
+  return std::make_pair(AddDispatcherNoSizeCheck(dispatcher0),
+                        AddDispatcherNoSizeCheck(dispatcher1));
+}
+
+bool HandleTable::AddDispatcherVector(
+    const std::vector<scoped_refptr<Dispatcher> >& dispatchers,
+    MojoHandle* handles) {
+  DCHECK_LE(dispatchers.size(), kMaxMessageNumHandles);
+  DCHECK(handles);
+  // TODO(vtl): |std::numeric_limits<size_t>::max()| isn't a compile-time
+  // expression in C++03.
+  COMPILE_ASSERT(
+      static_cast<uint64_t>(kMaxHandleTableSize) + kMaxMessageNumHandles <
+          (sizeof(size_t) == 8 ? kuint64max :
+                                 static_cast<uint64_t>(kuint32max)),
+      addition_may_overflow);
+
+  if (handle_to_entry_map_.size() + dispatchers.size() > kMaxHandleTableSize)
+    return false;
+
+  for (size_t i = 0; i < dispatchers.size(); i++) {
+    if (dispatchers[i]) {
+      handles[i] = AddDispatcherNoSizeCheck(dispatchers[i]);
+    } else {
+      LOG(WARNING) << "Invalid dispatcher at index " << i;
+      handles[i] = MOJO_HANDLE_INVALID;
+    }
+  }
+  return true;
+}
+
+MojoResult HandleTable::MarkBusyAndStartTransport(
+    MojoHandle disallowed_handle,
+    const MojoHandle* handles,
+    uint32_t num_handles,
+    std::vector<DispatcherTransport>* transports) {
+  DCHECK_NE(disallowed_handle, MOJO_HANDLE_INVALID);
+  DCHECK(handles);
+  DCHECK_LE(num_handles, kMaxMessageNumHandles);
+  DCHECK(transports);
+
+  std::vector<Entry*> entries(num_handles);
+
+  // First verify all the handles and get their dispatchers.
+  uint32_t i;
+  MojoResult error_result = MOJO_RESULT_INTERNAL;
+  for (i = 0; i < num_handles; i++) {
+    // Sending your own handle is not allowed (and, for consistency, returns
+    // "busy").
+    if (handles[i] == disallowed_handle) {
+      error_result = MOJO_RESULT_BUSY;
+      break;
+    }
+
+    HandleToEntryMap::iterator it = handle_to_entry_map_.find(handles[i]);
+    if (it == handle_to_entry_map_.end()) {
+      error_result = MOJO_RESULT_INVALID_ARGUMENT;
+      break;
+    }
+
+    entries[i] = &it->second;
+    if (entries[i]->busy) {
+      error_result = MOJO_RESULT_BUSY;
+      break;
+    }
+    // Note: By marking the handle as busy here, we're also preventing the
+    // same handle from being sent multiple times in the same message.
+    entries[i]->busy = true;
+
+    // Try to start the transport.
+    DispatcherTransport transport =
+        Dispatcher::HandleTableAccess::TryStartTransport(
+            entries[i]->dispatcher.get());
+    if (!transport.is_valid()) {
+      // Unset the busy flag (since it won't be unset below).
+      entries[i]->busy = false;
+      error_result = MOJO_RESULT_BUSY;
+      break;
+    }
+
+    // Check if the dispatcher is busy (e.g., in a two-phase read/write).
+    // (Note that this must be done after the dispatcher's lock is acquired.)
+    if (transport.IsBusy()) {
+      // Unset the busy flag and end the transport (since it won't be done
+      // below).
+      entries[i]->busy = false;
+      transport.End();
+      error_result = MOJO_RESULT_BUSY;
+      break;
+    }
+
+    // Hang on to the transport (which we'll need to end the transport).
+    (*transports)[i] = transport;
+  }
+  if (i < num_handles) {
+    DCHECK_NE(error_result, MOJO_RESULT_INTERNAL);
+
+    // Unset the busy flags and release the locks.
+    for (uint32_t j = 0; j < i; j++) {
+      DCHECK(entries[j]->busy);
+      entries[j]->busy = false;
+      (*transports)[j].End();
+    }
+    return error_result;
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+
+//////////////
+
+MojoHandle HandleTable::AddDispatcherNoSizeCheck(
+    const scoped_refptr<Dispatcher>& dispatcher) {
+  DCHECK(dispatcher);
+  DCHECK_LT(handle_to_entry_map_.size(), kMaxHandleTableSize);
+  DCHECK_NE(next_handle_, MOJO_HANDLE_INVALID);
+
+  // TODO(vtl): Maybe we want to do something different/smarter. (Or maybe try
+  // assigning randomly?)
+  while (handle_to_entry_map_.find(next_handle_) !=
+             handle_to_entry_map_.end()) {
+    next_handle_++;
+    if (next_handle_ == MOJO_HANDLE_INVALID)
+      next_handle_++;
+  }
+
+  MojoHandle new_handle = next_handle_;
+  handle_to_entry_map_[new_handle] = Entry(dispatcher);
+
+  next_handle_++;
+  if (next_handle_ == MOJO_HANDLE_INVALID)
+    next_handle_++;
+
+  return new_handle;
+}
+
+void HandleTable::RemoveBusyHandles(const MojoHandle* handles,
+                                    uint32_t num_handles) {
+  DCHECK(handles);
+  DCHECK_LE(num_handles, kMaxMessageNumHandles);
+
+  for (uint32_t i = 0; i < num_handles; i++) {
+    HandleToEntryMap::iterator it = handle_to_entry_map_.find(handles[i]);
+    DCHECK(it != handle_to_entry_map_.end());
+    DCHECK(it->second.busy);
+    it->second.busy = false;  // For the sake of a |DCHECK()|.
+    handle_to_entry_map_.erase(it);
+  }
+}
+
+void HandleTable::RestoreBusyHandles(const MojoHandle* handles,
+                                     uint32_t num_handles) {
+  DCHECK(handles);
+  DCHECK_LE(num_handles, kMaxMessageNumHandles);
+
+  for (uint32_t i = 0; i < num_handles; i++) {
+    HandleToEntryMap::iterator it = handle_to_entry_map_.find(handles[i]);
+    DCHECK(it != handle_to_entry_map_.end());
+    DCHECK(it->second.busy);
+    it->second.busy = false;
+  }
+}
+
+}  // namespace system
+}  // namespace mojo
diff --git a/mojo/system/handle_table.h b/mojo/system/handle_table.h
new file mode 100644
index 0000000..4bcf02c
--- /dev/null
+++ b/mojo/system/handle_table.h
@@ -0,0 +1,126 @@
+// Copyright 2014 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_HANDLE_TABLE_H_
+#define MOJO_SYSTEM_HANDLE_TABLE_H_
+
+#include <utility>
+#include <vector>
+
+#include "base/containers/hash_tables.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "mojo/public/c/system/core.h"
+#include "mojo/system/system_impl_export.h"
+
+namespace mojo {
+namespace system {
+
+class CoreImpl;
+class Dispatcher;
+class DispatcherTransport;
+
+// Test-only function (defined/used in embedder/test_embedder.cc). Declared here
+// so it can be friended.
+namespace internal {
+bool ShutdownCheckNoLeaks(CoreImpl*);
+}
+
+// This class provides the (global) handle table (owned by |CoreImpl|), which
+// maps (valid) |MojoHandle|s to |Dispatcher|s. This is abstracted so that,
+// e.g., caching may be added.
+//
+// This class is NOT thread-safe; locking is left to |CoreImpl| (since it may
+// need to make several changes -- "atomically" or in rapid successsion, in
+// which case the extra locking/unlocking would be unnecessary overhead).
+
+class MOJO_SYSTEM_IMPL_EXPORT HandleTable {
+ public:
+  HandleTable();
+  ~HandleTable();
+
+  // Gets the dispatcher for a given handle (which should not be
+  // |MOJO_HANDLE_INVALID|). Returns null if there's no dispatcher for the given
+  // handle.
+  // WARNING: For efficiency, this returns a dumb pointer. If you're going to
+  // use the result outside |CoreImpl|'s lock, you MUST take a reference (e.g.,
+  // by storing the result inside a |scoped_refptr|).
+  Dispatcher* GetDispatcher(MojoHandle handle);
+
+  // On success, gets the dispatcher for a given handle (which should not be
+  // |MOJO_HANDLE_INVALID|) and removes it. (On failure, returns an appropriate
+  // result (and leaves |dispatcher| alone), namely
+  // |MOJO_RESULT_INVALID_ARGUMENT| if there's no dispatcher for the given
+  // handle or |MOJO_RESULT_BUSY| if the handle is marked as busy.)
+  MojoResult GetAndRemoveDispatcher(MojoHandle handle,
+                                    scoped_refptr<Dispatcher>* dispatcher);
+
+  // Adds a dispatcher (which must be valid), returning the handle for it.
+  // Returns |MOJO_HANDLE_INVALID| on failure (if the handle table is full).
+  MojoHandle AddDispatcher(const scoped_refptr<Dispatcher>& dispatcher);
+
+  // Adds a pair of dispatchers (which must be valid), return a pair of handles
+  // for them. On failure (if the handle table is full), the first (and second)
+  // handles will be |MOJO_HANDLE_INVALID|, and neither dispatcher will be
+  // added.
+  std::pair<MojoHandle, MojoHandle> AddDispatcherPair(
+      const scoped_refptr<Dispatcher>& dispatcher0,
+      const scoped_refptr<Dispatcher>& dispatcher1);
+
+  // Adds the given vector of dispatchers (of size at most
+  // |kMaxMessageNumHandles|). |handles| must point to an array of size at least
+  // |dispatchers.size()|. Unlike the other |AddDispatcher...()| functions, some
+  // of the dispatchers may be invalid (null). Returns true on success and false
+  // on failure (if the handle table is full), in which case it leaves
+  // |handles[...]| untouched (and all dispatchers unadded).
+  bool AddDispatcherVector(
+      const std::vector<scoped_refptr<Dispatcher> >& dispatchers,
+      MojoHandle* handles);
+
+  // Tries to mark the given handles as busy and start transport on them (i.e.,
+  // take their dispatcher locks); |transports| must be sized to contain
+  // |num_handles| elements. On failure, returns them to their original
+  // (non-busy, unlocked state).
+  MojoResult MarkBusyAndStartTransport(
+      MojoHandle disallowed_handle,
+      const MojoHandle* handles,
+      uint32_t num_handles,
+      std::vector<DispatcherTransport>* transports);
+
+  // Remove the given handles, which must all be present and which should have
+  // previously been marked busy by |MarkBusyAndStartTransport()|.
+  void RemoveBusyHandles(const MojoHandle* handles, uint32_t num_handles);
+
+  // Restores the given handles, which must all be present and which should have
+  // previously been marked busy by |MarkBusyAndStartTransport()|, to a non-busy
+  // state.
+  void RestoreBusyHandles(const MojoHandle* handles, uint32_t num_handles);
+
+ private:
+  friend bool internal::ShutdownCheckNoLeaks(CoreImpl*);
+
+  struct Entry {
+    Entry();
+    explicit Entry(const scoped_refptr<Dispatcher>& dispatcher);
+    ~Entry();
+
+    scoped_refptr<Dispatcher> dispatcher;
+    bool busy;
+  };
+  typedef base::hash_map<MojoHandle, Entry> HandleToEntryMap;
+
+  // Adds the given dispatcher to the handle table, not doing any size checks.
+  MojoHandle AddDispatcherNoSizeCheck(
+      const scoped_refptr<Dispatcher>& dispatcher);
+
+  HandleToEntryMap handle_to_entry_map_;
+  MojoHandle next_handle_;  // Invariant: never |MOJO_HANDLE_INVALID|.
+
+  DISALLOW_COPY_AND_ASSIGN(HandleTable);
+};
+
+}  // namespace system
+}  // namespace mojo
+
+#endif  // MOJO_SYSTEM_HANDLE_TABLE_H_