[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..21dcfb2ab1fa7cbf1f3fb619e81226c03ec683d8 100644
--- a/mojo/edk/system/data_pipe.h
+++ b/mojo/edk/system/data_pipe.h
@@ -6,9 +6,14 @@
 #define MOJO_EDK_SYSTEM_DATA_PIPE_H_
 
 #include <stddef.h>
+#include <vector>
 
+#include "base/callback_forward.h"
 #include "base/compiler_specific.h"
+#include "base/logging.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 +24,19 @@ namespace mojo {
 namespace edk {
 class RawChannel;
 
+enum DataPipeCommand : uint32_t { 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 +52,97 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe {
       const MojoCreateDataPipeOptions* in_options,
       MojoCreateDataPipeOptions* out_options);
 
+  explicit DataPipe(const MojoCreateDataPipeOptions& options);
+
+  void Init();
+
   // 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);
-  static ScopedPlatformHandle Deserialize(
+  void StartSerialize(size_t* max_size, size_t* max_platform_handles);
+  void EndSerialize(void* destination,
+                    size_t* actual_size,
+                    PlatformHandleVector* platform_handles);
+  static scoped_refptr<DataPipe> Deserialize(
       const void* source,
       size_t size,
-      PlatformHandleVector* platform_handles,
-      MojoCreateDataPipeOptions* options,
-      ScopedPlatformHandle* shared_memory_handle,
-      size_t* shared_memory_size);
+      PlatformHandleVector* platform_handles);
+
+  // 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* channel() { return channel_; }
+  void set_channel(RawChannel* channel) {
+    DCHECK(!channel_);
+    channel_ = channel;
+  }
+  void set_shared_buffer(scoped_refptr<PlatformSharedBuffer> shared_buffer) {
+    shared_buffer_ = shared_buffer;
+  }
+  const MojoCreateDataPipeOptions& options() { 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();
+  // Releases the raw channel.
+  void Serialize();
+
+  RawChannel* channel_;
+
+  MojoCreateDataPipeOptions options_;
+
+  // True if we've released the channel because we're about to get serialized
+  // 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_;
 };
 
 }  // namespace edk