[ON HOLD] Implement pull-based design for content decoding

net/filter/stream_source.cc

Index: net/filter/stream_source.cc
diff --git a/net/filter/stream_source.cc b/net/filter/stream_source.cc
new file mode 100644
index 0000000000000000000000000000000000000000..a70361babfde97e1febb4851211ad54a8a1a024d
--- /dev/null
+++ b/net/filter/stream_source.cc
@@ -0,0 +1,222 @@
+// Copyright 2016 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.
+
+#include "net/filter/stream_source.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/strings/string_util.h"
+#include "net/filter/block_buffer.h"
+#include "net/filter/brotli_stream_source.h"
+#include "net/filter/gzip_stream_source.h"
+#include "net/filter/sdch_stream_source.h"
+
+namespace net {
+
+namespace {
+
+const char kDeflate[] = "deflate";
+const char kGZip[] = "gzip";
+const char kSdch[] = "sdch";
+const char kXGZip[] = "x-gzip";
+const char kBrotli[] = "br";
+
+std::vector<StreamSource::SourceType> SourceTypeNamesToTypes(
+    const std::vector<std::string>& types) {
+  std::vector<StreamSource::SourceType> result;
+  for (const auto& type : types) {
+    if (base::LowerCaseEqualsASCII(type, kBrotli)) {
+      result.push_back(StreamSource::SOURCE_BROTLI);
+    } else if (base::LowerCaseEqualsASCII(type, kDeflate)) {
+      result.push_back(StreamSource::SOURCE_DEFLATE);
+    } else if (base::LowerCaseEqualsASCII(type, kGZip) ||
+               base::LowerCaseEqualsASCII(type, kXGZip)) {
+      result.push_back(StreamSource::SOURCE_GZIP);
+    } else if (base::LowerCaseEqualsASCII(type, kSdch)) {
+      result.push_back(StreamSource::SOURCE_SDCH);
+    }
+  }

[mmenke, 2016/03/04 21:15:57]

If we see an unrecognized scheme, we should just give up, right?

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

Define "give up"?  The current behavior (which I don't like, but think we should
change in a separate CL) appears to me to be that if we fail to recognize any
encoding type, we don't do any filtering at all but let the request go through. 
I'd recommend keeping that behavior in this CL (or changing it in a dependent
CL), which I think requires changing this code.

[mmenke, 2016/03/09 23:11:50]

That's what I meant by "give up".  :)

[xunjieli, 2016/04/20 19:16:10]

Acknowledged.

+  return result;
+}
+
+std::string SourceTypeAsString(StreamSource::SourceType type) {
+  switch (type) {
+    case StreamSource::SOURCE_BROTLI:

[mmenke, 2016/03/04 21:15:57]

We presumably won't change these often, but I suggest a separate file with a
list of these, and then include it in two other files:  One with the enum class,
and one that converts it to a string.  Less likely to break when adding a
filter.

[xunjieli, 2016/04/20 19:16:09]

Done.

+      return "SOURCE_TYPE_BROTLI";
+    case StreamSource::SOURCE_DEFLATE:
+      return "SOURCE_TYPE_DEFLATE";
+    case StreamSource::SOURCE_GZIP:
+      return "SOURCE_TYPE_GZIP";
+    case StreamSource::SOURCE_SDCH:
+      return "SOURCE_TYPE_SDCH";
+    case StreamSource::SOURCE_GZIP_FALLBACK:
+      return "SOURCE_TYPE_GZIP_FALLBACK";
+    case StreamSource::SOURCE_INVALID:
+      return "SOURCE_TYPE_INVALID";
+    case StreamSource::SOURCE_NONE:
+      return "SOURCE_TYPE_NONE";
+  }

[mmenke, 2016/03/04 21:15:56]

NOTREACHED()?

[xunjieli, 2016/04/20 19:16:09]

Done.

+  return "";
+}
+
+scoped_ptr<StreamSource> BuildSource(scoped_ptr<StreamSource> current,

[mmenke, 2016/03/04 21:15:57]

Randy may have an opinion here, but I think it's a little weird to have
stream_source dependent on (almost) all of its subclasses.  May make sense to
have a separate file to handle construction.  Or not, don't feel strongly about
this.

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

This matches the (IMO bad for the reasons you articulate) pattern that currently
exists for filter.cc.  I'd be happy to see it changed in this CL, but ok if it's
not; I think it would take some real work to actually change it (since I think
that means pretty much removing awareness of the StreamSource type from
StreamSource).  If we change it, I'd vote more in favor of putting the build
function in either url_request_job.cc or url_request_http_job.cc (probably
url_request_job.cc).

[xunjieli, 2016/04/20 19:16:10]

Done. I moved them to a util file.

+                                     StreamSource::SourceType type,
+                                     SdchStreamSourceDelegate* delegate) {
+  if (type == StreamSource::SOURCE_BROTLI) {
+    scoped_ptr<BrotliStreamSource> next(
+        new BrotliStreamSource(std::move(current)));
+    current = std::move(next);

[mmenke, 2016/03/04 21:15:57]

overwriting current with next seems weird.  Suggest renaming current to
previous, and reutrning next.

[xunjieli, 2016/04/20 19:16:10]

Done.

+  } else if (type == StreamSource::SOURCE_SDCH) {
+    scoped_ptr<SdchStreamSource> next(
+        new SdchStreamSource(std::move(current), delegate));
+    if (next->Init())
+      current = std::move(next);

[mmenke, 2016/03/04 21:15:57]

On failure, just return NULL?  I'd also like to see tests for the failure case
here (Not sure we can make a gzip init failure, but I assume we can make an SDCH
one)

[xunjieli, 2016/04/20 19:16:09]

Partially DONE. How do we make SDCH's init() fail? Do we mock out that method? I
am not quite sure how to do that.

+  } else if (type == StreamSource::SOURCE_GZIP ||
+             type == StreamSource::SOURCE_DEFLATE ||
+             type == StreamSource::SOURCE_GZIP_FALLBACK) {
+    scoped_ptr<GzipStreamSource> next(new GzipStreamSource(std::move(current)));
+    GzipStreamSource::GzipStreamSourceMode mode =
+        type == StreamSource::SOURCE_DEFLATE
+            ? GzipStreamSource::GZIP_STREAM_SOURCE_DEFLATE
+            : GzipStreamSource::GZIP_STREAM_SOURCE_GZIP;
+    bool fallback = type == StreamSource::SOURCE_GZIP_FALLBACK;
+    if (next->Init(mode, fallback))
+      current = std::move(next);

[mmenke, 2016/03/04 21:15:56]

On failure, just return NULL?

[xunjieli, 2016/04/20 19:16:09]

Done.

+  }
+  return current;
+}
+
+}  // namespace
+
+StreamSource::StreamSource(SourceType type, scoped_ptr<StreamSource> previous)
+    : type_(type), previous_(std::move(previous)) {
+  // Initializes |buffer_| if only the stream source needs to read input data
+  // from |previous_|.
+  if (previous_)
+    buffer_.reset(new BlockBuffer());

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

Could get rid of this magic here by allocating it on first read.

[xunjieli, 2016/04/20 19:16:10]

Done.

+}
+
+StreamSource::~StreamSource() {}
+
+Error StreamSource::Read(IOBuffer* dest_buffer,
+                         size_t buffer_size,
+                         size_t* bytes_read,
+                         const OnReadCompleteCallback& callback) {
+  *bytes_read = 0;
+
+  Error error = OK;
+  while (error == OK) {
+    size_t single_bytes_read = 0;
+    error = ReadInternal(dest_buffer, buffer_size, &single_bytes_read);
+    *bytes_read += single_bytes_read;
+
+    // ReadInternal() returns synchronously or an error occurred, return right
+    // here.
+    if ((error == OK && single_bytes_read > 0) ||
+        (error != OK && error != ERR_IO_PENDING))

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

Is ReadInternal *allowed* to return ERR_IO_PENDING?  I guess it's valid for the
first filter in the stream, but any others?  If not (i.e. if we deal with the
first filter in some other fashion as discussed elsewhere in these reviews) we
should probably DCHECK() against ERR_IO_PENDING and document its illegality in
the header file.

[xunjieli, 2016/04/20 19:16:09]

Done.

+      return error;
+
+    // Needs more input, it has consumed all existing input.
+    DCHECK(!buffer_ || !buffer_->HasMoreBytes());
+
+    if (!previous_)
+      break;

[mmenke, 2016/03/04 21:15:57]

This previous_ magic (And the magic in the constructor) seems weird.  Suggest
either making this method part of a StreamSource subclass that the
URLRequestJob's StreamSource does not inherit from, or having URLRequestJob's
StreamSource class override Read.

The first option seems to better get rid of the previous_ maginc in the
constructor.

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

Based on my current understanding, I'd prefer the second option with the
block_buffer_ being allocated on first read.  But that does mean that the Read()
method has to be documented to be dependent on a non-null previous_ value passed
to the constructor.

[xunjieli, 2016/04/20 19:16:09]

Done.

[xunjieli, 2016/04/20 19:16:09]

Done.

+
+    // Dispatch a read to refill the input buffer.
+    size_t previous_bytes_read = 0;
+    error = previous_->Read(
+        buffer_->buffer(), buffer_->size(), &previous_bytes_read,
+        base::Bind(&StreamSource::OnReadComplete, base::Unretained(this),

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

nit, knee-jerk response: Use of Unretained should have a comment as to why it's
safe.

[xunjieli, 2016/04/20 19:16:10]

Done.

+                   base::Unretained(dest_buffer), buffer_size));
+
+    // OK with 0 bytes read means EOF. Since the buffer is already empty, and
+    // Decompress already failed to return any more data, this source is also

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

nit: I think you mean ReadInternal, not Decompress.

[xunjieli, 2016/04/20 19:16:10]

Done.

+    // at EOF. Just return that synchronously.
+    if (error == OK && previous_bytes_read == 0)
+      return OK;
+
+    // If the underlying read completed synchronously, mark the buffer as
+    // refilled and try again.
+    if (error == OK)
+      buffer_->WasRefilled(previous_bytes_read);
+  }
+
+  DCHECK(!buffer_ || !buffer_->HasMoreBytes());
+
+  if (error == ERR_IO_PENDING) {
+    callback_ = callback;
+    pending_read_buffer_ = dest_buffer;

[mmenke, 2016/03/04 21:15:57]

Hrm...seems a little weird to set these when this is called from
StreamSource::OnReadComplete.  Could just make them OnReadComplete arguments and
base::Bind them.  We're already doing that for dest_buffer, actually, and
doesn't seem like we need it both as an argument to OnReadComplete, and as a
member of |this|.

Randy may disagree on style here.

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

No, I think I agree.  I'd vote for making the callback hold the reference (which
is closer to where/how it's going to be used) and getting rid of
pending_read_buffer_.

[xunjieli, 2016/04/20 19:16:10]

Done.

[xunjieli, 2016/04/20 19:16:10]

Done.

+  }
+  return error;
+}
+
+void StreamSource::OnReadComplete(IOBuffer* dest_buffer,
+                                  size_t dest_buffer_size,
+                                  Error error,
+                                  size_t bytes_read) {
+  DCHECK(!buffer_ || !buffer_->HasMoreBytes());
+  DCHECK_EQ(dest_buffer, pending_read_buffer_.get());
+  DCHECK(!callback_.is_null());
+
+  // Take a ref for the lifetime of this function.
+  scoped_refptr<IOBuffer> dest_ref(dest_buffer);
+  pending_read_buffer_ = nullptr;
+
+  // If the underlying read failed, fail this read directly.
+  if (error != OK) {
+    base::ResetAndReturn(&callback_).Run(error, bytes_read);
+    return;
+  }
+
+  if (bytes_read == 0) {
+    // EOF. Since the buffer is empty, there is no more data to decompress (any
+    // internally buffered data would have been drained already before calling
+    // the previous stream's Read). Return EOF to our caller.
+    if (!callback_.is_null()) {

[mmenke, 2016/03/04 21:15:56]

Why check this here, but not in the "error != OK" case?  Also, suggest combining
this and the error != OK case.  (Know I suggested splitting up combined cases
elsewhere, but here they share more code, and the checks are a little simpler)

[xunjieli, 2016/04/20 19:16:10]

Done.

+      base::ResetAndReturn(&callback_).Run(error, bytes_read);
+    }
+    return;
+  }
+
+  // Mark the buffer as refilled and try decompressing.
+  buffer_->WasRefilled(bytes_read);
+
+  // Recurse. Read runs the callback if completes synchronously,

[Randy Smith (Not in Mondays), 2016/03/09 23:03:56]

I think this comment is wrong?  The code is calling the callback on synchronous
return below, and I don't see Read() calling it above.

[xunjieli, 2016/04/20 19:16:10]

Done.

+  // Otherwise, Read will have posted an asynchronous read that
+  // will later re-invoke OnReadComplete to run the callback.
+  error = Read(dest_buffer, dest_buffer_size, &bytes_read, callback_);
+  if (error != ERR_IO_PENDING)
+    base::ResetAndReturn(&callback_).Run(error, bytes_read);
+}
+
+scoped_ptr<StreamSource> StreamSource::BuildSourceChain(
+    scoped_ptr<StreamSource> current,
+    const std::vector<std::string>& type_names,
+    SdchStreamSourceDelegate* sdch_delegate) {

[mmenke, 2016/03/04 21:15:56]

Suggest just passing in the headers, and walking through them here.  That means
we don't have to assembly type_names.  And in the future, we can grab the SDCH
dictionary from it, too, instead of using the sdch_delegate to get it.

Beyond that, we take in a set of headers, walk through them, make strings, walk
through them, get types, split based on type, and actually do stuff based on
them.  Can we just go straight from headers to to filters?  Still may need types
for logging purposes, though, I suppose.

[xunjieli, 2016/04/20 19:16:10]

Done.

+  std::vector<SourceType> types = SourceTypeNamesToTypes(type_names);
+
+  // SDCH-specific hack: if the first filter is SDCH, add a gzip filter in front
+  // of it in fallback mode.
+  if (!types.empty() && types.at(0) == SOURCE_SDCH)
+    types.insert(types.begin(), SOURCE_GZIP_FALLBACK);
+
+  for (const auto& type : types) {
+    current = BuildSource(std::move(current), type, sdch_delegate);
+    if (current == nullptr)
+      return nullptr;
+  }
+
+  return current;
+}
+
+std::string StreamSource::OrderedStreamSourceList() const {
+  if (previous_) {
+    return SourceTypeAsString(type_) + "," +
+           previous_->OrderedStreamSourceList();
+  } else {
+    return SourceTypeAsString(type_);
+  }
+}
+
+}  // namespace net