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

net/filter/filter_source_stream.cc

Index: net/filter/filter_source_stream.cc
diff --git a/net/filter/filter_source_stream.cc b/net/filter/filter_source_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..e0666fd6c2a346d7d006a3c3620b358a90e8c09c
--- /dev/null
+++ b/net/filter/filter_source_stream.cc
@@ -0,0 +1,162 @@
+// 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/filter_source_stream.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/strings/string_util.h"
+
+namespace net {
+
+namespace {
+
+const size_t kBufferSize = 32 * 1024;
+
+}  // namespace
+
+FilterSourceStream::FilterSourceStream(
+    SourceType type,
+    std::unique_ptr<SourceStream> next_stream)
+    : SourceStream(type),
+      next_stream_(std::move(next_stream)),
+      next_state_(STATE_NONE),
+      output_buffer_size_(0),
+      next_stream_end_reached_(false) {
+  DCHECK(next_stream_);
+}
+
+FilterSourceStream::~FilterSourceStream() {}
+
+int FilterSourceStream::Read(IOBuffer* read_buffer,
+                             size_t read_buffer_size,
+                             const CompletionCallback& callback) {
+  // Allocate a BlockBuffer during first Read().
+  if (!input_buffer_)
+    input_buffer_ = new IOBufferWithSize(kBufferSize);

[Randy Smith (Not in Mondays), 2016/08/09 20:28:30]

Why not in the constructor?

[xunjieli, 2016/08/15 15:19:07]

This is so that we don't initialize one unnecessarily. For instance, if caller
cancels the request before reading the response body. There's no need to
allocate a 32kB buffer in that case.

[Randy Smith (Not in Mondays), 2016/08/22 18:12:31]

Hmmm.  Ok.  My shoot-from-the-hip is that it's a premature optimization (i.e.
that that case won't occur often enough to be a real issue).  But I don't feel
strongly.  Put in a comment, though?

+
+  // Start with filtering data, which tells us whether it needs input data.
+  next_state_ = STATE_FILTER_DATA;
+
+  output_buffer_ = read_buffer;
+  output_buffer_size_ = read_buffer_size;
+  int rv = DoLoop(OK);
+  if (rv > OK) {
+    return rv;
+  } else if (rv == ERR_IO_PENDING) {
+    callback_ = callback;
+  }
+  return rv;
+}
+
+std::string FilterSourceStream::OrderedTypeStringList() const {
+  std::string next_type_string = next_stream_->OrderedTypeStringList();
+  if (next_type_string.empty())
+    return GetTypeAsString();
+  return next_type_string + "," + GetTypeAsString();
+}
+
+int FilterSourceStream::DoLoop(int result) {
+  DCHECK(this);
+  DCHECK_NE(STATE_NONE, next_state_);
+  int rv = result;
+  do {
+    State state = next_state_;
+    next_state_ = STATE_NONE;
+    switch (state) {
+      case STATE_READ_DATA:
+        rv = DoReadData();
+        break;
+      case STATE_READ_DATA_COMPLETE:
+        rv = DoReadDataComplete(rv);
+        break;
+      case STATE_FILTER_DATA:
+        rv = DoFilterData(rv);
+        break;
+      default:
+        NOTREACHED() << "bad state: " << state;
+        rv = ERR_UNEXPECTED;
+        break;
+    }
+  } while (rv != ERR_IO_PENDING && next_state_ != STATE_NONE);
+  return rv;
+}
+
+int FilterSourceStream::DoReadData() {
+  // Read more data means subclasses have consumed all input or this is the
+  // first read in which case the |drainable_input_buffer_| is not initialized.
+  DCHECK(drainable_input_buffer_ == nullptr ||
+         0 == drainable_input_buffer_->BytesRemaining());
+  // Use base::Unretained here is safe because |this| owns |next_stream_|.
+  int rv =
+      next_stream_->Read(input_buffer_.get(), kBufferSize,
+                         base::Bind(&FilterSourceStream::OnNextReadCompleted,
+                                    base::Unretained(this)));
+
+  if (rv != ERR_IO_PENDING)
+    next_state_ = STATE_READ_DATA_COMPLETE;
+  return rv;
+}
+
+int FilterSourceStream::DoReadDataComplete(int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  // If EOF is read (result == OK), also pass that to the the filter.
+  if (result >= OK) {
+    drainable_input_buffer_ =
+        new DrainableIOBuffer(input_buffer_.get(), result);
+    next_state_ = STATE_FILTER_DATA;
+  }
+  if (result <= OK)
+    next_stream_end_reached_ = true;
+  return result;
+}
+
+void FilterSourceStream::OnNextReadCompleted(int result) {
+  next_state_ = STATE_READ_DATA_COMPLETE;
+  int rv = DoLoop(result);
+  if (rv != ERR_IO_PENDING)
+    DoCallback(rv);
+}
+
+int FilterSourceStream::DoFilterData(int result) {
+  DCHECK(output_buffer_);
+  DCHECK_LE(0, result);
+
+  // This is first Read(), short circuit it and go straight to read data from
+  // |next_stream_|.
+  if (drainable_input_buffer_ == nullptr) {
+    next_state_ = STATE_READ_DATA;
+    return OK;
+  }
+
+  int bytes_output = FilterData(output_buffer_.get(), output_buffer_size_,
+                                drainable_input_buffer_.get());
+  if (bytes_output == ERR_CONTENT_DECODING_FAILED) {
+    UMA_HISTOGRAM_ENUMERATION("Net.ContentDecodingFailed.FilterType", type(),
+                              TYPE_MAX);
+  }
+  // FilterData() is not allowed to return ERR_IO_PENDING.
+  DCHECK_NE(ERR_IO_PENDING, bytes_output);
+
+  // If can still read data from |next_stream_| and filter did not return any

[Randy Smith (Not in Mondays), 2016/08/09 20:28:30]

nit, rephrasing: "If data can still be read from |next_stream_| ..."

[xunjieli, 2016/08/15 15:19:07]

Done.

+  // data, it is likely that the filter needs more input.
+  if (bytes_output == OK && !next_stream_end_reached_)
+    next_state_ = STATE_READ_DATA;

[Randy Smith (Not in Mondays), 2016/08/09 20:28:30]

Would an assertion that drainable_input_buffer_ has nothing left in it be
appropriate here?

[xunjieli, 2016/08/15 15:19:07]

DoReadData() already has a DCHECK.

[Randy Smith (Not in Mondays), 2016/08/22 18:12:31]

Acknowledged.

+  return bytes_output;
+}
+
+void FilterSourceStream::DoCallback(int result) {
+  DCHECK_NE(ERR_IO_PENDING, result);
+  DCHECK(!callback_.is_null());
+
+  output_buffer_ = nullptr;
+  output_buffer_size_ = 0;
+
+  base::ResetAndReturn(&callback_).Run(result);
+}
+
+}  // namespace net