net: Implement canceling of all async operations in FileStream.

net/base/file_stream.cc

Index: net/base/file_stream.cc
diff --git a/net/base/file_stream.cc b/net/base/file_stream.cc
index df18ad098b3a8318dac45624ae7ab328d5624527..05f146013feea9aee4fa835db0bf25bc77673543 100644
--- a/net/base/file_stream.cc
+++ b/net/base/file_stream.cc
@@ -4,99 +4,261 @@
 
 #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_context.h"
+#include "net/base/file_stream_net_log_parameters.h"
+#include "net/base/net_errors.h"
+
 namespace net {
 
-FileStream::FileStream(net::NetLog* net_log)
-    : impl_(net_log) {
+FileStream::FileStream(NetLog* net_log)
+      /* To allow never opened stream to be destroyed on any thread we set flags
+         as if stream was opened asynchronously. */
+    : open_flags_(base::PLATFORM_FILE_ASYNC),
+      bound_net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_FILESTREAM)),
+      context_(new Context(bound_net_log_)) {
+  bound_net_log_.BeginEvent(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, NetLog* net_log)
+    : open_flags_(flags),
+      bound_net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_FILESTREAM)),
+      context_(new Context(file, bound_net_log_, open_flags_)) {
+  bound_net_log_.BeginEvent(NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 FileStream::~FileStream() {
-}
-
-void FileStream::Close(const CompletionCallback& callback) {
-  impl_.Close(callback);
-}
+  if (!is_async()) {
+    base::ThreadRestrictions::AssertIOAllowed();
+    context_->CloseSync();
+    context_.reset();
+  } else {
+    context_.release()->Orphan();
+  }
 
-void FileStream::CloseSync() {
-  impl_.CloseSync();
+  bound_net_log_.EndEvent(NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 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(const CompletionCallback& callback) {
-  return impl_.Flush(callback);
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
+  // Make sure we're async.
+  DCHECK(is_async());
+
+  context_->FlushAsync(callback);
+  return ERR_IO_PENDING;
 }
 
 int FileStream::FlushSync() {
-  return impl_.FlushSync();
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (!IsOpen())
+    return ERR_UNEXPECTED;
+
+  DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
+  return context_->FlushSync();
 }
 
 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);
+void FileStream::SetBoundNetLogSource(const BoundNetLog& owner_bound_net_log) {
+  if ((owner_bound_net_log.source().id == NetLog::Source::kInvalidId) &&
+      (bound_net_log_.source().id == 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(NetLog::TYPE_FILE_STREAM_BOUND_TO_OWNER,
+      owner_bound_net_log.source().ToEventParametersCallback());
+
+  owner_bound_net_log.AddEvent(NetLog::TYPE_FILE_STREAM_SOURCE,
+      bound_net_log_.source().ToEventParametersCallback());
 }
 
 base::PlatformFile FileStream::GetPlatformFileForTesting() {
-  return impl_.GetPlatformFileForTesting();
+  return context_->file();
 }
 
 }  // namespace net