MTP Streaming: Readahead Buffer

chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc

Index: chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc
diff --git a/chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc b/chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc
new file mode 100644
index 0000000000000000000000000000000000000000..2c9738f35bc8d5e0b125559fec840e1fb3d62d00
--- /dev/null
+++ b/chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc
@@ -0,0 +1,159 @@
+// Copyright 2014 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 "chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.h"
+
+#include <algorithm>
+
+#include "base/message_loop/message_loop.h"
+#include "base/numerics/safe_conversions.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+using webkit_blob::FileStreamReader;
+
+namespace {
+
+const int kDesiredNumberOfBuffers = 2; // So we are always one buffer ahead.
+const int kBufferSize = 1024*1024; // 1MB to minimize transaction costs.
+
+}  // namespace
+
+ReadaheadFileStreamReader::ReadaheadFileStreamReader(FileStreamReader* source)
+    : source_(source),
+      source_error_(0),
+      current_offset_(0),
+      source_has_pending_read_(false),
+      weak_factory_(this) {
+}
+
+ReadaheadFileStreamReader::~ReadaheadFileStreamReader() {}
+
+int ReadaheadFileStreamReader::Read(
+    net::IOBuffer* buf, int buf_len, const net::CompletionCallback& callback) {
+  DCHECK(!pending_sink_buffer_.get());
+  DCHECK(pending_read_callback_.is_null());
+
+  ReadFromSourceIfNeeded(scoped_refptr<net::DrainableIOBuffer>());
+
+  int result = FinishReadFromCacheOrStoredError(buf, buf_len);
+
+  // We are waiting for an source read to complete, so save the request.
+  if (result == net::ERR_IO_PENDING) {
+    DCHECK(!pending_sink_buffer_.get());
+    DCHECK(pending_read_callback_.is_null());
+    pending_sink_buffer_ = new net::DrainableIOBuffer(buf, buf_len);
+    pending_read_callback_ = callback;
+  }
+
+  return result;
+}
+
+int64 ReadaheadFileStreamReader::GetLength(
+    const net::Int64CompletionCallback& callback) {
+  return source_->GetLength(callback);
+}
+
+int ReadaheadFileStreamReader::FinishReadFromCacheOrStoredError(
+    net::IOBuffer* buf, int buf_len) {
+  // If we don't have any ready cache, return the pending read code, or
+  // the stored error code.
+  if (buffers_.empty()) {
+    if (source_.get()) {
+      DCHECK(source_has_pending_read_);
+      return net::ERR_IO_PENDING;
+    } else {
+      return source_error_;
+    }
+  }
+
+  scoped_refptr<net::DrainableIOBuffer> sink(
+      new net::DrainableIOBuffer(buf, buf_len));
+
+  while (sink->BytesRemaining() > 0 && !buffers_.empty()) {
+    scoped_refptr<net::DrainableIOBuffer> source_buffer = buffers_.front();
+
+    DCHECK(source_buffer->BytesRemaining() > 0);
+
+    int copy_len = std::min(source_buffer->BytesRemaining(),
+                            sink->BytesRemaining());
+    std::copy(source_buffer->data(), source_buffer->data() + copy_len,
+              sink->data());
+
+    source_buffer->DidConsume(copy_len);
+    sink->DidConsume(copy_len);
+
+    if (source_buffer->BytesRemaining() == 0) {
+      buffers_.pop();
+
+      // Get a new buffer to replace the one we just used up.
+      ReadFromSourceIfNeeded(source_buffer);
+    }
+  }
+
+  return sink->BytesConsumed();
+}
+
+void ReadaheadFileStreamReader::ReadFromSourceIfNeeded(
+    scoped_refptr<net::DrainableIOBuffer> reuse) {
+  if (!source_.get() || source_has_pending_read_ ||
+      buffers_.size() >= kDesiredNumberOfBuffers) {
+    return;
+  }
+
+  source_has_pending_read_ = true;
+
+  scoped_refptr<net::DrainableIOBuffer> buf;
+  if (reuse.get()) {
+    reuse->SetOffset(0);
+    reuse->SetSize(kBufferSize);
+    buf = reuse;
+  } else {
+    buf = new net::DrainableIOBuffer(
+        new net::IOBuffer(kBufferSize), kBufferSize);
+  }
+
+  int result = source_->Read(
+      buf,
+      kBufferSize,
+      base::Bind(&ReadaheadFileStreamReader::OnFinishReadFromSource,
+                 weak_factory_.GetWeakPtr(), buf));
+
+  if (result != net::ERR_IO_PENDING) {
+    OnFinishReadFromSource(buf, result);
+  }
+}
+
+void ReadaheadFileStreamReader::OnFinishReadFromSource(
+    net::DrainableIOBuffer* buf, int result) {
+  DCHECK(result != net::ERR_IO_PENDING);
+  DCHECK(source_has_pending_read_);
+  source_has_pending_read_ = false;
+
+  // Either store the data read from |source_|, or store the error code.
+  if (result > 0) {
+    buf->SetSize(result);
+    buffers_.push(make_scoped_refptr(buf));
+  } else {
+    source_.reset();
+    source_error_ = result;
+  }
+
+  // If there's a read request waiting for the source FileStreamReader to
+  // finish reading, fulfill that request now from the cache or stored error.
+  if (pending_sink_buffer_.get()) {
+    DCHECK(!pending_read_callback_.is_null());
+
+    int result = FinishReadFromCacheOrStoredError(
+        pending_sink_buffer_, pending_sink_buffer_->BytesRemaining());
+    DCHECK_NE(net::ERR_IO_PENDING, result);

[vandebo (ex-Chrome), 2014/03/04 21:55:12]

What happens if you exhaust the current buffer and then get an EOF on the
replacement request?  I think you just need to move 154 and 155 above this and
then not DCHECK?

[tommycli, 2014/03/04 23:03:44]

Done.

+
+    // Free the pending callback before running it, as the callback often
+    // dispatches another read.
+    pending_sink_buffer_ = NULL;
+    net::CompletionCallback completion_callback = pending_read_callback_;
+    pending_read_callback_.Reset();
+    completion_callback.Run(result);
+  }
+}