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 145483)
+++ net/base/file_stream_posix.cc	(working copy)
@@ -20,12 +20,12 @@
 #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/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"
@@ -49,320 +49,487 @@
                FROM_CURRENT == SEEK_CUR &&
                FROM_END     == SEEK_END, whence_matches_system);
 
-namespace {
 

[wtc, 2012/07/13 23:40:21]

Nit: delete a blank line.

-int RecordAndMapError(int error,
-                      FileErrorSource source,
-                      bool record_uma,
-                      const net::BoundNetLog& bound_net_log) {
-  net::Error net_error = MapSystemError(error);
+class FileStreamPosix::AsyncContext {
+ public:
+  explicit AsyncContext(const BoundNetLog& bound_net_log);
+  AsyncContext(base::PlatformFile file, const BoundNetLog& bound_net_log);
 
-  bound_net_log.AddEvent(
-      net::NetLog::TYPE_FILE_STREAM_ERROR,
-      base::Bind(&NetLogFileStreamErrorCallback,
-                 source, error, net_error));
+  // Destroys the context. It can be deleted in the method or deletion can be
+  // deferred to WorkerPool if some asynchronous operation is now in progress
+  // or if auto-closing is needed.
+  void Destroy();
 
-  RecordFileError(error, source, record_uma);
+  bool record_uma() { return record_uma_; }
+  void set_record_uma(bool value) { record_uma_ = value; }
+  base::PlatformFile file() { return file_; }
+  bool auto_closed() { return auto_closed_; }
+  void set_auto_closed(bool value) { auto_closed_ = value; }
+  bool async_in_progress() { return async_in_progress_; }
 
-  return net_error;
+  int RecordAndMapError(int error, FileErrorSource source);
+
+  // Sync and async versions of all operations
+  void OpenAsync(const FilePath& path,
+                 int open_flags,
+                 const CompletionCallback& callback);
+  int OpenSync(const FilePath& path, int open_flags);
+
+  void CloseAsync(const CompletionCallback& callback);
+  void CloseSync();
+
+  void SeekAsync(Whence whence,
+                 int64 offset,
+                 const Int64CompletionCallback& callback);
+  int64 SeekSync(Whence whence, int64 offset);
+
+  void ReadAsync(IOBuffer* buf,
+                 int buf_len,
+                 const CompletionCallback& callback);
+  int ReadSync(char* buf, int buf_len);
+
+  void WriteAsync(IOBuffer* in_buf,
+                  int buf_len,
+                  const CompletionCallback& callback);
+  int WriteSync(const char* in_buf, int buf_len);
+
+ private:
+  // Map system error into network error code and log it with |bound_net_log_|.
+  // Method should be called with |net_log_lock_| locked.
+  int MapAndLogError(int error, FileErrorSource source);
+
+  // Opens a file with some network logging.
+  // The result code is written to |result|.
+  void OpenFileImpl(const FilePath& path, int open_flags, int* result);
+
+  // Closes a file with some network logging.
+  void CloseFileImpl();
+
+  // Adjusts the position from where the data is read.
+  void SeekFileImpl(Whence whence, int64 offset, int64* result);
+
+  // ReadFile() is a simple wrapper around read() that handles EINTR signals
+  // and calls RecordAndMapError() to map errno to net error codes.
+  void ReadFileImpl(scoped_refptr<IOBuffer> buf, int buf_len, int* result);
+
+  // 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 WriteFileImpl(scoped_refptr<IOBuffer> buf, int buf_len, int* result);
+
+  // Called when asynchronous Close(), Open(), Read(), Write() or Seek()
+  // is completed. |result| contains the result or a network error code.
+  template <typename R>
+  void OnAsyncCompleted(const base::Callback<void(R)>& callback, R* result);

[wtc, 2012/07/13 23:40:21]

Since OnAsyncCompleted is only instantianted with R=int and
R=int64, it seems that we should be able to define OnAsyncCompleted
so that it takes "R result" instead of "R* result".  This
should avoid the need to do "new int(OK)".  I didn't study
the code well enough to know if this can be done.

[pivanof, 2012/07/14 00:13:58]

No, this can't be done because value of result is bound inside of Closure at the
time of scheduling of async task but async task during execution should be able
to change which result value OnAsyncCompleted() receives.

+
+  // Delete the context with asynchronous closing if necessary.
+  void DeleteAbandoned();
+
+  base::PlatformFile file_;
+  bool auto_closed_;
+  bool record_uma_;
+  bool async_in_progress_;
+  bool destroyed_;
+  base::Lock net_log_lock_;
+  BoundNetLog bound_net_log_;
+};
+
+
+FileStreamPosix::AsyncContext::AsyncContext(const BoundNetLog& bound_net_log)
+  : file_(base::kInvalidPlatformFileValue),
+    auto_closed_(true),
+    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));
+FileStreamPosix::AsyncContext::AsyncContext(base::PlatformFile file,
+                                            const BoundNetLog& bound_net_log)
+  : file_(file),
+    auto_closed_(false),
+    record_uma_(false),
+    async_in_progress_(false),
+    destroyed_(false),
+    bound_net_log_(bound_net_log) {
+}
 
