Added primitives to IMC-based MessageQueue to allow clients,

rather than O3D, to allocate and register shared memory segments.
This is required in order to work in Protected Mode Internet
Explorer on Windows Vista and later. Fixed preexisting bugs in
MessageQueue related to shared memory mapping failures and
corruption of concurrent incoming messages.

Added the MessageQueue unit tests and Chrome's ConditionVariable
class to the build. These tests required restructuring and
multithreading. Wrote simple framework which detects test
failures and timeouts in child threads and reports them to the
main thread. Restructured existing MessageQueue tests. Added unit
test for Register/UnregisterSharedMemory and stress test for
above concurrency bug.

Buganizer ID: 1997023.

Review URL: http://codereview.chromium.org/155947

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

6 files changed, 784 insertions, 95 deletions

diff --git a/o3d/base/build.scons b/o3d/base/build.scons
index 499b56d..45fa4bd 100644
--- a/o3d/base/build.scons
+++ b/o3d/base/build.scons
@@ -65,6 +65,7 @@ if env.Bit('windows'):
   env.Append(CCFLAGS=['/FIstdlib.h'])
   chrome_base_inputs += [
     'base_paths_win',
+    'condition_variable_win',
     'debug_util_win',
     'file_util_win',
     'lock_impl_win',
@@ -103,6 +104,7 @@ if env.Bit('windows'):
 
 chrome_base_inputs_posix = [
   'atomicops_internals_x86_gcc',
+  'condition_variable_posix',
   'debug_util_posix',
   'file_util_posix',
   'lock_impl_posix',
diff --git a/o3d/core/core.gyp b/o3d/core/core.gyp
index 4e14c88..7bd7926 100644
--- a/o3d/core/core.gyp
+++ b/o3d/core/core.gyp
@@ -375,8 +375,7 @@
           'cross/matrix4_composition_test.cc',
           'cross/matrix4_scale_test.cc',
           'cross/matrix4_translation_test.cc',
-  # TODO(gspencer): fix this test so it can be re-enabled.
-  #       'cross/message_queue_test.cc',
+          'cross/message_queue_test.cc',
           'cross/object_base_test.cc',
           'cross/pack_test.cc',
           'cross/param_array_test.cc',
diff --git a/o3d/core/cross/message_queue.cc b/o3d/core/cross/message_queue.cc
index 17e981d..eb4cc04 100644
--- a/o3d/core/cross/message_queue.cc
+++ b/o3d/core/cross/message_queue.cc
@@ -98,6 +98,22 @@ void ConnectedClient::RegisterSharedMemory(int32 buffer_id,
   shared_memory_array_.push_back(shared_mem);
 }
 
+// Unregisters a shared memory buffer for a given client-allocated
+// memory region, unmapping and closing it in the process.
+bool ConnectedClient::UnregisterSharedMemory(int32 buffer_id) {
+  std::vector<SharedMemoryInfo>::iterator iter;
+  for (iter = shared_memory_array_.begin(); iter < shared_memory_array_.end();
+       ++iter) {
+    if (iter->buffer_id_ == buffer_id) {
+      nacl::Unmap(iter->mapped_address_, iter->size_);
+      nacl::Close(iter->shared_memory_handle_);
+      shared_memory_array_.erase(iter);
+      return true;
+    }
+  }
+  return false;
+}
+
 // Returns the SharedMemoryInfo corresponding to the given shared
 // memory buffer id.  The buffer must first be created by the
 // MessageQueue on behalf of this ConnectedClient.
@@ -190,8 +206,6 @@ bool MessageQueue::CheckForNewMessages() {
   nacl::IOVec io_vec[1];
   io_vec[0].base = message_buffer;
   io_vec[0].length = kBufferLength;
-  // Clear out the buffer.
-  memset(message_buffer, 0, kBufferLength);
 
   nacl::MessageHeader header;
   header.iov = io_vec;
@@ -226,6 +240,10 @@ bool MessageQueue::CheckForNewMessages() {
   // messages.
   std::vector<ConnectedClient*>::iterator iter;
   for (iter = connected_clients_.begin(); iter < connected_clients_.end();) {
+    // Must reset the available buffer length and number of handles each time
+    // so NaCl's IMC knows how much space is available for reading
+    io_vec[0].length = kBufferLength;
+    header.handle_count = kMaxNumHandles;
     if (ReceiveMessageFromSocket((*iter)->client_handle(),
                                  &header,
                                  &message_id,
@@ -337,6 +355,18 @@ bool MessageQueue::ProcessClientRequest(ConnectedClient* client,
                                     message_id,
                                     header,
                                     handles);
+    case REGISTER_SHARED_MEMORY:
+      return ProcessRegisterSharedMemory(client,
+                                         message_length,
+                                         message_id,
+                                         header,
+                                         handles);
+    case UNREGISTER_SHARED_MEMORY:
+      return ProcessUnregisterSharedMemory(client,
+                                           message_length,
+                                           message_id,
+                                           header,
+                                           handles);
     default:
       LOG(ERROR) << "Unrecognized message id " << message_id;
       return false;
@@ -450,7 +480,7 @@ bool MessageQueue::ProcessAllocateSharedMemory(ConnectedClient* client,
                                   shared_memory,
                                   0);
 
-  if (shared_region == NULL) {
+  if (shared_region == nacl::kMapFailed) {
     LOG_IMC_ERROR("Failed to map shared memory");
     nacl::Close(shared_memory);
     return false;
@@ -580,5 +610,112 @@ bool MessageQueue::ProcessUpdateTexture2D(ConnectedClient* client,
   return true;
 }
 
+// Processes a request to register a client-allocated shared memory
+// buffer on behalf of a connected client.  Parses the arguments of
+// the message to determine how much space is being passed. It maps
+// the shared memory buffer into the local address space and sends a
+// message back to the client with the newly allocated shared memory
+// ID.
+bool MessageQueue::ProcessRegisterSharedMemory(ConnectedClient* client,
+                                               int message_length,
+                                               MessageId message_id,
+                                               nacl::MessageHeader* header,
+                                               nacl::Handle* handles) {
+  int32 mem_size = 0;
+  int expected_message_length = sizeof(message_id) + sizeof(mem_size);
+
+  if (message_length != expected_message_length ||
+      header->iov_length != 1 ||
+      header->handle_count != 1) {
+    LOG(ERROR) << "Malformed message for REGISTER_SHARED_MEMORY";
+    return false;
+  }
+
+  char* message_buffer = static_cast<char*>(header->iov[0].base);
+  message_buffer += sizeof(message_id);
+  mem_size = *(reinterpret_cast<int32*>(message_buffer));
+  const int32 kMaxSharedMemSize = 1024 * 1024 * 100;   // 100MB
+  if (mem_size <= 0 || mem_size > kMaxSharedMemSize) {
+    LOG(ERROR) << "Invalid mem size sent: " << mem_size
+               << "(max size = " << kMaxSharedMemSize << ")";
+    return false;
+  }
+
+  // Fetch the handle to the preexisting shared memory object.
+  nacl::Handle shared_memory = header->handles[0];
+  if (shared_memory == nacl::kInvalidHandle) {
+    LOG_IMC_ERROR("Invalid shared memory object registered");
+    return false;
+  }
+
+  // Map it in local address space.
+  void* shared_region = nacl::Map(0,
+                                  mem_size,
+                                  nacl::kProtRead | nacl::kProtWrite,
+                                  nacl::kMapShared,
+                                  shared_memory,
+                                  0);
+  if (shared_region == nacl::kMapFailed) {
+    LOG_IMC_ERROR("Failed to map shared memory");
+    nacl::Close(shared_memory);
+    return false;
+  }
+
+  // Create a unique id for the shared memory buffer.
+  int32 buffer_id = next_shared_memory_id_++;
+
+  // Send the buffer id back to the client.
+  nacl::MessageHeader response_header;
+  nacl::IOVec id_vec;
+  id_vec.base = &buffer_id;
+  id_vec.length = sizeof(buffer_id);
+
+  response_header.iov = &id_vec;
+  response_header.iov_length = 1;
+  response_header.handles = NULL;
+  response_header.handle_count = 0;
+  int result = nacl::SendDatagram(client->client_handle(), &response_header, 0);
+
+  if (result != sizeof(buffer_id)) {
+    LOG_IMC_ERROR("Failed to send shared memory ID back to the client");
+    nacl::Unmap(shared_region, mem_size);
+    nacl::Close(shared_memory);
+    return false;
+  }
+
+  // Register the newly mapped shared memory with the connected client.
+  client->RegisterSharedMemory(buffer_id,
+                               shared_memory,
+                               shared_region,
+                               mem_size);
+
+  return true;
+}
+
+// Processes a request to unregister a client-allocated shared memory
+// buffer, referenced by ID.
+bool MessageQueue::ProcessUnregisterSharedMemory(ConnectedClient* client,
+                                                 int message_length,
+                                                 MessageId message_id,
+                                                 nacl::MessageHeader* header,
+                                                 nacl::Handle* handles) {
+  int32 buffer_id = 0;
+  int expected_message_length = sizeof(message_id) + sizeof(buffer_id);
+
+  if (message_length != expected_message_length ||
+      header->iov_length != 1 ||
+      header->handle_count != 0) {
+    LOG(ERROR) << "Malformed message for UNREGISTER_SHARED_MEMORY";
+    return false;
+  }
+
+  char* message_buffer = static_cast<char*>(header->iov[0].base);
+  message_buffer += sizeof(message_id);
+  buffer_id = *(reinterpret_cast<int32*>(message_buffer));
+
+  bool res = client->UnregisterSharedMemory(buffer_id);
+  SendBooleanResponse(client->client_handle(), res);
+}
+
 
 }  // namespace o3d
diff --git a/o3d/core/cross/message_queue.h b/o3d/core/cross/message_queue.h
index e32d471..50e7f4d 100644
--- a/o3d/core/cross/message_queue.h
+++ b/o3d/core/cross/message_queue.h
@@ -79,6 +79,13 @@ class ConnectedClient {
                             void *address,
                             int32 size);
 
+  // Unregisters a client-allocated shared memory segment, referenced by ID.
+  // Parameters:
+  //  id - the unique id of the shared memory buffer.
+  // Returns:
+  //  true if the ID was valid, false otherwise
+  bool UnregisterSharedMemory(int id);
+
   // Returns the socket handle the client uses to talk to the server.
   nacl::Handle client_handle() { return client_handle_; }
 
@@ -101,9 +108,13 @@ class MessageQueue {
  public:
   enum MessageId {
     INVALID_ID = 0,
-    HELLO,                   // Handshake between the client and the server
-    ALLOCATE_SHARED_MEMORY,  // Request to allocate a shared memory buffer
-    UPDATE_TEXTURE2D,        // Request to update a 2D texture bitmap
+    HELLO,                     // Handshake between the client and the server
+    ALLOCATE_SHARED_MEMORY,    // Request to allocate a shared memory buffer
+    UPDATE_TEXTURE2D,          // Request to update a 2D texture bitmap
+    REGISTER_SHARED_MEMORY,    // Register a client-allocated shared memory
+                               // buffer, returning a shared memory ID
+    UNREGISTER_SHARED_MEMORY,  // Unregister a client-allocated shared
+                               // memory ID
     MAX_NUM_IDS,
 
     ID_FORCE_DWORD = 0x7fffffff  // Forces a 32-bit size enum
@@ -161,7 +172,7 @@ class MessageQueue {
 
   // Processes a request by a connected client to allocate a shared memory
   // buffer.  The size of the requested buffer is determined from the data
-  // passed in the message.  Once the memory is allocated, a message containting
+  // passed in the message.  Once the memory is allocated, a message containing
   // the shared memory handle is sent back to the client.
   // Parameters:
   //   client - pointer to the ConnectedClient the request came from.
@@ -197,6 +208,49 @@ class MessageQueue {
                               nacl::MessageHeader* header,
                               nacl::Handle* handles);
 
+  // Processes a request by a connected client to register a
+  // client-allocated shared memory buffer with O3D.  The size of the
+  // buffer is determined from the data passed in the message.  Once
+  // the shared memory buffer has been mapped and registered, a
+  // message containing the shared memory ID is sent back to the
+  // client. This ID can be used to update the contents of a Texture2D
+  // object.
+  // Parameters:
+  //   client - pointer to the ConnectedClient the request came from.
+  //   message_length - length of the received message in bytes.
+  //   message_id - id of the request received by the message
+  //   header - message header containing information about the received
+  //            message.
+  //   handles - the array of handles referenced by the header.
+  // Returns:
+  //   true if the message is properly formed and is succesfully handled by the
+  //   Client.
+  bool ProcessRegisterSharedMemory(ConnectedClient* client,
+                                   int message_length,
+                                   MessageId message_id,
+                                   nacl::MessageHeader* header,
+                                   nacl::Handle* handles);
+
+  // Processes a request by a connected client to unregister a shared
+  // memory buffer previously registered with O3D. The shared memory
+  // buffer is referenced by the ID returned from
+  // RegisterSharedMemory.
+  // Parameters:
+  //   client - pointer to the ConnectedClient the request came from.
+  //   message_length - length of the received message in bytes.
+  //   message_id - id of the request received by the message
+  //   header - message header containing information about the received
+  //            message.
+  //   handles - the array of handles referenced by the header.
+  // Returns:
+  //   true if the message is properly formed and is succesfully handled by the
+  //   Client.
+  bool ProcessUnregisterSharedMemory(ConnectedClient* client,
+                                     int message_length,
+                                     MessageId message_id,
+                                     nacl::MessageHeader* header,
+                                     nacl::Handle* handles);
+
   // Sends a true of false (1 or 0) message using the given socket handle.
   // Parameters:
   //   client_handle - handle of socket to send the response to.
diff --git a/o3d/core/cross/message_queue_test.cc b/o3d/core/cross/message_queue_test.cc
index 065e1a1..9915629 100644
--- a/o3d/core/cross/message_queue_test.cc
+++ b/o3d/core/cross/message_queue_test.cc
@@ -36,49 +36,301 @@
 #include "core/cross/client.h"
 #include "core/cross/types.h"
 #include "tests/common/win/testing_common.h"
+#include "base/at_exit.h"
+#include "base/condition_variable.h"
+#include "base/lock.h"
+#include "base/platform_thread.h"
+#include "base/time.h"
 
+using ::base::AtExitManager;
+using ::base::Time;
+using ::base::TimeDelta;
 
 namespace o3d {
 
-// Name of socket address used in this test
-nacl::SocketAddress my_address = {
-  "test-client"
+//----------------------------------------------------------------------
+// These are helper classes for the little multithreaded test harness
+// below.
+
+class TimeSource {
+ public:
+  virtual ~TimeSource() {}
+  virtual TimeDelta TimeSinceConstruction() = 0;
 };
 
-class MessageQueueTest : public testing::Test {
+class WallClockTimeSource : public TimeSource {
+ private:
+  Time construction_time_;
+
+ public:
+  WallClockTimeSource() {
+    construction_time_ = Time::Now();
+  }
+
+  virtual TimeDelta TimeSinceConstruction() {
+    return Time::Now() - construction_time_;
+  }
+};
+
+// The TestWatchdog expects to be signalled a certain number of times
+// within a certain period of time. If it is not signalled this number
+// of times, this indicates one failure mode of the test.
+class TestWatchdog {
+ private:
+  Lock lock_;
+  ConditionVariable condition_;
+  int expected_num_signals_;
+  TimeDelta time_to_run_;
+  TimeSource* time_source_;
+
+ public:
+  TestWatchdog(int expected_num_signals,
+               TimeDelta time_to_run,
+               TimeSource* time_source)
+      : lock_(),
+        condition_(&lock_),
+        expected_num_signals_(expected_num_signals),
+        time_to_run_(time_to_run),
+        time_source_(time_source) {}
+
+  void Signal() {
+    AutoLock locker(lock_);
+    ASSERT_GE(expected_num_signals_, 0);
+    --expected_num_signals_;
+    condition_.Broadcast();
+  }
+
+  // Pause the current thread briefly waiting for a signal so we don't
+  // consume all CPU
+  void WaitBrieflyForSignal() {
+    AutoLock locker(lock_);
+    condition_.TimedWait(TimeDelta::FromMilliseconds(5));
+  }
+
+  bool Expired() {
+    return time_source_->TimeSinceConstruction() > time_to_run_;
+  }
+
+  bool Succeeded() {
+    return expected_num_signals_ == 0;
+  }
+
+  bool Done() {
+    return Succeeded() || Expired();
+  }
+};
+
+// This is the base class for the multithreaded tests which are
+// executed via MessageQueueTest::RunTests(). Each instance is run in
+// its own thread. Override the Run() method with the body of the
+// test.
+class PerThreadConnectedTest : public PlatformThread::Delegate {
+ private:
+  MessageQueue* queue_;
+  nacl::Handle socket_handle_;
+  TestWatchdog* watchdog_;
+  volatile bool completed_;
+  volatile bool passed_;
+  String file_;
+  int line_;
+  String failure_message_;
+
  protected:
+  void Pass() {
+    completed_ = true;
+    passed_ = true;
+    watchdog_->Signal();
+  }
+
+  void Fail(String file,
+            int line,
+            String failure_message) {
+    completed_ = true;
+    passed_ = false;
+    file_ = file;
+    line_ = line;
+    failure_message_ = failure_message;
+    watchdog_->Signal();
+  }
+
+ public:
+  PerThreadConnectedTest()
+      : queue_(NULL),
+        socket_handle_(nacl::kInvalidHandle),
+        watchdog_(NULL),
+        completed_(false),
+        passed_(false),
+        line_(0) {}
+
+  // Override this with the particular test's functionality
+  virtual void Run(MessageQueue* queue,
+                   nacl::Handle socket_handle) = 0;
+
+  void Configure(MessageQueue* queue,
+                 nacl::Handle socket_handle,
+                 TestWatchdog* watchdog) {
+    queue_ = queue;
+    watchdog_ = watchdog;
+    socket_handle_ = socket_handle;
+  }
+
+  // Indicates whether or not the PerThreadTest should be deleted;
+  // if it is hanging then to avoid crashes we do not delete it
+  bool Completed() {
+    return completed_;
+  }
+
+  bool Passed() {
+    return passed_;
+  }
+
+  const String FailureMessage() {
+    std::ostringstream oss;
+    oss << file_ << ", line " << line_ << ": " + failure_message_;
+    return oss.str();
+  }
 
+  // This overrides the functionality in PlatformThread::Delegate;
+  // don't override this in subclasses.
+  virtual void ThreadMain() {
+    Run(queue_, socket_handle_);
+  }
+};
+
+#define FAIL_TEST(message) Fail(__FILE__, __LINE__, message); return
+
+class TestProvider {
+ public:
+  virtual ~TestProvider() {}
+  virtual PerThreadConnectedTest* CreateTest() = 0;
+};
+
+//----------------------------------------------------------------------
+// This is the main class containing all of the other ones. It knows
+// how to run multiple concurrent PerThreadConnectedTests.
+class MessageQueueTest : public testing::Test {
+ protected:
   MessageQueueTest()
-      : object_manager_(g_service_locator) {}
+      : object_manager_(g_service_locator),
+        socket_handles_(NULL),
+        num_socket_handles_(0) {}
 
   virtual void SetUp();
   virtual void TearDown();
 
+  // This is the entry point for test cases that need to be run in one
+  // or more threads.
+  void RunTests(int num_threads,
+                TimeDelta timeout,
+                TestProvider* test_provider);
+
   Pack* pack() { return pack_; }
-  nacl::Handle my_socket_handle() { return my_socket_handle_; }
 
  private:
   ServiceDependency<ObjectManager> object_manager_;
   Pack *pack_;
-  nacl::Handle my_socket_handle_;
+  int num_socket_handles_;
+  nacl::Handle* socket_handles_;
+
+  // This can't be part of SetUp since it needs to be called from each
+  // individual test.
+  void ConfigureSockets(int number_of_clients);
+
+  nacl::Handle GetSocketHandle(int i) {
+    // We would use ASSERT_ here, but that doesn't seem to work from
+    // within methods that have non-void return types.
+    EXPECT_TRUE(socket_handles_ != NULL);
+    EXPECT_LT(i, num_socket_handles_);
+    return socket_handles_[i];
+  }
 };
 
 void MessageQueueTest::SetUp() {
-  pack_ = object_manager_->CreatePack(L"MessageQueueTest");
-
-  // Create a bound socket to connect to the MessageQueue.
-  my_socket_handle_ = nacl::BoundSocket(&my_address);
+  pack_ = object_manager_->CreatePack();
+  pack_->set_name("MessageQueueTest pack");
+}
 
-  ASSERT_TRUE(my_socket_handle_ != nacl::kInvalidHandle);
+void MessageQueueTest::ConfigureSockets(int number_of_clients) {
+  ASSERT_GT(number_of_clients, 0);
+  num_socket_handles_ = number_of_clients;
+  socket_handles_ = new nacl::Handle[num_socket_handles_];
+  ASSERT_TRUE(socket_handles_ != NULL);
+  for (int i = 0; i < num_socket_handles_; i++) {
+    nacl::SocketAddress socket_address;
+    ::base::snprintf(socket_address.path,
+                     sizeof(socket_address.path),
+                     "%s%d",
+                     "test-client",
+                     i);
+    socket_handles_[i] = nacl::BoundSocket(&socket_address);
+    ASSERT_NE(nacl::kInvalidHandle, socket_handles_[i]);
+  }
 }
 
 void MessageQueueTest::TearDown() {
-  nacl::Close(my_socket_handle_);
+  if (socket_handles_ != NULL) {
+    for (int i = 0; i < num_socket_handles_; i++) {
+      nacl::Close(socket_handles_[i]);
+    }
+    delete[] socket_handles_;
+    socket_handles_ = NULL;
+    num_socket_handles_ = 0;
+  }
+
   object_manager_->DestroyPack(pack_);
 }
 
-// Helper class that handles connecting to the MessageQueue and issuing
-// commands to it.
+void MessageQueueTest::RunTests(int num_threads,
+                                TimeDelta timeout,
+                                TestProvider* provider) {
+  MessageQueue* message_queue = new MessageQueue(g_service_locator);
+  message_queue->Initialize();
+
+  AtExitManager manager;
+  TimeSource* time_source = new WallClockTimeSource();
+  TestWatchdog* watchdog = new TestWatchdog(num_threads,
+                                            timeout,
+                                            time_source);
+  PerThreadConnectedTest** tests = new PerThreadConnectedTest*[num_threads];
+  ASSERT_TRUE(tests != NULL);
+  PlatformThreadHandle* thread_handles = new PlatformThreadHandle[num_threads];
+  ASSERT_TRUE(thread_handles != NULL);
+  ConfigureSockets(num_threads);
+  for (int i = 0; i < num_threads; i++) {
+    tests[i] = provider->CreateTest();
+    ASSERT_TRUE(tests[i] != NULL);
+    tests[i]->Configure(message_queue, GetSocketHandle(i), watchdog);
+  }
+  // Now that all tests are created, start them up
+  for (int i = 0; i < num_threads; i++) {
+    ASSERT_TRUE(PlatformThread::Create(0, tests[i], &thread_handles[i]));
+  }
+  // Wait for completion
+  while (!watchdog->Done()) {
+    ASSERT_TRUE(message_queue->CheckForNewMessages());
+    watchdog->WaitBrieflyForSignal();
+  }
+  ASSERT_FALSE(watchdog->Expired());
+  ASSERT_TRUE(watchdog->Succeeded());
+  for (int i = 0; i < num_threads; i++) {
+    PerThreadConnectedTest* test = tests[i];
+    ASSERT_TRUE(test->Passed()) << test->FailureMessage();
+    // Only join the thread and delete the test if it completed
+    if (test->Completed()) {
+      PlatformThread::Join(thread_handles[i]);
+      delete test;
+    }
+  }
+  delete[] thread_handles;
+  delete[] tests;
+  delete watchdog;
+  delete time_source;
+  delete message_queue;
+}
+
+//----------------------------------------------------------------------
+// This is a helper class that handles connecting to the MessageQueue
+// and issuing commands to it.
 class TextureUpdateHelper {
  public:
   TextureUpdateHelper() : o3d_handle_(nacl::kInvalidHandle) {}
@@ -99,6 +351,16 @@ class TextureUpdateHelper {
                             size_t offset,
                             size_t number_of_bytes);
 
+  // Registers a client-allocated shared memory segment with O3D,
+  // returning O3D's shared memory ID for later updating. Returns -1
+  // upon failure.
+  int RegisterSharedMemory(nacl::Handle shared_memory,
+                           size_t shared_memory_size);
+
+  // Unregisters a previously-registered client-allocated shared
+  // memory segment.
+  bool UnregisterSharedMemory(int shared_memory_id);
+
  private:
   // Handle of the socket that's connected to o3d.
   nacl::Handle o3d_handle_;
@@ -152,8 +414,10 @@ bool TextureUpdateHelper::ConnectToO3D(const char* o3d_address,
   vec.base = buffer;
   vec.length = message_size;
 
-  nacl::SocketAddress o3d_address;
-  sprintf_s(o3d_address.path, sizeof(o3d_address.path), "%s", o3d_address);
+  nacl::SocketAddress socket_address;
+  sprintf_s(socket_address.path,
+            sizeof(socket_address.path),
+            "%s", o3d_address);
 
   header.iov = &vec;
   header.iov_length = 1;
@@ -162,7 +426,7 @@ bool TextureUpdateHelper::ConnectToO3D(const char* o3d_address,
   int result = nacl::SendDatagramTo(my_socket_handle,
                                     &header,
                                     0,
-                                    &o3d_address);
+                                    &socket_address);
 
   EXPECT_EQ(message_size, result);
   if (result != message_size) {
@@ -237,7 +501,7 @@ bool TextureUpdateHelper::RequestSharedMemory(size_t requested_size,
   result = nacl::ReceiveDatagram(o3d_handle_, &header, 0);
 
   EXPECT_LT(0, result);
-  EXPECT_FALSE(header.flags & nacl::kMessageTruncated);
+  EXPECT_EQ(0, header.flags & nacl::kMessageTruncated);
   EXPECT_EQ(1, header.handle_count);
   EXPECT_EQ(1, header.iov_length);
 
@@ -327,7 +591,99 @@ bool TextureUpdateHelper::RequestTextureUpdate(unsigned int texture_id,
   return texture_updated;
 }
 
+// Registers a client-allocated shared memory segment with O3D. It
+// receives back a shared memory ID for later texture updating.
+// Returns -1 upon failure.
+int TextureUpdateHelper::RegisterSharedMemory(nacl::Handle shared_memory,
+                                              size_t shared_memory_size) {
+  if (o3d_handle_ == nacl::kInvalidHandle) {
+    return false;
+  }
+
+  MessageQueue::MessageId message_id = MessageQueue::REGISTER_SHARED_MEMORY;
+
+  nacl::MessageHeader header;
+  nacl::IOVec vec;
+  char buffer[256];
+  char *buffer_ptr = &buffer[0];
+
+  // Message contains the ID and one argument (the size of the shared memory
+  // buffer which has been allocated).
+  *(reinterpret_cast<MessageQueue::MessageId*>(buffer_ptr)) = message_id;
+  buffer_ptr += sizeof(message_id);
+  *(reinterpret_cast<size_t*>(buffer_ptr)) = shared_memory_size;
+  buffer_ptr += sizeof(shared_memory_size);
+
+  vec.base = buffer;
+  vec.length = buffer_ptr - &buffer[0];
+
+  header.iov = &vec;
+  header.iov_length = 1;
+  header.handles = &shared_memory;
+  header.handle_count = 1;
+
+  // Send message.
+  int result = nacl::SendDatagram(o3d_handle_, &header, 0);
+  EXPECT_EQ(vec.length, result);
+  if (result != vec.length) {
+    return -1;
+  }
+
+  // Wait for a message back from the server containing the ID of the
+  // shared memory object.
+  nacl::MessageHeader reply_header;
+  int shared_memory_id = -1;
+  nacl::IOVec shared_memory_vec;
+  shared_memory_vec.base = &shared_memory_id;
+  shared_memory_vec.length = sizeof(shared_memory_id);
+  reply_header.iov = &shared_memory_vec;
+  reply_header.iov_length = 1;
+  reply_header.handles = NULL;
+  reply_header.handle_count = 0;
+
+  result = nacl::ReceiveDatagram(o3d_handle_, &reply_header, 0);
+  EXPECT_EQ(shared_memory_vec.length, result);
+  EXPECT_EQ(0, reply_header.flags & nacl::kMessageTruncated);
+  EXPECT_EQ(0, reply_header.handle_count);
+  EXPECT_EQ(1, reply_header.iov_length);
+
+  return shared_memory_id;
+}
+
+// Unregisters a previously-registered client-allocated shared
+// memory segment.
+bool TextureUpdateHelper::UnregisterSharedMemory(int shared_memory_id) {
+  MessageQueue::MessageId message_id = MessageQueue::UNREGISTER_SHARED_MEMORY;
+  nacl::MessageHeader header;
+  nacl::IOVec vec;
+  char buffer[256];
+  char *buffer_ptr = &buffer[0];
+
+  // Message contains the message ID and the ID of the shared memory
+  // segment to release
+  *(reinterpret_cast<MessageQueue::MessageId*>(buffer_ptr)) = message_id;
+  buffer_ptr += sizeof(message_id);
+  *(reinterpret_cast<int*>(buffer_ptr)) = shared_memory_id;
+  buffer_ptr += sizeof(shared_memory_id);
+
+  vec.base = buffer;
+  vec.length = buffer_ptr - buffer;
+  header.iov = &vec;
+  header.iov_length = 1;
+  header.handles = NULL;
+  header.handle_count = 0;
+
+  // Send message.
+  int result = nacl::SendDatagram(o3d_handle_, &header, 0);
+  EXPECT_EQ(static_cast<int>(vec.length), result);
+  // Read back the boolean reply from the O3D plugin
+  bool reply = ReceiveBooleanResponse();
+  EXPECT_TRUE(reply);
+  return reply;
+}
 
+//----------------------------------------------------------------------
+// Test cases follow.
 
 // Tests that the message queue socket is properly initialized.
 TEST_F(MessageQueueTest, Initialize) {
@@ -337,8 +693,8 @@ TEST_F(MessageQueueTest, Initialize) {
 
   String socket_addr = message_queue->GetSocketAddress();
 
-  // Make sure the name starts with "google-o3d"
-  EXPECT_EQ(0, socket_addr.find("google-o3d"));
+  // Make sure the name starts with the expected value
+  EXPECT_EQ(0, socket_addr.find("o3d"));
 
   delete message_queue;
 }
@@ -346,85 +702,227 @@ TEST_F(MessageQueueTest, Initialize) {
 // Tests that the a client can actually establish a connection to the
 // MessageQueue.
 TEST_F(MessageQueueTest, TestConnection) {
-  MessageQueue* message_queue = new MessageQueue(g_service_locator);
-  message_queue->Initialize();
-
-  String socket_addr = message_queue->GetSocketAddress();
-  std::string socket_addr_utf8;
-  StringToUTF8(socket_addr, &socket_addr_utf8);
-
-  TextureUpdateHelper helper;
-  EXPECT_TRUE(helper.ConnectToO3D(socket_addr_utf8.c_str(),
-                                  my_socket_handle()));
-
-  delete message_queue;
+  class ConnectionTest : public PerThreadConnectedTest {
+   public:
+    void Run(MessageQueue* queue,
+             nacl::Handle socket_handle) {
+      String socket_addr = queue->GetSocketAddress();
+      TextureUpdateHelper helper;
+      if (helper.ConnectToO3D(socket_addr.c_str(),
+                              socket_handle)) {
+        Pass();
+      } else {
+        FAIL_TEST("Failed to connect to O3D");
+      }
+    }
+  };
+
+  class Provider : public TestProvider {
+   public:
+    virtual PerThreadConnectedTest* CreateTest() {
+      return new ConnectionTest();
+    }
+  };
+
+  RunTests(1, TimeDelta::FromSeconds(1),
+           scoped_ptr<Provider>(new Provider()).get());
 }
 
 // Tests a request for shared memory.
 TEST_F(MessageQueueTest, GetSharedMemory) {
-  MessageQueue* message_queue = new MessageQueue(g_service_locator);
-  message_queue->Initialize();
-
-  String socket_addr = message_queue->GetSocketAddress();
-  std::string socket_addr_utf8;
-  StringToUTF8(socket_addr, &socket_addr_utf8);
-
-  TextureUpdateHelper helper;
-  ASSERT_TRUE(helper.ConnectToO3D(socket_addr_utf8.c_str(),
-                                  my_socket_handle()));
-
-  void *shared_mem_address = NULL;
-  int shared_mem_id = -1;
-  bool memory_ok = helper.RequestSharedMemory(65536,
-                                              &shared_mem_id,
-                                              &shared_mem_address);
-  EXPECT_NE(-1, shared_mem_id);
-  EXPECT_TRUE(shared_mem_address != NULL);
-
-  EXPECT_TRUE(memory_ok);
-
-  delete message_queue;
+  class SharedMemoryTest : public PerThreadConnectedTest {
+   public:
+    void Run(MessageQueue* queue,
+             nacl::Handle socket_handle) {
+      String socket_addr = queue->GetSocketAddress();
+      TextureUpdateHelper helper;
+      if (!helper.ConnectToO3D(socket_addr.c_str(),
+                               socket_handle)) {
+        FAIL_TEST("Failed to connect to O3D");
+      }
+
+      void *shared_mem_address = NULL;
+      int shared_mem_id = -1;
+      bool memory_ok = helper.RequestSharedMemory(65536,
+                                                  &shared_mem_id,
+                                                  &shared_mem_address);
+      if (shared_mem_id == -1) {
+        FAIL_TEST("Shared memory id was -1");
+      }
+
+      if (shared_mem_address == NULL) {
+        FAIL_TEST("Shared memory address was NULL");
+      }
+
+      if (!memory_ok) {
+        FAIL_TEST("Memory request failed");
+      }
+
+      Pass();
+    }
+  };
+
+  class Provider : public TestProvider {
+   public:
+    virtual PerThreadConnectedTest* CreateTest() {
+      return new SharedMemoryTest();
+    }
+  };
+
+  RunTests(1, TimeDelta::FromSeconds(1),
+           scoped_ptr<Provider>(new Provider()).get());
 }
 
 // Tests a request to update a texture.
 TEST_F(MessageQueueTest, UpdateTexture2D) {
-  MessageQueue* message_queue = new MessageQueue(g_service_locator);
-  message_queue->Initialize();
-
-  String socket_addr = message_queue->GetSocketAddress();
-  std::string socket_addr_utf8;
-  StringToUTF8(socket_addr, &socket_addr_utf8);
-
-  TextureUpdateHelper helper;
-  ASSERT_TRUE(helper.ConnectToO3D(socket_addr_utf8.c_str(),
-                                  my_socket_handle()));
-
-  void *shared_mem_address = NULL;
-  int shared_mem_id = -1;
-  bool memory_ok = helper.RequestSharedMemory(65536,
-                                              &shared_mem_id,
-                                              &shared_mem_address);
-
-  ASSERT_TRUE(memory_ok);
-
-  Texture2D* texture = pack()->CreateTexture2D("test_texture",
-                                               128,
+  class UpdateTexture2DTest : public PerThreadConnectedTest {
+   private:
+    int texture_id_;
+
+   public:
+    explicit UpdateTexture2DTest(int texture_id) {
+      texture_id_ = texture_id;
+    }
+
+    void Run(MessageQueue* queue,
+             nacl::Handle socket_handle) {
+      String socket_addr = queue->GetSocketAddress();
+      TextureUpdateHelper helper;
+      if (!helper.ConnectToO3D(socket_addr.c_str(),
+                               socket_handle)) {
+        FAIL_TEST("Failed to connect to O3D");
+      }
+
+      void *shared_mem_address = NULL;
+      int shared_mem_id = -1;
+      bool memory_ok = helper.RequestSharedMemory(65536,
+                                                  &shared_mem_id,
+                                                  &shared_mem_address);
+      if (shared_mem_id == -1) {
+        FAIL_TEST("Shared memory id was -1");
+      }
+
+      if (shared_mem_address == NULL) {
+        FAIL_TEST("Shared memory address was NULL");
+      }
+
+      if (!memory_ok) {
+        FAIL_TEST("Memory request failed");
+      }
+
+      int texture_buffer_size = 128*128*4;
+
+      if (!helper.RequestTextureUpdate(texture_id_,
+                                       0,
+                                       shared_mem_id,
+                                       0,
+                                       texture_buffer_size)) {
+        FAIL_TEST("RequestTextureUpdate failed");
+      }
+
+      Pass();
+    }
+  };
+
+  class Provider : public TestProvider {
+   private:
+    int texture_id_;
+
+   public:
+    explicit Provider(int texture_id) : texture_id_(texture_id) {}
+
+    virtual PerThreadConnectedTest* CreateTest() {
+      return new UpdateTexture2DTest(texture_id_);
+    }
+  };
+
+  Texture2D* texture = pack()->CreateTexture2D(128,
                                                128,
                                                Texture::ARGB8,
-                                               0);
+                                               0,
+                                               false);
 
   ASSERT_TRUE(texture != NULL);
 
-  int texture_buffer_size = 128*128*4;
+  RunTests(1, TimeDelta::FromSeconds(1),
+           scoped_ptr<Provider>(new Provider(texture->id())).get());
+}
 
-  EXPECT_TRUE(helper.RequestTextureUpdate(texture->id(),
-                                          0,
-                                          shared_mem_id,
-                                          0,
-                                          texture_buffer_size));
+// This helper class is used for both single-threaded and concurrent
+// shared memory registration / unregistration tests.
+class SharedMemoryRegisterUnregisterTest : public PerThreadConnectedTest {
+ private:
+  int num_iterations_;
 
-  delete message_queue;
+ public:
+  explicit SharedMemoryRegisterUnregisterTest(int num_iterations) {
+    num_iterations_ = num_iterations;
+  }
+
+  void Run(MessageQueue* queue,
+           nacl::Handle socket_handle) {
+    String socket_addr = queue->GetSocketAddress();
+    TextureUpdateHelper helper;
+    if (!helper.ConnectToO3D(socket_addr.c_str(),
+                             socket_handle)) {
+      FAIL_TEST("Failed to connect to O3D");
+    }
+
+    // Allocate a shared memory segment
+    size_t mem_size = nacl::kMapPageSize;
+    nacl::Handle shared_memory = nacl::CreateMemoryObject(mem_size);
+    if (shared_memory == nacl::kInvalidHandle) {
+      FAIL_TEST("Failed to allocate shared memory object");
+    }
+
+    // Note that we don't actually have to map it in our process in
+    // order to test the failure mode (corrupted messages) this test
+    // exercises
+
+    for (int i = 0; i < num_iterations_; i++) {
+      int shared_mem_id =
+        helper.RegisterSharedMemory(shared_memory, mem_size);
+      if (shared_mem_id < 0) {
+        FAIL_TEST("Failed to register shared memory with server");
+      }
+      bool result = helper.UnregisterSharedMemory(shared_mem_id);
+      if (!result) {
+        FAIL_TEST("Failed to unregister shared memory from server");
+      }
+    }
+
+    nacl::Close(shared_memory);
+
+    Pass();
+  }
+};
+
+// Tests that a simple shared memory registration and unregistration
+// pair appear to work.
+TEST_F(MessageQueueTest, RegisterAndUnregisterSharedMemory) {
+  class Provider : public TestProvider {
+   public:
+    virtual PerThreadConnectedTest* CreateTest() {
+      return new SharedMemoryRegisterUnregisterTest(1);
+    }
+  };
+
+  RunTests(1, TimeDelta::FromSeconds(1),
+           scoped_ptr<Provider>(new Provider()).get());
 }
 
+// Tests that multiple concurrent clients of the MessageQueue don't
+// break its deserialization operations.
+TEST_F(MessageQueueTest, ConcurrentSharedMemoryOperations) {
+  class Provider : public TestProvider {
+   public:
+    virtual PerThreadConnectedTest* CreateTest() {
+      return new SharedMemoryRegisterUnregisterTest(100);
+    }
+  };
+
+  RunTests(2, TimeDelta::FromSeconds(6),
+           scoped_ptr<Provider>(new Provider()).get());
+}
 
 }  // namespace o3d
diff --git a/o3d/tests/build.scons b/o3d/tests/build.scons
index 67fdd1d..b622edb 100644
--- a/o3d/tests/build.scons
+++ b/o3d/tests/build.scons
@@ -191,8 +191,7 @@ tests = [
     'core/cross/matrix4_composition_test.cc',
     'core/cross/matrix4_scale_test.cc',
     'core/cross/matrix4_translation_test.cc',
-# Test disabled, pending investigation into NaCl API.
-#   'core/cross/message_queue_test.cc',
+    'core/cross/message_queue_test.cc',
     'core/cross/object_base_test.cc',
     'core/cross/pack_test.cc',
     'core/cross/param_array_test.cc',