IPC to pass media data using a lock free circular fifo.

chromecast/media/cma/ipc/media_message_fifo_unittest.cc

+// 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 "base/basictypes.h"
+#include "base/bind.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "chromecast/media/cma/ipc/media_memory_chunk.h"
+#include "chromecast/media/cma/ipc/media_message.h"
+#include "chromecast/media/cma/ipc/media_message_fifo.h"
+#include "chromecast/media/cma/ipc/media_message_type.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace chromecast {
+namespace media {
+
+namespace {
+
+class FifoMemoryChunk : public MediaMemoryChunk {
+ public:
+  FifoMemoryChunk(void* mem, size_t size)
+      : mem_(mem), size_(size) {}
+  virtual ~FifoMemoryChunk() {}
+
+  virtual void* data() const OVERRIDE { return mem_; }
+  virtual size_t size() const OVERRIDE { return size_; }
+  virtual bool valid() const OVERRIDE { return true; }
+
+ private:
+  void* mem_;
+  size_t size_;
+
+  DISALLOW_COPY_AND_ASSIGN(FifoMemoryChunk);
+};
+
+void MsgProducer(scoped_ptr<MediaMessageFifo> fifo,
+                 int msg_count,
+                 base::WaitableEvent* event) {
+
+  for (int k = 0; k < msg_count; k++) {
+    uint32 msg_type = 0x2 + (k % 5);
+    uint32 max_msg_content_size = k % 64;
+    do {
+      scoped_ptr<MediaMessage> msg1(
+          MediaMessage::CreateMessage(
+              msg_type,
+              base::Bind(&MediaMessageFifo::ReserveMemory,
+                         base::Unretained(fifo.get())),
+              max_msg_content_size));
+      if (msg1)
+        break;
+      base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(10));
+    } while(true);
+  }
+
+  fifo.reset();
+
+  event->Signal();
+}
+
+void MsgConsumer(scoped_ptr<MediaMessageFifo> fifo,
+                 int msg_count,
+                 base::WaitableEvent* event) {
+
+  int k = 0;
+  while (k < msg_count) {
+    uint32 msg_type = 0x2 + (k % 5);
+    do {
+      scoped_ptr<MediaMessage> msg2(fifo->Pop());
+      if (msg2) {
+        if (msg2->type() != PaddingMediaMsg) {
+          EXPECT_EQ(msg2->type(), msg_type);
+          k++;
+        }
+        break;
+      }
+      base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(10));
+    } while(true);
+  }
+
+  fifo.reset();
+
+  event->Signal();
+}
+
+void MsgProducerConsumer(
+    scoped_ptr<MediaMessageFifo> producer_fifo,
+    scoped_ptr<MediaMessageFifo> consumer_fifo,
+    base::WaitableEvent* event) {
+  for (int k = 0; k < 2048; k++) {
+    // Should have enough space to create a message.
+    uint32 msg_type = 0x2 + (k % 5);
+    uint32 max_msg_content_size = k % 64;
+    scoped_ptr<MediaMessage> msg1(
+        MediaMessage::CreateMessage(
+            msg_type,
+            base::Bind(&MediaMessageFifo::ReserveMemory,
+                       base::Unretained(producer_fifo.get())),
+            max_msg_content_size));
+    EXPECT_TRUE(msg1);
+
+    // Make sure the message is commited.
+    msg1.reset();
+
+    // At this point, we should have a message to read.
+    scoped_ptr<MediaMessage> msg2(consumer_fifo->Pop());
+    EXPECT_TRUE(msg2);
+  }
+
+  producer_fifo.reset();
+  consumer_fifo.reset();
+
+  event->Signal();
+}
+
+}  // namespace
+
+TEST(MediaMessageFifoTest, AlternateWriteRead) {
+  size_t buffer_size = 64 * 1024;
+  scoped_ptr<uint64[]> buffer(new uint64[buffer_size / sizeof(uint64)]);
+
+  scoped_ptr<base::Thread> thread(
+      new base::Thread("FeederConsumerThread"));
+  thread->Start();
+
+  scoped_ptr<MediaMessageFifo> producer_fifo(new MediaMessageFifo(
+      scoped_ptr<MediaMemoryChunk>(
+          new FifoMemoryChunk(&buffer[0], buffer_size)),
+      true));
+  scoped_ptr<MediaMessageFifo> consumer_fifo(new MediaMessageFifo(
+      scoped_ptr<MediaMemoryChunk>(
+          new FifoMemoryChunk(&buffer[0], buffer_size)),
+      false));
+
+  base::WaitableEvent event(false, false);
+  thread->message_loop_proxy()->PostTask(
+      FROM_HERE,
+      base::Bind(&MsgProducerConsumer,
+                 base::Passed(&producer_fifo),
+                 base::Passed(&consumer_fifo),
+                 &event));
+  event.Wait();
+
+  thread.reset();
+}
+
+TEST(MediaMessageFifoTest, MultiThreaded) {
+  size_t buffer_size = 64 * 1024;
+  scoped_ptr<uint64[]> buffer(new uint64[buffer_size / sizeof(uint64)]);
+
+  scoped_ptr<base::Thread> producer_thread(
+      new base::Thread("FeederThread"));
+  scoped_ptr<base::Thread> consumer_thread(
+      new base::Thread("ConsumerThread"));
+  producer_thread->Start();
+  consumer_thread->Start();
+
+  scoped_ptr<MediaMessageFifo> producer_fifo(new MediaMessageFifo(
+      scoped_ptr<MediaMemoryChunk>(
+          new FifoMemoryChunk(&buffer[0], buffer_size)),
+      true));
+  scoped_ptr<MediaMessageFifo> consumer_fifo(new MediaMessageFifo(
+      scoped_ptr<MediaMemoryChunk>(
+          new FifoMemoryChunk(&buffer[0], buffer_size)),
+      false));
+
+  base::WaitableEvent producer_event_done(false, false);
+  base::WaitableEvent consumer_event_done(false, false);
+
+  const int msg_count = 2048;
+  producer_thread->message_loop_proxy()->PostTask(
+      FROM_HERE,
+      base::Bind(&MsgProducer,
+                 base::Passed(&producer_fifo),
+                 msg_count,
+                 &producer_event_done));
+  consumer_thread->message_loop_proxy()->PostTask(
+      FROM_HERE,
+      base::Bind(&MsgConsumer,
+                 base::Passed(&consumer_fifo),
+                 msg_count,
+                 &consumer_event_done));
+
+  producer_event_done.Wait();
+  consumer_event_done.Wait();
+
+  producer_thread.reset();
+  consumer_thread.reset();
+}
+
+}  // namespace media
+}  // namespace chromecast
+