-  *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);
+void FileStreamPosix::AsyncContext::Destroy() {
+  {
+    // By locking we don't allow any operation with |bound_net_log_| to be
+    // in progress while this method is executed. Attempt to do something
+    // with |bound_net_log_| will be done either before this method or after
+    // we switch |context_->destroyed_| which will prohibit any operation on
+    // |bound_net_log_|.
+    base::AutoLock locked(net_log_lock_);
+    destroyed_ = true;
   }
+  if (!async_in_progress_)
+    DeleteAbandoned();
 }
 
-// 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();
+int FileStreamPosix::AsyncContext::RecordAndMapError(int error,
+                                                     FileErrorSource source) {
+  int net_error;
+  {
+    base::AutoLock locked(net_log_lock_);
+    net_error = MapAndLogError(error, source);
+  }
+  RecordFileError(error, source, record_uma_);
+  return net_error;
 }
 
-// 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 FileStreamPosix::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);
+  int* result = new int(OK);
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&AsyncContext::OpenFileImpl, base::Unretained(this),
+                 path, open_flags, result),
+      base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+                 base::Unretained(this),
+                 callback, base::Owned(result)),
+      true /* task_is_slow */);
+  DCHECK(posted);
+
+  async_in_progress_ = true;
 }
 
-// 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();
+int FileStreamPosix::AsyncContext::OpenSync(const FilePath& path,
+                                            int open_flags) {
+  int result = OK;
+  OpenFileImpl(path, open_flags, &result);
+  return result;
 }
 
-// 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;
+void FileStreamPosix::AsyncContext::CloseAsync(
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  // Value OK will never be changed in AsyncContext::CloseFile() and is needed
+  // here just to use the same AsyncContext::OnAsyncCompleted().
+  int* result = new int(OK);
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&AsyncContext::CloseFileImpl, base::Unretained(this)),
+      base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+                 base::Unretained(this),
+                 callback, base::Owned(result)),
+      true /* task_is_slow */);
+  DCHECK(posted);
+
+  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();
+void FileStreamPosix::AsyncContext::CloseSync() {
+  DCHECK(!async_in_progress_);
+  CloseFileImpl();
 }
 
