Make SharedMemoryDataConsumerHandle thread-safe.

content/child/shared_memory_data_consumer_handle_unittest.cc

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "content/child/shared_memory_data_consumer_handle.h"
 
 #include <string.h>
 #include <string>
 #include <vector>
 
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/threading/thread.h"
 #include "content/public/child/fixed_received_data.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace content {
 
 namespace {
 using blink::WebDataConsumerHandle;
 using Result = WebDataConsumerHandle::Result;
 using Writer = SharedMemoryDataConsumerHandle::Writer;
@@ skipping 58 matching lines @@
 class MockClient : public WebDataConsumerHandle::Client {
  public:
   MOCK_METHOD0(didGetReadable, void());
 };
 
 std::string ToString(const void* p, size_t size) {
   const char* q = static_cast<const char*>(p);
   return std::string(q, q + size);
 }
 
+class ThreadedSharedMemoryDataConsumerHandleTest : public ::testing::Test {
+ protected:
+  class ReadDataOperation;
+  class ClientImpl final : public WebDataConsumerHandle::Client {
+   public:
+    explicit ClientImpl(ReadDataOperation* operation) : operation_(operation) {}
+
+    void didGetReadable() override { operation_->ReadData(); }
+
+   private:
+    ReadDataOperation* operation_;
+  };
+
+  class ReadDataOperation final {
+   public:
+    typedef WebDataConsumerHandle::Result Result;
+    ReadDataOperation(scoped_ptr<WebDataConsumerHandle> handle,
+                      base::MessageLoop* main_message_loop,
+                      const base::Closure& on_done)
+        : handle_(handle.Pass()),
+          main_message_loop_(main_message_loop),
+          on_done_(on_done) {}
+
+    const std::string& result() const { return result_; }
+
+    void ReadData() {
+      if (!client_) {
+        client_.reset(new ClientImpl(this));
+        handle_->registerClient(client_.get());
+      }
+
+      Result rv = kOk;
+      size_t read_size = 0;
+
+      while (true) {
+        char buffer[16];
+        rv = handle_->read(&buffer, sizeof(buffer), kNone, &read_size);
+        if (rv != kOk)
+          break;
+        result_.insert(result_.size(), &buffer[0], read_size);
+      }
+
+      if (rv == kShouldWait) {
+        // Wait a while...
+        return;
+      }
+
+      if (rv != kDone) {
+        // Something is wrong.
+        result_ = "error";
+      }
+
+      // The operation is done.
+      main_message_loop_->PostTask(FROM_HERE, on_done_);
+    }
+
+   private:
+    scoped_ptr<WebDataConsumerHandle> handle_;
+    scoped_ptr<WebDataConsumerHandle::Client> client_;
+    base::MessageLoop* main_message_loop_;
+    base::Closure on_done_;
+    std::string result_;
+  };
+
+  void SetUp() override {
+    handle_.reset(
+        new SharedMemoryDataConsumerHandle(kApplyBackpressure, &writer_));
+  }
+
+  StrictMock<MockClient> client_;
+  scoped_ptr<WebDataConsumerHandle> handle_;
+  scoped_ptr<Writer> writer_;
+  base::MessageLoop loop_;
+};
+
 class SharedMemoryDataConsumerHandleTest
     : public ::testing::TestWithParam<BackpressureMode> {
  protected:
   void SetUp() override {
     handle_.reset(new SharedMemoryDataConsumerHandle(GetParam(), &writer_));
   }
   scoped_ptr<FixedReceivedData> NewFixedData(const char* s) {
     return make_scoped_ptr(new FixedReceivedData(s, strlen(s), strlen(s)));
   }
 
   StrictMock<MockClient> client_;
   scoped_ptr<SharedMemoryDataConsumerHandle> handle_;
   scoped_ptr<Writer> writer_;
+  base::MessageLoop loop_;
 };
 
 TEST_P(SharedMemoryDataConsumerHandleTest, ReadFromEmpty) {
   char buffer[4];
   size_t read = 88;
   Result result = handle_->read(buffer, 4, kNone, &read);
 
   EXPECT_EQ(result, kShouldWait);
   EXPECT_EQ(0u, read);
 }
@@ skipping 338 matching lines @@
 
   writer_->Close();
 
   result = handle_->beginRead(&buffer, kNone, &size);
   EXPECT_EQ(kDone, result);
   EXPECT_EQ(0u, size);
   EXPECT_EQ(nullptr, buffer);
 }
 
 TEST(SharedMemoryDataConsumerHandleBackpressureTest, Read) {
+  base::MessageLoop loop;
   char buffer[20];
   Result result;
   size_t size;
 
   scoped_ptr<Writer> writer;
   auto handle = make_scoped_ptr(
       new SharedMemoryDataConsumerHandle(kApplyBackpressure, &writer));
   scoped_refptr<Logger> logger(new Logger);
   writer->AddData(
       make_scoped_ptr(new LoggingFixedReceivedData("data1", "Once ", logger)));
@@ skipping 23 matching lines @@
       "2\n"
       "data1 is destructed.\n"
       "3\n"
       "data2 is destructed.\n"
       "data3 is destructed.\n"
       "4\n",
       logger->log());
 }
 
 TEST(SharedMemoryDataConsumerHandleBackpressureTest, CloseAndReset) {
+  base::MessageLoop loop;
   char buffer[20];
   Result result;
   size_t size;
 
   scoped_ptr<Writer> writer;
   auto handle = make_scoped_ptr(
       new SharedMemoryDataConsumerHandle(kApplyBackpressure, &writer));
   scoped_refptr<Logger> logger(new Logger);
   writer->AddData(
       make_scoped_ptr(new LoggingFixedReceivedData("data1", "Once ", logger)));
@@ skipping 17 matching lines @@
       "2\n"
       "3\n"
       "data1 is destructed.\n"
       "data2 is destructed.\n"
       "data3 is destructed.\n"
       "4\n",
       logger->log());
 }
 
 TEST(SharedMemoryDataConsumerHandleWithoutBackpressureTest, AddData) {
+  base::MessageLoop loop;
   scoped_ptr<Writer> writer;
   auto handle = make_scoped_ptr(
       new SharedMemoryDataConsumerHandle(kDoNotApplyBackpressure, &writer));
   scoped_refptr<Logger> logger(new Logger);
 
   logger->Add("1");
   writer->AddData(
       make_scoped_ptr(new LoggingFixedReceivedData("data1", "Once ", logger)));
   logger->Add("2");
   writer->AddData(
       make_scoped_ptr(new LoggingFixedReceivedData("data2", "upon ", logger)));
   logger->Add("3");
 
   EXPECT_EQ(
       "1\n"
       "data1 is destructed.\n"
       "2\n"
       "data2 is destructed.\n"
       "3\n",
       logger->log());
 }
 
