[GCM] Initial work to set up directory structure and introduce socket integration

google_apis/gcm/base/socket_stream.h

+// Copyright (c) 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.
+//
+// Protobuf ZeroCopy[Input/Output]Stream implementations capable of using a
+// net::StreamSocket. Built to work with Protobuf CodedStreams.
+// Note that any caller must first check the state() of the stream before
+// using it. It is not valid to use a CLOSED stream.
+
+#ifndef GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
+#define GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
+
+#include "base/basictypes.h"
+#include "base/callback_forward.h"
+#include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "google/protobuf/io/zero_copy_stream.h"
+#include "google_apis/gcm/base/gcm_export.h"
+
+namespace base {
+class RunLoop;
+}
+
+namespace net {
+class DrainableIOBuffer;
+class IOBufferWithSize;
+class StreamSocket;
+}
+
+namespace gcm {
+
+class GCM_EXPORT_PRIVATE SocketInputStream
+    : public google::protobuf::io::ZeroCopyInputStream {

[Ryan Sleevi, 2013/09/03 23:04:08]

So, as I mentioned over Chat, I think we need to be careful here.

In order to really leverage Chrome's network stack, you'll want to be doing the
IO reads on the actual IO thread. Even Chromoting is designed around having an
IO thread that doesn't block.

However, the requirement for the ZCIS is that it *does* block, particularly when
Next is called. With StreamSocket, you always have the chance that it will
enqueue a read operation and return ERR_IO_PENDING (due to the normalizing of
the Reactor/Proactor patterns of Windows vs Posix I/O that net/ does).

In an ideal world, we could fix Protobuf so that it could do something more
intelligent here (such as advising message lengths / expected consumption side).
Barring that, if we could change GCM to have explicit message lengths before the
protobufs, we could make intelligent decisions about how much data to read from
the socket (This is what Chromoting does, FWIW).

Absent both of those things, we'll need to split the ZCIS to be something that
can live on the 'blocking' thread (GCM thread?), and communicates with a
TaskRunner that it can issue Read/Write requests to. The pre-condition for the
TaskRunner is !RunsTasksOnCurrentThread().

You basically have three classes at play:
A ZCIS implementation [that lives on the GCM thread]
A (inner class) shared ZCIS<->network object that holds the synchronization
primitives used to coordinate [either WeakPtr or RefCountedThreadSafe]
A (inner class) network object that handles the reading and writing from the
network and signalling back to the shared state object

When ZCIS::Next() is called, the ZCIS would first check if it has any data
internally buffered (eg: from a prior call to BackUp?). If it does, it can
return that immediately.
If not, ZCIS needs to PostTask to the network TaskRunner to schedule a Read. It
then waits on a CV for your read timeout duration.
When the Read completes, it signals the CV, waking up the GCM thread.

The tricky thing here is going to be maintaining ownership and thread safety.
Requirements:
1) Any call to StreamSocket Read/Write happens on the IO thread
2) Ergo, any callbacks to Read/Write need to remain valid until the StreamSocket
is destroyed
3) The StreamSocket can only be destroyed on the IO thread
4) The CV needs to be accessible to both the network task runner and the GCM
task runner
5) [figure out which task runner lives longer. Likely network > GCM]
6) When the GCM ZCIS is destroyed, it should clean up the StreamSocket

You should be able to do this by having an (inner, private) state object that
lives on the network task runner, that holds onto things like the CV, the
socket, etc. The CV & a buffer is shared between the two, but all other
members/methods 'live' on the network task runner. Wait on the CV for the
timeout. If the CV times out, post to the network task runner a DeleteSoon of
that object, then return. Otherwise, if the CV is signalled, snag the buffer and
use that for the ZCIS.

[Nicolas Zea, 2013/09/04 00:43:33]

How does Chromoting make use of the message length? GCM does in fact send the
protobuf length before the actual data, and I use this via a call to SetLimit.
|limit_| is then used to tell the socket how much data is left to read, and once
limit_ bytes have been consumed by Next(..) it will immediately return false.
But in
the meantime the thread will block until those bytes have been received.

https://codereview.chromium.org/23181011/diff/13001/google_apis/gcm/engine/gc...
shows how this is done (see ReadProto(..)).

Although I suppose I could delay attempting to process _any_ data in the
protobuf
until all of it is available, with a timeout of course, and just make as many
Socket::Read
calls as it takes to get all the data. WDYT?

[Nicolas Zea, 2013/09/07 01:07:52]

Went ahead and made SocketStream completely asynchronous. SetLimit has been
replaced with GetNextMessage, and Next(..) no longer triggers refreshes.

