Multi reader/writer cache/buffer

media/blink/multibuffer.h

+// 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.
+
+#ifndef MEDIA_BLINK_MULTIBUFFER_H_
+#define MEDIA_BLINK_MULTIBUFFER_H_
+
+#include <stdint.h>
+
+#include <limits>
+#include <map>
+
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "media/base/data_buffer.h"
+#include "media/blink/lru.h"
+#include "media/blink/rangemap.h"
+#include "media/blink/url_index.h"
+#include "media/blink/waiter_index.h"
+
+namespace media {
+
+typedef scoped_refptr<UrlData> MultiBufferUrlId;
+typedef int32_t MultiBufferBlockNum;
+
+const int kMaxFreesPerAdd = 10;
+const int kMaxWaitForWriterOffset = 5;
+#define kUnknownUrlId nullptr
+
+// This is the key for the multibuffer cache, it's made up of an
+// refcounted UrlData pointer and a block ID.

[liberato (no reviews please), 2015/10/09 17:36:09]

if !url_id has special meaning, might want to mention it here.

[hubbe, 2015/10/13 23:08:17]

Done.

+class MultiBufferBlockId {
+ public:
+  MultiBufferBlockId();
+  MultiBufferBlockId(const MultiBufferBlockId& block_id);
+  MultiBufferBlockId(MultiBufferUrlId url_id,
+                     MultiBufferBlockNum block_num);
+  ~MultiBufferBlockId();

[liberato (no reviews please), 2015/10/09 17:36:10]

why declare this?

[hubbe, 2015/10/13 23:08:17]

Because chrome-style demands it?

+  bool SameUrl(const MultiBufferBlockId& other) const {
+    return url_id_ == other.url_id_;
+  }
+  bool operator<(const MultiBufferBlockId& other) const {
+    if (SameUrl(other)) {
+      return block_num_ < other.block_num_;
+    }
+    return url_id_cmp() < other.url_id_cmp();
+  }
+  bool operator>(const MultiBufferBlockId& other) const {
+    if (SameUrl(other)) {
+      return block_num_ > other.block_num_;
+    }
+    return url_id_cmp() > other.url_id_cmp();
+  }
+  bool operator<=(const MultiBufferBlockId& other) const {
+    if (SameUrl(other)) {
+      return block_num_ <= other.block_num_;
+    }
+    return url_id_cmp() <= other.url_id_cmp();
+  }
+  bool operator>=(const MultiBufferBlockId& other) const {
+    if (SameUrl(other)) {
+      return block_num_ >= other.block_num_;
+    }
+    return url_id_cmp() >= other.url_id_cmp();
+  }
+  bool operator==(const MultiBufferBlockId& other) const {
+    return SameUrl(other) && block_num_ == other.block_num_;
+  }
+  bool operator!=(const MultiBufferBlockId& other) const {
+    return !SameUrl(other) || block_num_ != other.block_num_;
+  }
+  int64_t operator-(const MultiBufferBlockId& other) const {
+    if (SameUrl(other)) {
+      return block_num_ - other.block_num_;
+    }
+    // Blocks from different URLs are "infinitely" distant.
+    return std::numeric_limits<int64_t>::max();
+  }
+  MultiBufferBlockId operator-(int32_t n) const {
+    return MultiBufferBlockId(url_id_, block_num_ - n);
+  }
+  MultiBufferBlockId operator+(int32_t n) const {
+    return MultiBufferBlockId(url_id_, block_num_ + n);
+  }
+
+  void operator++() {
+    block_num_++;
+  }
+
+  MultiBufferUrlId url_id() const { return url_id_; }
+  MultiBufferBlockNum block_num() const { return block_num_; }
+
+ private:
+  // We represent the minimum block ID as {0, 0} and the
+  // minimum as {0, max_int64}. This helper function lets

[liberato (no reviews please), 2015/10/09 17:36:09]

maximum

[hubbe, 2015/10/13 23:08:17]

Done.

+  // us compare the URLs and get the sort order right.
+  uint64_t url_id_cmp() const {
+    DCHECK_GE(sizeof(uint64_t), sizeof(UrlData*));
+    if (url_id_) return reinterpret_cast<uint64_t>(url_id_.get());
+    if (block_num_ == std::numeric_limits<MultiBufferBlockNum>::max())
+      return std::numeric_limits<uint64_t>::max();
+    return 0;
+  }
+  MultiBufferUrlId url_id_;
+  MultiBufferBlockNum block_num_;
+};
+
+}   // namespace media
+
+namespace std {
+ostream& operator<<(ostream& o, const media::MultiBufferBlockId& id);
+
+template<>
+class numeric_limits<media::MultiBufferBlockId> {
+  public:
+  static media::MultiBufferBlockId min() {
+    return media::MultiBufferBlockId(
+        numeric_limits<media::MultiBufferUrlId>::min(),
+        numeric_limits<media::MultiBufferBlockNum>::min());
+  }
+
+  static media::MultiBufferBlockId max() {
+    return media::MultiBufferBlockId(
+        numeric_limits<media::MultiBufferUrlId>::max(),

[liberato (no reviews please), 2015/10/09 17:36:09]

should this be 0 rather than ::max?  (probably also in min() too).  if not, then
i don't see what the special case in url_id_cmp() is for.

[hubbe, 2015/10/13 23:08:17]

Hmm, not sure what is best here.
Originally, MultiBufferUrlId was an integer, but now it's a scoped_refptr<>

Unfortunately std::numeric_limits<T>::min() and std::numeric_limits<T>::max()
default to returning T(), which is confusing and error-inducing, but I kept this
code the way it was in case I wanted to go back to an integer.

The reason min and max are both represented using kUnknownUrlID is because
scoped_refptr<UrlData>((UrlData*)-1); will crash, so having a real "largest
value" is not possible.

+        numeric_limits<media::MultiBufferBlockNum>::max());
+  }
+};
+
+}  // namespace std
+
+namespace media {
+
+
+// MultiBufferDataProvider is the interface that MultiBuffer
+// uses to get data into the cache.
+class MultiBufferDataProvider {
+ public:
+  virtual ~MultiBufferDataProvider();

[liberato (no reviews please), 2015/10/09 17:36:09]

why not {}?

[hubbe, 2015/10/13 23:08:17]

Done.

+
+  // Returns the block number that is to be returned
+  // by the next Read() call.
+  virtual MultiBufferBlockId Tell() const = 0;
+
+  // Returns true if one (or more) blocks are
+  // availble to read.
+  virtual bool Available() const = 0;
+
+  // Returns the next block. Only valid if Available()
+  // returns true. Last block might be of a smaller size
+  // and after the last block we will get an end-of-stream
+  // DataBuffer.
+  virtual scoped_refptr<DataBuffer> Read() = 0;
+
+  // |cb| is called every time Available() becomes true.
+  virtual void SetAvailableCallback(base::Closure cb) = 0;

[liberato (no reviews please), 2015/10/09 17:36:09]

const base::Closure& ?

[hubbe, 2015/10/13 23:08:17]

Done.

+
+  // Ask the data provider to stop giving us data.

[liberato (no reviews please), 2015/10/09 17:36:09]

does this mean that it may no longer call the available callback, or just best
effort?

[hubbe, 2015/10/13 23:08:17]

Clarified.

+  virtual void SetDeferred(bool deferred) = 0;
+};
+
+class Reader {

[liberato (no reviews please), 2015/10/09 17:36:10]

comment here would help.  also, MultiBufferReader?  Reader is pretty generic. 
alternatively, move this into MultiBuffer.

[hubbe, 2015/10/13 23:08:16]

Comment added, moved into MultiBuffer.

+public:
+  Reader() {}

[liberato (no reviews please), 2015/10/09 17:36:09]

why?

[hubbe, 2015/10/13 23:08:17]

because:

../../media/blink/multibuffer.cc:398:20: error: constructor for
'media::MultiBufferReader' must explicitly initialize the base class
'MultiBuffer::Reader' which does not have a default constructor

+  virtual ~Reader() {}
+  // Tell the reader that we have a new URL ID. There may be several
+  // reasons for this:
+  //  o We encountered a redirect.
+  //  o An error occured, in this case the new_url_id will be kUnknownUrlId
+  //  o We discovered a better UrlID. This happens when expired data
+  //    exists in the cache, but the expired data was found to be still
+  //    valid after sending the first request.
+  virtual void UpdateURLId(MultiBufferUrlId old_url_id,

[liberato (no reviews please), 2015/10/09 17:36:09]

might want to include what a typical reader action would be.  i suspect it's to
call stopwaitfor(old) and then waitfor(new).

[hubbe, 2015/10/13 23:08:17]

Done.

+                           MultiBufferUrlId new_url_id) = 0;
+
+  // Notifies the reader that the range of available blocks has changed.
+  // The reader must call Observe() to activate this callback.
+  virtual void NotifyAvailableRange(
+      const Range<MultiBufferBlockId>& range) = 0;
+private:
+  DISALLOW_COPY_AND_ASSIGN(Reader);
+};
+
+
+class ReaderIndex {

[liberato (no reviews please), 2015/10/09 17:36:09]

please comment.  Also, MultiBufferReaderIndex?

[hubbe, 2015/10/13 23:08:17]

Gone.

+ public:
+  ReaderIndex();
+  ~ReaderIndex();
+  // Register a waiter at position |pos|.
+  // Returns true if this is the first Reader waiting for
+  // this particular position.
+  bool WaitFor(MultiBufferBlockId pos, Reader* waiter);
+
+  // Unregister a waiter, |pos| should be the same
+  // position as given to WaitFor().
+  // Returns true if this is the last Reader waiting for
+  // this particular position.
+  bool StopWaitFor(MultiBufferBlockId pos, Reader* waiter);
+
+  // Return the distance from pos to the next waiting reader.
+  int64_t DistToNext(MultiBufferBlockId pos) const;
+
+  // Tell all readers working on |old_url_id| to go read from
+  // |new_url_id| instead. See Reader::UpdateURLId for more info.
+  void UpdateURLId(MultiBufferUrlId old_url_id,
+                   MultiBufferUrlId new_url_id);
+
+  // Call NotifyAvailableRange(new_range) on all readers waiting
+  // for a block in |observer_range|
+  void NotifyAvailableRange(
+      const Range<MultiBufferBlockId>& observer_range,
+      const Range<MultiBufferBlockId>& new_range);
+
+  const std::map<MultiBufferBlockId, std::set<Reader*> >& map() const {
+    return readers_;
+  }
+
+ private:
+  std::map<MultiBufferBlockId, std::set<Reader*> > readers_;
+};
+
+// MultiBuffers are multi-reader multi-writer cache/buffers with
+// prefetching and pinning. Data is stored internally in ref-counted
+// blocks of identical size. |block_size_shift| is log2 of the block
+// size.
+//
+// Users should inherit this class and implement StartWriter().

[liberato (no reviews please), 2015/10/09 17:36:10]

s/Start/Create/

also, this might be better if a MultiBufferWriterFactory was provided, rather
than inheriting.

[hubbe, 2015/10/13 23:08:17]

Fixed.
Why?

+// TODO(hubbe): Make the multibuffer respond to memory pressure.
+class MultiBuffer {
+ public:
+  explicit MultiBuffer(int32_t block_size_shift);
+  virtual ~MultiBuffer();
+
+  // Identifies a block in the cache.
+  // Block numbers can be calculated from byte positions as:
+  // block_num = byte_pos >> block_size_shift
+  typedef MultiBufferBlockId BlockId;
+  typedef std::map<BlockId, scoped_refptr<DataBuffer> > DataMap;
+
+  // Wait for block |pos| to become available. Starts new Writers as needed.
+  void WaitFor(BlockId pos, Reader* reader);
+
+  // Let the buffer that we're going to want to read block |pos|, but we're ok
+  // with the current level of buffering.

[liberato (no reviews please), 2015/10/09 17:36:09]

"ok with the current level of buffering" isn't very clear.  maybe something like
"provides a hint to pause rather than destroy data providers when they've filled
all outstanding WaitFor requests.".

[hubbe, 2015/10/13 23:08:17]

DeferredWaitFor is gone.

+  void DeferredWaitFor(BlockId pos, Reader* reader);
+
+  // Stop waiting for block |pos|.
+  // Often followed by a call to WaitFor(pos + 1, ...);
+  // Cancels a previous WaitFor() or DeferredWaitFor().
+  // Idempotent.
+  void StopWaitFor(BlockId pos, Reader* reader);
+
+  // Immediately remove writers at or before |pos| if nobody needs them.
+  void CleanupWriters(BlockId pos);
+
+  // Register |reader| for NotifyAvailableRange calls. If the range of available
+  // blocks before or after |pos| changes, we'll call NotifyAvailableRange on

[liberato (no reviews please), 2015/10/09 17:36:10]

this isn't clear.  all blocks are either before, after, or equal to |pos|.

[hubbe, 2015/10/13 23:08:17]

Updated comment, better?

+  // the reader to let it know.
+  void Observe(BlockId pos, Reader* reader);
+  void StopObserve(BlockId pos, Reader* reader);
+
+  // Returns true if block |pos| is available in the cache.
+  bool Contains(BlockId pos) const;
+
+  // Returns the next unavailable block at or after |pos|.
+  BlockId FindNextUnavailable(BlockId pos) const;
+
+  // Change the pin count for a range of data blocks.
+  // Note that blocks do not have to be present in the
+  // cache to be pinned.
+  // Examples:
+  // Pin block 3, 4 & 5: PinRange(3, 6, 1);
+  // Unpin block 4 & 5: PinRange(4, 6, -1);
+  void PinRange(BlockId from, BlockId to, int32_t howmuch);
+
+  // Increment max cache size by |size| (counted in blocks).
+  void IncrementMaxSize(int32_t size);
+
+  // Caller takes ownership of 'provider', cache will
+  // not call it anymore.
+  scoped_ptr<MultiBufferDataProvider> RemoveProvider(
+      MultiBufferDataProvider* provider);
+
+  // Add a writer to this cache. Cache takes ownership and
+  // may choose to destroy it.
+  void AddProvider(scoped_ptr<MultiBufferDataProvider> provider);
+
+  // Tell all reader workin on |old_url_id| about a new url.
+  // See Reader::UpdateURLId for more info.
+  void UpdateURLId(MultiBufferUrlId old_url_id,
+                   MultiBufferUrlId new_url_id);
+
+  // Accessors.
+  const DataMap& map() const { return data_; }
+  int32_t block_size_shift() const { return block_size_shift_; }
+
+ protected:
+  // Create a new writer at |pos| and return it.
+  // Users needs to implemement this method.
+  virtual MultiBufferDataProvider* CreateWriter(BlockId pos) = 0;
+
+ private:
+  // For testing.
+  friend class TestMultiBuffer;
+
+  // Callback which notifies us that a data provider has
+  // some data for us. Also called when it might be apprperiate
+  // for a provider in a deferred state to wake up.
+  void DataProviderEvent(MultiBufferDataProvider *provider);
+
+  // Free elements from cache if needed and possible.
+  // Don't free more than |max_to_free| blocks.
+  void Prune(size_t max_to_free);
+
+  // Max number of blocks.
+  size_t max_size_;
+
+  // log2 of block size.
+  int32_t block_size_shift_;
+
+  // Stores the actual data.
+  DataMap data_;
+
+  // Keeps track of readers waiting for data.
+  ReaderIndex reader_index_;
+
+  // Keeps track of readers that will want data soon.
+  ReaderIndex deferred_reader_index_;
+
+  // Observers are notified when the surrounding ranges
+  // of available blocks change.
+  ReaderIndex observer_index_;
+
+  // Keeps track of writers by their position.
+  // The writers are owned by this class.
+  std::map<BlockId, MultiBufferDataProvider*> writer_index_;
+
+  // The LRU should contain all blocks which are not pinned.
+  LRU<BlockId> lru_;
+
+  // Keeps track of what blocks are pinned. If block p is pinned,
+  // then pinned_[p] > 0. Pinned blocks cannot be freed and should not
+  // be present in |lru_|.
+  RangeMap<BlockId, int32_t> pinned_;
+
+  // present_[block] should be 1 for all blocks that are present
+  // and 0 for all blocks that are not. Used to quickly figure out
+  // ranges of available blocks without iterating.
+  RangeMap<BlockId, int32_t> present_;
+};
+
+// Wrapper for MultiBuffer that offers a simple byte-reading
+// interface with prefetch.
+class MultiBufferReader : public Reader {

[liberato (no reviews please), 2015/10/09 17:36:10]

this is a confusing name.  StreamingMultiBufferReaderImpl?

also, this should get its own header.

[hubbe, 2015/10/13 23:08:17]

I like the shorter name, but I moved it to a separate file.

+ public:
+  // Note that |progress_callback| is guaranteed to be called if
+  // a redirect happens and the url_id is updated. Otherwise
+  // origin checks will become insecure.

[liberato (no reviews please), 2015/10/09 17:36:10]

please document processs_callback args.

[hubbe, 2015/10/13 23:08:17]

Done.

+  MultiBufferReader(MultiBuffer* multibuffer,
+                    MultiBufferUrlId url_id,
+                    int64_t start,
+                    int64_t end,
+                    int64_t preload_high,
+                    int64_t preload_low,
+                    int64_t max_buffer_forward,
+                    int64_t max_buffer_backward,
+                    base::Callback<void(int64_t, int64_t)> progress_callback);
+
+  ~MultiBufferReader() override;
+
+  // Returns number of bytes available for reading. When the rest of the file
+  // is available, the number returned will be greater than the number
+  // or readable bytes. If an error occurs, -1 is returned.
+  int64_t Available() const;
+
+  // Seek to a different position.
+  // If there is a pending Wait(), it will be cancelled.
+  void Seek(int64_t pos);
+
+  // Returns the current position.
+  int64_t Tell() const { return pos_; }
+
+  // Tries to read |len| bytes and advance position.
+  // Returns number of bytes read.
+  // If there is a pending Wait(), it will be cancelled.
+  int64_t TryRead(unsigned char *data, int64_t len);
+
+  // Wait until |len| bytes are available for reading.
+  // Returns net::OK if |len| bytes are already available, otherwise it will
+  // return net::ERR_IO_PENDING and call |cb| at some point later.
+  // |len| must be smaller or equal to max_buffer_forward.
+  int Wait(int64_t len, base::Closure cb);
+
+  // Set how much data we try to preload into the cache ahead of our current
+  // location. Normally, we preload until we have preload_high bytes, then
+  // stop until we fall below preload_low bytes. Note that preload can be
+  // set higher than max_buffer_forward, but the result is usually that
+  // some blocks will be freed between the current position and the preload
+  // position.
+  void SetPreload(int64_t preload_high, int64_t preload_low);
+
+  // Adjusts max_buffer_forward and max_buffer_backsard.
+  void SetMaxBuffer(int64_t backward, int64_t forward);
+
+  // Returns true if we are currently loading data.
+  bool IsLoading() const;
+
+  // Reader implementation.
+  void NotifyAvailableRange(
+      const Range<MultiBufferBlockId>& range) override;
+  void UpdateURLId(MultiBufferUrlId old_url_id,
+                   MultiBufferUrlId new_url_id) override;
+
+  // Getters
+  MultiBufferUrlId GetURLId() const { return url_id_; }
+  int64_t preload_high() const { return preload_high_; }
+  int64_t preload_low() const { return preload_low_; }
+
+ private:
+  // Returns the block for a particular byte position.
+  MultiBufferBlockId block(int64_t byte_pos) const;
+
+  // Returns the block for a particular byte position, rounding up.
+  MultiBufferBlockId block_ceil(int64_t byte_pos) const;
+
+  // Check if wait operation can complete now.
+  // Returns false if this object was destroyed while calling the
+  // wait callback, true otherwise.
+  bool CheckWait();
+
+  // Increment preload position if data has been added to the buffer.
+  // Returns false if this object was destroyed, true otherwise.
+  bool IncrementPreloadPos();
+
+  // The ID of the url we're accessing.
+  MultiBufferUrlId url_id_;
+
+  // We're not interested in reading past this position.
+  int64_t end_;
+
+  // Defer reading once we have this much data.
+  int64_t preload_high_;
+  // Stop deferring once we have this much data.
+  int64_t preload_low_;
+
+  // Pin this much data in the cache from the current position.
+  int64_t max_buffer_forward_;
+  int64_t max_buffer_backward_;
+
+  // Current position in bytes.
+  int64_t pos_;
+
+  // [block(pos_)..preload_pos_) are known to be in the cache.
+  // preload_pos_ is only allowed to point to a filled
+  // cache position if it is equal to end_ or pos_+preload_.
+  // This is a pointer to a slot in the cache, so the unit is
+  // blocks.
+  MultiBufferBlockId preload_pos_;
+
+  // True if we've requested data from the cache by calling WaitFor().
+  bool loading_;
+
+  // When Available() > current_wait_size_ we call cb_.
+  int64_t current_wait_size_;
+  base::Closure cb_;
+
+  // Progress callback.
+  base::Callback<void(int64_t, int64_t)> progress_callback_;
+
+  // Hack to let us detect if we've been deleted when calling a callback.
+  bool* clear_on_delete_;
+
+  // The multibuffer we're wrapping, not owned.
+  MultiBuffer* multibuffer_;
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BLINK_MULTIBUFFER_H_