Multi reader/writer cache/buffer

media/blink/multibuffer.h

Index: media/blink/multibuffer.h
diff --git a/media/blink/multibuffer.h b/media/blink/multibuffer.h
new file mode 100644
index 0000000000000000000000000000000000000000..fbce55497ac2817bca170e66af1329d0cae3e543
--- /dev/null
+++ b/media/blink/multibuffer.h
@@ -0,0 +1,222 @@
+// 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 <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/waiter_index.h"
+
+namespace media {
+
+const int kMaxFreesPerAdd = 10;
+const int kMaxWaitForWriterOffset = 5;
+
+// 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.
+class MultiBuffer {

[liberato (no reviews please), 2015/06/09 15:00:54]

the name "MultiBuffer" is a little confusing.  maybe something that contains the
word "cache", else it sounds like it should be a thing with getStart(),
getLength(), and little else.

[hubbe, 2015/06/09 21:23:48]

It's meant to replace multiple preloading buffers.
I couldn't come up with a GOOD name, so I picked a short unique one.

Perhaps StreamingBlockCache would be better though?

+ public:
+  explicit MultiBuffer(int32_t block_size_shift);
+  virtual ~MultiBuffer();
+
+  typedef std::map<int32_t, scoped_refptr<DataBuffer> > DataMap;
+
+  // Adds a writer. This writer becomes responsible
+  // for writing data at |pos| and forwards.
+  // Takes ownership of |writer|.
+  void AddWriter(int32_t pos, Waiter* writer);

[liberato (no reviews please), 2015/06/09 15:00:53]

are all the positions in byte offsets or block offsets?

also not sure that int32 is the right type for it.  might be a good place for a
typedef to make its interpretation clearer.

[hubbe, 2015/06/09 21:23:48]

Yes. (the new typedef makes that clear)
Done.

When you say "I'm not sure...", can you please tell me why you're not sure so
that I can address it better?

(With a block size of 32kb, this class can handle 64Tb files....)

+
+  // Remove a writer.
+  // Caller takes ownership of |writer|.
+  void RemoveWriter(int32_t pos, Waiter* writer);
+
+  // Wait for block |pos| to become available.
+  // Starts new Writers as needed.
+  void WaitFor(int32_t pos, Waiter* reader);
+
+  // Let the buffer that we're going to want to read
+  // block |pos|, but we're ok with the current level
+  // of buffering, so no worries.
+  void DeferredWaitFor(int32_t pos, Waiter* reader);

[liberato (no reviews please), 2015/06/09 15:00:53]

why do we need WaitFor vs DeferredWaitFor?  shouldn't the preload / max / min
specs cover this?

[hubbe, 2015/06/09 21:23:48]

This class doesn't know about any preload specs.

+
+  // Stop waiting for block |pos|.
+  // Often followed by a call to WaitFor(pos + 1, ...);
+  void StopWaitFor(int32_t pos, Waiter* reader);
+
+  // Returns true if block |pos| is available in the cache.
+  bool Contains(int32_t pos) const;
+
+  // Returns true if some reader is currently waiting for
+  // block |pos|.
+  bool WantNow(int32_t pos) const;
+
+  // Returns true if some reader is will want block |pos| soon.
+  bool WantSoon(int32_t pos) const;
+
+  // Add a data block to the cache, notifying readers as
+  // needed and prune old blocks if cache is too big.
+  void AddData(int32_t pos, scoped_refptr<DataBuffer> data);
+
+  // 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(int32_t from, int32_t to, int32_t howmuch);
+
+  // Increment max cache size by |size|.
+  void IncrementMaxSize(int32_t size);
+
+  // Accessors.
+  const DataMap& map() const { return data_; }
+  int32_t block_size_shift() const { return block_size_shift_; }
+
+ protected:
+  // Start a new writer at |pos|.
+  // Should call AddWriter().
+  virtual void StartWriter(int32_t pos) = 0;

[liberato (no reviews please), 2015/06/09 15:00:53]

MultiBuffer::StartWriter => new Writer => MultiBuffer::AddWriter().

maybe multibuffer is doing too much -- it seems like it's a complicated cache
data structure, and also trying to control who's accessing it.

maybe this can be split into two parts: the cache data structure (add / contains
/ remove) and the cache manager (start writers / prune).  plus, we might want to
experiment with different eviction or loading strategies, for example.  not sure
where 'pin' goes.

[hubbe, 2015/06/09 21:23:48]

I could break out the cache itself into something like "PinnableBlockCache"
which would contain pinned_, lru_, data_ and PinRange().  That doesn't provide a
whole lot of simplification, and makes it harder to explain why PinRange
traverses the cache backwards.

I'm not sure what you mean by "control who's access it". The idea was to create
a cache where consumers simply ask for blocks and the cache decides how to fill
those requests by instantiating streaming writers. However, I wanted to abstract
out the writers themselves for testing and possible re-use.

+
+ private:
+  // Free elements from cache if needed and possible.
+  void Prune();
+
+  // Current number of blocks.
+  int32_t size_;
+
+  // Max number of blocks.
+  int32_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.
+  WaiterIndex reader_index_;
+
+  // Keeps track of readers that will want data soon.
+  WaiterIndex deferred_reader_index_;
+
+  // Keeps track of what writers are waiting.
+  WaiterIndex writer_index_;
+
+  // The LRU should contain all blocks which are not pinned.
+  LRU<int32_t> lru_;
+
+  // Keeps track of what blocks are pinned. If block p is pinned,
+  // then pinned_[p] > 0. Pinned blocks cannot be freed and are
+  // not present in |lru_|.
+  RangeMap<int32_t, int32_t> pinned_;
+};
+
+// Typically, calling MultiBuffer::StartWriter() will
+// instantiate a subclass of this class. Writers are
+// responsible for writing data into the cache.
+// Writers are owned by the MultiBuffer.
+class MultiBufferWriter : public Waiter {

[liberato (no reviews please), 2015/06/09 15:00:54]

this might be better off in another .h

[hubbe, 2015/06/09 21:23:48]

They really are meant to be used as a set.
Originally this was an internal class in MultiBuffer, but I moved it out because
it seemed easier to read that way.
However, I'll move it if you feel strongly about it.

+  MultiBufferWriter(MultiBuffer* multibuffer, int32_t pos);
+
+  ~MultiBufferWriter() override;
+
+  // Waiter implementation.
+  void Continue() override;
+
+ protected:
+  virtual void SetDeferred(bool deferred) = 0;
+
+  // Write |data| to |pos_| in the cache and update |pos_|.
+  void Write(scoped_refptr<DataBuffer> data);
+
+  int32_t pos_;
+  MultiBuffer* multibuffer_;
+};
+
+// Wrapper for MultiBuffer that offers a simple byte-reading
+// interface with prefetch.
+class MultiBufferReader : public Waiter {
+ public:
+  MultiBufferReader(MultiBuffer* multibuffer,
+                    int64_t start,
+                    int64_t end,
+                    int64_t preload,
+                    int64_t max_buffer_forward,
+                    int64_t max_buffer_backward);
+
+  ~MultiBufferReader() override;
+
+  // Returns number of bytes available for reading.
+  // If the actual number of bytes available is greater
+  // than |preload|, then it might return a number between
+  // |preload| and and the actual number.
+  int64_t Available() const;
+
+  // Seek to a different position.
+  void Seek(int64_t pos);
+
+  // Tries to read |len| bytes and advance position.
+  // Returns true if successful.
+  bool TryRead(unsigned char *data, int64_t len);
+
+  // Wait until |len| bytes are available for reading.
+  void Wait(int64_t len, base::Closure cb);
+
+  // Waiter implementation.
+  void Continue() override;
+
+ private:
+  // Returns the block for a particular byte position.
+  int32_t block(int64_t byte_pos) const;
+
+  // Returns the block for a particular byte position, rounding up.
+  int32_t block_ciel(int64_t byte_pos) const;
+
+  // Check if wait operation can complete now.
+  void CheckWait();
+
+  // Increment preload position if data has been added to the buffer.
+  void IncrementPreloadPos();
+
+  // We're not interested in reading past this position.
+  const int64_t end_;
+
+  // Defer reading once we have this much data.
+  const int64_t preload_;
+
+  // Pin this much data in the cache from the current position.
+  const int64_t max_buffer_forward_;
+  const 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.
+  int32_t preload_pos_;
+
+  // When Available() > current_wait_size_ we call cb_.
+  int64_t current_wait_size_;
+  base::Closure cb_;
+
+  // The multibuffer we're wrapping, not owned.
+  MultiBuffer* multibuffer_;
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BLINK_MULTIBUFFER_H_