Initial implementation of the ByteStream refactor.

content/browser/download/byte_stream.h

+// Copyright (c) 2012 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 CONTENT_BROWSER_DOWNLOAD_BYTE_STREAM_H_
+#define CONTENT_BROWSER_DOWNLOAD_BYTE_STREAM_H_
+#pragma once
+
+#include <set>
+#include <utility>
+#include <deque>
+
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "base/synchronization/lock.h"
+#include "base/task_runner.h"
+#include "content/public/browser/download_interrupt_reasons.h"
+#include "net/base/io_buffer.h"
+
+namespace content {
+
+// ByteStream is a class which implements the concept of a pipe between
+// a source and a sink, which may be on different threads.  It is intended
+// to be the only connection between source and sink; they need have
+// no awareness of each other aside from the ByteStream.  Each of
+// source and sink will maintain a reference on the ByteStream while
+// they are in existence.
+//
+// The source will add bytes to the bytestream via |AddData| and the
+// sync will retrieve the data already written via |ReleaseContents|.
+//
+// When the source has no more data to add, it will call
+// SourceComplete to indicate that and release its reference.  Errors
+// at the source are indicated to the sync via a
+// non-DOWNLOAD_INTERRUPT_REASON_NONE code.
+//
+// Normally the source is not managed after the relationship is setup;
+// it is expected to provide data and then close itself.  If an error
+// occurs on the sink, it is not signalled to the source via this
+// mechanism; instead, the source will write data until it exausts the
+// available space.  Instead, it is the responsibility of the sink,
+// usually through notifying whatever controller setup the
+// relationship, to signal the source in some other fashion.
+//
+// Callback lifetime management: No lifetime management is done in this
+// class to prevent registered callbacks from being called after any
+// objects to which they may refer have been destroyed.  It is the
+// responsibility of the callers to avoid use-after-free references.
+// This may be done by any of several mechanisms, including weak
+// pointers, scoped_refptr references, or calling the registration
+// function with a null callback from a destructor.  To avoid PostTask
+// races, the callback will always be evaluated on the thread on which
+// it is going to be executed immediately before execution.
+
+class CONTENT_EXPORT ByteStream
+    : public base::RefCountedThreadSafe<ByteStream> {
+ public:
+  typedef enum { STREAM_EMPTY, STREAM_NON_EMPTY, STREAM_COMPLETE } StreamState;

[ahendrickson, 2012/04/16 15:14:27]

STREAM_NON_EMPTY -> STREAM_HAS_DATA ?

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+  typedef base::Closure ByteStreamCallback;
+
+  static const size_t kDefaultBufferSize = 100*1024;
+
+  ByteStream();
+
+  // |buffer_size| specifies the advisory limit on the amount of data
+  // the ByteStream can hold.  If, at the end of an AddData() call, the
+  // total memory held by a ByteStream is greater than |buffer_size|,
+  // AddData() will return false.  This has no other effect.
+  // The default buffer size is kDefaultBufferSize.
+  void SetBufferSize(size_t buffer_size);
+
+  // **** Source interface
+
+  // Always adds the data passed into the ByteStream.  Returns true
+  // if more data may be added without exceeding the class limit
+  // on data.
+  // Takes ownership of the passed buffer.
+  bool AddData(scoped_refptr<net::IOBuffer> buffer, size_t byte_count);
+
+  // Signal that all data that is going to be sent, has been sent,
+  // and provide a status.

[ahendrickson, 2012/04/16 15:14:27]

Indicate what status should be provided in the normal completion case?

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Done.

+  void SourceComplete(DownloadInterruptReason status);
+
+  // Probe whether the stream is full (== has more data than the configured
+  // buffer size).
+  bool IsFull();
+
+  // Register a callback to be called on the specified TaskRunner
+  // when the stream transitions from full to having space available.
+  // This callback will only be called if a call to AddData returns
+  // false.
+  // Multiple calls to this function are supported, but they may result
+  // in dispatched source callbacks never arriving if they race with
+  // the callback update.
+  void RegisterSourceCallback(

[ahendrickson, 2012/04/16 15:14:27]

I think this needs a better name.  RegisterSourceCanAddCallback?

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

I'd prefer not, mostly because there isn't really anything else that the source
can be notified about and it's a longer name.  How strongly do you feel?

+      scoped_refptr<base::TaskRunner> source_task_runner,
+      ByteStreamCallback source_callback);
+
+  // **** Sink interface
+
+  // Returns STREAM_EMPTY if there is no data on the ByteStream and
+  // SourceComplete() has not been called, and STREAM_COMPLETE if there
+  // is no data on the ByteStream and SourceComplete() has been called.
+  // If there is data on the ByteStream, returns STREAM_NON_EMPTY
+  // and fills in |*data| with a pointer to the data, and |*length|
+  // with its length.
+  StreamState GetData(scoped_refptr<net::IOBuffer>* data, size_t* length);
+
+  // Only valid to call if GetData() has returned STREAM_COMPLETE.

[ahendrickson, 2012/04/16 15:14:27]

What will this return if called too early?

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

Undefined.  I don't want to promise to assert, but assertion is what I generally
think of as happening when you violate an interface contract.  

(I've added a DCHECK in the code.)

+  DownloadInterruptReason GetSourceResult();
+
+  // Register a callback to be called on the specified TaskRunner
+  // on any transition out of the state (no data, source not complete).
+  // I.e. the callback will be called if there is no data in the
+  // stream and either any data is added or the source completes.
+  // Multiple calls to this function are supported, but they may result
+  // in dispatched sink callbacks never arriving if they race with
+  // the callback update.
+  void RegisterSinkCallback(scoped_refptr<base::TaskRunner> sink_task_runner,

[ahendrickson, 2012/04/16 15:14:27]

RegisterDataAvailableCallback?

[Randy Smith (Not in Mondays), 2012/04/18 19:10:38]

But this is also called when there's just a result code.

+                            ByteStreamCallback sink_callback);
+
+ private:
+  friend class base::RefCountedThreadSafe<ByteStream>;
+
+  typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>,size_t> >
+      ContentVector;
+
+  virtual ~ByteStream();
+
+  void RunSinkCallback(scoped_refptr<base::TaskRunner> target_runner);
+  void RunSourceCallback(scoped_refptr<base::TaskRunner> target_runner);
+
+  // Must be acquired for all accesses other than construction/destruction.
+  base::Lock lock_;
+
+  // Size at which we start pushing back.
+  size_t buffer_size_;
+
+  // Current data.  Note that ByteSTream owns and is responsible for
+  // deletion of all memory pointed to by this member.  We do not use
+  // scoped_array<> because that has no copy constructor and STL
+  // containers require copy constructors.
+  ContentVector contents_;
+
+  // Total number of bytes held.  Redundant with (and must be kept in
+  // sync with) the sum of the individual ContentVector elements.
+  size_t data_size_;
+
+  // Completion status
+  bool is_complete_;
+  DownloadInterruptReason source_status_;
+
+  // Source callback; to be called on transition from
+  // a state in which |data_size_ >= buffer_size_| to a state in which
+  // |data_size_ < buffer_size_|.
+  scoped_refptr<base::TaskRunner> source_task_runner_;
+  ByteStreamCallback source_callback_;
+
+  // Sink callback; to be called on transition from empty to
+  // non-empty data_.
+  scoped_refptr<base::TaskRunner> sink_task_runner_;
+  ByteStreamCallback sink_callback_;
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_DOWNLOAD_BYTE_STREAM_H_