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 |Read()| and |WriteNoLock()|.
+  enum IOResult {
+    IO_SUCCEEDED,
+    IO_FAILED,
+    IO_PENDING
+  };
 
-  Delegate* delegate() { return delegate_; }
+  // 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 {

[viettrungluu, 2014/02/25 01:30:06]

I think I'd move this to the Windows-specific class, and just do what's
necessary to allow OnShutdownNoLock() to take ownership of the buffers. (You can
then document the reason for |OnShutdownNoLock()'s existence, as well as any
other methods.)

[yzshen1, 2014/02/25 05:16:34]

(Maybe I haven't fully understood your idea.) IMO, it seems reasonable to have
this class here:

Currently, I don't provide the subclasses with access to the read/write buffer
members (i.e., |read_buffer_|, |write_message_queue_|, etc.) Because they don't
need to care about how buffer management is done.  (Accordingly,
IOBufferPreserver doesn't provide any accessor. It is an opaque object to the
subclasses.) I personally like this encapsulation.

If we move IOBufferPreserver into the Windows-specific class, we will have to
expose buffer members to the subclasses. Besides, the subclasses need to
understand the buffer management details in order to preserve buffers by
themselves. For example, they need to know that the pending write buffer is
within the first element of |write_message_queue_|; after |Shutdown()|,
|write_message_queue_| won't be used by the base anymore, so it is safe to
mutate it; etc.

For subclasses (POSIX) that don't need to preserve the buffer beyond
OnShutdownNoLock(), there isn't much burden for them -- they just ignore
|buffer_preserver|, which will get destroyed when OnShutdownNoLock() exits.

What do you think? 

[viettrungluu, 2014/02/25 16:37:07]

I like the encapsulation too, but I have to weight it against having a very
specialized class hanging around for a single platform (namely Windows; or
really just XP).

I'd prefer that you expose the guts to subclasses (if you're feeling paranoid,
you could even add accessors inside an ifdef, with an explanatory note). The
reason is that it becomes much more obvious that a) this code is really only
used on Windows (and can only be tested/debugged on Windows), b) this code isn't
really a part of the core functionality (and perhaps can be removed at some
point).

Admittedly, either way, there's a cost.

[yzshen1, 2014/02/26 21:47:10]

Done.

+   public:
+    IOBufferPreserver(scoped_ptr<std::vector<char> > read_buffer,
+                      scoped_ptr<MessageInTransit> write_buffer);
+
+    ~IOBufferPreserver();
+
+   private:
+    scoped_ptr<std::vector<char> > read_buffer_;
+    scoped_ptr<MessageInTransit> write_buffer_;
+    DISALLOW_COPY_AND_ASSIGN(IOBufferPreserver);
+  };
+
+  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_; }
+
+  // If |schedule_for_later| is true, this method won't succeed synchronously,
+  // i.e., it is guaranteed to only return IO_FAILED or IO_PENDING. |buffer|
+  // must stay valid until read completion. |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()| 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,

[viettrungluu, 2014/02/25 01:30:06]

I think it may be better to split this apart into Read()/ScheduleRead(), rather
than having a boolean.

[yzshen1, 2014/02/25 05:16:34]

After the split, Read() can still return IO_PENDING. Does that sound okay to
you?
(I don't have a strong opinion, just to double check.)

[viettrungluu, 2014/02/25 16:37:07]

Yes, that's fine. (The main reason for having "ScheduleRead(...)" instead of
"Read(true, ...)" is that the former is clearer/more readable at the call site.
(An alternative might be to use an enum and have "Read(SCHEDULE_ONLY, ...)", but
I think that's probably overkill.)

[yzshen1, 2014/02/26 21:47:10]

Done.

+                        char* buffer,
+                        size_t bytes_to_read,
+                        size_t* bytes_read) = 0;
+
+  // If |schedule_for_later| is true, this method won't succeed synchronously,
+  // i.e., it is guaranteed to only returns IO_FAILED or IO_PENDING. |buffer|
+  // must stay valid until write completion. |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(bool schedule_for_later,

[viettrungluu, 2014/02/25 01:30:06]

(ditto)

[yzshen1, 2014/02/26 21:47:10]

Done.

+                               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:
+  // 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_;
+
+  // 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_;
+
+  // The following members are 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<RawChannel> weak_ptr_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(RawChannel);
 };
 
 }  // namespace system
 }  // namespace mojo
 
 #endif  // MOJO_SYSTEM_RAW_CHANNEL_H_