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 size_t 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),
      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> sink =
      new net::DrainableIOBuffer(buf, buf_len);
  int result = FinishReadFromCacheOrStoredError(sink);

  // 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_ = sink;
    pending_read_callback_ = callback;
  }

  return result;
}

int64 ReadaheadFileStreamReader::GetLength(
    const net::Int64CompletionCallback& callback) {
  return source_->GetLength(callback);
}

int ReadaheadFileStreamReader::FinishReadFromCacheOrStoredError(
    net::DrainableIOBuffer* sink) {
  // 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_;
    }
  }

  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)
    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);
    ReadFromSourceIfNeeded();
  } 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));
  }
}