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..52cd35a676a910c7084b4b12e8b26b0adbc13340 100644
--- a/mojo/system/raw_channel.h
+++ b/mojo/system/raw_channel.h
@@ -5,10 +5,14 @@
 #ifndef MOJO_SYSTEM_RAW_CHANNEL_H_
 #define MOJO_SYSTEM_RAW_CHANNEL_H_
 
+#include <deque>
+#include <utility>
 #include <vector>
 
 #include "base/macros.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "base/synchronization/lock.h"
 #include "mojo/system/constants.h"
 #include "mojo/system/embedder/scoped_platform_handle.h"
 #include "mojo/system/system_impl_export.h"
@@ -39,7 +43,7 @@ class MessageInTransit;
 // on which |Init()| is called).
 class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
  public:
-  virtual ~RawChannel() {}
+  virtual ~RawChannel();
 
   // The |Delegate| is only accessed on the same thread as the message loop
   // (passed in on creation).
@@ -78,26 +82,156 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
 
   // This must be called (on an I/O thread) before this object is used. Returns
   // true on success. On failure, |Shutdown()| should *not* be called.
-  virtual bool Init() = 0;
+  bool Init();
 
   // This must be called (on the I/O thread) before this object is destroyed.
-  virtual void Shutdown() = 0;
+  void Shutdown();
 
   // This is thread-safe. It takes ownership of |message| (always, even on
   // failure). Returns true on success.
-  virtual bool WriteMessage(scoped_ptr<MessageInTransit> message) = 0;
+  bool WriteMessage(scoped_ptr<MessageInTransit> message);
 
  protected:
-  RawChannel(Delegate* delegate, base::MessageLoopForIO* message_loop_for_io)
-      : delegate_(delegate), message_loop_for_io_(message_loop_for_io) {}
+  // Return values of |[Schedule]Read()| and |[Schedule]WriteNoLock()|.
+  enum IOResult {
+    IO_SUCCEEDED,
+    IO_FAILED,
+    IO_PENDING
+  };
+
+  struct ReadBuffer {
+   public:
+    ReadBuffer();
+    ~ReadBuffer();
+
+    void GetBuffer(char** buffer, size_t* bytes_to_read);
+
+   private:
+    friend class RawChannel;
+
+    // We store data from |[Schedule]Read()|s in |buffer_|. The start of
+    // |buffer_| is always aligned with a message boundary (we will copy memory
+    // to ensure this), but |buffer_| may be larger than the actual number of
+    // bytes we have.
+    std::vector<char> buffer_;
+    size_t num_valid_bytes_;
+
+    DISALLOW_COPY_AND_ASSIGN(ReadBuffer);
+  };
+
+  struct WriteBuffer {
+   public:
+    typedef std::vector<std::pair<const char*, size_t> > BufferVector;

[viettrungluu, 2014/02/27 05:22:15]

I'd generally prefer a

struct Buffer {
  const void* buffer;
  size_t size;
};

(or whatever), than an std::pair (because "first" and "second" aren't
descriptive).

[yzshen1, 2014/02/27 06:56:18]

Done.

+
+    WriteBuffer();
+    ~WriteBuffer();
+
+    void GetBuffers(BufferVector* buffers) const;
+    // Returns the total size of all buffers returned by |GetBuffers()|.
+    size_t GetTotalBytesToWrite() const;
+
+   private:
+    friend class RawChannel;
+
+    // TODO(vtl): When C++11 is available, switch this to a deque of
+    // |scoped_ptr|/|unique_ptr|s.
+    std::deque<MessageInTransit*> message_queue_;
+    // The first message may have been partially sent. |offset_| indicates the
+    // position in the first message where to start the next write.
+    size_t offset_;
+
+    DISALLOW_COPY_AND_ASSIGN(WriteBuffer);
+  };
+
+  RawChannel(Delegate* delegate, base::MessageLoopForIO* 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_; }
+
+  // Only accessed on the I/O thread.
+  ReadBuffer* read_buffer();
+
+  // Only accessed under |write_lock_|.
+  WriteBuffer* write_buffer_no_lock();
+
+  // Reads into |read_buffer()|.
+  // This class guarantees that:
+  // - the area indicated by |GetBuffer()| will stay valid until read completion
+  //   (but please also see the comments for |OnShutdownNoLock()|);
+  // - a second read is not started if there is a pending read;
+  // - the method is called on the I/O thread WITHOUT |write_lock_| held.
+  //
+  // The implementing subclass must guarantee that:
+  // - |bytes_read| is untouched if the method returns values other than
+  //   IO_SUCCEEDED;
+  // - if the method returns IO_PENDING, |OnReadCompleted()| will be called on
+  //   the I/O thread to report the result, unless |Shutdown()| is called.
+  virtual IOResult Read(size_t* bytes_read) = 0;
+  // Similar to |Read()|, except that the implementing subclass must also
+  // guarantee that the method doesn't succeed synchronously, i.e., it only
+  // returns IO_FAILED or IO_PENDING.
+  virtual IOResult ScheduleRead() = 0;
+
+  // Writes contents in |write_buffer_no_lock()|.
+  // This class guarantees that:
+  // - the area indicated by |GetBuffers()| will stay valid until write
+  //   completion (but please also see the comments for |OnShutdownNoLock()|);
+  // - a second write is not started if there is a pending write;
+  // - the method is called under |write_lock_|.
+  //
+  // The implementing subclass must guarantee that:
+  // - |bytes_written| is untouched if the method returns values other than
+  //   IO_SUCCEEDED;
+  // - if the method returns IO_PENDING, |OnWriteCompleted()| will be called on
+  //   the I/O thread to report the result, unless |Shutdown()| is called.
+  virtual IOResult WriteNoLock(size_t* bytes_written) = 0;
+  // Similar to |WriteNoLock()|, except that the implementing subclass must also
+  // guarantee that the method doesn't succeed synchronously, i.e., it only
+  // returns IO_FAILED or IO_PENDING.
+  virtual IOResult ScheduleWriteNoLock() = 0;
+
+  // Must be called on the I/O thread WITHOUT |write_lock_| held.
+  virtual bool OnInit() = 0;
+  // On shutdown, passes the ownership of the buffers to subclasses, who may
+  // want to preserve them if there are pending read/write.
+  // Must be called on the I/O thread under |write_lock_|.
+  virtual void OnShutdownNoLock(
+      scoped_ptr<ReadBuffer> read_buffer,
+      scoped_ptr<WriteBuffer> write_buffer) = 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:
+  // Calls |delegate_->OnFatalError(fatal_error)|. Must be called on the I/O
+  // thread WITHOUT |write_lock_| held.
+  void CallOnFatalError(Delegate::FatalError fatal_error);
+
+  // If |result| is true, updates the write buffer and schedules a write
+  // operation to run later if there are more contents to write. If |result| is
+  // false or any error occurs during the method execution, cancels pending
+  // writes and returns false.
+  // Must be called only if |write_stopped_| is false and under |write_lock_|.
+  bool OnWriteCompletedNoLock(bool result, size_t bytes_written);
+
   Delegate* const delegate_;
   base::MessageLoopForIO* const message_loop_for_io_;
 
+  // Only used on the I/O thread:
+  bool read_stopped_;
+  scoped_ptr<ReadBuffer> read_buffer_;
+
+  base::Lock write_lock_;  // Protects the following members.
+  bool write_stopped_;
+  scoped_ptr<WriteBuffer> write_buffer_;
+
+  // 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<RawChannel> weak_ptr_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(RawChannel);
 };