Multi reader/writer cache/buffer

media/blink/multibuffer_reader.cc

Index: media/blink/multibuffer_reader.cc
diff --git a/media/blink/multibuffer_reader.cc b/media/blink/multibuffer_reader.cc
new file mode 100644
index 0000000000000000000000000000000000000000..c0daf3ca927f265064b05fba426d2fe43026c088
--- /dev/null
+++ b/media/blink/multibuffer_reader.cc
@@ -0,0 +1,268 @@
+// Copyright 2015 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 "base/callback_helpers.h"
+#include "media/blink/multibuffer_reader.h"
+#include "net/base/net_errors.h"
+
+namespace media {
+
+MultiBufferReader::MultiBufferReader(
+    MultiBuffer* multibuffer,
+    MultiBufferUrlData url_data,
+    int64_t start,
+    int64_t end,
+    base::Callback<void(int64_t, int64_t)> progress_callback) :
+    multibuffer_(multibuffer),
+    url_data_(url_data),
+    end_(end == -1LL ? (1LL << (multibuffer->block_size_shift() + 30)) : end),
+    preload_high_(0),
+    preload_low_(0),
+    max_buffer_forward_(0),
+    max_buffer_backward_(0),
+    pos_(start),
+    preload_pos_(),
+    loading_(true),
+    current_wait_size_(0),
+    progress_callback_(progress_callback),
+    clear_on_delete_(nullptr) {
+  DCHECK_GE(start, 0);
+  DCHECK_GE(end_, 0);
+}
+
+MultiBufferReader::~MultiBufferReader() {
+  if (clear_on_delete_) *clear_on_delete_ = false;
+  multibuffer_->RemoveReader(preload_pos_, this);
+  multibuffer_->IncrementMaxSize(
+      -block_ceil(max_buffer_forward_ + max_buffer_backward_).block_num());
+  multibuffer_->PinRange(block(pos_ - max_buffer_backward_),
+                         block_ceil(pos_ + max_buffer_forward_),
+                         -1);
+  multibuffer_->CleanupWriters(preload_pos_);
+}
+
+MultiBuffer::BlockId MultiBufferReader::block(int64_t byte_pos) const {
+  return MultiBufferBlockId(
+      url_data_,
+      byte_pos >> multibuffer_->block_size_shift());
+}
+
+MultiBuffer::BlockId MultiBufferReader::block_ceil(int64_t byte_pos) const {
+  return block(byte_pos + (1LL << multibuffer_->block_size_shift()) - 1);
+}
+
+void MultiBufferReader::Seek(int64_t pos) {
+  DCHECK_GE(pos, 0);
+  // Use a rangemap to compute the diff in pinning.
+  RangeMap<MultiBuffer::BlockId, int32_t> tmp;
+  tmp.IncrementRange(block(pos_ - max_buffer_backward_),
+                     block_ceil(pos_ + max_buffer_forward_),
+                     -1);
+  tmp.IncrementRange(block(pos - max_buffer_backward_),
+                     block_ceil(pos + max_buffer_forward_),
+                     1);
+
+  multibuffer_->PinRanges(tmp);
+
+  multibuffer_->RemoveReader(preload_pos_, this);
+  MultiBufferBlockId old_preload_pos = preload_pos_;
+  preload_pos_ = block(pos);
+  pos_ = pos;
+  if (IncrementPreloadPos()) {
+    multibuffer_->CleanupWriters(old_preload_pos);
+  }
+}
+
+void MultiBufferReader::SetMaxBuffer(int64_t backward, int64_t forward) {
+  // Safe, because we know this doesn't actually prune the cache right away.
+  multibuffer_->IncrementMaxSize(
+      -block_ceil(max_buffer_forward_ + max_buffer_backward_).block_num());
+  // Use a rangemap to compute the diff in pinning.
+  RangeMap<MultiBuffer::BlockId, int32_t> tmp;
+  tmp.IncrementRange(block(pos_ - max_buffer_backward_),
+                     block_ceil(pos_ + max_buffer_forward_),
+                     -1);
+  max_buffer_backward_ = backward;
+  max_buffer_forward_ = forward;
+  tmp.IncrementRange(block(pos_ - max_buffer_backward_),
+                     block_ceil(pos_ + max_buffer_forward_),
+                     1);
+  multibuffer_->PinRanges(tmp);
+
+  multibuffer_->IncrementMaxSize(
+      block_ceil(max_buffer_forward_ + max_buffer_backward_).block_num());
+}
+
+
+int64_t MultiBufferReader::Available() const {
+  if (url_data_ == kUnknownUrlData)
+    return -1;
+
+  MultiBufferBlockId current_block = block(pos_);
+  int64_t unavailable_byte_pos =
+      static_cast<int64_t>(
+          multibuffer_->FindNextUnavailable(block(pos_)).block_num()) <<
+      multibuffer_->block_size_shift();
+  return std::max<int64_t>(0, unavailable_byte_pos - pos_);
+}
+
+int64_t MultiBufferReader::TryRead(unsigned char *data, int64_t len) {
+  DCHECK_GT(len, 0);
+  current_wait_size_ = 0;
+  cb_ = base::Closure();
+  DCHECK_LE(pos_ + len, end_);
+  const MultiBuffer::DataMap& data_map = multibuffer_->map();
+  MultiBuffer::DataMap::const_iterator i = data_map.find(block(pos_));
+  int64_t p = pos_;
+  int64_t bytes_read = 0;
+  while (bytes_read < len) {
+    if (i == data_map.end()) break;
+    if (i->first != block(p)) break;
+    if (i->second->end_of_stream()) break;
+    size_t offset = p & ((1LL << multibuffer_->block_size_shift()) - 1);
+    size_t tocopy = std::min<size_t>(len - bytes_read,
+                                     i->second->data_size() - offset);
+    memcpy(data, i->second->data() + offset, tocopy);
+    data += tocopy;
+    bytes_read += tocopy;
+    p += tocopy;
+    ++i;
+  }
+  Seek(p);
+  return bytes_read;
+}
+
+int MultiBufferReader::Wait(int64_t len, base::Closure cb) {
+  DCHECK_LE(pos_ + len, end_);
+  DCHECK_NE(Available(), -1);
+  DCHECK_LE(len, max_buffer_forward_);
+  current_wait_size_ = len;
+
+  if (!IncrementPreloadPos())
+    return net::ERR_ABORTED;
+
+  if (Available() >= current_wait_size_) {
+    return net::OK;
+  } else {
+    cb_ = cb;
+    return net::ERR_IO_PENDING;
+  }
+}
+
+void MultiBufferReader::UpdateUrlData(const MultiBufferUrlData& old_url_data,
+                                      const MultiBufferUrlData& new_url_data) {
+  DCHECK_EQ(old_url_data, url_data_);
+  multibuffer_->RemoveReader(preload_pos_, this);
+  multibuffer_->PinRange(block(pos_ - max_buffer_backward_),
+                         block_ceil(pos_ + max_buffer_forward_),
+                         1);
+  url_data_ = new_url_data;
+  if (url_data_ != kUnknownUrlData)
+    multibuffer_->PinRange(block(pos_ - max_buffer_backward_),
+                           block_ceil(pos_ + max_buffer_forward_),
+                           1);
+
+  MultiBufferBlockId old_preload_pos = preload_pos_;
+  multibuffer_->CleanupWriters(old_preload_pos);
+
+  preload_pos_ = block(pos_);
+  if (IncrementPreloadPos()) {
+    if (!progress_callback_.is_null()) {
+      MultiBuffer::BlockId unavailable_pos =
+          multibuffer_->FindNextUnavailable(preload_pos_);
+      progress_callback_.Run(
+          pos_,
+          static_cast<int64_t>(unavailable_pos.block_num()) <<
+          multibuffer_->block_size_shift());
+    }
+  }
+}
+
+void MultiBufferReader::SetPreload(int64_t preload_high, int64_t preload_low) {
+  multibuffer_->RemoveReader(preload_pos_, this);
+  preload_pos_ = block(pos_);
+  preload_high_ = preload_high;
+  preload_low_ = preload_low;
+  IncrementPreloadPos();
+}
+
+bool MultiBufferReader::IsLoading() const {
+  return loading_;
+}
+
+bool MultiBufferReader::CheckWait() {
+  bool still_alive = true;
+  if (!cb_.is_null() && (Available() >= current_wait_size_ ||
+                         Available() == -1)) {
+    current_wait_size_ = 0;
+    clear_on_delete_ = &still_alive;

[DaleCurtis, 2015/10/19 21:45:25]

This is really messy, can you remove this in favor weakptrs?

[hubbe, 2015/10/20 00:31:39]

I could have a weak pointer factory, get a pointer to myself and then check if
it's still valid when I return, but it doesn't really change anything except add
more overhead I think.

I tried really hard to think up ways to not have to do something like this, but
failed I'm afraid. :(

[DaleCurtis, 2015/10/20 00:49:44]

Hmm, that usage of WeakPtr wouldn't fly either. Lets discuss on Wednesday so I
can better understand what you're trying to do here.

+    base::ResetAndReturn(&cb_).Run();
+    if (still_alive)
+      clear_on_delete_ = NULL;
+  }
+  return still_alive;
+}
+
+void MultiBufferReader::NotifyAvailableRange(
+    const Range<MultiBufferBlockId>& range) {
+  // Update end_ if we can.
+  if (range.end > range.begin) {
+    auto i = multibuffer_->map().find(range.end - 1);
+    DCHECK(i != multibuffer_->map().end());
+    if (i->second->end_of_stream()) {
+      // This is an upper limit because the last-to-one block is allowed
+      // to be smaller than the rest of the blocks.
+      int64_t size_upper_limit =
+          static_cast<int64_t>(range.end.block_num()) <<
+          multibuffer_->block_size_shift();
+      end_ = std::min(end_, size_upper_limit);
+    }
+  }
+  if (IncrementPreloadPos()) {
+    if (!progress_callback_.is_null()) {
+      progress_callback_.Run(static_cast<int64_t>(range.begin.block_num()) <<
+                             multibuffer_->block_size_shift(),
+                             static_cast<int64_t>(range.end.block_num()) <<
+                             multibuffer_->block_size_shift());
+    }
+  }
+}
+
+bool MultiBufferReader::IncrementPreloadPos() {
+  int64_t effective_preload = loading_ ? preload_high_ : preload_low_;
+
+  loading_ = false;
+  if (url_data_ != kUnknownUrlData) {
+    if (preload_pos_ == MultiBuffer::BlockId()) {
+      preload_pos_ = block(pos_);
+      DCHECK_GE(preload_pos_.block_num(), 0);
+    }
+    MultiBuffer::BlockId max_preload = block_ceil(
+        std::min(end_, pos_ + std::max(effective_preload, current_wait_size_)));
+
+    multibuffer_->RemoveReader(preload_pos_, this);
+
+    preload_pos_ = multibuffer_->FindNextUnavailable(preload_pos_);
+    DCHECK_GE(preload_pos_.block_num(), 0);
+
+    DVLOG(3) << "IncrementPreloadPos"
+             << " pp = " << preload_pos_
+             << " block_ceil(end_) = " << block_ceil(end_)
+             << " end_ = " << end_
+             << " max_preload " << max_preload;
+
+    if (preload_pos_ < block_ceil(end_)) {
+      if (preload_pos_ < max_preload) {
+        loading_ = true;
+        multibuffer_->AddReader(preload_pos_, this);
+      } else if (multibuffer_->Contains(preload_pos_ - 1)) {
+        --preload_pos_;
+        multibuffer_->AddReader(preload_pos_, this);
+      }
+    }
+  }
+  return CheckWait();
+}
+
+}  // namespace media