net: Implement canceling of all async operations in FileStream.

net/base/file_stream_posix.cc

Index: net/base/file_stream_posix.cc
===================================================================
--- net/base/file_stream_posix.cc	(revision 151422)
+++ net/base/file_stream_posix.cc	(working copy)
@@ -5,7 +5,7 @@
 // For 64-bit file access (off_t = off64_t, lseek64, etc).
 #define _FILE_OFFSET_BITS 64
 
-#include "net/base/file_stream.h"
+#include "net/base/file_stream_posix.h"
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -20,13 +20,13 @@
 #include "base/eintr_wrapper.h"
 #include "base/file_path.h"
 #include "base/logging.h"
+#include "base/memory/ref_counted.h"
 #include "base/message_loop.h"
 #include "base/metrics/histogram.h"
 #include "base/string_util.h"
+#include "base/task_runner_util.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/threading/worker_pool.h"
-#include "base/synchronization/waitable_event.h"
-#include "net/base/file_stream_metrics.h"
 #include "net/base/file_stream_net_log_parameters.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
@@ -49,599 +49,339 @@
                FROM_CURRENT == SEEK_CUR &&
                FROM_END     == SEEK_END, whence_matches_system);
 
-namespace {
-
-int RecordAndMapError(int error,
-                      FileErrorSource source,
-                      bool record_uma,
-                      const net::BoundNetLog& bound_net_log) {
-  net::Error net_error = MapSystemError(error);
-
-  bound_net_log.AddEvent(
-      net::NetLog::TYPE_FILE_STREAM_ERROR,
-      base::Bind(&NetLogFileStreamErrorCallback,
-                 source, error, net_error));
-
-  RecordFileError(error, source, record_uma);
-
-  return net_error;
+FileStream::AsyncContext::AsyncContext(const BoundNetLog& bound_net_log)
+    : file_(base::kInvalidPlatformFileValue),
+      record_uma_(false),
+      async_in_progress_(false),
+      destroyed_(false),
+      bound_net_log_(bound_net_log) {
 }
 
-// Opens a file with some network logging.
-// The opened file and the result code are written to |file| and |result|.
-void OpenFile(const FilePath& path,
-              int open_flags,
-              bool record_uma,
-              base::PlatformFile* file,
-              int* result,
-              const net::BoundNetLog& bound_net_log) {
-  std::string file_name = path.AsUTF8Unsafe();
-  bound_net_log.BeginEvent(
-      net::NetLog::TYPE_FILE_STREAM_OPEN,
-      NetLog::StringCallback("file_name", &file_name));
-
-  *result = OK;
-  *file = base::CreatePlatformFile(path, open_flags, NULL, NULL);
-  if (*file == base::kInvalidPlatformFileValue) {
-    bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
-    *result = RecordAndMapError(errno, FILE_ERROR_SOURCE_OPEN, record_uma,
-                                bound_net_log);
-  }
+FileStream::AsyncContext::AsyncContext(base::PlatformFile file,
+                                       const BoundNetLog& bound_net_log,
+                                       int /* open_flags */)
+    : file_(file),
+      record_uma_(false),
+      async_in_progress_(false),
+      destroyed_(false),
+      bound_net_log_(bound_net_log) {
 }
 
-// Opens a file using OpenFile() and then signals the completion.
-void OpenFileAndSignal(const FilePath& path,
-                       int open_flags,
-                       bool record_uma,
-                       base::PlatformFile* file,
-                       int* result,
-                       base::WaitableEvent* on_io_complete,
-                       const net::BoundNetLog& bound_net_log) {
-  OpenFile(path, open_flags, record_uma, file, result, bound_net_log);
-  on_io_complete->Signal();
+void FileStream::AsyncContext::Destroy() {

[willchan no longer on Chromium, 2012/08/27 06:59:10]

Oops, earlier I neglected to mention how this bit would have to change if we
switched to Orphan(). Here, we wouldn't do DeleteAbandoned() anymore. But in all
the places below where the async IO completes, we would need to check
|orphaned_| and then, if the file were still open, begin a CloseAsync()
operation. If the file is not open, then we can just delete this. Note that when
we post the CloseAsync() operation, and that completes, then we'll again check
|orphaned_| and then delete this.

+  destroyed_ = true;
+  if (!async_in_progress_)
+    DeleteAbandoned();
 }
 
-// Closes a file with some network logging.
-void CloseFile(base::PlatformFile file,
-               const net::BoundNetLog& bound_net_log) {
-  bound_net_log.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
-  if (file == base::kInvalidPlatformFileValue)
-    return;
+void FileStream::AsyncContext::OpenAsync(
+    const FilePath& path,
+    int open_flags,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  if (!base::ClosePlatformFile(file))
-    NOTREACHED();
-  bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
-}
+  BeginOpenEvent(path);
 
-// 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();
-}
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
+      FROM_HERE,
+      base::Bind(&AsyncContext::OpenFileImpl,
+                 base::Unretained(this), path, open_flags),
+      base::Bind(&AsyncContext::OnOpenCompleted,
+                 base::Unretained(this), callback));
+  DCHECK(posted);
 
-// 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;
+  async_in_progress_ = true;
 }
 
-// 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();
+int FileStream::AsyncContext::OpenSync(const FilePath& path, int open_flags) {
+  BeginOpenEvent(path);
+  int result = OpenFileImpl(path, open_flags);
+  CheckForOpenError(&result);
+  return result;
 }
 
-// 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.
+void FileStream::AsyncContext::CloseAsync(const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  // Loop in the case of getting interrupted by a signal.
-  ssize_t res = HANDLE_EINTR(read(file, buf, static_cast<size_t>(buf_len)));
-  if (res == -1) {
-    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_READ,
-                            record_uma, bound_net_log);
-  }
-  *result = res;
-}
+  bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
 
-// Reads a file using ReadFile() and signals the completion.
-void ReadFileAndSignal(base::PlatformFile file,
-                       scoped_refptr<IOBuffer> buf,
-                       int buf_len,
-                       bool record_uma,
-                       int* result,
-                       base::WaitableEvent* on_io_complete,
-                       const net::BoundNetLog& bound_net_log) {
-  ReadFile(file, buf->data(), buf_len, record_uma, result, bound_net_log);
-  on_io_complete->Signal();
-}
+  if (file_ == base::kInvalidPlatformFileValue) {
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+                   base::Unretained(this), callback, OK));
+  } else {
+    const bool posted = base::WorkerPool::PostTaskAndReply(
+        FROM_HERE,
+        base::Bind(&AsyncContext::CloseFileImpl,
+                   base::Unretained(this)),
+        base::Bind(&AsyncContext::OnCloseCompleted,
+                   base::Unretained(this), callback),
+        true /* task_is_slow */);
+    DCHECK(posted);
 
-// WriteFile() is a simple wrapper around write() that handles EINTR signals and
-// calls MapSystemError() to map errno to net error codes.  It tries to write to
-// completion.
-void WriteFile(base::PlatformFile file,
-               const char* buf,
-               int buf_len,
-               bool record_uma,
-               int* result,
-               const net::BoundNetLog& bound_net_log) {
-  base::ThreadRestrictions::AssertIOAllowed();
-
-  ssize_t res = HANDLE_EINTR(write(file, buf, buf_len));
-  if (res == -1) {
-    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_WRITE, record_uma,
-                            bound_net_log);
+    async_in_progress_ = true;
   }
-  *result = res;
 }
 