-// 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 FileStreamPosix::AsyncContext::SeekAsync(
+    Whence whence,
+    int64 offset,
+    const Int64CompletionCallback& 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);
+  int64* result = new int64(-1);
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&AsyncContext::SeekFileImpl, base::Unretained(this),
+                 whence, offset, result),
+      base::Bind(&AsyncContext::OnAsyncCompleted<int64>,
+                 base::Unretained(this),
+                 callback, base::Owned(result)),
+      true /* task is slow */);
+  DCHECK(posted);
+
+  async_in_progress_ = true;
+}
+
+int64 FileStreamPosix::AsyncContext::SeekSync(Whence whence, int64 offset) {
+  off_t result = -1;
+  SeekFileImpl(whence, offset, &result);
+  return result;
+}
+
+void FileStreamPosix::AsyncContext::ReadAsync(
+    IOBuffer* in_buf,
+    int buf_len,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  int* result = new int(OK);
+  scoped_refptr<IOBuffer> buf = in_buf;
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&AsyncContext::ReadFileImpl, base::Unretained(this),
+                 buf, buf_len, result),
+      base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+                 base::Unretained(this),
+                 callback, base::Owned(result)),
+      true /* task is slow */);
+  DCHECK(posted);
+
+  async_in_progress_ = true;
+}
+
+int FileStreamPosix::AsyncContext::ReadSync(char* in_buf, int buf_len) {
+  int result = OK;
+  scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+  ReadFileImpl(buf, buf_len, &result);
+  return result;
+}
+
+void FileStreamPosix::AsyncContext::WriteAsync(
+    IOBuffer* in_buf,
+    int buf_len,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  int* result = new int(OK);
+  scoped_refptr<IOBuffer> buf = in_buf;
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&AsyncContext::WriteFileImpl, base::Unretained(this),
+                 buf, buf_len, result),
+      base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+                 base::Unretained(this),
+                 callback, base::Owned(result)),
+      true /* task is slow */);
+  DCHECK(posted);
+
+  async_in_progress_ = true;
+}
+
+int FileStreamPosix::AsyncContext::WriteSync(const char* in_buf, int buf_len) {
+  int result = OK;
+  scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+  WriteFileImpl(buf, buf_len, &result);
+  return result;
+}
+
+int FileStreamPosix::AsyncContext::MapAndLogError(int error,
+                                                  FileErrorSource source) {
+  net_log_lock_.AssertAcquired();
+  // 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);
+
+  if (!destroyed_) {
+    bound_net_log_.AddEvent(
+        net::NetLog::TYPE_FILE_STREAM_ERROR,
+        base::Bind(&NetLogFileStreamErrorCallback,
+                   source, error, net_error));
   }
-  *result = res;
+
+  return net_error;
 }
 
-// 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();
+void FileStreamPosix::AsyncContext::OpenFileImpl(const FilePath& path,
+                                                 int open_flags,
+                                                 int* result) {
+  std::string file_name = path.AsUTF8Unsafe();
+  {
+    base::AutoLock locked(net_log_lock_);
+    // Bail out quickly if operation was already canceled
+    if (destroyed_)
+      return;
+
+    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) {
+    int error = errno;
+    {
+      base::AutoLock locked(net_log_lock_);
+      if (!destroyed_)
+        bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+      *result = MapAndLogError(error, FILE_ERROR_SOURCE_OPEN);
+    }
+    RecordFileError(error, FILE_ERROR_SOURCE_OPEN, record_uma_);
+  }
 }
 
-// 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) {
+void FileStreamPosix::AsyncContext::CloseFileImpl() {
+  {
+    base::AutoLock locked(net_log_lock_);
+    if (!destroyed_)
+      bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
+  }
+
+  if (file_ == base::kInvalidPlatformFileValue)
+    return;
+  if (!base::ClosePlatformFile(file_))
+    NOTREACHED();
+  file_ = base::kInvalidPlatformFileValue;
+
+  {
+    base::AutoLock locked(net_log_lock_);
+    if (!destroyed_)
+      bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+  }
+}
+
+void FileStreamPosix::AsyncContext::SeekFileImpl(Whence whence,
+                                                 int64 offset,
+                                                 int64* result) {
   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);
-  }
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return;
+
+  off_t res = lseek(file_, static_cast<off_t>(offset),
+                    static_cast<int>(whence));
+  if (res == static_cast<off_t>(-1))
+    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_SEEK);
   *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();
+void FileStreamPosix::AsyncContext::ReadFileImpl(scoped_refptr<IOBuffer> buf,
+                                                 int buf_len,
+                                                 int* result) {
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return;
+
+  // 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)
+    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_READ);
+  *result = res;
 }
 
