net: Implement canceling of all async operations in FileStream.

net/base/file_stream.cc

Index: net/base/file_stream.cc
===================================================================
--- net/base/file_stream.cc	(revision 151422)
+++ net/base/file_stream.cc	(working copy)
@@ -4,95 +4,485 @@
 
 #include "net/base/file_stream.h"
 
+#include "base/location.h"
+#include "base/message_loop_proxy.h"
+#include "base/task_runner_util.h"
+#include "base/threading/thread_restrictions.h"
+#include "base/threading/worker_pool.h"
+#include "net/base/file_stream_net_log_parameters.h"
+#include "net/base/net_errors.h"
+
+#if defined(OS_WIN)
+#include "net/base/file_stream_win.h"
+#elif defined(OS_POSIX)
+#include "net/base/file_stream_posix.h"
+#endif
+
+namespace {
+
+void CallInt64ToInt(const net::CompletionCallback& callback, int64 result) {
+  callback.Run(static_cast<int>(result));
+}
+
+}
+
 namespace net {
 
+void FileStream::Context::Orphan() {
+  DCHECK(!orphaned_);
+
+  orphaned_ = true;
+  if (async_in_progress_) {
+    // Do we need this here? It can race with CloseAsync() and result in
+    // calling CancelIo() on the file handle that was already closed and
+    // reopened again.
+    //if (file_ != base::kInvalidPlatformFileValue)
+    //  CancelIo(file_);
+  } else {
+    CloseAsync(CompletionCallback());
+  }
+}
+
+void FileStream::Context::OpenAsync(const FilePath& path,
+                                    int open_flags,
+                                    const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  BeginOpenEvent(path);
+
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
+      FROM_HERE,
+      base::Bind(&Context::OpenFileImpl,
+                 base::Unretained(this), path, open_flags),
+      base::Bind(&Context::OnOpenCompleted,
+                 base::Unretained(this), callback));
+  DCHECK(posted);
+
+  async_in_progress_ = true;
+}
+
+int FileStream::Context::OpenSync(const FilePath& path, int open_flags) {
+  DCHECK(!async_in_progress_);
+
+  BeginOpenEvent(path);
+  OpenResult result = OpenFileImpl(path, open_flags);
+  file_ = result.file;
+  if (file_ == base::kInvalidPlatformFileValue) {
+    result.error_code = ProcessOpenError(result.error_code);
+  } else {
+    // TODO(satorux): Remove this once all async clients are migrated to use
+    // Open(). crbug.com/114783
+    if (open_flags & base::PLATFORM_FILE_ASYNC)
+      OnAsyncFileOpened();
+  }
+  return result.error_code;
+}
+
+void FileStream::Context::CloseAsync(const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
+
+  if (file_ == base::kInvalidPlatformFileValue) {
+    base::MessageLoopProxy::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&Context::OnCloseCompleted,
+                   base::Unretained(this), callback));
+  } else {
+    const bool posted = base::WorkerPool::PostTaskAndReply(
+        FROM_HERE,
+        base::Bind(base::IgnoreResult(&base::ClosePlatformFile), file_),
+        base::Bind(&Context::OnCloseCompleted,
+                   base::Unretained(this), callback),
+        true /* task_is_slow */);
+    DCHECK(posted);
+
+    async_in_progress_ = true;
+  }
+}
+
+void FileStream::Context::CloseSync() {
+  DCHECK(!async_in_progress_);
+  bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
+  if (file_ != base::kInvalidPlatformFileValue) {
+    base::ClosePlatformFile(file_);
+    file_ = base::kInvalidPlatformFileValue;
+    bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+  }
+}
+
+void FileStream::Context::SeekAsync(Whence whence,
+                                    int64 offset,
+                                    const Int64CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task is slow */),
+      FROM_HERE,
+      base::Bind(&Context::SeekFileImpl,
+                 base::Unretained(this), whence, offset),
+      base::Bind(&Context::ProcessAsyncResult,
+                 base::Unretained(this), callback, FILE_ERROR_SOURCE_SEEK));
+  DCHECK(posted);
+
+  async_in_progress_ = true;
+}
+
+int64 FileStream::Context::SeekSync(Whence whence, int64 offset) {
+  int64 result = SeekFileImpl(whence, offset);
+  CheckForIOError(&result, FILE_ERROR_SOURCE_SEEK);
+  return result;
+}
+
+int FileStream::Context::RecordAndMapError(int error,
+                                           FileErrorSource source) const {
+  // 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 != kErrorBadFile);
+  net::Error net_error = MapSystemError(error);
+
+  if (!orphaned_) {
+    bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_ERROR,
+                            base::Bind(&NetLogFileStreamErrorCallback,
+                                       source, error, net_error));
+  }
+  RecordFileError(error, source, record_uma_);
+  return net_error;
+}
+
+void FileStream::Context::BeginOpenEvent(const FilePath& path) {
+  std::string file_name = path.AsUTF8Unsafe();
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_OPEN,
+                            NetLog::StringCallback("file_name", &file_name));
+}
+
+FileStream::Context::OpenResult FileStream::Context::OpenFileImpl(
+    const FilePath& path, int open_flags) {
+  OpenResult result;
+  result.error_code = OK;
+  result.file = base::CreatePlatformFile(path, open_flags, NULL, NULL);
+  if (result.file == base::kInvalidPlatformFileValue)
+    result.error_code = GetLastErrno();
+
+  return result;
+}
+
+int FileStream::Context::ProcessOpenError(int error_code) {
+  bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+  return RecordAndMapError(error_code, FILE_ERROR_SOURCE_OPEN);
+}
+
+void FileStream::Context::OnOpenCompleted(const CompletionCallback& callback,
+                                          OpenResult result) {
+  file_ = result.file;
+  if (file_ == base::kInvalidPlatformFileValue)
+    result.error_code = ProcessOpenError(result.error_code);
+  else if (!orphaned_)
+    OnAsyncFileOpened();
+  OnAsyncCompleted(IntToInt64(callback), result.error_code);
+}
+
+void FileStream::Context::OnCloseCompleted(const CompletionCallback& callback) {
+  file_ = base::kInvalidPlatformFileValue;
+  if (!orphaned_) {
+    bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+    // Reset this before Run() as Run() may issue a new async operation.
+    async_in_progress_ = false;
+    callback.Run(OK);
+  } else {
+    delete this;
+  }
+}
+
+Int64CompletionCallback FileStream::Context::IntToInt64(
+    const CompletionCallback& callback) {
+  return base::Bind(&CallInt64ToInt, callback);
+}
+
+void FileStream::Context::CheckForIOError(int64* result,
+                                          FileErrorSource source) {
+  if (*result < 0)
+    *result = RecordAndMapError(static_cast<int>(*result), source);
+}
+
+void FileStream::Context::ProcessAsyncResult(
+    const Int64CompletionCallback& callback,
+    FileErrorSource source,
+    int64 result) {
+  CheckForIOError(&result, source);
+  OnAsyncCompleted(callback, result);
+}
+
+void FileStream::Context::OnAsyncCompleted(
+    const Int64CompletionCallback& callback,
+    int64 result) {
+  // Reset this before Run() as Run() may issue a new async operation. Also it
+  // should be reset before CloseAsync() because it shouldn't run if any async
+  // operation is in progress.
+  async_in_progress_ = false;
+  if (orphaned_)
+    CloseAsync(CompletionCallback());
+  else
+    callback.Run(result);
+}
+
 FileStream::FileStream(net::NetLog* net_log)
