net: Implement canceling of all async operations in FileStream.

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>
 #include <sys/stat.h>
 #include <fcntl.h>
 #include <unistd.h>
 #include <errno.h>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #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"
 #include "net/base/net_errors.h"
 
 #if defined(OS_ANDROID)
 // Android's bionic libc only supports the LFS transitional API.
 #define off_t off64_t
 #define lseek lseek64
 #define stat stat64
 #define fstat fstat64
 #endif
 
 namespace net {
 
 // We cast back and forth, so make sure it's the size we're expecting.
 COMPILE_ASSERT(sizeof(int64) == sizeof(off_t), off_t_64_bit);
 
 // Make sure our Whence mappings match the system headers.
 COMPILE_ASSERT(FROM_BEGIN   == SEEK_SET &&
                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);
-
+

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

Nit: delete a blank line.

+class FileStreamPosix::AsyncContext {
+ public:
+  explicit AsyncContext(const BoundNetLog& bound_net_log);
+  AsyncContext(base::PlatformFile file, const BoundNetLog& bound_net_log);
+
+  // 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();
+
+  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_; }
+
+  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) {
+}
+
+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) {
+}
+
+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();
+}
+
+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;
 }
 
-// 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);
-  }
-}
-
-// 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();
-}
-
-// 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;
-
-  if (!base::ClosePlatformFile(file))
-    NOTREACHED();
-  bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
-}
-
-// 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.
-
-  // 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;
-}
-
-// 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();
-}
-
-// 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);
-  }
-  *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);
-  }
-  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;
-
-  // 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,
-                                            net::NetLog::SOURCE_FILESTREAM)),
-      weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
-  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
-}
-
-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);
-}
-
-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 (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.
+void FileStreamPosix::AsyncContext::OpenAsync(
+    const FilePath& path,
+    int open_flags,
+    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  int* result = new int(OK);
   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),
+      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;
+}
+
+int FileStreamPosix::AsyncContext::OpenSync(const FilePath& path,
+                                            int open_flags) {
+  int result = OK;
+  OpenFileImpl(path, open_flags, &result);
+  return result;
+}
+
+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;
+}
+
+void FileStreamPosix::AsyncContext::CloseSync() {
+  DCHECK(!async_in_progress_);
+  CloseFileImpl();
+}
+
+void FileStreamPosix::AsyncContext::SeekAsync(
+    Whence whence,
+    int64 offset,
+    const Int64CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  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));
+  }
+
+  return net_error;
+}
+
+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_);
+  }
+}
+
+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();
+
+  // 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;
+}
+
+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;
+}
+
+void FileStreamPosix::AsyncContext::WriteFileImpl(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;
+
+  ssize_t res = HANDLE_EINTR(write(file_, buf->data(), buf_len));
+  if (res == -1)
+    res = RecordAndMapError(errno, FILE_ERROR_SOURCE_WRITE);
+  *result = res;
+}
+
+template <typename R>
+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);
+  }
+}
+
+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;
+  }
+}
+
+
+// FileStreamPosix ------------------------------------------------------------
+
+FileStreamPosix::FileStreamPosix(net::NetLog* net_log)
+    : context_(NULL),
+      open_flags_(0),
+      bound_net_log_(net::BoundNetLog::Make(net_log,
+                                            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)
+    : context_(NULL),
+      open_flags_(flags),
+      bound_net_log_(net::BoundNetLog::Make(net_log,
+                                            net::NetLog::SOURCE_FILESTREAM)) {
+  context_ = new AsyncContext(file, bound_net_log_);
+
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
+}
+
+FileStreamPosix::~FileStreamPosix() {
+  if (context_->auto_closed() && !is_async())
+    CloseSync();
+  context_->Destroy();
+
+  bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
+}
+
+void FileStreamPosix::Close(const CompletionCallback& callback) {
+  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
-
-  // 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 (context_->async_in_progress())
+    CloseAndCancelAsync();
+  else
+    context_->CloseSync();
+}
+
+void FileStreamPosix::CloseAndCancelAsync() {
+  if (!IsOpen())
+    return;
+
+  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) {
   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;
 }
 
 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
+  // 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() {
   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_);
-  }
+  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);
 
   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;
 }
 
 int FileStreamPosix::ReadSync(char* buf, int buf_len) {
   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) {
   int to_read = buf_len;
   int bytes_total = 0;
 
   do {
     int bytes_read = ReadSync(buf, to_read);
     if (bytes_read <= 0) {
       if (bytes_total == 0)
         return bytes_read;
 
       return bytes_total;
     }
 
     bytes_total += bytes_read;
     buf += bytes_read;
     to_read -= bytes_read;
   } while (bytes_total < buf_len);
 
   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) {
   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 = SeekSync(FROM_BEGIN, bytes);
   if (seek_position != bytes)
     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(
     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;
   }
 
   // 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();
-  }
+  return context_->file();
 }
 
 }  // namespace net