RawChannel refactoring

mojo/system/raw_channel.h

Index: mojo/system/raw_channel.h
diff --git a/mojo/system/raw_channel.h b/mojo/system/raw_channel.h
index 4b14edd07f57f1ec344a05bc3d617a4a28ac1c05..687fcac4909921fb6b9088bfd4b65892561caba2 100644
--- a/mojo/system/raw_channel.h
+++ b/mojo/system/raw_channel.h
@@ -5,10 +5,12 @@
 #ifndef MOJO_SYSTEM_RAW_CHANNEL_H_
 #define MOJO_SYSTEM_RAW_CHANNEL_H_
 
+#include <deque>
 #include <vector>
 
 #include "base/macros.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
 #include "mojo/system/constants.h"
 #include "mojo/system/embedder/scoped_platform_handle.h"
 #include "mojo/system/system_impl_export.h"
@@ -88,16 +90,102 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
   virtual bool WriteMessage(scoped_ptr<MessageInTransit> message) = 0;
 
  protected:
+  // Return values of |Read()| and |WriteNoLock()|.
+  enum IOResult {
+    IO_SUCCEEDED,
+    IO_FAILED,
+    IO_PENDING
+  };
+
+  // The buffer passed to |Read()| or |WriteNoLock()| must remain valid while
+  // the operation is pending. On some platforms (e.g., Windows), however, it
+  // may be costly if RawChannel shutdown or destruction has to wait for I/O
+  // completion.
+  // In order to solve the problem, when RawChannel is shutdown, an
+  // IOBufferPreserver object is created to manage the lifespan of the I/O
+  // buffers currently being used. As long as it stays alive, the buffers remain
+  // valid. RawChannel subclasses receive this object via |OnShutdownNoLock()|
+  // and determine when to destroy it.
+  class IOBufferPreserver {

[yzshen1, 2014/02/22 23:49:06]

I created this opaque object so that subclasses can decided how long the buffer
of the current pending read/write should live after |Shutdown()|. With its help,
I don't need to |WaitForIOCompletion()| or make an extra copy of the buffer.
What do you think?

+   public:
+    IOBufferPreserver(scoped_ptr<vector<char> > read_buffer,
+                      scoped_ptr<MessageInTransit> write_buffer);
+   private:
+    scoped_ptr<MessageInTransit> write_buffer;
+    scoped_ptr<vector<char> > read_buffer;
+    DISALLOW_COPY_AND_ASSIGN(IOBufferPreserver);
+  };
+
   RawChannel(Delegate* delegate, base::MessageLoopForIO* message_loop_for_io)
       : delegate_(delegate), message_loop_for_io_(message_loop_for_io) {}
 
-  Delegate* delegate() { return delegate_; }
   base::MessageLoopForIO* message_loop_for_io() { return message_loop_for_io_; }
+  base::Lock& write_lock() { return write_lock_; }
+
+  // If |schedule_for_later| is true, this method only returns IO_FAILED or
+  // IO_PENDING. |buffer| must stay valid until read completion.
+  // If the method returns IO_PENDING, |OnReadCompleted()| will be called on the
+  // I/O thread to report the result, unless |Shutdown()| happens.
+  // A second read shouldn't be started if there is a pending read.
+  // Must be called on the I/O thread WITHOUT |write_lock_| held.
+  virtual IOResult Read(bool schedule_for_later,
+                        char* buffer,
+                        size_t bytes_to_read,
+                        size_t* byte_read) = 0;
+
+  // If |schedule_for_later| is true, this method only returns IO_FAILED or
+  // IO_PENDING. |buffer| must stay valid until write completion.
+  // If the method returns IO_PENDING, |OnWriteCompleted()| will be called on
+  // the I/O thread to report the result, unless |Shutdown()| happens.
+  // A second write shouldn't be started if there is a pending write.
+  // Must be called under |write_lock_|.
+  virtual IOResult WriteNoLock(bool delayed,
+                               const char* buffer,
+                               size_t bytes_to_write,
+                               size_t* bytes_written) = 0;
+  // Must be called on the I/O thread WITHOUT |write_lock_| held.
+  virtual bool OnInit() = 0;
+  // Must be called on the I/O thread under |write_lock_|.
+  virtual void OnShutdownNoLock(
+      scoped_ptr<IOBufferPreserver> buffer_preserver) = 0;
+
+  // Must be called on the I/O thread WITHOUT |write_lock_| held.
+  void OnReadCompleted(bool result, size_t bytes_read);
+  // Must be called on the I/O thread WITHOUT |write_lock_| held.
+  void OnWriteCompleted(bool result, size_t bytes_written);
 
  private:
+  // TODO(yzshen): Other private methods...
+
+  // Only used on the I/O thread: ----------------------------------------------
+
   Delegate* const delegate_;
+
+  bool read_stopped_;
+
+  // We store data from |Read()|s in |read_buffer_|. The start of |read_buffer_|
+  // is always aligned with a message boundary (we will copy memory to ensure
+  // this), but |read_buffer_| may be larger than the actual number of bytes we
+  // have.
+  std::vector<char> read_buffer_;
+  size_t read_buffer_num_valid_bytes_;
+
+  // Used on mutiple threads: --------------------------------------------------
+
   base::MessageLoopForIO* const message_loop_for_io_;
 
+  base::Lock write_lock_;  // Protects the following members.
+  bool write_stopped_;
+
+  // TODO(vtl): When C++11 is available, switch this to a deque of
+  // |scoped_ptr|/|unique_ptr|s.
+  std::deque<MessageInTransit*> write_message_queue_;
+  size_t write_message_offset_;
+  // This is used for posting tasks from write threads to the I/O thread. It
+  // must only be accessed under |write_lock_|. The weak pointers it produces
+  // are only used/invalidated on the I/O thread.
+  base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(RawChannel);
 };