[mojo] Implement data pipe using a shared buffer.

mojo/edk/system/data_pipe.h

Index: mojo/edk/system/data_pipe.h
diff --git a/mojo/edk/system/data_pipe.h b/mojo/edk/system/data_pipe.h
index d2b90bf1ab6a138f26dd358d7d03c7ace4f3bf94..e9d57379aa2aa02e4d438b16f9a130e42765201b 100644
--- a/mojo/edk/system/data_pipe.h
+++ b/mojo/edk/system/data_pipe.h
@@ -6,9 +6,13 @@
 #define MOJO_EDK_SYSTEM_DATA_PIPE_H_
 
 #include <stddef.h>
+#include <vector>
 
+#include "base/callback_forward.h"
 #include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
 #include "mojo/edk/embedder/platform_handle_vector.h"
+#include "mojo/edk/embedder/platform_shared_buffer.h"
 #include "mojo/edk/embedder/scoped_platform_handle.h"
 #include "mojo/edk/system/system_impl_export.h"
 #include "mojo/public/c/system/data_pipe.h"
@@ -19,9 +23,23 @@ namespace mojo {
 namespace edk {
 class RawChannel;
 
+enum DataPipeCommand : uint32_t {
+  NOTIFY_SHARED_BUFFER,

[Anand Mistry (off Chromium), 2016/01/11 06:19:34]

I'm curious why you need this functionality. Since both sides are created at the
same time in the same process, why not just create the shared buffer and pass it
into both instances of DataPipe?

[Eliot Courtney, 2016/01/13 00:00:10]

Done.

+  DATA_WRITTEN,
+  DATA_READ
+};
+
+struct MOJO_ALIGNAS(8) DataPipeCommandHeader {
+  DataPipeCommand command;
+  uint32_t num_bytes;
+};
+
 // Shared code between DataPipeConsumerDispatcher and
-// DataPipeProducerDispatcher.
-class MOJO_SYSTEM_IMPL_EXPORT DataPipe {
+// DataPipeProducerDispatcher. This class is not thread safe -- all locking must
+// be performed by the caller. This class is only intended to be used as a
+// writer or a reader, not both.
+class MOJO_SYSTEM_IMPL_EXPORT DataPipe final
+    : public base::RefCountedThreadSafe<DataPipe> {
  public:
   // The default options for |MojoCreateDataPipe()|. (Real uses should obtain
   // this via |ValidateCreateOptions()| with a null |in_options|; this is
@@ -37,26 +55,113 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe {
       const MojoCreateDataPipeOptions* in_options,
       MojoCreateDataPipeOptions* out_options);
 
+  explicit DataPipe(const MojoCreateDataPipeOptions& options);
+
+  void Init(ScopedPlatformHandle message_pipe,

[Anand Mistry (off Chromium), 2016/01/11 06:19:34]

Most of these arguments are details about the serialisation. If deserialisation
was mostly pushed into DataPipe, you wouldn't need most of these.

[Eliot Courtney, 2016/01/13 00:00:10]

Done.

+            char* serialized_write_buffer,
+            uint32_t serialized_write_buffer_size,
+            char* serialized_read_buffer,
+            uint32_t serialized_read_buffer_size,
+            ScopedPlatformHandle shared_buffer_handle,
+            uint32_t ring_buffer_start,
+            uint32_t ring_buffer_size,
+            bool is_producer,
+            const base::Closure& init_callback);
+
   // Helper methods used by DataPipeConsumerDispatcher and
   // DataPipeProducerDispatcher for serialization and deserialization.
-  static void StartSerialize(bool have_channel_handle,
-                             bool have_shared_memory,
-                             size_t* max_size,
-                             size_t* max_platform_handles);
-  static void EndSerialize(const MojoCreateDataPipeOptions& options,
-                           ScopedPlatformHandle channel_handle,
-                           ScopedPlatformHandle shared_memory_handle,
-                           size_t shared_memory_size,
-                           void* destination,
-                           size_t* actual_size,
-                           PlatformHandleVector* platform_handles);
+  void StartSerialize(size_t* max_size, size_t* max_platform_handles);
+  void EndSerialize(void* destination,
+                    size_t* actual_size,
+                    PlatformHandleVector* platform_handles);
   static ScopedPlatformHandle Deserialize(
       const void* source,
       size_t size,
       PlatformHandleVector* platform_handles,
       MojoCreateDataPipeOptions* options,
-      ScopedPlatformHandle* shared_memory_handle,
-      size_t* shared_memory_size);
+      ScopedPlatformHandle* channel_shared_handle,
+      uint32_t* serialized_read_buffer_size,
+      uint32_t* serialized_write_buffer_size,
+      ScopedPlatformHandle* shared_buffer_handle,
+      uint32_t* ring_buffer_start,
+      uint32_t* ring_buffer_size);
+
+  // Returns the number of readable or writable bytes. If there is no valid
+  // shared buffer, returns 0.
+  uint32_t GetReadableBytes() const;
+  uint32_t GetWritableBytes() const;
+
+  // Two phase reads and writes. Note that since returned memory has to be
+  // contiguous, these may return less than the values returned by
+  // |GetReadableBytes| / |GetWritableBytes|.
+  void* GetWriteBuffer(uint32_t* num_bytes);
+  const void* GetReadBuffer(uint32_t* num_bytes);
+
+  // Normal reads and writes. Will return false if they can't read or write
+  // all the data given.
+  bool WriteDataIntoSharedBuffer(const void* data, uint32_t num_bytes);
+  bool ReadDataFromSharedBuffer(void* buf, uint32_t num_bytes);
+
+  // Notify other end of a read or write. Returns false if the other side was
+  // closed (on a RawChannel error).
+  bool NotifyWrite(uint32_t num_bytes);
+  bool NotifyRead(uint32_t num_bytes);
+
+  // Reading and writing are split into three parts: 1. reading / writing the
+  // data, telling the other side what we did, and updating the ring buffer to
+  // reflect what we did. The functions |UpdateFromRead| and |UpdateFromWrite|
+  // update the ring buffer state.
+  void UpdateFromRead(uint32_t num_bytes);
+  void UpdateFromWrite(uint32_t num_bytes);
+
+  RawChannel* GetChannel() { return channel_; }
+  const MojoCreateDataPipeOptions& GetOptions() { return options_; }
+
+  // Returns whether readable or writable state changed.
+  bool ProcessCommand(const DataPipeCommandHeader& command,
+                      ScopedPlatformHandleVectorPtr platform_handles);
+
+  // Shuts down the channel, and releases the shared buffer.
+  void Shutdown();
+
+  // Named after |CreateEquivalentDispatcherAndCloseImplNoLock|. This will
+  // Release() the RawChannel and save its returned read and write buffer, then
+  // swap ourselves into |out|.
+  void CreateEquivalentAndClose(DataPipe* out);
+
+ private:
+  friend class base::RefCountedThreadSafe<DataPipe>;
+  ~DataPipe();
+
+  uint8_t* GetSharedBufferBase();
+  // Sends the other side our shared buffer.
+  void NotifySharedBuffer();
+  // Set our shared buffer.
+  void UpdateSharedBuffer(scoped_refptr<PlatformSharedBuffer> shared_buffer);
+  // Releases the raw channel.
+  void Serialize();
+
+  RawChannel* channel_;
+
+  MojoCreateDataPipeOptions options_;
+
+  // True if we've released the channel because we're about to get serialised

[Anand Mistry (off Chromium), 2016/01/11 06:19:34]

Unfortunately, we use american spelling. So 'serialize'.

[Eliot Courtney, 2016/01/13 00:00:10]

Done.

+  // and transported. We also store the channel's read and write buffer, and
+  // handle.
+  bool channel_released_;
+  std::vector<char> serialized_read_buffer_;
+  std::vector<char> serialized_write_buffer_;
+  ScopedPlatformHandle serialized_channel_handle_;
+
+  scoped_refptr<PlatformSharedBuffer> shared_buffer_;
+  // Keep the full mapping of the shared buffer around so we don't have to
+  // recreate it.
+  scoped_ptr<PlatformSharedBufferMapping> mapping_;
+  uint32_t ring_buffer_start_;
+  uint32_t ring_buffer_size_;
+
+  // By default, the producer will attempt to create the shared buffer.
+  bool is_producer_;
 };
 
 }  // namespace edk