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/embedder/scoped_platform_handle.h"
#include "mojo/system/constants.h"
#include "mojo/system/message_in_transit.h"
#include "mojo/system/system_impl_export.h"

namespace base {
class MessageLoopForIO;
}

namespace mojo {
namespace system {

// |RawChannel| is an interface to objects that wrap an OS "pipe". It presents
// the following interface to users:
//  - Receives and dispatches messages on an I/O thread (running a
//    |MessageLoopForIO|.
//  - Provides a thread-safe way of writing messages (|WriteMessage()|);
//    writing/queueing messages will not block and is atomic from the point of
//    view of the caller. If necessary, messages are queued (to be written on
//    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();

  // 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
    };

    // Called when a message is read. This may call |Shutdown()| on the
    // |RawChannel|, but must not destroy it.
    virtual void OnReadMessage(const MessageInTransit::View& message_view) = 0;

    // Called when there's a fatal error, which leads to the channel no longer
    // being viable.
    // For each raw channel, at most one |FATAL_ERROR_FAILED_READ| and one
    // |FATAL_ERROR_FAILED_WRITE| notification will be issued. (And it is
    // possible to get both.)
    // After |OnFatalError(FATAL_ERROR_FAILED_READ)| there won't be further
    // |OnReadMessage()| calls.
    virtual void OnFatalError(FatalError fatal_error) = 0;

   protected:
    virtual ~Delegate() {}
  };

  // 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.
  bool Init();

  // This must be called (on the I/O thread) before this object is destroyed.
  void Shutdown();

  // This is thread-safe. It takes ownership of |message| (always, even on
  // failure). Returns true on success.
  bool WriteMessage(scoped_ptr<MessageInTransit> message);

 protected:
  // Return values of |[Schedule]Read()| and |[Schedule]WriteNoLock()|.
  enum IOResult {
    IO_SUCCEEDED,
    IO_FAILED,
    IO_PENDING
  };

  class ReadBuffer {
   public:
    ReadBuffer();
    ~ReadBuffer();

    void GetBuffer(char** addr, size_t* size);

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

  class WriteBuffer {
   public:
    struct Buffer {
      const char* addr;
      size_t size;
    };

    WriteBuffer();
    ~WriteBuffer();

    void GetBuffers(std::vector<Buffer>* 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);

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

}  // namespace system
}  // namespace mojo

#endif  // MOJO_SYSTEM_RAW_CHANNEL_H_