RawChannel refactoring

mojo/system/raw_channel.h

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #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"
 
 namespace base {
 class MessageLoopForIO;
 }
 
 namespace mojo {
 namespace system {
@@ skipping 10 matching lines @@
 //    the aforementioned thread).
 //
 // OS-specific implementation subclasses are to be instantiated using the
 // |Create()| static factory method.
 //
 // With the exception of |WriteMessage()|, this class is thread-unsafe (and in
 // general its methods should only be used on the I/O thread, i.e., the thread
 // 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).
   class MOJO_SYSTEM_IMPL_EXPORT Delegate {
    public:
     enum FatalError {
       FATAL_ERROR_UNKNOWN = 0,
       FATAL_ERROR_FAILED_READ,
       FATAL_ERROR_FAILED_WRITE
     };
@@ skipping 18 matching lines @@
   // Static factory method. |handle| should be a handle to a
   // (platform-appropriate) bidirectional communication channel (e.g., a socket
   // on POSIX, a named pipe on Windows). Does *not* take ownership of |delegate|
   // and |message_loop_for_io|, which must remain alive while this object does.
   static RawChannel* Create(embedder::ScopedPlatformHandle handle,
                             Delegate* delegate,
                             base::MessageLoopForIO* message_loop_for_io);
 
   // 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
+  };
 
-  Delegate* delegate() { return delegate_; }
+  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);
+
   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);
 };
 
 }  // namespace system
 }  // namespace mojo
 
 #endif  // MOJO_SYSTEM_RAW_CHANNEL_H_