[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..13f8d1d4ac43663ed0477d7b1da4471f3c54d9c2
--- /dev/null
+++ b/net/filter/brotli_stream_source.cc
@@ -0,0 +1,159 @@
+// 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/memory/ptr_util.h"
+#include "base/metrics/histogram_macros.h"
+#include "net/filter/block_buffer.h"
+
+namespace net {
+
+class BrotliStreamSource : public FilterStreamSource {
+ public:
+  BrotliStreamSource(std::unique_ptr<StreamSource> previous)
+      : FilterStreamSource(StreamSource::TYPE_BROTLI, std::move(previous)),
+        decoding_status_(DecodingStatus::DECODING_IN_PROGRESS),
+        used_memory_(0),
+        used_memory_maximum_(0),
+        produced_bytes_(0),
+        consumed_bytes_(0) {
+    brotli_state_ = BrotliCreateState(AllocateMemory, FreeMemory, this);
+    CHECK(brotli_state_);
+  }
+
+  ~BrotliStreamSource() override {
+    BrotliDestroyState(brotli_state_);
+    brotli_state_ = nullptr;
+    DCHECK(used_memory_ == 0);
+
+    UMA_HISTOGRAM_ENUMERATION(
+        "BrotliFilter.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(
+            "BrotliFilter.CompressionPercent",
+            static_cast<int>((consumed_bytes_ * 100) / produced_bytes_));
+      }
+    }
+    // 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("BrotliFilter.UsedMemoryKB",
+                                used_memory_maximum_ / 1024, 1, kMaxKb,
+                                kBuckets);
+  }
+
+ private:
+  // Reported in UMA and must be kept in sync with the histograms.xml file.
+  enum class DecodingStatus : int {
+    DECODING_IN_PROGRESS = 0,
+    DECODING_DONE,
+    DECODING_ERROR,
+
+    DECODING_STATUS_COUNT
+    // DECODING_STATUS_COUNT must always be the last element in this enum.
+  };
+
+  // StreamSource implementation
+  Error ReadInternal(IOBuffer* dest_buffer,
+                     size_t dest_buffer_size,
+                     size_t* bytes_read) override {
+    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;
+    CHECK_GE(bytes_used, 0u);
+    CHECK_GE(bytes_written, 0u);
+    *bytes_read = bytes_written;
+    produced_bytes_ += bytes_written;
+    consumed_bytes_ += bytes_used;
+
+    buffer_->WasDrained(bytes_used);
+
+    switch (result) {
+      case BROTLI_RESULT_NEEDS_MORE_OUTPUT:
+      // Fall through.
+      case BROTLI_RESULT_SUCCESS:
+        decoding_status_ = DecodingStatus::DECODING_DONE;
+        return OK;
+      case BROTLI_RESULT_NEEDS_MORE_INPUT:
+        decoding_status_ = DecodingStatus::DECODING_IN_PROGRESS;
+        if (bytes_written > 0) {
+          return OK;
+        }
+        break;
+      // If the decompressor threw an error, fail synchronously.
+      default:
+        decoding_status_ = DecodingStatus::DECODING_ERROR;
+        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());
+
+    return OK;
+  }
+
+  static void* AllocateMemory(void* opaque, size_t size) {
+    BrotliStreamSource* filter = reinterpret_cast<BrotliStreamSource*>(opaque);
+    return filter->AllocateMemoryInternal(size);
+  }
+
+  static void FreeMemory(void* opaque, void* address) {
+    BrotliStreamSource* filter = reinterpret_cast<BrotliStreamSource*>(opaque);
+    filter->FreeMemoryInternal(address);
+  }
+
+  void* 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 FreeMemoryInternal(void* address) {
+    if (!address)
+      return;
+    size_t* array = reinterpret_cast<size_t*>(address);
+    used_memory_ -= array[-1];
+    free(&array[-1]);
+  }
+
+  BrotliState* brotli_state_;
+
+  DecodingStatus decoding_status_;
+
+  size_t used_memory_;
+  size_t used_memory_maximum_;
+  size_t produced_bytes_;
+  size_t consumed_bytes_;
+
+  DISALLOW_COPY_AND_ASSIGN(BrotliStreamSource);
+};
+
+std::unique_ptr<FilterStreamSource> CreateBrotliStreamSource(
+    std::unique_ptr<StreamSource> previous) {
+  return base::WrapUnique(new BrotliStreamSource(std::move(previous)));
+}
+
+}  // namespace net