Make FileStream::Seek async and add FileStream::SeekSync for sync operation

net/base/file_stream_posix.cc

 // Copyright (c) 2012 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.
 
 // For 64-bit file access (off_t = off64_t, lseek64, etc).
 #define _FILE_OFFSET_BITS 64
 
 #include "net/base/file_stream.h"
 
 #include <sys/types.h>
@@ skipping 107 matching lines @@
 
 // Closes a file with CloseFile() and signals the completion.
 void CloseFileAndSignal(base::PlatformFile* file,
                         base::WaitableEvent* on_io_complete,
                         const net::BoundNetLog& bound_net_log) {
   CloseFile(*file, bound_net_log);
   *file = base::kInvalidPlatformFileValue;
   on_io_complete->Signal();
 }
 
+// Adjusts the position from where the data is read.
+void SeekFile(base::PlatformFile file,
+              Whence whence,
+              int64 offset,
+              int64* result,
+              bool record_uma,
+              const net::BoundNetLog& bound_net_log) {
+  off_t res = lseek(file, static_cast<off_t>(offset),
+                    static_cast<int>(whence));
+  if (res == static_cast<off_t>(-1)) {
+    *result = RecordAndMapError(errno,
+                                FILE_ERROR_SOURCE_SEEK,
+                                record_uma,
+                                bound_net_log);
+    return;
+  }
+  *result = res;
+}
+
+// Seeks a file by calling SeekSync() and signals the completion.
+void SeekFileAndSignal(base::PlatformFile file,
+                       Whence whence,
+                       int64 offset,
+                       int64* result,
+                       bool record_uma,
+                       base::WaitableEvent* on_io_complete,
+                       const net::BoundNetLog& bound_net_log) {
+  SeekFile(file, whence, offset, result, record_uma, bound_net_log);
+  on_io_complete->Signal();
+}
+
 // ReadFile() is a simple wrapper around read() that handles EINTR signals and
 // calls MapSystemError() to map errno to net error codes.
 void ReadFile(base::PlatformFile file,
               char* buf,
               int buf_len,
               bool record_uma,
               int* result,
               const net::BoundNetLog& bound_net_log) {
   base::ThreadRestrictions::AssertIOAllowed();
   // read(..., 0) returns 0 to indicate end-of-file.
@@ skipping 58 matching lines @@
               const net::BoundNetLog& bound_net_log) {
   base::ThreadRestrictions::AssertIOAllowed();
   ssize_t res = HANDLE_EINTR(fsync(file));
   if (res == -1) {
     res = RecordAndMapError(errno, FILE_ERROR_SOURCE_FLUSH, record_uma,
                             bound_net_log);
   }
   return res;
 }
 
+// Called when Read(), Write() or Seek() is completed.
+// |result| contains the result or a network error code.
+template <typename R>
+void OnIOComplete(const base::WeakPtr<FileStreamPosix>& stream,
+                  const base::Callback<void(R)>& callback,
+                  R* result) {
+  if (!stream.get())

[willchan no longer on Chromium, 2012/04/05 15:12:28]

You don't need this, base::Bind() will take care of it for you when you pass a
WeakPtr. See
http://code.google.com/searchframe#OAMlx_jo-ck/src/base/bind_helpers.h&exact_...
and
http://code.google.com/searchframe#OAMlx_jo-ck/src/base/bind_internal.h&l=873.

[kinuko, 2012/04/06 12:42:37]

In this case stream is not a target object and I doubt Bind takes care of this
case.  (Actually without this if test fails with null point reference)

[willchan no longer on Chromium, 2012/04/09 12:33:57]

Woops, totally misread that :)