-// 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) {
+void FileStreamPosix::AsyncContext::WriteFileImpl(scoped_refptr<IOBuffer> buf,
+                                                  int buf_len,
+                                                  int* result) {
   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;
+
+  // If context has been already destroyed nobody waits for operation results.
+  if (destroyed_)
+    return;
+
+  ssize_t res = HANDLE_EINTR(write(file_, buf->data(), buf_len));
+  if (res == -1)
+    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_WRITE);
+  *result = 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 FileStreamPosix::AsyncContext::OnAsyncCompleted(
+    const base::Callback<void(R)>& callback,
+    R* result) {
+  if (destroyed_) {
+    DeleteAbandoned();
+  } else {
+    // Reset this before Run() as Run() may issue a new async operation.
+    async_in_progress_ = false;
+    callback.Run(*result);
+  }
+}
 
-  // Reset this before Run() as Run() may issue a new async operation.
-  stream->ResetOnIOComplete();
-  callback.Run(*result);
+void FileStreamPosix::AsyncContext::DeleteAbandoned() {
+  if (auto_closed_ && 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;
+  }
 }
 
-}  // namespace
 
 // FileStreamPosix ------------------------------------------------------------
 
 FileStreamPosix::FileStreamPosix(net::NetLog* net_log)
-    : file_(base::kInvalidPlatformFileValue),
+    : context_(NULL),
       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)) {
+                                            net::NetLog::SOURCE_FILESTREAM)) {
+  context_ = new AsyncContext(bound_net_log_);
+
   bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
-FileStreamPosix::FileStreamPosix(
-    base::PlatformFile file, int flags, net::NetLog* net_log)
-    : file_(file),
+FileStreamPosix::FileStreamPosix(base::PlatformFile file,
+                                 int flags,
+                                 net::NetLog* net_log)
+    : context_(NULL),
       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)) {
+                                            net::NetLog::SOURCE_FILESTREAM)) {
+  context_ = new AsyncContext(file, bound_net_log_);
+
   bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 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();
-  }
+  if (context_->auto_closed() && !is_async())
+    CloseSync();
+  context_->Destroy();
 
-  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);
 }
 
 void FileStreamPosix::Close(const CompletionCallback& callback) {
-  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 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(),
-                 callback),
-      true /* task_is_slow */);
-
-  DCHECK(posted);
+  DCHECK(is_async());
+  context_->CloseAsync(callback);
 }
 
 void FileStreamPosix::CloseSync() {
+  // CloseSync() should be called on the correct thread even if it eventually
+  // ends up inside CloseAndCancelAsync().
+  base::ThreadRestrictions::AssertIOAllowed();
+
   // TODO(satorux): Replace the following async stuff with a
-  // DCHECK(open_flags & ASYNC) once once all async clients are migrated to
+  // DCHECK(open_flags & ASYNC) once all async clients are migrated to
   // use Close(). crbug.com/114783
+  if (context_->async_in_progress())
+    CloseAndCancelAsync();
+  else
+    context_->CloseSync();
+}
 
-  // 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();
+void FileStreamPosix::CloseAndCancelAsync() {
+  if (!IsOpen())
+    return;
 
-  CloseFile(file_, bound_net_log_);
-  file_ = base::kInvalidPlatformFileValue;
+  DCHECK(is_async());
+
+  AsyncContext* old_ctx = context_;
+  context_ = new AsyncContext(bound_net_log_);
+  context_->set_record_uma(old_ctx->record_uma());
+  old_ctx->set_auto_closed(true);
+  old_ctx->Destroy();
 }
 
-int FileStreamPosix::Open(const FilePath& path, int open_flags,
-                     const CompletionCallback& callback) {
+int FileStreamPosix::Open(const FilePath& path,
+                          int open_flags,
+                          const CompletionCallback& callback) {
   if (IsOpen()) {
     DLOG(FATAL) << "File is already open!";
     return ERR_UNEXPECTED;
   }
 
   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(
-      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 */);
-  DCHECK(posted);
+  DCHECK(is_async());
+  context_->OpenAsync(path, open_flags, callback);
   return ERR_IO_PENDING;
 }
 
@@ -373,59 +540,36 @@
   }
 
   open_flags_ = open_flags;