-// Writes a file using WriteFile() and signals the completion.
-void WriteFileAndSignal(base::PlatformFile file,
-                        scoped_refptr<IOBuffer> buf,
-                        int buf_len,
-                        bool record_uma,
-                        int* result,
-                        base::WaitableEvent* on_io_complete,
-                        const net::BoundNetLog& bound_net_log) {
-  WriteFile(file, buf->data(), buf_len, record_uma, result, bound_net_log);
-  on_io_complete->Signal();
-}
-
-// FlushFile() is a simple wrapper around fsync() that handles EINTR signals and
-// calls MapSystemError() to map errno to net error codes.  It tries to flush to
-// completion.
-int FlushFile(base::PlatformFile file,
-              bool record_uma,
-              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);
+void FileStream::AsyncContext::CloseSync() {
+  DCHECK(!async_in_progress_);
+  bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
+  if (file_ != base::kInvalidPlatformFileValue) {
+    CloseFileImpl();
+    bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
   }
-  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())
-    return;
+void FileStream::AsyncContext::SeekAsync(
+    Whence whence,
+    int64 offset,
+    const Int64CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  // Reset this before Run() as Run() may issue a new async operation.
-  stream->ResetOnIOComplete();
-  callback.Run(*result);
-}
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task is slow */),
+      FROM_HERE,
+      base::Bind(&AsyncContext::SeekFileImpl,
+                 base::Unretained(this), whence, offset),
+      base::Bind(&AsyncContext::OnIOCompleted<int64>,
+                 base::Unretained(this), callback, FILE_ERROR_SOURCE_SEEK));
+  DCHECK(posted);
 
