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

net/filter/filter_stream_source.cc

Index: net/filter/filter_stream_source.cc
diff --git a/net/filter/filter_stream_source.cc b/net/filter/filter_stream_source.cc
new file mode 100644
index 0000000000000000000000000000000000000000..efcc24103b330512e2cf924fa532f4520af01938
--- /dev/null
+++ b/net/filter/filter_stream_source.cc
@@ -0,0 +1,118 @@
+// 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_stream_source.h"
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/strings/string_util.h"
+#include "net/base/io_buffer.h"
+#include "net/filter/block_buffer.h"
+
+namespace net {
+
+FilterStreamSource::FilterStreamSource(SourceType type,
+                                       std::unique_ptr<StreamSource> previous)
+    : StreamSource(type, std::move(previous)) {
+  DCHECK(previous_);
+}
+
+FilterStreamSource::~FilterStreamSource() {}
+
+Error FilterStreamSource::Read(IOBuffer* dest_buffer,
+                               size_t buffer_size,
+                               size_t* bytes_read,
+                               const OnReadCompleteCallback& callback) {
+  *bytes_read = 0;
+
+  Error error = OK;
+  // Allocate a BlockBuffer during first Read().
+  if (!buffer_)
+    buffer_.reset(new BlockBuffer());
+  // Loop on reading until either:
+  // * ReadInternal() returns some data, in which case this method completes
+  //   synchronously, or
+  // * previous_->Read() does not complete synchronously, in which case
+  //   OnReadComplete() is responsible for finishing the decompression.
+  while (error == OK) {

[mmenke, 2016/04/29 19:10:54]

Hrm...Wonder if a do loop approach would be a little clearer.  The API to talk
to the subclass is pretty much the same, but the transitions may be a little
clearer.  What do you two think?

Read() {
  DCHECK_EQ(NONE, next_state_);

  // Start with the filter.  It tells us it needs data if it doesn't have any
yet.
  next_state_ = FILTER_DATA;

  return DoLoop(OK);
}

DoLoop(result) {
  ...
  switch(state){
  case READ_DATA:
    result = DoReadData();
    break;
  case READ_DATA_COMPLETE:
    result = DoReadDataComplete();
    break;
  case FILTER_DATA:
    result = DoFilterData();
    break;
  case FILTER_DATA_COMPLETE:
    result = DoFilterDataComplete();
    break;
  }
  ..
}

int DoReadData() {
  next_state_ = READ_DATA_COMPLETE;
  return previous_->Read();
}

int DoReadDataComplete(int result) {
  // No more data left. May have to check with the filter if it has to have more
data,
  // or may always be an error, not sure.
  if (result == OK)
    return ...

  // error (If we go an ERR_IO_PENDING, we don't
  if (result < 0)
    return result;

  AddPrefilterData(read_buffer_, result);
  next_state_ = FILTER_DATA;
  return result;
}

// |result| is number of bytes read, if previous state was READ_DATA_COMPLETE,
OK if
// previous state was FILTER_DATA_COMPLETE (In which case we're coming directly
from a Read call).
int DoFilterData(int result) {
  next_state_ = FILTER_DATA_COMPLETE;
  // Pass entire buffer to subclass.  Make read_buffer a DrainableIOBuffer, so
subclass can call its consume method.
  // Call even when read_buffer is empty, before going back to read more data,
to make sure we need more data, and
  // that there's no left over decoded bytes to get.
  return FilterData(read_buffer.get(), write_buffer_, write_buffer_length_);
}

int DoFilterDataComplete(int result) {
  if (result == OK) {
    // Done filtering or needs more data. Need a method to distinguish the two.
    if (NeedsMoreData())
      next_state_ = READ_DATA;
    else
      next_state_ = DONE;
    return OK;
  }

  // Failed.
  if (result < 0) {
    next_state_ = DONE;
    return result;
  }

  // next_state_ defaults to NONE.

  return result;
}

[xunjieli, 2016/07/20 21:00:47]

Done.

+    size_t single_bytes_read = 0;
+    error = ReadInternal(dest_buffer, buffer_size, &single_bytes_read);
+
+    if (error == ERR_CONTENT_DECODING_FAILED) {
+      UMA_HISTOGRAM_ENUMERATION("Net.ContentDecodingFailed.FilterType", type(),
+                                TYPE_MAX);
+    }
+
+    *bytes_read += single_bytes_read;
+
+    // ReadInternal() returns synchronously or an error occurred, return right
+    // here. ReadInternal() is not allowed to return ERR_IO_PENDING.
+    DCHECK_NE(ERR_IO_PENDING, error);
+    if ((error == OK && single_bytes_read > 0) || error != OK)

[Randy Smith (Not in Mondays), 2016/04/26 21:54:01]

I believe what this means is that this function will only change *bytes_read
once, because if single_bytes_read is non-zero, the function exits here.  If
that's accurate, I'd suggest getting rid of single_bytes_read and just passing
bytes_read into ReadInternal() for simplicity (one less variable, and one that I
experience as having a confusing name).

[xunjieli, 2016/07/20 21:00:47]

Done.

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

[mmenke, 2016/04/29 19:10:54]

nit:  I'd say it's not worth DCHECKing on buffer_ here, or below.  The second
value in the DCHECK will crash is buffer_ is null, anyways.

[xunjieli, 2016/07/20 21:00:47]

Done.

+
+    // Dispatch a read to refill the input buffer.
+    size_t previous_bytes_read = 0;
+    // Using base::Unretained(this) is safe, because |this| owns |previous_|.
+    error = previous_->Read(
+        buffer_->buffer(), buffer_->size(), &previous_bytes_read,
+        base::Bind(&FilterStreamSource::OnReadComplete, base::Unretained(this),
+                   callback, base::Unretained(dest_buffer), buffer_size));
+
+    // OK with 0 bytes read means EOF. Since the buffer is already empty, and
+    // ReadInternal already failed to return any more data, this source is also
+    // 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());
+
+  return error;
+}
+
+bool FilterStreamSource::Init() {
+  return true;
+}
+
+void FilterStreamSource::OnReadComplete(const OnReadCompleteCallback& callback,
+                                        IOBuffer* dest_buffer,
+                                        size_t dest_buffer_size,
+                                        Error error,
+                                        size_t bytes_read) {
+  DCHECK(!buffer_ || !buffer_->HasMoreBytes());
+
+  // Take a ref for the lifetime of this function.
+  scoped_refptr<IOBuffer> dest_ref(dest_buffer);
+
+  // (1) If the underlying read failed, fail this read directly.
+  // (2) 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 (error != OK || (error == OK && bytes_read == 0)) {
+    callback.Run(error, bytes_read);
+    return;
+  }
+
+  // Mark the buffer as refilled and try Read again.
+  buffer_->WasRefilled(bytes_read);
+
+  // Recurse. If Read completes synchronously, run the |callback|,
+  // 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)
+    callback.Run(error, bytes_read);
+}
+
+}  // namespace net