MTP Streaming: Readahead Buffer

chrome/browser/media_galleries/fileapi/readahead_file_stream_reader.cc

+// 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.

[vandebo (ex-Chrome), 2014/03/05 17:31:31]

nit: two spaces before a comment.

[tommycli, 2014/03/05 18:07:29]

Done.

+const int kBufferSize = 1024*1024; // 1MB to minimize transaction costs.
+
+}  // namespace
+
+ReadaheadFileStreamReader::ReadaheadFileStreamReader(FileStreamReader* source)
+    : source_(source),
+      source_error_(0),
+      source_has_pending_read_(false),
+      current_offset_(0),
+      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();
+
+  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);

[vandebo (ex-Chrome), 2014/03/05 17:31:31]

You don't actually use pending_sink_buffer_ as a DrainableIOBuffer.  Two
options: just store the IO buffer and len, change FinishRead... to take a
DrainableIOBuffer.

[tommycli, 2014/03/05 18:07:29]

Done.

+    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()) {
+    net::DrainableIOBuffer* source_buffer = buffers_.front().get();
+
+    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();
+    }
+  }
+
+  return sink->BytesConsumed();
+}
+
+void ReadaheadFileStreamReader::ReadFromSourceIfNeeded() {
+  if (!source_.get() || source_has_pending_read_ ||
+      buffers_.size() >= kDesiredNumberOfBuffers) {
+    return;
+  }
+
+  source_has_pending_read_ = true;
+
+  scoped_refptr<net::IOBuffer> buf(new net::IOBuffer(kBufferSize));
+  int result = source_->Read(
+      buf,
+      kBufferSize,
+      base::Bind(&ReadaheadFileStreamReader::OnFinishReadFromSource,
+                 weak_factory_.GetWeakPtr(), buf));
+
+  if (result != net::ERR_IO_PENDING) {

[vandebo (ex-Chrome), 2014/03/05 17:31:31]

nit: predominate style in this directory is no {}'s on one line if's

[tommycli, 2014/03/05 18:07:29]

Done.

+    OnFinishReadFromSource(buf, result);
+  }
+}
+
+void ReadaheadFileStreamReader::OnFinishReadFromSource(net::IOBuffer* 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) {
+    scoped_refptr<net::DrainableIOBuffer> drainable_buffer(
+        new net::DrainableIOBuffer(buf, result));
+    buffers_.push(drainable_buffer);

[vandebo (ex-Chrome), 2014/03/05 17:31:31]

This should probably call ReadFromSourceIfNeeded(). I think it all works out
fine right now because kDesiredNumberOfBuffers is only 2, but if it were set to
3, things wouldn't work right and it doesn't hurt anything to call it here. 
It's also easier to understand.

[tommycli, 2014/03/05 18:07:29]

Done.

+  } 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());
+
+    // Free the pending callback before running it, as the callback often
+    // dispatches another read.
+    scoped_refptr<net::DrainableIOBuffer> sink = pending_sink_buffer_;
+    pending_sink_buffer_ = NULL;
+    net::CompletionCallback completion_callback = pending_read_callback_;
+    pending_read_callback_.Reset();
+
+    completion_callback.Run(
+        FinishReadFromCacheOrStoredError(sink, sink->BytesRemaining()));
+  }
+}