-}  // 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,
-                                            net::NetLog::SOURCE_FILESTREAM)),
-      weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
-  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
+  async_in_progress_ = true;
 }
 
-FileStreamPosix::FileStreamPosix(
-    base::PlatformFile file, int flags, net::NetLog* net_log)
-    : file_(file),
-      open_flags_(flags),
-      auto_closed_(false),
-      record_uma_(false),
-      bound_net_log_(net::BoundNetLog::Make(net_log,
-                                            net::NetLog::SOURCE_FILESTREAM)),
-      weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
-  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
+int64 FileStream::AsyncContext::SeekSync(Whence whence, int64 offset) {
+  off_t result = SeekFileImpl(whence, offset);
+  CheckForIOError(&result, FILE_ERROR_SOURCE_SEEK);
+  return result;
 }
 
-FileStreamPosix::~FileStreamPosix() {
-  if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
-    // Block until the last open/close/read/write operation is complete.
-    // TODO(satorux): Ideally we should not block. crbug.com/115067
-    WaitForIOCompletion();
-  }
+int64 FileStream::AsyncContext::GetFileSize() {
+  struct stat info;
+  if (fstat(file_, &info) != 0)
+    return RecordAndMapError(errno, FILE_ERROR_SOURCE_GET_SIZE);
 
-  if (auto_closed_) {
-    if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
-      // Close the file in the background.
-      if (IsOpen()) {
-        const bool posted = base::WorkerPool::PostTask(
-            FROM_HERE,
-            base::Bind(&CloseFile, file_, bound_net_log_),
-            true /* task_is_slow */);
-        DCHECK(posted);
-      }
-    } else {
-      CloseSync();
-    }
-  }
-
-  bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
+  return static_cast<int64>(info.st_size);
 }
 
-void FileStreamPosix::Close(const CompletionCallback& callback) {
-  DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+int FileStream::AsyncContext::ReadAsync(
+    IOBuffer* in_buf,
+    int buf_len,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  DCHECK(!weak_ptr_factory_.HasWeakPtrs());
-  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(
+  scoped_refptr<IOBuffer> buf = in_buf;
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task is slow */),
       FROM_HERE,
-      base::Bind(&CloseFileAndSignal, &file_, on_io_complete_.get(),
-                 bound_net_log_),
-      base::Bind(&FileStreamPosix::OnClosed,
-                 weak_ptr_factory_.GetWeakPtr(),
-                 callback),
-      true /* task_is_slow */);
-
+      base::Bind(&AsyncContext::ReadFileImpl,
+                 base::Unretained(this), buf, buf_len),
+      base::Bind(&AsyncContext::OnIOCompleted<int>,
+                 base::Unretained(this), callback, FILE_ERROR_SOURCE_READ));
   DCHECK(posted);
+
+  async_in_progress_ = true;
+  return ERR_IO_PENDING;
 }
 
-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 FileStream::AsyncContext::ReadSync(char* in_buf, int buf_len) {
+  scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+  int result = ReadFileImpl(buf, buf_len);
+  CheckForIOError(&result, FILE_ERROR_SOURCE_READ);
+  return result;
 }
 
-int FileStreamPosix::Open(const FilePath& path, int open_flags,
-                     const CompletionCallback& callback) {
-  if (IsOpen()) {
-    DLOG(FATAL) << "File is already open!";
-    return ERR_UNEXPECTED;
-  }
+int FileStream::AsyncContext::WriteAsync(
+    IOBuffer* in_buf,
+    int buf_len,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  open_flags_ = open_flags;
-  DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
-
-  DCHECK(!weak_ptr_factory_.HasWeakPtrs());
-  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(
+  scoped_refptr<IOBuffer> buf = in_buf;
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task is slow */),
       FROM_HERE,
-      base::Bind(&OpenFileAndSignal,
-                 path, open_flags, record_uma_, &file_, result,
-                 on_io_complete_.get(), bound_net_log_),
-      base::Bind(&OnIOComplete<int>, weak_ptr_factory_.GetWeakPtr(),
-                 callback, base::Owned(result)),
-      true /* task_is_slow */);
+      base::Bind(&AsyncContext::WriteFileImpl,
+                 base::Unretained(this), buf, buf_len),
+      base::Bind(&AsyncContext::OnIOCompleted<int>,
+                 base::Unretained(this), callback, FILE_ERROR_SOURCE_WRITE));
   DCHECK(posted);