+ public:
+  enum State {
+    // No valid data to read. This means the buffer is either empty or all data
+    // in the buffer has already been consumed.
+    EMPTY,
+    // Valid data to read.
+    READY,
+    // In the process of reading new data from the socket.
+    READING,
+    // An permanent error occurred and the stream is now closed.
+    CLOSED,
+  };
+
+  // |socket| should already be connected.
+  SocketInputStream(base::TimeDelta read_timeout,
+                    net::StreamSocket* socket);
+  virtual ~SocketInputStream();
+
+  // ZeroCopyInputStream implementation.
+  // Note: it is okay to poll on Next(..) calls while it returns true if more
+  // data is expected. The input stream will internally time out if no data
+  // arrives, at which point Next(..) calls will return false. The socket must
+  // then be reconnected.
+  virtual bool Next(const void** data, int* size) OVERRIDE;
+  virtual void BackUp(int count) OVERRIDE;
+  virtual bool Skip(int count) OVERRIDE;
+  virtual int64 ByteCount() const OVERRIDE;
+
+  // Reads and appends the next set of data from the socket into the existing
+  // buffer (max read size is dependent on the current limit and available
+  // buffer space).
+  void Refresh(const base::Closure& callback);
+
+  // Resets the buffer state by copying over any unread data to the beginning
+  // of the buffer and resetting the buffer position.
+  // Note: it is not valid to call Reset() if state() == CLOSED. The stream
+  // must be recreated in such a scenario.
+  void Reset();
+
+  // Sets the read limit which upon having read |limit| bytes Next(..) will
+  // return false. This is useful when the size of a message is known ahead of
+  // time. The stream will remain valid after the limit has been reached, but
+  // Reset() must be called or a new limit must be set before more data can
+  // be consumed.
+  void SetLimit(int limit);
+
+  // Returns the current buffer position (in bytes) within the stream.
+  int GetCurrentPosition() const;
+
+  // Returns the last fatal error encountered. Only valid if state() == CLOSED.
+  int last_error() const;
+
+  // Returns the current state.
+  State state() const;
+
+ private:
+  // Callback for Socket::Read calls.
+  void RefreshCompletionCallback(const base::Closure& callback, int result);
+
+  // Clears the local state.
+  void ResetInternal();
+
+  // Permanently closes the stream.
+  void CloseStream(int error, const base::Closure& callback);
+
+  // Internal net components.
+  net::StreamSocket* socket_;
+  scoped_refptr<net::IOBufferWithSize> io_buffer_;
+  scoped_refptr<net::DrainableIOBuffer> drainable_io_buffer_;
+
+  // The amount of data within |io_buffer_| that has already been read.
+  int buffer_pos_;
+
+  // The amount of valid data within |io_buffer_|.
+  int buffer_size_;
+
+  // The |buffer_pos_| limit at which Next(..) will no longer trigger Refresh
+  // calls.
+  int limit_;
+
+  // The number of bytes we've backed up by. These bytes need to be returned
+  // on the subsequent Next(..) call. 0 <= backup_bytes_ <= buffer_pos_
+  int backup_bytes_;
+
+  // The number of bytes to skip from the next successful refresh.
+  int skipped_bytes_;
+
+  // If < net::OK, the last net error received.
+  int last_error_;
+
+  // The current state.
+  State state_;
+
+  // Run loop for handling callers who would otherwise continuously poll
+  // Next(..).
+  scoped_ptr<base::RunLoop> run_loop_;
+
+  // Timeout when performing synchronous reads.
+  base::TimeDelta read_timeout_;
+
+  base::WeakPtrFactory<SocketInputStream> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketInputStream);
+};
+
+class GCM_EXPORT_PRIVATE SocketOutputStream
+    : public google::protobuf::io::ZeroCopyOutputStream {
+ public:
+  enum State {
+    // No valid data yet.
+    EMPTY,
+    // Ready for flushing (some data is present).
+    READY,
+    // In the process of flushing into the socket.
+    FLUSHING,
+    // A permanent error occurred, and the stream is now closed.
+    CLOSED,
+  };
+
+  // |socket| should already be connected.
+  SocketOutputStream(net::StreamSocket* socket);
+  virtual ~SocketOutputStream();
+
+  // ZeroCopyOutputStream implementation.
+  virtual bool Next(void** data, int* size) OVERRIDE;
+  virtual void BackUp(int count) OVERRIDE;
+  virtual int64 ByteCount() const OVERRIDE;
+
+  // Writes the buffer into the Socket.
+  void Flush(const base::Closure& callback);
+
+  // Returns the current state.
+  State state() const;
+
+  // Returns the last fatal error encountered. Only valid if state() == CLOSED.
+  int last_error() const;
+
+ private:
+  void FlushCompletionCallback(const base::Closure& callback, int result);
+
+  net::StreamSocket* socket_;
+  scoped_refptr<net::IOBufferWithSize> io_buffer_;
+  scoped_refptr<net::DrainableIOBuffer> drainable_io_buffer_;
+
+  // The amount of valid buffer data. This is the amount of data written on the
+  // next call to Flush().
+  int buffer_used_;
+
+  // If < net::OK, the last net error received.
+  int last_error_;
+
+  // The current state.
+  State state_;
+
+  base::WeakPtrFactory<SocketOutputStream> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketOutputStream);
+};
+
+}  // namespace gcm
+
+#endif  // GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_