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.
+const int kBufferSize = 1024*1024; // 1MB to minimize transaction costs.
+
+}  // namespace
+
+ReadaheadFileStreamReader::ReadaheadFileStreamReader(
+    FileStreamReader* underlying)
+    : underlying_(underlying),
+      underlying_error_(0),
+      current_offset_(0),
+      underlying_has_pending_read_(false),
+      weak_factory_(this) {
+}
+
+ReadaheadFileStreamReader::~ReadaheadFileStreamReader() {}
+
+int ReadaheadFileStreamReader::Read(
+    net::IOBuffer* buf, int buf_len, const net::CompletionCallback& callback) {
+  DCHECK(!pending_read_.get());
+
+  // Dispatch a request to fill up our buffers if needed.
+  if (underlying_.get() && !underlying_has_pending_read_ &&
+      buffers_.size() < kDesiredNumberOfBuffers) {
+    ReadFromUnderlying();
+  }
+
+  // Consume from our existing buffers and return immediately if possible.
+  if (!buffers_.empty())
+    return ConsumeFromBuffer(buf, buf_len);
+
+  // Pass through the stored underlying error or EOF if existent.
+  if (!underlying_.get())
+    return underlying_error_;
+
+  // We are waiting for an underlying read to complete, so save the request.

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

DCHECK(!pending_read.get());

[tommycli, 2014/03/04 00:39:40]

Done.

+  pending_read_.reset(new Request(buf, buf_len, callback));
+  return net::ERR_IO_PENDING;
+}
+
+int64 ReadaheadFileStreamReader::GetLength(
+    const net::Int64CompletionCallback& callback) {
+  return underlying_->GetLength(callback);
+}
+
+ReadaheadFileStreamReader::Request::Request(
+    net::IOBuffer* buf, int buf_len, const net::CompletionCallback& callback)
+    : buf(buf), buf_len(buf_len), callback(callback) {

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

http://google-styleguide.googlecode.com/svn/trunk/cppguide.xml?showone=Constr...

Multiple lines - I know git cl format does it this way...

[tommycli, 2014/03/04 00:39:40]

Done.

+}
+
+ReadaheadFileStreamReader::Request::~Request() {}
+
+int ReadaheadFileStreamReader::ConsumeFromBuffer(net::IOBuffer* buf,
+                                                 int buf_len) {
+  DCHECK(!buffers_.empty());
+

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

Probably want a while loop here - there might only be one byte left in the
current buffer with a request for 2MB, which should consume both buffers.

[tommycli, 2014/03/04 00:39:40]

Done.

+  net::DrainableIOBuffer* head_buffer = buffers_.front().get();
+
+  DCHECK(head_buffer->BytesRemaining() > 0);
+
+  int copy_len = std::min(buf_len, head_buffer->BytesRemaining());
+  std::copy(head_buffer->data(), head_buffer->data() + copy_len, buf->data());
+  head_buffer->DidConsume(copy_len);
+
+  if (head_buffer->BytesRemaining() == 0) {
+    buffers_.pop();
+
+    // Get a new buffer to replace the one we just used up.
+    if (!underlying_has_pending_read_ && underlying_.get())
+      ReadFromUnderlying();
+  }
+
+  return copy_len;
+}
+
+void ReadaheadFileStreamReader::ReadFromUnderlying() {
+  DCHECK(!underlying_has_pending_read_);
+  underlying_has_pending_read_ = true;
+
+  scoped_refptr<net::IOBuffer> buf(new net::IOBuffer(kBufferSize));

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

Why not make a DrainableIOBuffer here instead of copying the data in the
completion callback?

[tommycli, 2014/03/04 00:39:40]

Added a comment. DrainableIOBuffer just wraps the buffer without a copy right?

[vandebo (ex-Chrome), 2014/03/04 18:51:32]

Right.  I figured that out, but forgot to remove my comment.  The comment you
added isn't necessary.

[tommycli, 2014/03/04 21:14:58]

Done.

+  int result = underlying_->Read(
+      buf,
+      kBufferSize,
+      base::Bind(&ReadaheadFileStreamReader::OnFinishReadFromUnderlying,
+                 weak_factory_.GetWeakPtr(), buf));
+
+  if (result != net::ERR_IO_PENDING) {

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

If we get ERR_IO_PENDING, you'll leave the request hanging

[tommycli, 2014/03/04 00:39:40]

As intended. If we get ERR_IO_PENDING, OnFinishReadFromSource doesn't happen
until later.

If there isn't enough buffer to complete the request, it gets saved in
pending_read_.

+    OnFinishReadFromUnderlying(buf, result);
+  }
+}
+
+void ReadaheadFileStreamReader::OnFinishReadFromUnderlying(net::IOBuffer* buf,
+                                                           int result) {
+  DCHECK(result != net::ERR_IO_PENDING);
+  DCHECK(underlying_has_pending_read_);
+  underlying_has_pending_read_ = false;
+
+  if (result <= 0) {
+    underlying_.reset();
+    underlying_error_ = result;

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

Doesn't this leave request hanging?  You probably want a FinishRequest method
that you can call anywhere you encounter an error condition.

[tommycli, 2014/03/04 00:39:40]

Done. I didn't use a FinishRequest, as I couldn't find another place to use it.

+    return;
+  }
+
+  scoped_refptr<net::DrainableIOBuffer> drainable_buffer(
+      new net::DrainableIOBuffer(buf, result));
+  buffers_.push(drainable_buffer);
+

[vandebo (ex-Chrome), 2014/03/03 22:24:54]

Should this kick off another read if buffers_.size < kDesiredNumberOfBuffers ?

[tommycli, 2014/03/04 00:39:40]

The ConsumeFromBuffer call within this if-block should kick off another read if
necessary.

+  // If there's a read request waiting for the underlying FileStreamReader to
+  // finish reading, fulfill that request now.
+  if (pending_read_.get()) {
+    // Free up pending read immediately, as the read completion callback often
+    // dispatches another read.
+    scoped_ptr<Request> request(pending_read_.Pass());
+
+    request->callback.Run(ConsumeFromBuffer(request->buf, request->buf_len));
+  }
+}