+
+  async_in_progress_ = true;
   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_);
+int FileStream::AsyncContext::WriteSync(const char* in_buf, int buf_len) {
+  scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+  int result = WriteFileImpl(buf, buf_len);
+  CheckForIOError(&result, FILE_ERROR_SOURCE_WRITE);
   return result;
 }
 
-bool FileStreamPosix::IsOpen() const {
-  return file_ != base::kInvalidPlatformFileValue;
+int FileStream::AsyncContext::Flush() {
+  ssize_t res = HANDLE_EINTR(fsync(file_));
+  if (res == -1)
+    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_FLUSH);
+  return res;
 }
 
-int FileStreamPosix::Seek(Whence whence, int64 offset,
-                          const Int64CompletionCallback& callback) {
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
+int FileStream::AsyncContext::Truncate(int64 bytes) {
+  int result = ftruncate(file_, bytes);
+  if (result == 0)
+    return bytes;
 
-  // Make sure we're async and we have no other in-flight async operations.
-  DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
-  DCHECK(!weak_ptr_factory_.HasWeakPtrs());
-  DCHECK(!on_io_complete_.get());
+  return RecordAndMapError(errno, FILE_ERROR_SOURCE_SET_EOF);
+}
 
-  on_io_complete_.reset(new base::WaitableEvent(
-      false  /* manual_reset */, false  /* initially_signaled */));
+int FileStream::AsyncContext::RecordAndMapError(int error,
+                                                FileErrorSource source) {
+  // The following check is against incorrect use or bug. File descriptor
+  // shouldn't ever be closed outside of FileStream while it still tries to do
+  // something with it.
+  DCHECK(error != EBADF);
+  net::Error net_error = MapSystemError(error);
 
-  int64* result = new int64(-1);
-  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;
+  if (!destroyed_) {
+    bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_ERROR,
+                            base::Bind(&NetLogFileStreamErrorCallback,
+                                       source, error, net_error));
+  }
+  RecordFileError(error, source, record_uma_);
+  return net_error;
 }
 
-int64 FileStreamPosix::SeekSync(Whence whence, int64 offset) {
-  base::ThreadRestrictions::AssertIOAllowed();
+void FileStream::AsyncContext::BeginOpenEvent(const FilePath& path) {
+  std::string file_name = path.AsUTF8Unsafe();
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_OPEN,
+                            NetLog::StringCallback("file_name", &file_name));
+}
 
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
+int FileStream::AsyncContext::OpenFileImpl(const FilePath& path,
+                                           int open_flags) {
+  if (destroyed_)

[willchan no longer on Chromium, 2012/08/27 06:24:06]

Remove this. It's arguably a benign race, but as Hans Boehm taught us, no data
race is truly benign since compilers can miscompile them:
http://static.usenix.org/event/hotpar11/tech/final_files/Boehm.pdf.

Once it's been posted to the WorkerPool thread, don't bother trying to cancel
it. It's OK to just check |orphaned_| in the reply callback.

+    return OK;
 
-  // If we're in async, make sure we don't have a request in flight.
-  DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC) ||
-         !on_io_complete_.get());
+  file_ = base::CreatePlatformFile(path, open_flags, NULL, NULL);
+  if (file_ == base::kInvalidPlatformFileValue)
+    return errno;
 
-  off_t result = -1;
-  SeekFile(file_, whence, offset, &result, record_uma_, bound_net_log_);
-  return result;
+  return OK;
 }
 
-int64 FileStreamPosix::Available() {
-  base::ThreadRestrictions::AssertIOAllowed();
-
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
-
-  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_);
+void FileStream::AsyncContext::CheckForOpenError(int* result) {
+  if (file_ == base::kInvalidPlatformFileValue) {
+    bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+    *result = RecordAndMapError(*result, FILE_ERROR_SOURCE_OPEN);
   }
+}
 