-    : impl_(net_log) {
+    : open_flags_(0),
+      bound_net_log_(net::BoundNetLog::Make(net_log,
+                                            net::NetLog::SOURCE_FILESTREAM)),
+      context_(new Context(bound_net_log_)) {
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
-FileStream::FileStream(
-    base::PlatformFile file, int flags, net::NetLog* net_log)
-    : impl_(file, flags, net_log) {
+FileStream::FileStream(base::PlatformFile file, int flags, net::NetLog* net_log)
+    : open_flags_(flags),
+      bound_net_log_(net::BoundNetLog::Make(net_log,
+                                            net::NetLog::SOURCE_FILESTREAM)),
+      context_(new Context(file, bound_net_log_, open_flags_)) {
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 FileStream::~FileStream() {
+  if (IsOpen() && !is_async()) {
+    CloseSync();
+    context_.reset();
+  } else {
+    context_.release()->Orphan();
+  }
+
+  bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 void FileStream::Close(const CompletionCallback& callback) {
-  impl_.Close(callback);
+  DCHECK(is_async());
+  context_->CloseAsync(callback);
 }
 
 void FileStream::CloseSync() {
-  impl_.CloseSync();
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  // TODO(satorux): Replace the following async stuff with a
+  // DCHECK(is_async()) once all async clients are migrated to
+  // use Close(). crbug.com/114783
+  if (!context_->async_in_progress()) {
+    context_->CloseSync();
+  } else {
+    Context* old_context = context_.release();
+    context_.reset(new Context(bound_net_log_));
+    context_->set_record_uma(old_context->record_uma());
+    old_context->Orphan();
+  }
 }
 
 int FileStream::Open(const FilePath& path, int open_flags,
                      const CompletionCallback& callback) {
-  return impl_.Open(path, open_flags, callback);
+  if (IsOpen()) {
+    DLOG(FATAL) << "File is already open!";
+    return ERR_UNEXPECTED;
+  }
+
+  open_flags_ = open_flags;
+  DCHECK(is_async());
+  context_->OpenAsync(path, open_flags, callback);
+  return ERR_IO_PENDING;
 }
 
 int FileStream::OpenSync(const FilePath& path, int open_flags) {
-  return impl_.OpenSync(path, open_flags);
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (IsOpen()) {
+    DLOG(FATAL) << "File is already open!";
+    return ERR_UNEXPECTED;
+  }
+
+  open_flags_ = open_flags;
+  // TODO(satorux): Put a DCHECK once all async clients are migrated
+  // to use Open(). crbug.com/114783
+  //
+  // DCHECK(!is_async());
+  return context_->OpenSync(path, open_flags_);
 }
 
 bool FileStream::IsOpen() const {
-  return impl_.IsOpen();
+  return context_->file() != base::kInvalidPlatformFileValue;
 }
 
-int FileStream::Seek(Whence whence, int64 offset,
+int FileStream::Seek(Whence whence,
+                     int64 offset,
                      const Int64CompletionCallback& callback) {
-  return impl_.Seek(whence, offset, callback);
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  // Make sure we're async.
+  DCHECK(is_async());
+  context_->SeekAsync(whence, offset, callback);
+  return ERR_IO_PENDING;
 }
 
 int64 FileStream::SeekSync(Whence whence, int64 offset) {
-  return impl_.SeekSync(whence, 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(!is_async() || !context_->async_in_progress());
+  return context_->SeekSync(whence, offset);
 }
 
 int64 FileStream::Available() {
-  return impl_.Available();
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  int64 cur_pos = SeekSync(FROM_CURRENT, 0);
+  if (cur_pos < 0)
+    return cur_pos;
+
+  int64 size = context_->GetFileSize();
+  if (size < 0)
+    return size;
+
+  DCHECK_GT(size, cur_pos);
+  return size - cur_pos;
 }
 
-int FileStream::Read(
-    IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
-  return impl_.Read(in_buf, buf_len, callback);
+int FileStream::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(is_async());
+
+  return context_->ReadAsync(buf, buf_len, callback);
 }
 
 int FileStream::ReadSync(char* buf, int buf_len) {
-  return impl_.ReadSync(buf, buf_len);
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  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);
+
+  return context_->ReadSync(buf, buf_len);
 }
 
 int FileStream::ReadUntilComplete(char *buf, int buf_len) {
-  return impl_.ReadUntilComplete(buf, buf_len);
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  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 FileStream::Write(
-    IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
-  return impl_.Write(buf, buf_len, callback);
+int FileStream::Write(IOBuffer* buf,
+                      int buf_len,
+                      const CompletionCallback& callback) {
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  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);
+
+  return context_->WriteAsync(buf, buf_len, callback);
 }
 
 int FileStream::WriteSync(const char* buf, int buf_len) {
-  return impl_.WriteSync(buf, buf_len);
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  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);
+
+  return context_->WriteSync(buf, buf_len);
 }
 
 int64 FileStream::Truncate(int64 bytes) {
-  return impl_.Truncate(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.
+  return context_->Truncate(bytes);
 }
 
 int FileStream::Flush() {
-  return impl_.Flush();
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
+  return context_->Flush();
 }
 
 void FileStream::EnableErrorStatistics() {
-  impl_.EnableErrorStatistics();
+  context_->set_record_uma(true);
 }
 
 void FileStream::SetBoundNetLogSource(
     const net::BoundNetLog& owner_bound_net_log) {
-  impl_.SetBoundNetLogSource(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 FileStream::GetPlatformFileForTesting() {
-  return impl_.GetPlatformFileForTesting();
+  return context_->file();
 }
 
 }  // namespace net