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

BUG=408189

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

Cr-Commit-Position: refs/heads/master@{#293451}

10 files changed, 1399 insertions, 0 deletions

diff --git a/chromecast/media/cma/ipc/media_memory_chunk.cc b/chromecast/media/cma/ipc/media_memory_chunk.cc
new file mode 100644
index 0000000..0d68969
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_memory_chunk.cc
@@ -0,0 +1,14 @@
+// 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 "chromecast/media/cma/ipc/media_memory_chunk.h"
+
+namespace chromecast {
+namespace media {
+
+MediaMemoryChunk::~MediaMemoryChunk() {
+}
+
+}  // namespace media
+}  // namespace chromecast
diff --git a/chromecast/media/cma/ipc/media_memory_chunk.h b/chromecast/media/cma/ipc/media_memory_chunk.h
new file mode 100644
index 0000000..1b91c84
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_memory_chunk.h
@@ -0,0 +1,39 @@
+// 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 CHROMECAST_MEDIA_CMA_IPC_MEDIA_MEMORY_CHUNK_H_
+#define CHROMECAST_MEDIA_CMA_IPC_MEDIA_MEMORY_CHUNK_H_
+
+#include "base/basictypes.h"
+
+namespace chromecast {
+namespace media {
+
+// MediaMemoryChunk represents a block of memory without doing any assumption
+// about the type of memory (e.g. shared memory) nor about the underlying
+// memory ownership.
+// The block of memory can be invalidated under the cover (e.g. if the derived
+// class does not own the underlying memory),
+// in that case, MediaMemoryChunk::valid() will return false.
+class MediaMemoryChunk {
+ public:
+  virtual ~MediaMemoryChunk();
+
+  // Returns the start of the block of memory.
+  virtual void* data() const = 0;
+
+  // Returns the size of the block of memory.
+  virtual size_t size() const = 0;
+
+  // Returns whether the underlying block of memory is valid.
+  // Since MediaMemoryChunk does not specify a memory ownership model,
+  // the underlying block of memory might be invalidated by a third party.
+  // In that case, valid() will return false.
+  virtual bool valid() const = 0;
+};
+
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_CMA_IPC_MEDIA_MEMORY_CHUNK_H_
diff --git a/chromecast/media/cma/ipc/media_message.cc b/chromecast/media/cma/ipc/media_message.cc
new file mode 100644
index 0000000..8dfcd7c
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message.cc
@@ -0,0 +1,198 @@
+// 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 "chromecast/media/cma/ipc/media_message.h"
+
+#include <limits>
+
+#include "base/basictypes.h"
+#include "base/logging.h"
+#include "chromecast/media/cma/ipc/media_memory_chunk.h"
+
+namespace chromecast {
+namespace media {
+
+// static
+scoped_ptr<MediaMessage> MediaMessage::CreateDummyMessage(
+    uint32 type) {
+  return scoped_ptr<MediaMessage>(
+      new MediaMessage(type, std::numeric_limits<size_t>::max()));
+}
+
+// static
+scoped_ptr<MediaMessage> MediaMessage::CreateMessage(
+    uint32 type,
+    const MemoryAllocatorCB& memory_allocator,
+    size_t msg_content_capacity) {
+  size_t msg_size = minimum_msg_size() + msg_content_capacity;
+
+  // Make the message size a multiple of the alignment
+  // so that if we have proper alignment for array of messages.
+  size_t end_alignment = msg_size % ALIGNOF(SerializedMsg);
+  if (end_alignment != 0)
+    msg_size += ALIGNOF(SerializedMsg) - end_alignment;
+
+  scoped_ptr<MediaMemoryChunk> memory(memory_allocator.Run(msg_size));
+  if (!memory)
+    return scoped_ptr<MediaMessage>();
+
+  return scoped_ptr<MediaMessage>(new MediaMessage(type, memory.Pass()));
+}
+
+// static
+scoped_ptr<MediaMessage> MediaMessage::CreateMessage(
+    uint32 type,
+    scoped_ptr<MediaMemoryChunk> memory) {
+  return scoped_ptr<MediaMessage>(new MediaMessage(type, memory.Pass()));
+}
+
+// static
+scoped_ptr<MediaMessage> MediaMessage::MapMessage(
+    scoped_ptr<MediaMemoryChunk> memory) {
+  return scoped_ptr<MediaMessage>(new MediaMessage(memory.Pass()));
+}
+
+MediaMessage::MediaMessage(uint32 type, size_t msg_size)
+  : is_dummy_msg_(true),
+    cached_header_(&cached_msg_.header),
+    msg_(&cached_msg_),
+    msg_read_only_(&cached_msg_),
+    rd_offset_(0) {
+  cached_header_->size = msg_size;
+  cached_header_->type = type;
+  cached_header_->content_size = 0;
+}
+
+MediaMessage::MediaMessage(uint32 type, scoped_ptr<MediaMemoryChunk> memory)
+  : is_dummy_msg_(false),
+    cached_header_(&cached_msg_.header),
+    msg_(static_cast<SerializedMsg*>(memory->data())),
+    msg_read_only_(msg_),
+    mem_(memory.Pass()),
+    rd_offset_(0) {
+  CHECK(mem_->valid());
+  CHECK_GE(mem_->size(), minimum_msg_size());
+
+  // Check memory alignment:
+  // needed to cast properly |msg_dst| to a SerializedMsg.
+  CHECK_EQ(
+      reinterpret_cast<uintptr_t>(mem_->data()) % ALIGNOF(SerializedMsg), 0u);
+
+  // Make sure that |mem_->data()| + |mem_->size()| is also aligned correctly.
+  // This is needed if we append a second serialized message next to this one.
+  // The second serialized message must be aligned correctly.
+  // It is similar to what a compiler is doing for arrays of structures.
+  CHECK_EQ(mem_->size() % ALIGNOF(SerializedMsg), 0u);
+
+  cached_header_->size = mem_->size();
+  cached_header_->type = type;
+  cached_header_->content_size = 0;
+  msg_->header = *cached_header_;
+}
+
+MediaMessage::MediaMessage(scoped_ptr<MediaMemoryChunk> memory)
+  : is_dummy_msg_(false),
+    cached_header_(&cached_msg_.header),
+    msg_(NULL),
+    msg_read_only_(static_cast<SerializedMsg*>(memory->data())),
+    mem_(memory.Pass()),
+    rd_offset_(0) {
+  CHECK(mem_->valid());
+
+  // Check memory alignment.
+  CHECK_EQ(
+      reinterpret_cast<uintptr_t>(mem_->data()) % ALIGNOF(SerializedMsg), 0u);
+
+  // Cache the message header which cannot be modified while reading.
+  CHECK_GE(mem_->size(), minimum_msg_size());
+  *cached_header_ = msg_read_only_->header;
+  CHECK_GE(cached_header_->size, minimum_msg_size());
+
+  // Make sure if we have 2 consecutive serialized messages in memory,
+  // the 2nd message is also aligned correctly.
+  CHECK_EQ(cached_header_->size % ALIGNOF(SerializedMsg), 0u);
+
+  size_t max_content_size = cached_header_->size - minimum_msg_size();
+  CHECK_LE(cached_header_->content_size, max_content_size);
+}
+
+MediaMessage::~MediaMessage() {
+}
+
+bool MediaMessage::IsSerializedMsgAvailable() const {
+  return !is_dummy_msg_ && mem_->valid();
+}
+
+bool MediaMessage::WriteBuffer(const void* src, size_t size) {
+  // No message to write into.
+  if (!msg_)
+    return false;
+
+  // The underlying memory was invalidated.
+  if (!is_dummy_msg_ && !mem_->valid())
+    return false;
+
+  size_t max_content_size = cached_header_->size - minimum_msg_size();
+  if (cached_header_->content_size + size > max_content_size) {
+    cached_header_->content_size = max_content_size;
+    msg_->header.content_size = cached_header_->content_size;
+    return false;
+  }
+
+  // Write the message only for non-dummy messages.
+  if (!is_dummy_msg_) {
+    uint8* wr_ptr = &msg_->content + cached_header_->content_size;
+    memcpy(wr_ptr, src, size);
+  }
+
+  cached_header_->content_size += size;
+  msg_->header.content_size = cached_header_->content_size;
+  return true;
+}
+
+bool MediaMessage::ReadBuffer(void* dst, size_t size) {
+  // No read possible for a dummy message.
+  if (is_dummy_msg_)
+    return false;
+
+  // The underlying memory was invalidated.
+  if (!mem_->valid())
+    return false;
+
+  if (rd_offset_ + size > cached_header_->content_size) {
+    rd_offset_ = cached_header_->content_size;
+    return false;
+  }
+
+  const uint8* rd_ptr = &msg_read_only_->content + rd_offset_;
+  memcpy(dst, rd_ptr, size);
+  rd_offset_ += size;
+  return true;
+}
+
+void* MediaMessage::GetWritableBuffer(size_t size) {
+  // No read possible for a dummy message.
+  if (is_dummy_msg_)
+    return NULL;
+
+  // The underlying memory was invalidated.
+  if (!mem_->valid())
+    return NULL;
+
+  if (rd_offset_ + size > cached_header_->content_size) {
+    rd_offset_ = cached_header_->content_size;
+    return NULL;
+  }
+
+  uint8* rd_ptr = &msg_read_only_->content + rd_offset_;
+  rd_offset_ += size;
+  return rd_ptr;
+}
+
+const void* MediaMessage::GetBuffer(size_t size) {
+  return GetWritableBuffer(size);
+}
+
+}  // namespace media
+}  // namespace chromecast
diff --git a/chromecast/media/cma/ipc/media_message.h b/chromecast/media/cma/ipc/media_message.h
new file mode 100644
index 0000000..066409b
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message.h
@@ -0,0 +1,165 @@
+// 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 CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_H_
+#define CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_H_
+
+#include <stddef.h>
+
+#include "base/basictypes.h"
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+
+namespace chromecast {
+namespace media {
+class MediaMemoryChunk;
+
+// MediaMessage -
+// Represents a media message, including:
+// - a message header that gives for example the message size or its type,
+// - the content of the message,
+// - and some possible padding if the content does not occupy the whole
+//   reserved space.
+//
+class MediaMessage {
+ public:
+  // Memory allocator: given a number of bytes to allocate,
+  // return the pointer to the allocated block if successful
+  // or NULL if allocation failed.
+  typedef base::Callback<scoped_ptr<MediaMemoryChunk>(size_t)>
+      MemoryAllocatorCB;
+
+  // Creates a message with no associated memory for its content, i.e.
+  // each write on this message is a dummy operation.
+  // This type of message can be useful to calculate first the size of the
+  // message, before allocating the real message.
+  static scoped_ptr<MediaMessage> CreateDummyMessage(uint32 type);
+
+  // Creates a message with a capacity of at least |msg_content_capacity|
+  // bytes. The actual content size can be smaller than its capacity.
+  // The message can be populated with some Write functions.
+  static scoped_ptr<MediaMessage> CreateMessage(
+      uint32 type,
+      const MemoryAllocatorCB& memory_allocator,
+      size_t msg_content_capacity);
+
+  // Creates a message of type |type| whose serialized structure is stored
+  // in |mem|.
+  static scoped_ptr<MediaMessage> CreateMessage(
+      uint32 type,
+      scoped_ptr<MediaMemoryChunk> mem);
+
+  // Creates a message from a memory area which already contains
+  // the serialized structure of the message.
+  // Only Read functions can be invoked on this type of message.
+  static scoped_ptr<MediaMessage> MapMessage(
+      scoped_ptr<MediaMemoryChunk> mem);
+
+  // Return the minimum size of a message.
+  static size_t minimum_msg_size() {
+    return offsetof(SerializedMsg, content);
+  }
+
+  ~MediaMessage();
+
+  // Indicate whether the underlying serialized structure of the message is
+  // available.
+  // Note: the serialized structure might be unavailable in case of a dummy
+  // message or if the underlying memory has been invalidated.
+  bool IsSerializedMsgAvailable() const;
+
+  // Return the message and the total size of the message
+  // incuding the header, the content and the possible padding.
+  const void* msg() const { return msg_read_only_; }
+  size_t size() const { return cached_msg_.header.size; }
+
+  // Return the size of the message without padding.
+  size_t actual_size() const {
+    return minimum_msg_size() + cached_msg_.header.content_size;
+  }
+
+  // Return the size of the content of the message.
+  size_t content_size() const { return cached_msg_.header.content_size; }
+
+  // Return the type of the message.
+  uint32 type() const { return cached_msg_.header.type; }
+
+  // Append a POD to the message.
+  // Return true if the POD has been succesfully written.
+  template<typename T> bool WritePod(T* const& pod);
+  template<typename T> bool WritePod(const T& pod) {
+    return WriteBuffer(&pod, sizeof(T));
+  }
+
+  // Append a raw buffer to the message.
+  bool WriteBuffer(const void* src, size_t size);
+
+  // Read a POD from the message.
+  template<typename T> bool ReadPod(T* pod) {
+    return ReadBuffer(pod, sizeof(T));
+  }
+
+  // Read |size| bytes from the message from the last read position
+  // and write it to |dst|.
+  bool ReadBuffer(void* dst, size_t size);
+
+  // Return a pointer to a buffer of size |size|.
+  // Return NULL if not successful.
+  const void* GetBuffer(size_t size);
+  void* GetWritableBuffer(size_t size);
+
+ private:
+  MediaMessage(uint32 type, size_t msg_size);
+  MediaMessage(uint32 type, scoped_ptr<MediaMemoryChunk> memory);
+  MediaMessage(scoped_ptr<MediaMemoryChunk> memory);
+
+  struct Header {
+    // Total size of the message (including both header & content).
+    uint32 size;
+    // Indicate the message type.
+    uint32 type;
+    // Actual size of the content in the message.
+    uint32 content_size;
+  };
+
+  struct SerializedMsg {
+    // Message header.
+    Header header;
+
+    // Start of the content of the message.
+    // Use uint8_t since no special alignment is needed.
+    uint8 content;
+  };
+
+  // Indicate whether the message is a dummy message, i.e. a message without
+  // a complete underlying serialized structure: only the message header is
+  // available.
+  bool is_dummy_msg_;
+
+  // |cached_msg_| is used for 2 purposes:
+  // - to create a dummy message
+  // - for security purpose: cache the msg header to avoid browser security
+  // issues.
+  SerializedMsg cached_msg_;
+  Header* const cached_header_;
+
+  SerializedMsg* msg_;
+  SerializedMsg* msg_read_only_;
+
+  // Memory allocated to store the underlying serialized structure into memory.
+  // Note: a dummy message has no underlying serialized structure:
+  // |mem_| is a null pointer in that case.
+  scoped_ptr<MediaMemoryChunk> mem_;
+
+  // Read iterator into the message.
+  size_t rd_offset_;
+
+  DISALLOW_COPY_AND_ASSIGN(MediaMessage);
+};
+
+}  // namespace media
+}  // namespace chromecast
+
+#endif
diff --git a/chromecast/media/cma/ipc/media_message_fifo.cc b/chromecast/media/cma/ipc/media_message_fifo.cc
new file mode 100644
index 0000000..413afd2
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message_fifo.cc
@@ -0,0 +1,401 @@
+// 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 "chromecast/media/cma/ipc/media_message_fifo.h"
+
+#include "base/atomicops.h"
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "chromecast/media/cma/base/cma_logging.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_type.h"
+
+namespace chromecast {
+namespace media {
+
+class MediaMessageFlag
+    : public base::RefCountedThreadSafe<MediaMessageFlag> {
+ public:
+  // |offset| is the offset in the fifo of the media message.
+  explicit MediaMessageFlag(size_t offset);
+
+  bool IsValid() const;
+
+  void Invalidate();
+
+  size_t offset() const { return offset_; }
+
+ private:
+  friend class base::RefCountedThreadSafe<MediaMessageFlag>;
+  virtual ~MediaMessageFlag();
+
+  const size_t offset_;
+  bool flag_;
+
+  DISALLOW_COPY_AND_ASSIGN(MediaMessageFlag);
+};
+
+MediaMessageFlag::MediaMessageFlag(size_t offset)
+  : offset_(offset),
+    flag_(true) {
+}
+
+MediaMessageFlag::~MediaMessageFlag() {
+}
+
+bool MediaMessageFlag::IsValid() const {
+  return flag_;
+}
+
+void MediaMessageFlag::Invalidate() {
+  flag_ = false;
+}
+
+class FifoOwnedMemory : public MediaMemoryChunk {
+ public:
+  FifoOwnedMemory(void* data, size_t size,
+                  const scoped_refptr<MediaMessageFlag>& flag,
+                  const base::Closure& release_msg_cb);
+  virtual ~FifoOwnedMemory();
+
+  // MediaMemoryChunk implementation.
+  virtual void* data() const OVERRIDE { return data_; }
+  virtual size_t size() const OVERRIDE { return size_; }
+  virtual bool valid() const OVERRIDE { return flag_->IsValid(); }
+
+ private:
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  base::Closure release_msg_cb_;
+
+  void* const data_;
+  const size_t size_;
+  scoped_refptr<MediaMessageFlag> flag_;
+
+  DISALLOW_COPY_AND_ASSIGN(FifoOwnedMemory);
+};
+
+FifoOwnedMemory::FifoOwnedMemory(
+    void* data, size_t size,
+    const scoped_refptr<MediaMessageFlag>& flag,
+    const base::Closure& release_msg_cb)
+  : task_runner_(base::MessageLoopProxy::current()),
+    release_msg_cb_(release_msg_cb),
+    data_(data),
+    size_(size),
+    flag_(flag) {
+}
+
+FifoOwnedMemory::~FifoOwnedMemory() {
+  // Release the flag before notifying that the message has been released.
+  flag_ = scoped_refptr<MediaMessageFlag>();
+  if (!release_msg_cb_.is_null()) {
+    if (task_runner_->BelongsToCurrentThread()) {
+      release_msg_cb_.Run();
+    } else {
+      task_runner_->PostTask(FROM_HERE, release_msg_cb_);
+    }
+  }
+}
+
+MediaMessageFifo::MediaMessageFifo(
+    scoped_ptr<MediaMemoryChunk> mem, bool init)
+  : mem_(mem.Pass()),
+    weak_factory_(this) {
+  CHECK_EQ(reinterpret_cast<uintptr_t>(mem_->data()) % ALIGNOF(Descriptor),
+           0u);
+  CHECK_GE(mem_->size(), sizeof(Descriptor));
+  Descriptor* desc = static_cast<Descriptor*>(mem_->data());
+  base_ = static_cast<void*>(&desc->first_item);
+
+  // TODO(damienv): remove cast when atomic size_t is defined in Chrome.
+  // Currently, the sign differs.
+  rd_offset_ = reinterpret_cast<AtomicSize*>(&(desc->rd_offset));
+  wr_offset_ = reinterpret_cast<AtomicSize*>(&(desc->wr_offset));
+
+  size_t max_size = mem_->size() -
+      (static_cast<char*>(base_) - static_cast<char*>(mem_->data()));
+  if (init) {
+    size_ = max_size;
+    desc->size = size_;
+    internal_rd_offset_ = 0;
+    internal_wr_offset_ = 0;
+    base::subtle::Acquire_Store(rd_offset_, 0);
+    base::subtle::Acquire_Store(wr_offset_, 0);
+  } else {
+    size_ = desc->size;
+    CHECK_LE(size_, max_size);
+    internal_rd_offset_ = current_rd_offset();
+    internal_wr_offset_ = current_wr_offset();
+  }
+  CMALOG(kLogControl)
+      << "MediaMessageFifo:" << " init=" << init << " size=" << size_;
+  CHECK_GT(size_, 0) << size_;
+
+  weak_this_ = weak_factory_.GetWeakPtr();
+  thread_checker_.DetachFromThread();
+}
+
+MediaMessageFifo::~MediaMessageFifo() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+}
+
+void MediaMessageFifo::ObserveReadActivity(
+    const base::Closure& read_event_cb) {
+  read_event_cb_ = read_event_cb;
+}
+
+void MediaMessageFifo::ObserveWriteActivity(
+    const base::Closure& write_event_cb) {
+  write_event_cb_ = write_event_cb;
+}
+
+scoped_ptr<MediaMemoryChunk> MediaMessageFifo::ReserveMemory(
+    size_t size_to_reserve) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Capture first both the read and write offsets.
+  // and exit right away if not enough free space.
+  size_t wr_offset = internal_wr_offset();
+  size_t rd_offset = current_rd_offset();
+  size_t allocated_size = (size_ + wr_offset - rd_offset) % size_;
+  size_t free_size = size_ - 1 - allocated_size;
+  if (free_size < size_to_reserve)
+    return scoped_ptr<MediaMemoryChunk>();
+  CHECK_LE(MediaMessage::minimum_msg_size(), size_to_reserve);
+
+  // Note: in the next 2 conditions, we have:
+  // trailing_byte_count < size_to_reserve
+  // and since at this stage: size_to_reserve <= free_size
+  // we also have trailing_byte_count <= free_size
+  // which means that all the trailing bytes are free space in the fifo.
+  size_t trailing_byte_count = size_ - wr_offset;
+  if (trailing_byte_count < MediaMessage::minimum_msg_size()) {
+    // If there is no space to even write the smallest message,
+    // skip the trailing bytes and come back to the beginning of the fifo.
+    // (no way to insert a padding message).
+    if (free_size < trailing_byte_count)
+      return scoped_ptr<MediaMemoryChunk>();
+    wr_offset = 0;
+    CommitInternalWrite(wr_offset);
+
+  } else if (trailing_byte_count < size_to_reserve) {
+    // At this point, we know we have at least the space to write a message.
+    // However, to avoid splitting a message, a padding message is needed.
+    scoped_ptr<MediaMemoryChunk> mem(
+        ReserveMemoryNoCheck(trailing_byte_count));
+    scoped_ptr<MediaMessage> padding_message(
+        MediaMessage::CreateMessage(PaddingMediaMsg, mem.Pass()));
+  }
+
+  // Recalculate the free size and exit if not enough free space.
+  wr_offset = internal_wr_offset();
+  allocated_size = (size_ + wr_offset - rd_offset) % size_;
+  free_size = size_ - 1 - allocated_size;
+  if (free_size < size_to_reserve)
+    return scoped_ptr<MediaMemoryChunk>();
+
+  return ReserveMemoryNoCheck(size_to_reserve);
+}
+
+scoped_ptr<MediaMessage> MediaMessageFifo::Pop() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Capture the read and write offsets.
+  size_t rd_offset = internal_rd_offset();
+  size_t wr_offset = current_wr_offset();
+  size_t allocated_size = (size_ + wr_offset - rd_offset) % size_;
+
+  if (allocated_size < MediaMessage::minimum_msg_size())
+    return scoped_ptr<MediaMessage>();
+
+  size_t trailing_byte_count = size_ - rd_offset;
+  if (trailing_byte_count < MediaMessage::minimum_msg_size()) {
+    // If there is no space to even have the smallest message,
+    // skip the trailing bytes and come back to the beginning of the fifo.
+    // Note: all the trailing bytes correspond to allocated bytes since:
+    // trailing_byte_count < MediaMessage::minimum_msg_size() <= allocated_size
+    rd_offset = 0;
+    allocated_size -= trailing_byte_count;
+    trailing_byte_count = size_;
+    CommitInternalRead(rd_offset);
+  }
+
+  // The message should not be longer than the allocated size
+  // but since a message is a contiguous area of memory, it should also be
+  // smaller than |trailing_byte_count|.
+  size_t max_msg_size = std::min(allocated_size, trailing_byte_count);
+  if (max_msg_size < MediaMessage::minimum_msg_size())
+    return scoped_ptr<MediaMessage>();
+  void* msg_src = static_cast<uint8*>(base_) + rd_offset;
+
+  // Create a flag to protect the serialized structure of the message
+  // from being overwritten.
+  // The serialized structure starts at offset |rd_offset|.
+  scoped_refptr<MediaMessageFlag> rd_flag(new MediaMessageFlag(rd_offset));
+  rd_flags_.push_back(rd_flag);
+  scoped_ptr<MediaMemoryChunk> mem(
+      new FifoOwnedMemory(
+          msg_src, max_msg_size, rd_flag,
+          base::Bind(&MediaMessageFifo::OnRdMemoryReleased, weak_this_)));
+
+  // Create the message which wraps its the serialized structure.
+  scoped_ptr<MediaMessage> message(MediaMessage::MapMessage(mem.Pass()));
+  CHECK(message);
+
+  // Update the internal read pointer.
+  rd_offset = (rd_offset + message->size()) % size_;
+  CommitInternalRead(rd_offset);
+
+  return message.Pass();
+}
+
+void MediaMessageFifo::Flush() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  size_t wr_offset = current_wr_offset();
+
+  // Invalidate every memory region before flushing.
+  while (!rd_flags_.empty()) {
+    CMALOG(kLogControl) << "Invalidate flag";
+    rd_flags_.front()->Invalidate();
+    rd_flags_.pop_front();
+  }
+
+  // Flush by setting the read pointer to the value of the write pointer.
+  // Update first the internal read pointer then the public one.
+  CommitInternalRead(wr_offset);
+  CommitRead(wr_offset);
+}
+
+scoped_ptr<MediaMemoryChunk> MediaMessageFifo::ReserveMemoryNoCheck(
+    size_t size_to_reserve) {
+  size_t wr_offset = internal_wr_offset();
+
+  // Memory block corresponding to the serialized structure of the message.
+  void* msg_start = static_cast<uint8*>(base_) + wr_offset;
+  scoped_refptr<MediaMessageFlag> wr_flag(new MediaMessageFlag(wr_offset));
+  wr_flags_.push_back(wr_flag);
+  scoped_ptr<MediaMemoryChunk> mem(
+      new FifoOwnedMemory(
+          msg_start, size_to_reserve, wr_flag,
+          base::Bind(&MediaMessageFifo::OnWrMemoryReleased, weak_this_)));
+
+  // Update the internal write pointer.
+  wr_offset = (wr_offset + size_to_reserve) % size_;
+  CommitInternalWrite(wr_offset);
+
+  return mem.Pass();
+}
+
+void MediaMessageFifo::OnWrMemoryReleased() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  if (wr_flags_.empty()) {
+    // Sanity check: when there is no protected memory area,
+    // the external write offset has no reason to be different from
+    // the internal write offset.
+    DCHECK_EQ(current_wr_offset(), internal_wr_offset());
+    return;
+  }
+
+  // Update the external write offset.
+  while (!wr_flags_.empty() &&
+         (!wr_flags_.front()->IsValid() || wr_flags_.front()->HasOneRef())) {
+    // TODO(damienv): Could add a sanity check to make sure the offset is
+    // between the external write offset and the read offset (not included).
+    wr_flags_.pop_front();
+  }
+
+  // Update the read offset to the first locked memory area
+  // or to the internal read pointer if nothing prevents it.
+  size_t external_wr_offset = internal_wr_offset();
+  if (!wr_flags_.empty())
+    external_wr_offset = wr_flags_.front()->offset();
+  CommitWrite(external_wr_offset);
+}
+
+void MediaMessageFifo::OnRdMemoryReleased() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  if (rd_flags_.empty()) {
+    // Sanity check: when there is no protected memory area,
+    // the external read offset has no reason to be different from
+    // the internal read offset.
+    DCHECK_EQ(current_rd_offset(), internal_rd_offset());
+    return;
+  }
+
+  // Update the external read offset.
+  while (!rd_flags_.empty() &&
+         (!rd_flags_.front()->IsValid() || rd_flags_.front()->HasOneRef())) {
+    // TODO(damienv): Could add a sanity check to make sure the offset is
+    // between the external read offset and the write offset.
+    rd_flags_.pop_front();
+  }
+
+  // Update the read offset to the first locked memory area
+  // or to the internal read pointer if nothing prevents it.
+  size_t external_rd_offset = internal_rd_offset();
+  if (!rd_flags_.empty())
+    external_rd_offset = rd_flags_.front()->offset();
+  CommitRead(external_rd_offset);
+}
+
+size_t MediaMessageFifo::current_rd_offset() const {
+  DCHECK_EQ(sizeof(size_t), sizeof(AtomicSize));
+  size_t rd_offset = base::subtle::Acquire_Load(rd_offset_);
+  CHECK_LT(rd_offset, size_);
+  return rd_offset;
+}
+
+size_t MediaMessageFifo::current_wr_offset() const {
+  DCHECK_EQ(sizeof(size_t), sizeof(AtomicSize));
+
+  // When the fifo consumer acquires the write offset,
+  // we have to make sure that any possible following reads are actually
+  // returning results at least inline with the memory snapshot taken
+  // when the write offset was sampled.
+  // That's why an Acquire_Load is used here.
+  size_t wr_offset = base::subtle::Acquire_Load(wr_offset_);
+  CHECK_LT(wr_offset, size_);
+  return wr_offset;
+}
+
+void MediaMessageFifo::CommitRead(size_t new_rd_offset) {
+  // Add a memory fence to ensure the message content is completely read
+  // before updating the read offset.
+  base::subtle::Release_Store(rd_offset_, new_rd_offset);
+
+  // Make sure the read pointer has been updated before sending a notification.
+  if (!read_event_cb_.is_null()) {
+    base::subtle::MemoryBarrier();
+    read_event_cb_.Run();
+  }
+}
+
+void MediaMessageFifo::CommitWrite(size_t new_wr_offset) {
+  // Add a memory fence to ensure the message content is written
+  // before updating the write offset.
+  base::subtle::Release_Store(wr_offset_, new_wr_offset);
+
+  // Make sure the write pointer has been updated before sending a notification.
+  if (!write_event_cb_.is_null()) {
+    base::subtle::MemoryBarrier();
+    write_event_cb_.Run();
+  }
+}
+
+void MediaMessageFifo::CommitInternalRead(size_t new_rd_offset) {
+  internal_rd_offset_ = new_rd_offset;
+}
+
+void MediaMessageFifo::CommitInternalWrite(size_t new_wr_offset) {
+  internal_wr_offset_ = new_wr_offset;
+}
+
+}  // namespace media
+}  // namespace chromecast
diff --git a/chromecast/media/cma/ipc/media_message_fifo.h b/chromecast/media/cma/ipc/media_message_fifo.h
new file mode 100644
index 0000000..fe82721
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message_fifo.h
@@ -0,0 +1,208 @@
+// 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 CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_FIFO_H_
+#define CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_FIFO_H_
+
+#include <list>
+
+#include "base/atomicops.h"
+#include "base/basictypes.h"
+#include "base/callback.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "base/threading/thread_checker.h"
+
+namespace chromecast {
+namespace media {
+class MediaMemoryChunk;
+class MediaMessage;
+class MediaMessageFlag;
+
+// MediaMessageFifo is a lock free fifo implementation
+// to pass audio/video data from one thread to another or from one process
+// to another one (in that case using shared memory).
+//
+// Assuming the feeder and the consumer have a common block of shared memory
+// (representing the serialized structure of the fifo),
+// the feeder (which must be running on a single thread) instantiates its own
+// instance of MediaMessageFifo, same applies to the consumer.
+//
+// Example: assuming the block of shared memory is given by |mem|, a typical
+// feeder (using MediaMessageFifo instance fifo_feeder) will push messages
+// in the following way:
+//   // Create a dummy message to calculate the size of the serialized message.
+//   scoped_ptr<MediaMessage> dummy_msg(
+//     MediaMessage::CreateDummyMessage(msg_type));
+//   // ...
+//   // Write all the fields to the dummy message.
+//   // ...
+//
+//   // Create the real message, once the size of the serialized message
+//   // is known.
+//   scoped_ptr<MediaMessage> msg(
+//     MediaMessage::CreateMessage(
+//       msg_type,
+//       base::Bind(&MediaMessageFifo::ReserveMemory,
+//                  base::Unretained(fifo_feeder.get())),
+//       dummy_msg->content_size()));
+//   if (!msg) {
+//     // Not enough space for the message:
+//     // retry later (e.g. when receiving a read activity event, meaning
+//     // some enough space might have been release).
+//     return;
+//   }
+//   // ...
+//   // Write all the fields to the real message
+//   // in exactly the same way it was done for the dummy message.
+//   // ...
+//   // Once message |msg| is going out of scope, then MediaMessageFifo
+//   // fifo_feeder is informed that the message is not needed anymore
+//   // (i.e. it was fully written), and fifo_feeder can then update
+//   // the external write pointer of the fifo so that the consumer
+//   // can start consuming this message.
+//
+// A typical consumer (using MediaMessageFifo instance fifo_consumer)
+// will retrive messages in the following way:
+//   scoped_ptr<MediaMessage> msg(fifo_consumer->Pop());
+//   if (!msg) {
+//     // The fifo is empty, i.e. no message left.
+//     // Try reading again later (e.g. after receiving a write activity event.
+//     return;
+//   }
+//   // Parse the message using Read functions of MediaMessage:
+//   // ...
+//   // Once the message is going out of scope, MediaMessageFifo will receive
+//   // a notification that the underlying memory can be released
+//   // (i.e. the external read pointer can be updated).
+//
+//
+class MediaMessageFifo {
+ public:
+  // Creates a media message fifo using |mem| as the underlying serialized
+  // structure.
+  // If |init| is true, the underlying fifo structure is initialized.
+  MediaMessageFifo(scoped_ptr<MediaMemoryChunk> mem, bool init);
+  ~MediaMessageFifo();
+
+  // When the consumer and the feeder are living in two different processes,
+  // we might want to convey some messages between these two processes to notify
+  // about some fifo activity.
+  void ObserveReadActivity(const base::Closure& read_event_cb);
+  void ObserveWriteActivity(const base::Closure& write_event_cb);
+
+  // Reserves a writeable block of memory at the back of the fifo,
+  // corresponding to the serialized structure of the message.
+  // Returns NULL if the required size cannot be allocated.
+  scoped_ptr<MediaMemoryChunk> ReserveMemory(size_t size);
+
+  // Pop a message from the queue.
+  // Returns a null pointer if there is no message left.
+  scoped_ptr<MediaMessage> Pop();
+
+  // Flush the fifo.
+  void Flush();
+
+ private:
+  struct Descriptor {
+    size_t size;
+    size_t rd_offset;
+    size_t wr_offset;
+
+    // Ensure the first item has the same alignment as an int64.
+    int64 first_item;
+  };
+
+  // Add some accessors to ensure security on the browser process side.
+  size_t current_rd_offset() const;
+  size_t current_wr_offset() const;
+  size_t internal_rd_offset() const {
+    DCHECK_LT(internal_rd_offset_, size_);
+    return internal_rd_offset_;
+  }
+  size_t internal_wr_offset() const {
+    DCHECK_LT(internal_wr_offset_, size_);
+    return internal_wr_offset_;
+  }
+
+  // Reserve a block of free memory without doing any check on the available
+  // space. Invoke this function only when all the checks have been done.
+  scoped_ptr<MediaMemoryChunk> ReserveMemoryNoCheck(size_t size);
+
+  // Invoked each time there is a memory region in the free space of the fifo
+  // that has possibly been written.
+  void OnWrMemoryReleased();
+
+  // Invoked each time there is a memory region in the allocated space
+  // of the fifo that has possibly been released.
+  void OnRdMemoryReleased();
+
+  // Functions to modify the internal/external read/write pointers.
+  void CommitRead(size_t new_rd_offset);
+  void CommitWrite(size_t new_wr_offset);
+  void CommitInternalRead(size_t new_rd_offset);
+  void CommitInternalWrite(size_t new_wr_offset);
+
+  // An instance of MediaMessageFifo must be running on a single thread.
+  // If the fifo feeder and consumer are living on 2 different threads
+  // or 2 different processes, they must create their own instance
+  // of MediaMessageFifo using the same underlying block of (shared) memory
+  // in the constructor.
+  base::ThreadChecker thread_checker_;
+
+  // Callbacks invoked to notify either of some read or write activity on the
+  // fifo. This is especially useful when the feeder and consumer are living in
+  // two different processes.
+  base::Closure read_event_cb_;
+  base::Closure write_event_cb_;
+
+  // The serialized structure of the fifo.
+  scoped_ptr<MediaMemoryChunk> mem_;
+
+  // The size in bytes of the fifo is cached locally for security purpose.
+  // (the renderer process cannot modify the size and make the browser process
+  // access out of range addresses).
+  size_t size_;
+
+  // TODO(damienv): This is a work-around since atomicops.h does not define
+  // an atomic size_t type.
+#if SIZE_MAX == UINT32_MAX
+  typedef base::subtle::Atomic32 AtomicSize;
+#elif SIZE_MAX == UINT64_MAX
+  typedef base::subtle::Atomic64 AtomicSize;
+#elif
+#error "Unsupported size_t"
+#endif
+  AtomicSize* rd_offset_;
+  AtomicSize* wr_offset_;
+
+  // Internal read offset: this is where data is actually read from.
+  // The external offset |rd_offset_| is only used to protect data from being
+  // overwritten by the feeder.
+  // At any time, the internal read pointer must be between the external read
+  // offset and the write offset (circular fifo definition of "between").
+  size_t internal_rd_offset_;
+  size_t internal_wr_offset_;
+
+  // Note: all the memory read/write are followed by a memory fence before
+  // updating the rd/wr pointer.
+  void* base_;
+
+  // Protects the messages that are either being read or written.
+  std::list<scoped_refptr<MediaMessageFlag> > rd_flags_;
+  std::list<scoped_refptr<MediaMessageFlag> > wr_flags_;
+
+  base::WeakPtrFactory<MediaMessageFifo> weak_factory_;
+  base::WeakPtr<MediaMessageFifo> weak_this_;
+
+  DISALLOW_COPY_AND_ASSIGN(MediaMessageFifo);
+};
+
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_FIFO_H_
diff --git a/chromecast/media/cma/ipc/media_message_fifo_unittest.cc b/chromecast/media/cma/ipc/media_message_fifo_unittest.cc
new file mode 100644
index 0000000..bb1eef3
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message_fifo_unittest.cc
@@ -0,0 +1,195 @@
+// 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
+
diff --git a/chromecast/media/cma/ipc/media_message_type.h b/chromecast/media/cma/ipc/media_message_type.h
new file mode 100644
index 0000000..23ff847
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message_type.h
@@ -0,0 +1,15 @@
+// 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 CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_TYPE_H_
+#define CHROMECAST_MEDIA_CMA_IPC_MEDIA_MESSAGE_TYPE_H_
+
+enum MediaMessageType {
+  PaddingMediaMsg = 1,
+  FrameMediaMsg,
+  AudioConfigMediaMsg,
+  VideoConfigMediaMsg,
+};
+
+#endif
diff --git a/chromecast/media/cma/ipc/media_message_unittest.cc b/chromecast/media/cma/ipc/media_message_unittest.cc
new file mode 100644
index 0000000..07d9655
--- /dev/null
+++ b/chromecast/media/cma/ipc/media_message_unittest.cc
@@ -0,0 +1,146 @@
+// 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 "chromecast/media/cma/ipc/media_memory_chunk.h"
+#include "chromecast/media/cma/ipc/media_message.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace chromecast {
+namespace media {
+
+namespace {
+
+class ExternalMemoryBlock
+    : public MediaMemoryChunk {
+ public:
+  ExternalMemoryBlock(void* data, size_t size)
+      : data_(data), size_(size) {}
+  virtual ~ExternalMemoryBlock() {}
+
+  // MediaMemoryChunk implementation.
+  virtual void* data() const OVERRIDE { return data_; }
+  virtual size_t size() const OVERRIDE { return size_; }
+  virtual bool valid() const OVERRIDE { return true; }
+
+ private:
+  void* const data_;
+  const size_t size_;
+};
+
+scoped_ptr<MediaMemoryChunk> DummyAllocator(
+    void* data, size_t size, size_t alloc_size) {
+  CHECK_LE(alloc_size, size);
+  return scoped_ptr<MediaMemoryChunk>(
+      new ExternalMemoryBlock(data, alloc_size));
+}
+
+}
+
+TEST(MediaMessageTest, WriteRead) {
+  int buffer_size = 1024;
+  scoped_ptr<uint8[]> buffer(new uint8[buffer_size]);
+  MediaMessage::MemoryAllocatorCB mem_alloc_cb(
+      base::Bind(&DummyAllocator, buffer.get(), buffer_size));
+  uint32 type = 0x1;
+  int msg_content_capacity = 512;
+
+  // Write a message.
+  int count = 64;
+  scoped_ptr<MediaMessage> msg1(
+      MediaMessage::CreateMessage(type, mem_alloc_cb, msg_content_capacity));
+  for (int k = 0; k < count; k++) {
+    int v1 = 2 * k + 1;
+    EXPECT_TRUE(msg1->WritePod(v1));
+    uint8 v2 = k;
+    EXPECT_TRUE(msg1->WritePod(v2));
+  }
+  EXPECT_EQ(msg1->content_size(), count * (sizeof(int) + sizeof(uint8)));
+
+  // Verify the integrity of the message.
+  scoped_ptr<MediaMessage> msg2(
+      MediaMessage::MapMessage(scoped_ptr<MediaMemoryChunk>(
+          new ExternalMemoryBlock(&buffer[0], buffer_size))));
+  for (int k = 0; k < count; k++) {
+    int v1;
+    int expected_v1 = 2 * k + 1;
+    EXPECT_TRUE(msg2->ReadPod(&v1));
+    EXPECT_EQ(v1, expected_v1);
+    uint8 v2;
+    uint8 expected_v2 = k;
+    EXPECT_TRUE(msg2->ReadPod(&v2));
+    EXPECT_EQ(v2, expected_v2);
+  }
+}
+
+TEST(MediaMessageTest, WriteOverflow) {
+  int buffer_size = 1024;
+  scoped_ptr<uint8[]> buffer(new uint8[buffer_size]);
+  MediaMessage::MemoryAllocatorCB mem_alloc_cb(
+      base::Bind(&DummyAllocator, buffer.get(), buffer_size));
+  uint32 type = 0x1;
+  int msg_content_capacity = 8;
+
+  scoped_ptr<MediaMessage> msg1(
+      MediaMessage::CreateMessage(type, mem_alloc_cb, msg_content_capacity));
+  uint32 v1 = 0;
+  uint8 v2 = 0;
+  EXPECT_TRUE(msg1->WritePod(v1));
+  EXPECT_TRUE(msg1->WritePod(v1));
+
+  EXPECT_FALSE(msg1->WritePod(v1));
+  EXPECT_FALSE(msg1->WritePod(v2));
+}
+
+TEST(MediaMessageTest, ReadOverflow) {
+  int buffer_size = 1024;
+  scoped_ptr<uint8[]> buffer(new uint8[buffer_size]);
+  MediaMessage::MemoryAllocatorCB mem_alloc_cb(
+      base::Bind(&DummyAllocator, buffer.get(), buffer_size));
+  uint32 type = 0x1;
+  int msg_content_capacity = 8;
+
+  scoped_ptr<MediaMessage> msg1(
+      MediaMessage::CreateMessage(type, mem_alloc_cb, msg_content_capacity));
+  uint32 v1 = 0xcd;
+  EXPECT_TRUE(msg1->WritePod(v1));
+  EXPECT_TRUE(msg1->WritePod(v1));
+
+  scoped_ptr<MediaMessage> msg2(
+      MediaMessage::MapMessage(scoped_ptr<MediaMemoryChunk>(
+          new ExternalMemoryBlock(&buffer[0], buffer_size))));
+  uint32 v2;
+  EXPECT_TRUE(msg2->ReadPod(&v2));
+  EXPECT_EQ(v2, v1);
+  EXPECT_TRUE(msg2->ReadPod(&v2));
+  EXPECT_EQ(v2, v1);
+  EXPECT_FALSE(msg2->ReadPod(&v2));
+}
+
+TEST(MediaMessageTest, DummyMessage) {
+  int buffer_size = 1024;
+  scoped_ptr<uint8[]> buffer(new uint8[buffer_size]);
+  MediaMessage::MemoryAllocatorCB mem_alloc_cb(
+      base::Bind(&DummyAllocator, buffer.get(), buffer_size));
+  uint32 type = 0x1;
+
+  // Create first a dummy message to estimate the content size.
+  scoped_ptr<MediaMessage> msg1(
+      MediaMessage::CreateDummyMessage(type));
+  uint32 v1 = 0xcd;
+  EXPECT_TRUE(msg1->WritePod(v1));
+  EXPECT_TRUE(msg1->WritePod(v1));
+
+  // Create the real message and write the actual content.
+  scoped_ptr<MediaMessage> msg2(
+      MediaMessage::CreateMessage(type, mem_alloc_cb, msg1->content_size()));
+  EXPECT_TRUE(msg2->WritePod(v1));
+  EXPECT_TRUE(msg2->WritePod(v1));
+  EXPECT_FALSE(msg2->WritePod(v1));
+}
+
+}  // namespace media
+}  // namespace chromecast
diff --git a/chromecast/media/media.gyp b/chromecast/media/media.gyp
index 75f0547..f57033b 100644
--- a/chromecast/media/media.gyp
+++ b/chromecast/media/media.gyp
@@ -31,10 +31,26 @@
       ],
     },
     {
+      'target_name': 'cma_ipc',
+      'type': '<(component)',
+      'dependencies': [
+        '../../base/base.gyp:base',
+      ],
+      'sources': [
+        'cma/ipc/media_memory_chunk.cc',
+        'cma/ipc/media_memory_chunk.h',
+        'cma/ipc/media_message.cc',
+        'cma/ipc/media_message.h',
+        'cma/ipc/media_message_fifo.cc',
+        'cma/ipc/media_message_fifo.h',
+      ],
+    },
+    {
       'target_name': 'cast_media',
       'type': 'none',
       'dependencies': [
         'cma_base',
+        'cma_ipc',
       ],
     },
     {
@@ -53,6 +69,8 @@
         'cma/base/balanced_media_task_runner_unittest.cc',
         'cma/base/buffering_controller_unittest.cc',
         'cma/base/run_all_unittests.cc',
+        'cma/ipc/media_message_fifo_unittest.cc',
+        'cma/ipc/media_message_unittest.cc',
       ],
     },
   ],