-  int64 size = static_cast<int64>(info.st_size);
-  DCHECK_GT(size, cur_pos);
-
-  return size - cur_pos;
+void FileStream::AsyncContext::OnOpenCompleted(
+    const CompletionCallback& callback,
+    int result) {
+  CheckForOpenError(&result);
+  OnAsyncCompleted(callback, result);
 }
 
-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(!weak_ptr_factory_.HasWeakPtrs());
-  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(&OnIOComplete<int>,
-                 weak_ptr_factory_.GetWeakPtr(),
-                 callback, base::Owned(result)),
-      true /* task is slow */);
-  DCHECK(posted);
-  return ERR_IO_PENDING;
+void FileStream::AsyncContext::CloseFileImpl() {
+  if (!base::ClosePlatformFile(file_))
+    NOTREACHED();
+  file_ = base::kInvalidPlatformFileValue;
 }
 
-int FileStreamPosix::ReadSync(char* buf, int buf_len) {
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
-
-  DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
-  // read(..., 0) will return 0, which indicates end-of-file.
-  DCHECK_GT(buf_len, 0);
-  DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
-
-  int result = OK;
-  ReadFile(file_, buf, buf_len, record_uma_, &result, bound_net_log_);
-  return result;
+void FileStream::AsyncContext::OnCloseCompleted(
+    const CompletionCallback& callback) {
+  if (!destroyed_)
+    bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+  OnAsyncCompleted(callback, static_cast<int>(OK));
 }
 
-int FileStreamPosix::ReadUntilComplete(char *buf, int buf_len) {
-  int to_read = buf_len;
-  int bytes_total = 0;
+int64 FileStream::AsyncContext::SeekFileImpl(Whence whence, int64 offset) {
+  base::ThreadRestrictions::AssertIOAllowed();
 
-  do {
-    int bytes_read = ReadSync(buf, to_read);
-    if (bytes_read <= 0) {
-      if (bytes_total == 0)
-        return bytes_read;
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return 0;
 
-      return bytes_total;
-    }
+  off_t res = lseek(file_, static_cast<off_t>(offset),
+                    static_cast<int>(whence));
+  if (res == static_cast<off_t>(-1))
+    return errno;
 
-    bytes_total += bytes_read;
-    buf += bytes_read;
-    to_read -= bytes_read;
-  } while (bytes_total < buf_len);
-
-  return bytes_total;
+  return res;
 }
 
-int FileStreamPosix::Write(
-    IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
+int FileStream::AsyncContext::ReadFileImpl(scoped_refptr<IOBuffer> buf,
+                                           int buf_len) {
+  base::ThreadRestrictions::AssertIOAllowed();
 
-  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);
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return OK;
 
-  // Make sure we don't have a request in flight.
-  DCHECK(!weak_ptr_factory_.HasWeakPtrs());
-  DCHECK(!on_io_complete_.get());
-  on_io_complete_.reset(new base::WaitableEvent(
-      false  /* manual_reset */, false  /* initially_signaled */));
+  // Loop in the case of getting interrupted by a signal.
+  ssize_t res = HANDLE_EINTR(read(file_, buf->data(),
+                                  static_cast<size_t>(buf_len)));
+  if (res == -1)
+    return errno;
 
-  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(&OnIOComplete<int>,
-                 weak_ptr_factory_.GetWeakPtr(),
-                 callback, base::Owned(result)),
-      true /* task is slow */);
-  DCHECK(posted);
-  return ERR_IO_PENDING;
+  return res;
 }
 
-int FileStreamPosix::WriteSync(
-    const char* buf, int buf_len) {
-  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);
-
-  int result = OK;
-  WriteFile(file_, buf, buf_len, record_uma_, &result, bound_net_log_);
-  return result;
-}
-
-int64 FileStreamPosix::Truncate(int64 bytes) {
+int FileStream::AsyncContext::WriteFileImpl(scoped_refptr<IOBuffer> buf,
+                                            int buf_len) {
   base::ThreadRestrictions::AssertIOAllowed();
 
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return OK;
 
-  // We'd better be open for writing.
-  DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
+  ssize_t res = HANDLE_EINTR(write(file_, buf->data(), buf_len));
+  if (res == -1)
+    return errno;
 
-  // Seek to the position to truncate from.
-  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,
-                           record_uma_,
-                           bound_net_log_);
+  return res;
 }
 
