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

net/filter/brotli_stream_source.cc

Index: net/filter/brotli_stream_source.cc
diff --git a/net/filter/brotli_stream_source.cc b/net/filter/brotli_stream_source.cc
new file mode 100644
index 0000000000000000000000000000000000000000..1697e0f988a0beadfee1e4f18fd064ef02b66560
--- /dev/null
+++ b/net/filter/brotli_stream_source.cc
@@ -0,0 +1,197 @@
+// 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/brotli_stream_source.h"
+
+#include "base/bind.h"
+#include "base/bit_cast.h"
+#include "base/metrics/histogram_macros.h"
+#include "net/filter/block_buffer.h"
+
+namespace net {
+
+BrotliStreamSource::BrotliStreamSource(scoped_ptr<StreamSource> previous)
+    : buffer_(new BlockBuffer()),
+      previous_(std::move(previous)),
+      used_memory_(0),
+      used_memory_maximum_(0),
+      produced_bytes_(0) {
+  brotli_state_ = BrotliCreateState(AllocateMemory, FreeMemory, this);
+  CHECK(brotli_state_);
+}
+
+BrotliStreamSource::~BrotliStreamSource() {
+  BrotliDestroyState(brotli_state_);
+  brotli_state_ = nullptr;
+  DCHECK(used_memory_ == 0);
+
+#if 0
+    UMA_HISTOGRAM_ENUMERATION(
+        "BrotliStreamSource.Status", static_cast<int>(decoding_status_),
+        static_cast<int>(DecodingStatus::DECODING_STATUS_COUNT));
+    if (decoding_status_ == DecodingStatus::DECODING_DONE) {
+      // CompressionPercent is undefined when there is no output produced.
+      if (produced_bytes_ != 0) {
+        UMA_HISTOGRAM_PERCENTAGE(
+            "BrotliStreamSource.CompressionPercent",
+            static_cast<int>((consumed_bytes_ * 100) / produced_bytes_));
+      }
+    }
+#endif
+  // All code here is for gathering stats, and can be removed when
+  // BrotliStreamSource is considered stable.
+  static const int kBuckets = 48;
+  static const int64_t kMaxKb = 1 << (kBuckets / 3);  // 64MiB in KiB
+  UMA_HISTOGRAM_CUSTOM_COUNTS("BrotliStreamSource.UsedMemoryKB",
+                              used_memory_maximum_ / 1024, 1, kMaxKb, kBuckets);
+}
+
+Error BrotliStreamSource::Read(IOBuffer* dest_buffer,
+                               size_t dest_buffer_size,
+                               size_t* bytes_read,
+                               const OnReadCompleteCallback& callback) {
+  *bytes_read = 0;
+
+  // Loop on reading the previous source until either:
+  // * BrotliDecompressStream() returns some data, in which case this method
+  //   completes synchronously, or
+  // * Read() does not complete synchronously, in which case OnReadComplete()
+  //   is responsible for finishing the decompression.
+  Error error;
+  do {
+    const uint8_t* next_in = bit_cast<uint8_t*>(buffer_->bytes());
+    size_t available_in = buffer_->bytes_left();
+    uint8_t* next_out = bit_cast<uint8_t*>(dest_buffer->data());
+    size_t available_out = dest_buffer_size;
+    size_t total_out = 0;
+    BrotliResult result =
+        BrotliDecompressStream(&available_in, &next_in, &available_out,
+                               &next_out, &total_out, brotli_state_);
+
+    size_t bytes_used = buffer_->bytes_left() - available_in;
+    size_t bytes_written = dest_buffer_size - available_out;
+    if (bytes_written > 0) {
+      *bytes_read = bytes_written;
+      produced_bytes_ += *bytes_read;
+    }
+
+    buffer_->WasDrained(bytes_used);
+
+    switch (result) {
+      case BROTLI_RESULT_NEEDS_MORE_OUTPUT:
+      // Fall through.
+      case BROTLI_RESULT_SUCCESS:
+        return OK;
+      case BROTLI_RESULT_NEEDS_MORE_INPUT:
+        if (bytes_written > 0) {
+          return OK;
+        }
+        break;
+      // If the decompressor threw an error, fail synchronously.
+      default:
+        return ERR_CONTENT_DECODING_FAILED;
+    }
+
+    DCHECK_EQ(BROTLI_RESULT_NEEDS_MORE_INPUT, result);
+
+    // Since Decompress needs more input, it has consumed all existing input.
+    DCHECK(!buffer_->HasMoreBytes());
+
+    // Dispatch a read to refill the input buffer.
+    size_t previous_bytes_read;
+    error = previous_->Read(
+        buffer_->buffer(), buffer_->size(), &previous_bytes_read,
+        base::Bind(&BrotliStreamSource::OnReadComplete, base::Unretained(this),
+                   callback, base::Unretained(dest_buffer), dest_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
+    // 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);
+  } while (error == OK);
+
+  if (error == ERR_IO_PENDING)
+    pending_read_buffer_ = dest_buffer;
+
+  return error;
+}
+
+size_t BrotliStreamSource::GetBytesOutput() const {
+  return produced_bytes_;
+}
+
+void BrotliStreamSource::OnReadComplete(const OnReadCompleteCallback& callback,
+                                        IOBuffer* dest_buffer,
+                                        size_t dest_buffer_size,
+                                        Error error,
+                                        size_t bytes_read) {
+  DCHECK(!buffer_->HasMoreBytes());
+  DCHECK_EQ(dest_buffer, pending_read_buffer_.get());
+
+  // 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) {
+    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.
+    callback.Run(OK, 0);
+    return;
+  }
+
+  // Mark the buffer as refilled and try decompressing.
+  buffer_->WasRefilled(bytes_read);
+
+  // Recurse. If this Read completes synchronously, this method runs the
+  // callback; if it does not, 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);
+  }
+}
+
+void* BrotliStreamSource::AllocateMemory(void* opaque, size_t size) {
+  BrotliStreamSource* filter = reinterpret_cast<BrotliStreamSource*>(opaque);
+  return filter->AllocateMemoryInternal(size);
+}
+
+void BrotliStreamSource::FreeMemory(void* opaque, void* address) {
+  BrotliStreamSource* filter = reinterpret_cast<BrotliStreamSource*>(opaque);
+  filter->FreeMemoryInternal(address);
+}
+
+void* BrotliStreamSource::AllocateMemoryInternal(size_t size) {
+  size_t* array = reinterpret_cast<size_t*>(malloc(size + sizeof(size_t)));
+  if (!array)
+    return nullptr;
+  used_memory_ += size;
+  if (used_memory_maximum_ < used_memory_)
+    used_memory_maximum_ = used_memory_;
+  array[0] = size;
+  return &array[1];
+}
+
+void BrotliStreamSource::FreeMemoryInternal(void* address) {
+  if (!address)
+    return;
+  size_t* array = reinterpret_cast<size_t*>(address);
+  used_memory_ -= array[-1];
+  free(&array[-1]);
+}
+
+}  // namespace net