+    return;
+
+  // Reset this before Run() as Run() may issue a new async operation.
+  stream->ResetOnIOComplete();
+  callback.Run(*result);
+}
+
 }  // namespace
 
 // FileStreamPosix ------------------------------------------------------------
 
 FileStreamPosix::FileStreamPosix(net::NetLog* net_log)
     : file_(base::kInvalidPlatformFileValue),
       open_flags_(0),
       auto_closed_(true),
       record_uma_(false),
       bound_net_log_(net::BoundNetLog::Make(net_log,
@@ skipping 34 matching lines @@
     } else {
       CloseSync();
     }
   }
 
   bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE, NULL);
 }
 
 void FileStreamPosix::Close(const CompletionCallback& callback) {
   DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
-  DCHECK(callback_.is_null());

[willchan no longer on Chromium, 2012/04/05 15:12:28]

This DCHECK is useful. Let's keep it. It's basically asserting there's no other
async operation currently happening. Please
DCHECK(!weak_ptr_factory_.HasWeakPtrs());

Can you likewise add this for all the IO operations?

[kinuko, 2012/04/06 12:42:37]

Done.  I expect DCHECK(!on_io_complete_.get()) does the similar check as well.

-
-  callback_ = callback;
 
   DCHECK(!on_io_complete_.get());
   on_io_complete_.reset(new base::WaitableEvent(
       false  /* manual_reset */, false  /* initially_signaled */));
 
   // Passing &file_ to a thread pool looks unsafe but it's safe here as the
   // destructor ensures that the close operation is complete with
   // WaitForIOCompletion(). See also the destructor.
   const bool posted = base::WorkerPool::PostTaskAndReply(
       FROM_HERE,
       base::Bind(&CloseFileAndSignal, &file_, on_io_complete_.get(),
                  bound_net_log_),
       base::Bind(&FileStreamPosix::OnClosed,
-                 weak_ptr_factory_.GetWeakPtr()),
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback),
       true /* task_is_slow */);
 
   DCHECK(posted);
 }
 
 void FileStreamPosix::CloseSync() {
   // TODO(satorux): Replace the following async stuff with a
   // DCHECK(open_flags & ASYNC) once once all async clients are migrated to
   // use Close(). crbug.com/114783
 
   // Abort any existing asynchronous operations.
   weak_ptr_factory_.InvalidateWeakPtrs();
   // Block until the last open/read/write operation is complete.
   // TODO(satorux): Ideally we should not block. crbug.com/115067
   WaitForIOCompletion();
 
   CloseFile(file_, bound_net_log_);
   file_ = base::kInvalidPlatformFileValue;
 }
 
 int FileStreamPosix::Open(const FilePath& path, int open_flags,
                      const CompletionCallback& callback) {
   if (IsOpen()) {
     DLOG(FATAL) << "File is already open!";
     return ERR_UNEXPECTED;
   }
 
-  DCHECK(callback_.is_null());
-  callback_ = callback;
-
   open_flags_ = open_flags;
   DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
 
   DCHECK(!on_io_complete_.get());
   on_io_complete_.reset(new base::WaitableEvent(
       false  /* manual_reset */, false  /* initially_signaled */));
 
   // Passing &file_ to a thread pool looks unsafe but it's safe here as the
   // destructor ensures that the open operation is complete with
   // WaitForIOCompletion(). See also the destructor.
   int* result = new int(OK);
   const bool posted = base::WorkerPool::PostTaskAndReply(
       FROM_HERE,
       base::Bind(&OpenFileAndSignal,
                  path, open_flags, record_uma_, &file_, result,
                  on_io_complete_.get(), bound_net_log_),
-      base::Bind(&FileStreamPosix::OnIOComplete,
-                 weak_ptr_factory_.GetWeakPtr(),
-                 base::Owned(result)),
+      base::Bind(&OnIOComplete<int>, weak_ptr_factory_.GetWeakPtr(),
+                 callback, base::Owned(result)),
       true /* task_is_slow */);
   DCHECK(posted);
   return ERR_IO_PENDING;
 }
 
 int FileStreamPosix::OpenSync(const FilePath& path, int open_flags) {
   if (IsOpen()) {
     DLOG(FATAL) << "File is already open!";
     return ERR_UNEXPECTED;
   }
 
   open_flags_ = open_flags;
   // TODO(satorux): Put a DCHECK once once all async clients are migrated
   // to use Open(). crbug.com/114783
   //
   // DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
 
   int result = OK;
   OpenFile(path, open_flags_, record_uma_, &file_, &result, bound_net_log_);
   return result;
 }
 
 bool FileStreamPosix::IsOpen() const {
   return file_ != base::kInvalidPlatformFileValue;
 }
 
