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 148849)
+++ 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,472 @@
                FROM_CURRENT == SEEK_CUR &&
                FROM_END     == SEEK_END, whence_matches_system);
 
-namespace {
+class FileStreamPosix::AsyncContext {
+ public:
+  explicit AsyncContext(const BoundNetLog& bound_net_log);
+  AsyncContext(base::PlatformFile file, const BoundNetLog& bound_net_log);
 
-int RecordAndMapError(int error,
-                      FileErrorSource source,
-                      bool record_uma,
-                      const net::BoundNetLog& bound_net_log) {
-  net::Error net_error = MapSystemError(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();
 
-  bound_net_log.AddEvent(
-      net::NetLog::TYPE_FILE_STREAM_ERROR,
-      base::Bind(&NetLogFileStreamErrorCallback,
-                 source, error, net_error));
+  bool record_uma() { return record_uma_; }

[willchan no longer on Chromium, 2012/08/14 19:23:26]

All accessors should be const member functions.

+  void set_record_uma(bool value) { record_uma_ = value; }
+  base::PlatformFile file() { return file_; }
+  bool async_in_progress() { return async_in_progress_; }
 
-  RecordFileError(error, source, record_uma);
+  int RecordAndMapError(int error, FileErrorSource source);
 
-  return net_error;
+  // 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);

[willchan no longer on Chromium, 2012/08/14 19:23:26]

net/ convention is to name this function OnIOComplete().

I'm running off to lunch now, but you may want to look at
PostTaskAndReplyWithResult() instead of using a R* output param.

[pivanof, 2012/08/25 07:35:53]

There is already a function with this name in Win implementation and there is a
function with the same meaning as this one in Win implementation too. I wanted
to keep things symmetrical, so I needed to name the function here and in Win
with some different name.
But with new patch this should be moot as I needed to add one more function on
Linux that could be conveniently named OnIOCompleted().
It looks like I can't use PostTaskAndReplyWithResult because there's no way to
get TaskRunner* from WorkerPool.

[willchan no longer on Chromium, 2012/08/26 22:33:59]

http://code.google.com/searchframe#OAMlx_jo-ck/src/base/threading/worker_pool...
has a base::WorkerPool::GetTaskRunner().

+
+  // Delete the context with asynchronous closing if necessary.
+  void DeleteAbandoned();
+
+  base::PlatformFile file_;
+  bool record_uma_;
+  bool async_in_progress_;
+  bool destroyed_;

[willchan no longer on Chromium, 2012/08/14 19:23:26]

I think you should change it so that destroyed_ is not protected by the lock.
That's race prone. Only access it on the network thread.

+  base::Lock net_log_lock_;

[willchan no longer on Chromium, 2012/08/14 19:23:26]

This looks wrong. Note that the BoundNetLog is an object that writes to an
underlying NetLog object, which is a browser process wide singleton. So even if
you lock access to the BoundNetLog, we will have races on the underlying NetLog
object.

The code needs to be changed so we never access BoundNetLog except on the
network thread.

[mmenke, 2012/08/14 20:31:01]

It's safe to use a BoundNetLog on any thread, at least until the context is
destroyed (Actually lasts until the browser process is destroyed, but shouldn't
be reasoning about that in net/).

[willchan no longer on Chromium, 2012/08/14 20:34:12]

The problem is that the helper threads being used are unjoined WorkerPool
threads, by design, since we don't want to slow down shutdown on file IO. That
means during browser process shutdown, the BoundNetLog may be getting accessed
while/after the NetLog singleton is destroyed.

+  BoundNetLog bound_net_log_;
+};
+
+
+FileStreamPosix::AsyncContext::AsyncContext(const BoundNetLog& bound_net_log)
+  : file_(base::kInvalidPlatformFileValue),

[mmenke, 2012/08/14 20:31:01]

These should be using a 4-space indent.

[willchan no longer on Chromium, 2012/08/14 20:34:12]

google style guide says to indent 4 spaces for initializer lists

+    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),
+    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 (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 (IsOpen() && !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
-
-  // 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;
+  if (!context_->async_in_progress()) {
+    context_->CloseSync();
+  } else {
+    AsyncContext* old_ctx = context_;
+    context_ = new AsyncContext(bound_net_log_);
+    context_->set_record_uma(old_ctx->record_uma());
+    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 +525,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 +568,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 +577,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 +596,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 +625,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 +667,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 +711,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