BufferedDataSource uses media::SeekableBuffer

Use SeekableBuffer as internal cache for BufferedResourceLoader.
By switch to SeekableBuffer that supports short seek within
buffered data in both backward and forward direciton. The amount
of range request is greatly reduced.

Review URL: http://codereview.chromium.org/115453

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@16456 0039d316-1c4b-4281-b951-d872f2087c98

2 files changed, 119 insertions, 173 deletions

diff --git a/chrome/renderer/media/buffered_data_source.cc b/chrome/renderer/media/buffered_data_source.cc
index 23eb7e7..8eaba41 100644
--- a/chrome/renderer/media/buffered_data_source.cc
+++ b/chrome/renderer/media/buffered_data_source.cc
@@ -25,6 +25,12 @@ const int64 kPositionNotSpecified = -1;
 const int kHttpOK = 200;
 const int kHttpPartialContent = 206;
 
+// Backward capacity of the buffer, by default 2MB.
+const size_t kBackwardCapcity = 2048000;
+
+// Forward capacity of the buffer, by default 10MB.
+const size_t kForwardCapacity = 10240000;
+
 // A helper method that accepts only HTTP, HTTPS and FILE protocol.
 bool IsSchemeSupported(const GURL& url) {
   return url.SchemeIs(kHttpScheme) ||
@@ -44,9 +50,7 @@ BufferedResourceLoader::BufferedResourceLoader(int32 routing_id,
       bridge_(NULL),
       offset_(0),
       content_length_(kPositionNotSpecified),
-      buffered_bytes_(0),
-      buffer_limit_(10240000),  // By default 10MB.
-      buffer_event_(false, false),
+      buffer_(new media::SeekableBuffer(kBackwardCapcity, kForwardCapacity)),
       deferred_(false),
       stopped_(false),
       completed_(false),
@@ -56,11 +60,11 @@ BufferedResourceLoader::BufferedResourceLoader(int32 routing_id,
       url_(url),
       first_byte_position_(first_byte_position),
       last_byte_position_(last_byte_position),
-      render_loop_(RenderThread::current()->message_loop()) {
+      render_loop_(RenderThread::current()->message_loop()),
+      buffer_available_(&lock_) {
 }
 
 BufferedResourceLoader::~BufferedResourceLoader() {
-  STLDeleteElements<BufferQueue>(&buffers_);
 }
 
 int BufferedResourceLoader::Start(net::CompletionCallback* start_callback) {
@@ -112,22 +116,17 @@ int BufferedResourceLoader::Start(net::CompletionCallback* start_callback) {
   // Start() may get called on demuxer thread while Stop() is called on
   // pipeline thread, so we want to protect the posting of OnStart() task
   // with a lock.
-  bool task_posted = false;
   {
     AutoLock auto_lock(lock_);
     if (!stopped_) {
-      task_posted =  true;
       render_loop_->PostTask(FROM_HERE,
           NewRunnableMethod(this, &BufferedResourceLoader::OnStart));
-    }
-  }
 
-  // Wait for response to arrive we don't have an async start.
-  if (task_posted && !async_start_)
-    buffer_event_.Wait();
+      // Wait for response to arrive if we don't have an async start.
+      if (!async_start_)
+        buffer_available_.Wait();
+    }
 
-  {
-    AutoLock auto_lock(lock_);
     // We may have stopped because of a bad response from the server.
     if (stopped_)
       return net::ERR_ABORTED;
@@ -140,59 +139,43 @@ int BufferedResourceLoader::Start(net::CompletionCallback* start_callback) {
 }
 
 void BufferedResourceLoader::Stop() {
-  BufferQueue delete_queue;
   {
     AutoLock auto_lock(lock_);
     stopped_ = true;
-    // Use of |buffers_| is protected by the lock, we can destroy it safely.
-    delete_queue.swap(buffers_);
-    buffers_.clear();
+    buffer_.reset();
+
+    // Wakes up the waiting thread so they can catch the stop signal.
+    buffer_available_.Signal();
   }
-  STLDeleteElements<BufferQueue>(&delete_queue);
 
-  // Wakes up the waiting thread so they can catch the stop signal.
   render_loop_->PostTask(FROM_HERE,
       NewRunnableMethod(this, &BufferedResourceLoader::OnDestroy));
-  buffer_event_.Signal();
 }
 
 size_t BufferedResourceLoader::Read(uint8* data, size_t size) {
   size_t taken = 0;
-  while (taken < size) {
-    Buffer* buffer = NULL;
-    {
-      AutoLock auto_lock(lock_);
+  {
+    AutoLock auto_lock(lock_);
+    while (taken < size) {
+      // If stopped or the request has been completed, break the loop.
       if (stopped_)
         break;
-      if (!buffers_.empty())
-        buffer = buffers_.front();
-      else if (completed_)
+      // Read into |data|.
+      size_t unread_bytes = size - taken;
+      size_t bytes_read = buffer_->Read(unread_bytes, data + taken);
+      taken += bytes_read;
+      DCHECK_LE(taken, size);
+      if (taken == size)
         break;
-    }
-    if (buffer) {
-      size_t copy = std::min(size - taken, buffer->size - buffer->taken);
-      memcpy(data + taken, buffer->data.get() + buffer->taken, copy);
-      taken += copy;
-      buffer->taken += copy;
-      if (buffer->taken == buffer->size) {
-        // The buffer has been consumed, remove it.
-        {
-          AutoLock auto_lock(lock_);
-          buffers_.pop_front();
-        }
-        delete buffer;
-      }
-    } else {
-      buffer_event_.Wait();
+      if (completed_ && bytes_read < unread_bytes)
+        break;
+      buffer_available_.Wait();
     }
   }
+
+  // Adjust the offset and disable defer loading if needed.
   if (taken > 0) {
     offset_ += taken;
-    {
-      AutoLock auto_lock(lock_);
-      buffered_bytes_ -= taken;
-      DCHECK(buffered_bytes_ >= 0);
-    }
     if (ShouldDisableDefer()) {
       AutoLock auto_lock(lock_);
       if (!stopped_) {
@@ -205,53 +188,55 @@ size_t BufferedResourceLoader::Read(uint8* data, size_t size) {
   return taken;
 }
 
-bool BufferedResourceLoader::SeekForward(int64 position) {
+bool BufferedResourceLoader::Seek(int64 position) {
   // Use of |offset_| is safe without a lock, because it's modified only in
-  // Read() and this method after Start(). Read() and SeekForward() happens
+  // Read() and this method after Start(). Read() and Seek() happens
   // on the same thread.
-  // Seeking backward.
-  if (position < offset_)
-    return false;
-  // Done seeking forward.
-  else if (position == offset_)
+  if (position == offset_)
     return true;
 
+  // Use the conditions to avoid overflow, since |position| and |offset_| are
+  // int64, their difference can be greater than an int32.
+  if (position > offset_ + kint32max)
+    return false;
+  else if (position < offset_ + kint32min)
+    return false;
+
+  int32 offset = static_cast<int32>(position - offset_);
+  // Backward data are served directly from the buffer and will not be
+  // downloaded in the future so we perform a backward seek now.
+  if (offset < 0) {
+    AutoLock auto_lock(lock_);
+    if (buffer_->Seek(offset)) {
+      offset_ = position;
+      return true;
+    }
+    return false;
+  }
+
+  // If we are seeking too far ahead that the buffer cannot serve, return false.
+  // We only perform this check for forward seeking because we don't want to
+  // wait too long for data very far ahead to be downloaded, and of course we
+  // will never be able to seek to that position.
+  if (position >= offset_ + buffer_->forward_capacity())
+    return false;
+
+  // Perform seeking forward until we get to the offset.
+  AutoLock auto_lock(lock_);
   while(true) {
-    {
-      AutoLock auto_lock(lock_);
-      // Loader has stopped.
-      if (stopped_)
-        return false;
-      // Seek position exceeds bufferable range, buffer_limit_ can be changed.
-      if (position >= offset_ + buffer_limit_)
-        return false;
-      // Response completed and seek position exceeds buffered range.
-      if (completed_ && position >= offset_ + buffered_bytes_)
-        return false;
-
-      if (!buffers_.empty()) {
-        Buffer* buffer = buffers_.front();
-        int64 bytes_to_take = position - offset_;
-        if (!buffers_.empty()) {
-          size_t available_bytes_in_buffer = buffer->size - buffer->taken;
-          size_t taken = 0;
-          if (available_bytes_in_buffer <= bytes_to_take) {
-            taken = available_bytes_in_buffer;
-            buffers_.pop_front();
-            delete buffer;
-          } else {
-            taken = static_cast<size_t>(bytes_to_take);
-            buffer->taken += taken;
-          }
-          offset_ += taken;
-          if (bytes_to_take == taken)
-            return true;
-        }
-        continue;
-      }
+    // Loader has stopped.
+    if (stopped_)
+      return false;
+    // Response completed and seek position exceeds buffered range.
+    if (completed_ && position >= offset_ + buffer_->forward_bytes())
+      return false;
+    if (buffer_->Seek(offset)) {
+      offset_ = position;
+      break;
     }
-    buffer_event_.Wait();
+    buffer_available_.Wait();
   }
+  return true;
 }
 
 int64 BufferedResourceLoader::GetOffset() {
@@ -259,35 +244,6 @@ int64 BufferedResourceLoader::GetOffset() {
   return offset_;
 }
 
-int64 BufferedResourceLoader::GetBufferLimit() {
-  AutoLock auto_lock(lock_);
-  return buffer_limit_;
-}
-
-void BufferedResourceLoader::SetBufferLimit(size_t buffer_limit) {
-  {
-    AutoLock auto_lock(lock_);
-    buffer_limit_ = buffer_limit;
-  }
-
-  if (ShouldDisableDefer()) {
-    AutoLock auto_lock(lock_);
-    if (!stopped_) {
-      render_loop_->PostTask(FROM_HERE,
-          NewRunnableMethod(this,
-                            &BufferedResourceLoader::OnDisableDeferLoading));
-    }
-  }
-  if (ShouldEnableDefer()) {
-    AutoLock auto_lock(lock_);
-    if (!stopped_) {
-      render_loop_->PostTask(FROM_HERE,
-          NewRunnableMethod(this,
-                            &BufferedResourceLoader::OnEnableDeferLoading));
-    }
-  }
-}
-
 size_t BufferedResourceLoader::GetTimeout() {
   // TODO(hclam): implement.
   return 0;
@@ -351,14 +307,15 @@ void BufferedResourceLoader::OnReceivedResponse(
     // We only care about the first byte position if it's given by the server.
     if (first_byte_position != kPositionNotSpecified)
       offset_ = first_byte_position;
+
+    // If this is not an asynchronous start, signal the thread called Start().
+    if (!async_start_)
+      buffer_available_.Signal();
   }
 
-  // If we have started asynchronously we just need to invoke the callback or
-  // we need to signal the Start() method to wake up.
+  // If we have started asynchronously we need to invoke the callback.
   if (async_start_)
     InvokeAndResetStartCallback(net::OK);
-  else
-    buffer_event_.Signal();
 }
 
 void BufferedResourceLoader::OnReceivedData(const char* data, int len) {
@@ -383,31 +340,26 @@ void BufferedResourceLoader::OnCompletedRequest(
 /////////////////////////////////////////////////////////////////////////////
 // BufferedResourceLoader, private
 void BufferedResourceLoader::AppendToBuffer(const uint8* data, size_t size) {
-  {
-    AutoLock auto_lock(lock_);
-    if (!stopped_) {
-      Buffer* buffer = new Buffer(size);
-      memcpy(buffer->data.get(), data, size);
-      buffers_.push_back(buffer);
-      buffered_bytes_ += size;
-    }
-  }
-  buffer_event_.Signal();
+  AutoLock auto_lock(lock_);
+  if (!stopped_)
+    buffer_->Append(size, data);
+  buffer_available_.Signal();
 }
 
 void BufferedResourceLoader::SignalComplete() {
-  {
-    AutoLock auto_lock(lock_);
-    completed_ = true;
-  }
-  buffer_event_.Signal();
+  AutoLock auto_lock(lock_);
+  completed_ = true;
+  buffer_available_.Signal();
 }
 
 bool BufferedResourceLoader::ShouldEnableDefer() {
   AutoLock auto_lock(lock_);
-  if (deferred_) {
+
+  // If the resource loader has been stopped, we should not use |buffer_|.
+  if (stopped_)
     return false;
-  } else if (buffered_bytes_ >= buffer_limit_) {
+
+  if (!deferred_ && buffer_->forward_bytes() >= buffer_->forward_capacity()) {
     deferred_ = true;
     return true;
   }
@@ -416,10 +368,16 @@ bool BufferedResourceLoader::ShouldEnableDefer() {
 
 bool BufferedResourceLoader::ShouldDisableDefer() {
   AutoLock auto_lock(lock_);
-  if (deferred_ && buffered_bytes_ < buffer_limit_ / 2)
-    return true;
-  else
+
+  // If the resource loader has been stopped, we should not use |buffer_|.
+  if (stopped_)
     return false;
+
+  if (deferred_ && buffer_->forward_bytes() < buffer_->forward_capacity() / 2) {
+    deferred_ = false;
+    return true;
+  }
+  return false;
 }
 
 void BufferedResourceLoader::OnStart() {
@@ -431,6 +389,7 @@ void BufferedResourceLoader::OnStart() {
 void BufferedResourceLoader::OnDestroy() {
   DCHECK(MessageLoop::current() == render_loop_);
   if (bridge_.get()) {
+    // Cancel the resource request.
     bridge_->Cancel();
     bridge_.reset();
   }
@@ -527,7 +486,7 @@ bool BufferedDataSource::Initialize(const std::string& url) {
 
 size_t BufferedDataSource::Read(uint8* data, size_t size) {
   // We try two times here:
-  // 1. Use the existing resource loader to seek forward and read from it.
+  // 1. Use the existing resource loader to seek and read from it.
   // 2. If any of the above operations failed, we create a new resource loader
   //    starting with a new range. Goto 1.
   // TODO(hclam): change the logic here to do connection recovery and allow a
@@ -539,7 +498,7 @@ size_t BufferedDataSource::Read(uint8* data, size_t size) {
       resource_loader = buffered_resource_loader_;
     }
 
-    if (resource_loader && resource_loader->SeekForward(position_)) {
+    if (resource_loader && resource_loader->Seek(position_)) {
       size_t read = resource_loader->Read(data, size);
       if (read >= 0) {
         position_ += read;
@@ -592,8 +551,13 @@ bool BufferedDataSource::GetPosition(int64* position_out) {
 }
 
 bool BufferedDataSource::SetPosition(int64 position) {
-  position_ = position;
-  return true;
+  // |total_bytes_| can be -1 for pure streaming. There may be a problem with
+  // seeking for this case.
+  if (position < total_bytes_) {
+    position_ = position;
+    return true;
+  }
+  return false;
 }
 
 bool BufferedDataSource::GetSize(int64* size_out) {
diff --git a/chrome/renderer/media/buffered_data_source.h b/chrome/renderer/media/buffered_data_source.h
index 9072275..f3b3390 100644
--- a/chrome/renderer/media/buffered_data_source.h
+++ b/chrome/renderer/media/buffered_data_source.h
@@ -10,11 +10,12 @@
 
 #include "base/lock.h"
 #include "base/scoped_ptr.h"
-#include "base/waitable_event.h"
+#include "base/condition_variable.h"
 #include "media/base/factory.h"
 #include "media/base/filters.h"
 #include "media/base/media_format.h"
 #include "media/base/pipeline.h"
+#include "media/base/seekable_buffer.h"
 #include "net/base/completion_callback.h"
 #include "net/base/file_stream.h"
 #include "webkit/glue/resource_loader_bridge.h"
@@ -59,24 +60,20 @@ class BufferedResourceLoader :
   // the current position referred by calling GetOffset(). This method call is
   // synchronous, it returns only the required amount of bytes is read, the
   // loader is stopped, this resource loading has completed or the read has
-  // timed out. Read() and SeekForward() cannot be called concurrently.
+  // timed out. Read() and Seek() cannot be called concurrently.
   size_t Read(uint8* buffer, size_t size);
 
-  // Seek forward to |position| in bytes in the entire instance of the media
+  // Seek to |position| in bytes in the entire instance of the media
   // object, returns true if successful. If the seek operation cannot be
   // performed because it's seeking backward, the loader has been stopped,
   // the seek |position| exceed bufferable range or the seek operation has
   // timed out, returns false.
-  // There cannot be SeekForward() while another thread is calling Read().
-  bool SeekForward(int64 position);
+  // There cannot be Seek() while another thread is calling Read().
+  bool Seek(int64 position);
 
   // Returns the position in bytes that this loader is downloading from.
   int64 GetOffset();
 
-  // Gets and sets the buffering limit of this loader.
-  int64 GetBufferLimit();
-  void SetBufferLimit(size_t buffe_limit);
-
   // Gets and sets the timeout for the synchronous operations.
   size_t GetTimeout();
   void SetTimeout(size_t milliseconds);
@@ -113,26 +110,12 @@ class BufferedResourceLoader :
 
   void InvokeAndResetStartCallback(int error);
 
-  struct Buffer {
-    Buffer(size_t len) : taken(0), size(len), data(new uint8[len]) { }
-
-    // The amount of buffer in bytes consumed in this buffer starting from
-    // index 0.
-    size_t taken;
-    size_t size;
-    scoped_array<uint8> data;
-  };
-
   scoped_ptr<net::CompletionCallback> start_callback_;
   scoped_ptr<webkit_glue::ResourceLoaderBridge> bridge_;
   int64 offset_;
   int64 content_length_;
 
-  typedef std::deque<Buffer*> BufferQueue;
-  BufferQueue buffers_;
-  size_t buffered_bytes_;
-  size_t buffer_limit_;
-  base::WaitableEvent buffer_event_;
+  scoped_ptr<media::SeekableBuffer> buffer_;
 
   bool deferred_;
   bool stopped_;
@@ -147,13 +130,12 @@ class BufferedResourceLoader :
 
   MessageLoop* render_loop_;
   // A lock that protects usage of the following members:
-  // - buffers_
-  // - buffered_bytes_
-  // - buffered_limit_
+  // - buffer_
   // - deferred_
   // - stopped_
   // - completed_
   Lock lock_;
+  ConditionVariable buffer_available_;
 
   DISALLOW_COPY_AND_ASSIGN(BufferedResourceLoader);
 };