-int64 FileStreamPosix::Seek(Whence whence, int64 offset) {
+int FileStreamPosix::Seek(Whence whence, int64 offset,
+                          const Int64CompletionCallback& callback) {
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  // Make sure we're async.
+  DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+
+  int64* result = new int64(-1);
+  DCHECK(!on_io_complete_.get());
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&SeekFileAndSignal, file_, whence, offset, result,
+                 record_uma_, on_io_complete_.get(), bound_net_log_),
+      base::Bind(&OnIOComplete<int64>,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback, base::Owned(result)),
+      true /* task is slow */);
+  DCHECK(posted);
+  return ERR_IO_PENDING;
+}
+
+int64 FileStreamPosix::SeekSync(Whence whence, int64 offset) {
   base::ThreadRestrictions::AssertIOAllowed();
 
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   // If we're in async, make sure we don't have a request in flight.
   DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC) ||
-         callback_.is_null());
+         !on_io_complete_.get());
 
-  off_t res = lseek(file_, static_cast<off_t>(offset),
-                    static_cast<int>(whence));
-  if (res == static_cast<off_t>(-1)) {
-    return RecordAndMapError(errno,
-                             FILE_ERROR_SOURCE_SEEK,
-                             record_uma_,
-                             bound_net_log_);
-  }
-
-  return res;
+  off_t result = -1;
+  SeekFile(file_, whence, offset, &result, record_uma_, bound_net_log_);
+  return result;
 }
 
 int64 FileStreamPosix::Available() {
   base::ThreadRestrictions::AssertIOAllowed();
 
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  int64 cur_pos = Seek(FROM_CURRENT, 0);
+  int64 cur_pos = SeekSync(FROM_CURRENT, 0);
   if (cur_pos < 0)
     return cur_pos;
 
   struct stat info;
   if (fstat(file_, &info) != 0) {
     return RecordAndMapError(errno,
                              FILE_ERROR_SOURCE_GET_SIZE,
                              record_uma_,
                              bound_net_log_);
   }
 
   int64 size = static_cast<int64>(info.st_size);
   DCHECK_GT(size, cur_pos);
 
   return size - cur_pos;
 }
 
 int FileStreamPosix::Read(
     IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   // read(..., 0) will return 0, which indicates end-of-file.
   DCHECK_GT(buf_len, 0);
   DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
   DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+
   // Make sure we don't have a request in flight.
-  DCHECK(callback_.is_null());
-
-  callback_ = callback;
-  int* result = new int(OK);
-  scoped_refptr<IOBuffer> buf = in_buf;
   DCHECK(!on_io_complete_.get());
   on_io_complete_.reset(new base::WaitableEvent(
       false  /* manual_reset */, false  /* initially_signaled */));
+
+  int* result = new int(OK);
+  scoped_refptr<IOBuffer> buf = in_buf;
   const bool posted = base::WorkerPool::PostTaskAndReply(
       FROM_HERE,
       base::Bind(&ReadFileAndSignal, file_, buf, buf_len,
                  record_uma_, result, on_io_complete_.get(), bound_net_log_),
-      base::Bind(&FileStreamPosix::OnIOComplete,
+      base::Bind(&OnIOComplete<int>,
                  weak_ptr_factory_.GetWeakPtr(),
-                 base::Owned(result)),
+                 callback, base::Owned(result)),
       true /* task is slow */);
   DCHECK(posted);
   return ERR_IO_PENDING;
 }
 
 int FileStreamPosix::ReadSync(char* buf, int buf_len) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
