[fsp] Buffer consecutive Write() calls.

chrome/browser/chromeos/file_system_provider/fileapi/buffering_file_stream_writer.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/chromeos/file_system_provider/fileapi/buffering_file_stream_writer.h"
+
+#include <algorithm>
+
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace chromeos {
+namespace file_system_provider {
+
+BufferingFileStreamWriter::BufferingFileStreamWriter(
+    scoped_ptr<storage::FileStreamWriter> file_stream_writer,
+    int intermediate_buffer_length)
+    : file_stream_writer_(file_stream_writer.Pass()),
+      intermediate_buffer_length_(intermediate_buffer_length),
+      intermediate_buffer_(new net::IOBuffer(intermediate_buffer_length_)),
+      buffered_bytes_(0),
+      weak_ptr_factory_(this) {
+}
+
+BufferingFileStreamWriter::~BufferingFileStreamWriter() {
+}
+
+int BufferingFileStreamWriter::Write(net::IOBuffer* buffer,
+                                     int buffer_length,
+                                     const net::CompletionCallback& callback) {
+  // If |buffer_length| is larger than the intermediate buffer, then call the
+  // internal file stream writer directly. Note, that the intermediate buffer
+  // (used for buffering) must be flushed first.
+  if (buffer_length > intermediate_buffer_length_) {
+    if (buffered_bytes_) {
+      FlushIntermediateBuffer(
+          base::Bind(&BufferingFileStreamWriter::
+                         OnFlushIntermediateBufferForDirectWriteCompleted,
+                     weak_ptr_factory_.GetWeakPtr(),
+                     make_scoped_refptr(buffer),
+                     buffer_length,
+                     callback));
+    } else {
+      // Nothing to flush, so skip it.
+      OnFlushIntermediateBufferForDirectWriteCompleted(
+          make_scoped_refptr(buffer), buffer_length, callback, net::OK);
+    }
+    return net::ERR_IO_PENDING;
+  }
+
+  // Buffer consecutive writes to larger chunks.
+  const int buffer_bytes =
+      std::min(intermediate_buffer_length_ - buffered_bytes_, buffer_length);
+
+  CopyToIntermediateBuffer(
+      make_scoped_refptr(buffer), 0 /* offset */, buffer_bytes);
+  const int bytes_left = buffer_length - buffer_bytes;
+
+  if (buffered_bytes_ == intermediate_buffer_length_) {
+    FlushIntermediateBuffer(
+        base::Bind(&BufferingFileStreamWriter::
+                       OnFlushIntermediateBufferForBufferedWriteCompleted,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   make_scoped_refptr(buffer),
+                   buffer_bytes,
+                   bytes_left,
+                   callback));
+    return net::ERR_IO_PENDING;
+  }
+
+  // Optimistically return a success.
+  return buffer_length;
+}
+
+int BufferingFileStreamWriter::Cancel(const net::CompletionCallback& callback) {
+  return file_stream_writer_->Cancel(callback);

[hashimoto, 2014/09/11 07:52:16]

Could you add a comment to describe how this implementation is justified?
IIUC this implementation is correct as long as this class does not perform any
asynchronous operations other than performed by |file_stream_writer_|.

[mtomasz, 2014/09/17 04:59:40]

Done.
It is valid to call Cancel() if there is no ongoing in-flight operation on
file_stream_writer_. Saying that, we don't have any PostTasks in
buffering_file_stream_writer.cc, so we will not get any Cancel() invocation if
there is no in-flight operation on file_stream_writer_.

+}
+
+int BufferingFileStreamWriter::Flush(const net::CompletionCallback& callback) {
+  // Flush all the buffered bytes first, then invoke Flush() on the inner file
+  // stream writer.
+  FlushIntermediateBuffer(base::Bind(
+      &BufferingFileStreamWriter::OnFlushIntermediateBufferForFlushCompleted,
+      weak_ptr_factory_.GetWeakPtr(),
+      callback));
+  return net::ERR_IO_PENDING;
+}
+
+void BufferingFileStreamWriter::CopyToIntermediateBuffer(
+    scoped_refptr<net::IOBuffer> buffer,
+    int offset,

[hashimoto, 2014/09/11 07:52:16]

It's ambiguous whether |offset| is for the argument buffer or the intermediate
buffer.
How about passing char* instead of |buffer| and |offset| so that there is no
need to pass the offset?

[mtomasz, 2014/09/17 04:59:40]

It's clearly described in the header file.
I'd like to keep buffer as net::IOBuffer as I want to have all the ugly pointer
arithmetics in one place. I renamed the |offset| to |buffer_offset| and |length|
to |buffer_length|.

+    int length) {
+  DCHECK_GE(intermediate_buffer_length_, offset + length + buffered_bytes_);

[hashimoto, 2014/09/11 07:52:16]

What does the value of "offset + length + buffered_bytes_" correspond to?

[mtomasz, 2014/09/17 04:59:40]

Position of the last byte to be written. This DCHECK checks if we are not
writing more than size of the intermediate buffer.

[hashimoto, 2014/09/22 12:10:31]

Then shouldn't this be buffer_length + buffered_bytes_?

[mtomasz, 2014/09/24 01:45:31]

That's right. Fixed.

+  memcpy(intermediate_buffer_->data() + buffered_bytes_,
+         buffer->data() + offset,
+         length);
+  buffered_bytes_ += length;
+}
+
+void BufferingFileStreamWriter::FlushIntermediateBuffer(
+    const net::CompletionCallback& callback) {
+  const int result = file_stream_writer_->Write(
+      intermediate_buffer_,
+      buffered_bytes_,
+      base::Bind(&BufferingFileStreamWriter::OnFlushIntermediateBufferCompleted,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 buffered_bytes_,
+                 callback));
+  DCHECK_EQ(net::ERR_IO_PENDING, result);
+}
+
+void BufferingFileStreamWriter::OnFlushIntermediateBufferCompleted(
+    int length,
+    const net::CompletionCallback& callback,
+    int result) {
+  if (result < 0) {
+    callback.Run(result);
+    return;
+  }
+
+  buffered_bytes_ = 0;
+
+  DCHECK_EQ(length, result) << "Partial writes are not supported.";
+  callback.Run(net::OK);
+}
+
+void
+BufferingFileStreamWriter::OnFlushIntermediateBufferForDirectWriteCompleted(
+    scoped_refptr<net::IOBuffer> buffer,
+    int length,
+    const net::CompletionCallback& callback,
+    int result) {
+  if (result < 0) {
+    callback.Run(result);
+    return;
+  }
+
+  // Write all bytes directly.
+  DCHECK_EQ(net::OK, result);

[hashimoto, 2014/09/11 07:52:16]

nit: Why are you checking this here and in the method below? Can Flush() return
any non-negative value other than net::OK?

[mtomasz, 2014/09/17 04:59:40]

It's not possible, but since FlushIntermediateBuffer internally calls
file_stream_writer_->Write, this DCHECK makes sure that the |result| is properly
replaced with net::OK in #124. Hence I think it's worth keeping it, as it's only
a DCHECK.

[hashimoto, 2014/09/22 12:10:31]

I still don't see any reason to check |result| which is never used after the
"result < 0" conditional above.
Anyways, it seems harmless as long as this is DCHECK.

[mtomasz, 2014/09/24 01:45:31]

Yeah, seems not really helping much here. Removed.

+  DCHECK_EQ(0, buffered_bytes_);
+  const int write_result = file_stream_writer_->Write(buffer, length, callback);
+  DCHECK_EQ(net::ERR_IO_PENDING, write_result);
+}
+
+void
+BufferingFileStreamWriter::OnFlushIntermediateBufferForBufferedWriteCompleted(
+    scoped_refptr<net::IOBuffer> buffer,
+    int buffered_bytes,
+    int bytes_left,
+    const net::CompletionCallback& callback,
+    int result) {
+  if (result < 0) {
+    callback.Run(result);
+    return;
+  }
+
+  // Copy the rest of bytes to the buffer.
+  DCHECK_EQ(net::OK, result);
+  DCHECK_EQ(0, buffered_bytes_);
+  DCHECK_GE(intermediate_buffer_length_, bytes_left);
+  CopyToIntermediateBuffer(buffer, buffered_bytes, bytes_left);
+
+  callback.Run(buffered_bytes + bytes_left);
+}
+
+void BufferingFileStreamWriter::OnFlushIntermediateBufferForFlushCompleted(
+    const net::CompletionCallback& callback,
+    int result) {
+  if (result < 0) {
+    callback.Run(result);
+    return;
+  }
+
+  const int flush_result = file_stream_writer_->Flush(callback);
+  DCHECK_EQ(net::ERR_IO_PENDING, flush_result);
+}
+
+}  // namespace file_system_provider
+}  // namespace chromeos