-  // TODO(satorux): Put a DCHECK once once all async clients are migrated
+  // TODO(satorux): Put a DCHECK 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;
+  // DCHECK(!is_async());
+  return context_->OpenSync(path, open_flags_);
 }
 
 bool FileStreamPosix::IsOpen() const {
-  return file_ != base::kInvalidPlatformFileValue;
+  return context_->file() != base::kInvalidPlatformFileValue;
 }
 
-int 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 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());
-
-  on_io_complete_.reset(new base::WaitableEvent(
-      false  /* manual_reset */, false  /* initially_signaled */));
-
-  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);
+  // Make sure we're async.
+  DCHECK(is_async());
+  context_->SeekAsync(whence, offset, callback);
   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) ||
-         !on_io_complete_.get());
-
-  off_t result = -1;
-  SeekFile(file_, whence, offset, &result, record_uma_, bound_net_log_);
-  return result;
+  DCHECK(!is_async() || !context_->async_in_progress());
+  return context_->SeekSync(whence, offset);
 }
 
 int64 FileStreamPosix::Available() {
@@ -439,12 +583,8 @@
     return cur_pos;
 
   struct stat info;
-  if (fstat(file_, &info) != 0) {
-    return RecordAndMapError(errno,
-                             FILE_ERROR_SOURCE_GET_SIZE,
-                             record_uma_,
-                             bound_net_log_);
-  }
+  if (fstat(context_->file(), &info) != 0)
+    return context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_GET_SIZE);
 
   int64 size = static_cast<int64>(info.st_size);
   DCHECK_GT(size, cur_pos);
@@ -452,34 +592,18 @@
   return size - cur_pos;
 }
 
-int FileStreamPosix::Read(
-    IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
+int FileStreamPosix::Read(IOBuffer* 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);
+  DCHECK(is_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);
+  context_->ReadAsync(buf, buf_len, callback);
   return ERR_IO_PENDING;
 }
 
@@ -487,14 +611,12 @@
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
+  DCHECK(!is_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;
+  return context_->ReadSync(buf, buf_len);
 }
 
 int FileStreamPosix::ReadUntilComplete(char *buf, int buf_len) {
@@ -518,49 +640,31 @@
   return bytes_total;
 }
 
-int FileStreamPosix::Write(
-    IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
+int FileStreamPosix::Write(IOBuffer* buf,
+                           int buf_len,
+                           const CompletionCallback& callback) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+  DCHECK(is_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(!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(&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);
+  context_->WriteAsync(buf, buf_len, callback);
   return ERR_IO_PENDING;
 }
 
-int FileStreamPosix::WriteSync(
-    const char* buf, int buf_len) {
+int FileStreamPosix::WriteSync(const char* buf, int buf_len) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
+  DCHECK(!is_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;
+  return context_->WriteSync(buf, buf_len);
 }
 
 int64 FileStreamPosix::Truncate(int64 bytes) {
@@ -578,25 +682,27 @@
     return ERR_UNEXPECTED;
 
   // And truncate the file.
-  int result = ftruncate(file_, bytes);
+  int result = ftruncate(context_->file(), bytes);
   if (result == 0)
     return seek_position;
 
-  return RecordAndMapError(errno,
-                           FILE_ERROR_SOURCE_SET_EOF,
-                           record_uma_,
-                           bound_net_log_);
+  return context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_SET_EOF);
 }
 
 int FileStreamPosix::Flush() {
+  base::ThreadRestrictions::AssertIOAllowed();
+
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  return FlushFile(file_, record_uma_, bound_net_log_);
+  ssize_t res = HANDLE_EINTR(fsync(context_->file()));
+  if (res == -1)
+    res = context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_FLUSH);
+  return res;
 }
 
 void FileStreamPosix::EnableErrorStatistics() {
-  record_uma_ = true;
+  context_->set_record_uma(true);
 }
 
 void FileStreamPosix::SetBoundNetLogSource(
@@ -620,29 +726,7 @@
 }
 
 base::PlatformFile FileStreamPosix::GetPlatformFileForTesting() {
-  return file_;
+  return context_->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();
-  }
-}
-
 }  // namespace net