+TEST_F(ThreadedSharedMemoryDataConsumerHandleTest, Read) {
+  base::RunLoop run_loop;
+  auto operation = make_scoped_ptr(
+      new ReadDataOperation(handle_.Pass(), &loop_, run_loop.QuitClosure()));
+  scoped_refptr<Logger> logger(new Logger);
+
+  base::Thread t("DataConsumerHandle test thread");
+  ASSERT_TRUE(t.Start());
+
+  t.message_loop()->PostTask(FROM_HERE,
+                             base::Bind(&ReadDataOperation::ReadData,
+                                        base::Unretained(operation.get())));
+
+  logger->Add("1");
+  writer_->AddData(
+      make_scoped_ptr(new LoggingFixedReceivedData("data1", "Once ", logger)));
+  writer_->AddData(
+      make_scoped_ptr(new LoggingFixedReceivedData("data2", "upon ", logger)));
+  writer_->AddData(make_scoped_ptr(
+      new LoggingFixedReceivedData("data3", "a time ", logger)));
+  writer_->AddData(
+      make_scoped_ptr(new LoggingFixedReceivedData("data4", "there ", logger)));
+  writer_->AddData(
+      make_scoped_ptr(new LoggingFixedReceivedData("data5", "was ", logger)));
+  writer_->Close();
+  logger->Add("2");
+
+  run_loop.Run();
+  t.Stop();
+
+  EXPECT_EQ("Once upon a time there was ", operation->result());
+  EXPECT_EQ(
+      "1\n"
+      "2\n"
+      "data1 is destructed.\n"
+      "data2 is destructed.\n"
+      "data3 is destructed.\n"
+      "data4 is destructed.\n"
+      "data5 is destructed.\n",
+      logger->log());
+}
+
 INSTANTIATE_TEST_CASE_P(SharedMemoryDataConsumerHandleTest,
                         SharedMemoryDataConsumerHandleTest,
                         ::testing::Values(kApplyBackpressure,
                                           kDoNotApplyBackpressure));
 }  // namespace
 
 }  // namespace content