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
+
+// Represents an outstanding read request, waiting for the buffer to be filled
+// from the source FileStreamReader.
+struct ReadaheadFileStreamReader::Request {

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

You only have one of these...

There's no need to allocate and free it, so you could make it a class member
instead of a scoped_ptr.

As a class member, the only reason to but it into a struct is to bundle the
things up, but you don't really need that.

If you just inline this class's members, you can use |buf| or callback.is_null()
as an indicator of a pending request.

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

Done.

+  Request(net::IOBuffer* buf, int buf_len,
+          const net::CompletionCallback& callback)
+      : buf(buf),
+        buf_len(buf_len),
+        callback(callback) {
+  }
+
+  ~Request() {}
+
+  scoped_refptr<net::IOBuffer> buf;
+  const int buf_len;

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

Instead of storing the IOBuffer and length and wrapping it in a
DrainableIOBuffer each time you use it, you could just wrap it in a
DrainableIOBuffer immediately and store that as the class member.

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

Done.

+  const net::CompletionCallback callback;
+};
+
+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_read_.get());
+
+  // Dispatch a request to fill up our buffers if needed.
+  if (source_.get() && !source_has_pending_read_ &&

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

push these conditionals down into ReadFromSource(IfNeeded)

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

Done.

+      buffers_.size() < kDesiredNumberOfBuffers) {
+    ReadFromSource();
+  }
+
+  // Consume from our existing buffers and return immediately if possible.

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

Maybe push lines 60-66 into ConsumeFromBuffer and have it return ERR_IO_PENDING
if there's nothing to do?

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

Done.

+  if (!buffers_.empty())
+    return ConsumeFromBuffer(buf, buf_len);
+
+  // Pass through the stored source error or EOF if existent.
+  if (!source_.get())
+    return source_error_;
+
+  // We are waiting for an source read to complete, so save the request.
+  DCHECK(!pending_read_.get());
+  pending_read_.reset(new Request(buf, buf_len, callback));
+  return net::ERR_IO_PENDING;
+}
+
+int64 ReadaheadFileStreamReader::GetLength(
+    const net::Int64CompletionCallback& callback) {
+  return source_->GetLength(callback);
+}
+
+int ReadaheadFileStreamReader::ConsumeFromBuffer(net::IOBuffer* buf,
+                                                 int buf_len) {
+  DCHECK(!buffers_.empty());
+
+  // |buf| continues to exist after |sink| goes out of scope.

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

This comment doesn't seem necessary, since sink lives to the end of this method.

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

Done.

+  scoped_refptr<net::DrainableIOBuffer> sink(
+      new net::DrainableIOBuffer(buf, buf_len));
+
+  while (sink->BytesRemaining() > 0 && !buffers_.empty()) {
+    net::DrainableIOBuffer* head_buffer = buffers_.front().get();

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

If you're calling the destination sink, might as well call this source. Or
source_buffer if you're concerned about confusion with source_

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

Done.

+
+    DCHECK(head_buffer->BytesRemaining() > 0);
+
+    int copy_len = std::min(head_buffer->BytesRemaining(),
+                            sink->BytesRemaining());
+    std::copy(head_buffer->data(), head_buffer->data() + copy_len,
+              sink->data());
+
+    head_buffer->DidConsume(copy_len);
+    sink->DidConsume(copy_len);
+
+    if (head_buffer->BytesRemaining() == 0) {
+      buffers_.pop();

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

This is unfortunate - we're going to free a meg of memory and then allocate it
again.  Not sure it's worth a lot of gymnastics to prevent it, but if there's
something simple...

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

Last patchset adds something to reuse the just-exhausted buffer, but not sure if
it's legit.

+
+      // Get a new buffer to replace the one we just used up.
+      if (!source_has_pending_read_ && source_.get())
+        ReadFromSource();
+    }
+  }
+
+  return sink->BytesConsumed();
+}
+
+void ReadaheadFileStreamReader::ReadFromSource() {
+  DCHECK(!source_has_pending_read_);
+  source_has_pending_read_ = true;
+
+  // We don't create the DrainableIOBuffer wrapper until the callback, when
+  // we know the size of the content written to the buffer.
+  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) {
+    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;
+
+  if (result <= 0) {
+    source_.reset();
+    source_error_ = result;
+
+    // If there's a read waiting on this source-read, finish it with the error.
+    if (pending_read_.get()) {

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

If you push lines 65 and 66 into ComsumeFromBuffer, you can get rid of this
special case.

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

Done.

+      pending_read_->callback.Run(result);
+      pending_read_.reset();
+    }
+
+    return;
+  }
+
+  scoped_refptr<net::DrainableIOBuffer> drainable_buffer(
+      new net::DrainableIOBuffer(buf, result));
+  buffers_.push(drainable_buffer);
+
+  // If there's a read request waiting for the source 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));
+  }
+}