@@ skipping 29 matching lines @@
 
 int FileStreamPosix::Write(
     IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
   DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
   // write(..., 0) will return 0, which indicates end-of-file.
   DCHECK_GT(buf_len, 0);
+
   // Make sure we don't have a request in flight.
-  DCHECK(callback_.is_null());
-
-  callback_ = callback;
-  int* result = new int(OK);
-  scoped_refptr<IOBuffer> buf = in_buf;
   DCHECK(!on_io_complete_.get());
   on_io_complete_.reset(new base::WaitableEvent(
       false  /* manual_reset */, false  /* initially_signaled */));
+
+  int* result = new int(OK);
+  scoped_refptr<IOBuffer> buf = in_buf;
   const bool posted = base::WorkerPool::PostTaskAndReply(
       FROM_HERE,
       base::Bind(&WriteFileAndSignal, file_, buf, buf_len,
                  record_uma_, result, on_io_complete_.get(), bound_net_log_),
-      base::Bind(&FileStreamPosix::OnIOComplete,
+      base::Bind(&OnIOComplete<int>,
                  weak_ptr_factory_.GetWeakPtr(),
-                 base::Owned(result)),
+                 callback, base::Owned(result)),
       true /* task is slow */);
   DCHECK(posted);
   return ERR_IO_PENDING;
 }
 
 int FileStreamPosix::WriteSync(
     const char* buf, int buf_len) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
@@ skipping 10 matching lines @@
 int64 FileStreamPosix::Truncate(int64 bytes) {
   base::ThreadRestrictions::AssertIOAllowed();
 
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   // We'd better be open for writing.
   DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
 
   // Seek to the position to truncate from.
-  int64 seek_position = Seek(FROM_BEGIN, bytes);
+  int64 seek_position = SeekSync(FROM_BEGIN, bytes);
   if (seek_position != bytes)
     return ERR_UNEXPECTED;
 
   // And truncate the file.
   int result = ftruncate(file_, bytes);
   if (result == 0)
     return seek_position;
 
   return RecordAndMapError(errno,
                            FILE_ERROR_SOURCE_SET_EOF,
@@ skipping 33 matching lines @@
       net::NetLog::TYPE_FILE_STREAM_SOURCE,
       make_scoped_refptr(
           new net::NetLogSourceParameter("source_dependency",
                                          bound_net_log_.source())));
 }
 
 base::PlatformFile FileStreamPosix::GetPlatformFileForTesting() {
   return file_;
 }
 
-void FileStreamPosix::OnClosed() {
-  file_ = base::kInvalidPlatformFileValue;
-  int result = OK;
-  OnIOComplete(&result);
+void FileStreamPosix::ResetOnIOComplete() {

[willchan no longer on Chromium, 2012/04/05 15:12:28]

I'd invalidate weak pointers at the WeakPtrFactory now, because otherwise it'll
only be invalidated when the base::Callback goes out of scope, but since we're
going to call the "reply" callback on async completion right now, we may
re-enter FileStream. This would mean the
DCHECK(!weak_ptr_factory_.HasWeakPtrs()); assertion I suggested would fail.

[kinuko, 2012/04/06 12:42:37]

Good idea, added the invalidation and DCHECKs.

+  on_io_complete_.reset();
 }
 
-void FileStreamPosix::OnIOComplete(int* result) {
-  CompletionCallback temp_callback = callback_;
-  callback_.Reset();
+void FileStreamPosix::OnClosed(const CompletionCallback& callback) {
+  file_ = base::kInvalidPlatformFileValue;
 
   // Reset this before Run() as Run() may issue a new async operation.
   on_io_complete_.reset();
-  temp_callback.Run(*result);
+  callback.Run(OK);
 }
 
 void FileStreamPosix::WaitForIOCompletion() {
   if (on_io_complete_.get()) {
     on_io_complete_->Wait();
     on_io_complete_.reset();
   }
 }
 
 }  // namespace net