-int FileStreamPosix::Flush() {
-  if (!IsOpen())
-    return ERR_UNEXPECTED;
-
-  return FlushFile(file_, record_uma_, bound_net_log_);
+template <typename R>
+void FileStream::AsyncContext::CheckForIOError(R* result,
+                                               FileErrorSource source) {
+  if (*result < 0)
+    *result = RecordAndMapError(static_cast<int>(*result), source);
 }
 
-void FileStreamPosix::EnableErrorStatistics() {
-  record_uma_ = true;
+template <typename R>
+void FileStream::AsyncContext::OnIOCompleted(
+    const base::Callback<void(R)>& callback,
+    FileErrorSource source,
+    R result) {
+  CheckForIOError(&result, source);
+  OnAsyncCompleted(callback, result);
 }
 
-void FileStreamPosix::SetBoundNetLogSource(
-    const net::BoundNetLog& owner_bound_net_log) {
-  if ((owner_bound_net_log.source().id == net::NetLog::Source::kInvalidId) &&
-      (bound_net_log_.source().id == net::NetLog::Source::kInvalidId)) {
-    // Both |BoundNetLog|s are invalid.
-    return;
+template <typename R>
+void FileStream::AsyncContext::OnAsyncCompleted(
+    const base::Callback<void(R)>& callback,
+    R result) {
+  if (destroyed_) {

[willchan no longer on Chromium, 2012/08/27 06:24:06]

Change this to:
if (orphaned_) {
  delete this;
} ...

+    DeleteAbandoned();
+  } else {
+    // Reset this before Run() as Run() may issue a new async operation.
+    async_in_progress_ = false;
+    callback.Run(result);
   }
-
-  // Should never connect to itself.
-  DCHECK_NE(bound_net_log_.source().id, owner_bound_net_log.source().id);
-
-  bound_net_log_.AddEvent(
-      net::NetLog::TYPE_FILE_STREAM_BOUND_TO_OWNER,
-      owner_bound_net_log.source().ToEventParametersCallback());
-
-  owner_bound_net_log.AddEvent(
-      net::NetLog::TYPE_FILE_STREAM_SOURCE,
-      bound_net_log_.source().ToEventParametersCallback());
 }
 
-base::PlatformFile FileStreamPosix::GetPlatformFileForTesting() {
-  return file_;
-}
-
-void FileStreamPosix::ResetOnIOComplete() {
-  on_io_complete_.reset();
-  weak_ptr_factory_.InvalidateWeakPtrs();
-}
-
-void FileStreamPosix::OnClosed(const CompletionCallback& callback) {
-  file_ = base::kInvalidPlatformFileValue;
-
-  // Reset this before Run() as Run() may issue a new async operation.
-  ResetOnIOComplete();
-  callback.Run(OK);
-}
-
-void FileStreamPosix::WaitForIOCompletion() {
-  // http://crbug.com/115067
-  base::ThreadRestrictions::ScopedAllowWait allow_wait;
-  if (on_io_complete_.get()) {
-    on_io_complete_->Wait();
-    on_io_complete_.reset();
+void FileStream::AsyncContext::DeleteAbandoned() {

[willchan no longer on Chromium, 2012/08/27 06:24:06]

One good thing about removing DeleteAbandoned is this is racy, since WorkerPool
is not tied to a single thread. It can spawn multiple threads, so this posted
task may run while other async tasks are running for the same context.

[pivanof, 2012/08/27 08:43:31]

I don't follow. There are DCHECKs everywhere to ensure that at least in the
debug build there is only one async operation in progress. If there's several
async operations in progress there will be several races here and there. Are you
really talking about such situation?

[willchan no longer on Chromium, 2012/09/05 00:03:38]

You're right, there should not be multiple async operations at the same time.
The thing though is, DeleteAbandoned() may be triggered by the deletion of the
FileStream object while an async operation is in program. This means this
CloseFileImpl() and an async operation may be concurrently executing on the
WorkerPool, which is a race condition.

+  if (file_ != base::kInvalidPlatformFileValue) {
+    const bool posted = base::WorkerPool::PostTask(
+        FROM_HERE,
+        // Context should be deleted after closing, thus Owned().
+        base::Bind(&AsyncContext::CloseFileImpl, base::Owned(this)),
+        true /* task_is_slow */);
+    DCHECK(posted);
+  } else {
+    delete this;
   }
 }