Creation of ByteStream class.

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 "content/public/browser/download_interrupt_reasons.h"
+#include "net/base/io_buffer.h"
+
+namespace base {
+class SequencedTaskRunner;
+}
+
+namespace content {
+
+// A byte stream is a pipe to transfer bytes 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 byte stream.  The source and
+// the sink have different interfaces to a byte stream, |ByteStreamInput|
+// and |ByteStreamOutput|.  A pair of connected interfaces is generated by
+// calling |CreateByteStream|.
+//
+// The source adds bytes to the bytestream via |ByteStreamInput::AddData|
+// and the sync retrieves bytes already written via |ByteStreamOutput::GetData|.
+//
+// When the source has no more data to add, it will call
+// |ByteStreamInput::SourceComplete| to indicate that.  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 enable the null
+// callback strategy, callbacks will always be evaluated immediately
+// before execution.
+//
+// Class methods are virtual to allow mocking for tests; these classes
+// aren't intended to be base classes for other classes.
+class CONTENT_EXPORT ByteStreamInput {
+public:

[willchan no longer on Chromium, 2012/05/07 23:08:51]

I'm a bit surprised this interface isn't more like the sending side of a socket.
Why isn't it int Write(buffer, len, callback) and
Close(DownloadInterruptReason)?
I suggest an |int| because perhaps we should return ERR_CONNECTION_CLOSED or
something if the reading side is already closed.

[Randy Smith (Not in Mondays), 2012/05/07 23:26:01]

I'm happy to switch over to Write/Close; that's a good suggestion.  My instinct
is that I'd rather keep callback separate, because it's not associated with any
particular write; are you suggesting that because you envision the contract as
"either return true or promise to call the callback?"
Hmmm.  I rather value the one way propagation of error information; at least for
my use cases, I'm imagining the source can be written pretty simply and the sync
has the error processing responsibility.  But that interface would certainly
allow that programming model.  It just makes it a bit more complex interface,
and there isn't currently a use case for the functionality.  WDYT?

If we go that way, I'd rather return an enum than an int (i.e. create my own
namespace); that ok?

[Randy Smith (Not in Mondays), 2012/05/08 05:27:04]

A point in support of keeping the callback separate: If we associate the
callback with the Write call, we have the problem of how to spec multiple write
calls when the pipe is full (i.e. pick which callback we choose to call).  This
can be done, but sorta emphasizes that the callback is really a property of the
ByteStream, not of the individual Write calls.

[willchan no longer on Chromium, 2012/05/10 20:31:39]

I'm surprised there's no use case. What happens when the underlying sink is
Google Drive or another kind of network filesystem? Or the source? Do we really
want to keep reading from whatever (possibly the network) and writing to the
source even though the sink has gone away? I think I would be hard pressed to
find a pipe abstraction in a standard library that didn't account for this.
Sure. You may want to consider the error cases though. If sources/sinks are
backed by the network, and go through the network stack, it's likely they'll
return errors in net::Error form. If you want to ever want to expose that to a
higher layer, you may end up having to write mapping code between your namespace
and the net::Error namespace. You know your use case better than I, just be
aware of this possibility.

[willchan no longer on Chromium, 2012/05/10 20:31:39]

We avoid this problem in net::Socket code by disallowing multiple write calls
when the pipe is full. If you return an ERR_IO_PENDING type of return value,
then the caller should know not to write again.  This is a common design
paradigm.

As far as registering a callback once as opposed to per write/read, registering
it once is fine. You should probably call it "setting" as opposed to
"registering", since it's common to "register" multiple items, whereas you make
it sound like you only want to set it once.

One thing that makes me disagree with the comment about it always being a
property of the source's owner and not the individual write/read call, is
sometimes, for zero-copy purposes, you want to pass the pipe around so the
appropriate user can read/write directly from/to a buffer. That's a more
advanced use though, and if you don't want to support that, that's probably OK.

+  virtual ~ByteStreamInput() = 0;
+
+  // 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 |buffer|.
+  virtual bool AddData(scoped_refptr<net::IOBuffer> buffer,
+                       size_t byte_count) = 0;
+
+  // Signal that all data that is going to be sent, has been sent,
+  // and provide a status.  |DOWNLOAD_INTERRUPT_REASON_NONE| should be
+  // passed for successful completion.
+  virtual void SourceComplete(DownloadInterruptReason status) = 0;
+
+  // Register a callback to be called
+  // when the stream transitions from full to having space available.
+  // The callback will always be called on the task runner associated
+  // with the ByteStreamInput.
+  // This callback will only be called if a call to AddData has previously
+  // returned false (i.e. the ByteStream has been filled).
+  // 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 RegisterCallback(base::Closure source_callback) = 0;
+};
+
+class CONTENT_EXPORT ByteStreamOutput {
+ public:
+  typedef enum { STREAM_EMPTY, STREAM_HAS_DATA, STREAM_COMPLETE } StreamState;
+
+  virtual ~ByteStreamOutput() = 0;
+
+  // 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) = 0;
+
+  // Only valid to call if GetData() has returned STREAM_COMPLETE.
+  virtual DownloadInterruptReason GetSourceResult() const = 0;
+
+  // Register a callback to be called
+  // when data is added or the source completes.
+  // The callback will be always be called on the owning task runner.
+  // Multiple calls to this function are supported, but they may result
+  // in dispatched sink callbacks never arriving if they race with
+  // the callback update.
+  virtual void RegisterCallback(base::Closure sink_callback) = 0;
+
+  // Statistics.  On the output side of the byte stream to indicate

[willchan no longer on Chromium, 2012/05/07 23:08:51]

Why all these stats? Can't this be done in a composed object that also inherits
from ByteStreamOutput, if you really want it?

[Randy Smith (Not in Mondays), 2012/05/07 23:26:01]

Reasonable.  I'll make the change.  The reason for the stats is that I want to
put into UMA some notes on what the amortization tradeoffs that I'm making for
propagating data are costing me, but that's a download specific question rather
than a general ByteStream one, so I wanted to expose information about the
ByteStream so that I could implement the stats a layer up.  

(It was more obvious it was a download specific question when I had control
knobs on the ByteStream controlling the hysteresis & etc.; possibly I could just
stick the UMA into ByteStream now.)

+  // that they track only completed transfers, i.e. ones that have gone
+  // all the way through the pipe.
+  virtual size_t NumSourceCallbacks() const = 0;
+  virtual size_t NumSinkCallbacks() const = 0;
+  virtual size_t BytesRead() const = 0;
+  virtual size_t BuffersRead() const = 0;
+
+  // Time between any space becoming available for writing in the
+  // stream and triggering the source.
+  virtual base::TimeDelta TotalSourceTriggerWaitTime() const = 0;
+
+  // Time between any data becoming available for reading in the
+  // stream and triggering the sink.
+  virtual base::TimeDelta TotalSinkTriggerWaitTime() const = 0;
+};
+
+CONTENT_EXPORT void CreateByteStream(
+    scoped_ptr<ByteStreamInput>* input,
+    scoped_ptr<ByteStreamOutput>* output,
+    scoped_refptr<base::SequencedTaskRunner> input_task_runner,
+    scoped_refptr<base::SequencedTaskRunner> output_task_runner,
+    size_t buffer_size);
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_DOWNLOAD_BYTE_STREAM_H_