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..c9e3b39b66f9dd65b43266c947a0eff2bccc8d1e 100644
--- a/mojo/system/raw_channel.h
+++ b/mojo/system/raw_channel.h
@@ -5,10 +5,13 @@
 #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 "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 +42,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 +81,151 @@ 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();
+
+    char* GetPosition();
+    size_t GetBytesToRead() const;
+
+   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:
+    WriteBuffer();
+    ~WriteBuffer();
+
+    const char* GetPosition() const;
+    size_t GetBytesToWrite() 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();

[viettrungluu, 2014/02/26 23:03:39]

-> write_buffer_no_lock

[yzshen1, 2014/02/27 02:00:30]

Done.

+
+  // Reads into |read_buffer()|, the area indicated by |GetPosition()| and

[viettrungluu, 2014/02/26 23:03:39]

nit: comma-splice: , the -> . The

[yzshen1, 2014/02/27 02:00:30]

Done.

+  // |GetBytesToRead()| will stay valid until read completion. (But please also

[viettrungluu, 2014/02/26 23:03:39]

nit: "completion. (But [...] .)" -> "completion (but [...])."

[yzshen1, 2014/02/27 02:00:30]

Done. Thanks! :)

+  // see comments for |OnShutdownNoLock()|.)
+  // |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

[viettrungluu, 2014/02/26 23:03:39]

nit: -> If IO_PENDING is returned, |OnReadCompleted()| must be called later on
the I/O thread ....

You should make it clear in the comments what contract this class offers (e.g.,
that |read_buffer()| stays valid until completion), versus what contract the
implementing subclass has to satisfy (e.g., that it must call
|OnReadCOmpleted()|).

[yzshen1, 2014/02/27 02:00:30]

Good point. Done.

+  // I/O thread to report the result, unless |Shutdown()| happens.

[viettrungluu, 2014/02/26 23:03:39]

happens -> is called

[yzshen1, 2014/02/27 02:00:30]

Done.

+  // 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(size_t* bytes_read) = 0;
+  // Similar to |Read()|, except that this method won't succeed synchronously,
+  // i.e., it is guaranteed to only return IO_FAILED or IO_PENDING.
+  virtual IOResult ScheduleRead() = 0;
+
+  // Writes contents in |write_buffer()|, the area indicated by |GetPosition()|

[viettrungluu, 2014/02/26 23:03:39]

(similarly)

I suggest that you merge the comments for reading/writing to the extent
possible.

[yzshen1, 2014/02/27 02:00:30]

I made similar changes.
 
I find that it is not very clear after merging. I ended up with things like
|On{Read,Write}Completed()|, |bytes_{read,written}|. And I need to comment that
this block of comments applies to all four methods, the other applies to
.*Read(), etc.

If you don't mind, I will leave them as separate blocks. (Sigh, English is hard.
:))

+  // and |GetBytesToWrite()| will stay valid until write completion. (But please
+  // also see comments for |OnShutdownNoLock()|.)
+  // |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()| happens.
+  // A second write shouldn't be started if there is a pending write.
+  // Must be called under |write_lock_|.
+  virtual IOResult WriteNoLock(size_t* bytes_written) = 0;
+  // Similar to |WriteNoLock()|, except that this method won't succeed
+  // synchronously, i.e., it is guaranteed to 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);
+
+  // The following members are only used on the I/O thread:
   Delegate* const delegate_;
+
+  bool read_stopped_;
+
+  scoped_ptr<ReadBuffer> read_buffer_;
+
+  // The following members are used on mutiple threads:
   base::MessageLoopForIO* const message_loop_for_io_;

[viettrungluu, 2014/02/26 23:03:39]

nit: I think I'd prefer if you grouped the const pointers at the top. The
pointers themselves are de facto-thread-safe because they won't change. (Using
the pointers is, as always, subject to the thread-safety of the thing they point
to.)

[yzshen1, 2014/02/27 02:00:30]

Done.

 
+  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);
 };