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 methods are virtual and destructor is protected to allow mocking
+// for tests; this class isn't intended to be a base class for other classes.
+class CONTENT_EXPORT ByteStream
+    : public base::RefCountedThreadSafe<ByteStream> {
+ public:
+  typedef enum { STREAM_EMPTY, STREAM_HAS_DATA, STREAM_COMPLETE } StreamState;
+  typedef base::Closure ByteStreamCallback;
+
+  static const size_t kDefaultBufferSize = 100*1024;
+
+  ByteStream();
+
+  // *** Configuration
+
+  // |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.
+  virtual 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.
+  virtual 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.  |DOWNLOAD_INTERRUPT_REASON_NONE| will be
+  // passed for successful completion.
+  virtual void SourceComplete(DownloadInterruptReason status);
+
+  // Probe whether the stream is full (== has more data than the configured
+  // buffer size).
+  virtual bool IsFull() const;
+
+  // 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 has returned
+  // 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.
+  // |empty_percentage| is an integer in the range 0-100 that specifies how
+  // much of the space in the pipe must be available before the source
+  // callback is called.  If it is 0, the callback is called as soon as there
+  // is any space in the pipe; if 100, the pipe must be completely empty
+  // before it is called.
+  virtual void RegisterSourceCallback(
+      scoped_refptr<base::TaskRunner> source_task_runner,
+      ByteStreamCallback source_callback,
+      int empty_percentage);
+
+  // **** 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_HAS_DATA
+  // and fills in |*data| with a pointer to the data, and |*length|
+  // with its length.
+  virtual StreamState GetData(scoped_refptr<net::IOBuffer>* data,
+                              size_t* length);
+
+  // Only valid to call if GetData() has returned STREAM_COMPLETE.
+  virtual DownloadInterruptReason GetSourceResult() const;
+
+  // 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.
+  // |full_percentage| is an integer in the range 0-100 that specifies how
+  // much of the space in the pipe must be used before the sink
+  // callback is called.  If it is 0, the callback is called as soon as there
+  // is any data in the pipe; if 100, the pipe must be completely full
+  // before it is called.
+  virtual void RegisterSinkCallback(
+      scoped_refptr<base::TaskRunner> sink_task_runner,
+      ByteStreamCallback sink_callback,
+      int full_percentage);
+
+  // **** Statistics
+  virtual size_t num_source_callbacks() const { return num_source_callbacks_; }
+  virtual size_t num_sink_callbacks() const { return num_sink_callbacks_; }
+  virtual size_t bytes_read() const { return bytes_read_; }
+  virtual size_t buffers_read() const { return buffers_read_; }
+  virtual base::TimeDelta source_trigger_wait_time() const {
+    return source_trigger_wait_time_;
+  }
+  virtual base::TimeDelta sink_trigger_wait_time() const {
+    return sink_trigger_wait_time_;
+  }
+
+ protected:
+  virtual ~ByteStream();
+
+ private:
+  friend class base::RefCountedThreadSafe<ByteStream>;
+
+  typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>,size_t> >
+      ContentVector;
+
+  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.
+  mutable 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_;
+  int empty_percentage_;
+
+  // Sink callback; to be called on transition from empty to
+  // non-empty data_.
+  scoped_refptr<base::TaskRunner> sink_task_runner_;
+  ByteStreamCallback sink_callback_;
+  int full_percentage_;
+
+  // Stats
+  size_t num_source_callbacks_;
+  size_t num_sink_callbacks_;
+  size_t bytes_read_;
+  size_t buffers_read_;
+  base::Time last_non_empty_time_;
+  base::Time last_non_full_time_;
+  base::TimeDelta sink_trigger_wait_time_;
+  base::TimeDelta source_trigger_wait_time_;
+
+  DISALLOW_COPY_AND_ASSIGN(ByteStream);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_DOWNLOAD_BYTE_STREAM_H_