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)
@@ -2,97 +2,265 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "base/threading/thread_restrictions.h"
 #include "net/base/file_stream.h"
+#if defined(OS_WIN)
+#include "net/base/file_stream_win.h"
+#elif defined(OS_POSIX)
+#include "net/base/file_stream_posix.h"
+#endif
+#include "net/base/net_errors.h"
 
 namespace net {
 
 FileStream::FileStream(net::NetLog* net_log)
-    : impl_(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);
 }
 
-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)
+    : context_(NULL),
+      open_flags_(flags),
+      bound_net_log_(net::BoundNetLog::Make(net_log,
+                                            net::NetLog::SOURCE_FILESTREAM)) {
+  context_ = new AsyncContext(file, bound_net_log_, open_flags_);
+
+  bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
 }
 
 FileStream::~FileStream() {
+  if (IsOpen() && !is_async())
+    CloseSync();
+  context_->Destroy();
+
+  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();
+  // 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(is_async()) once all async clients are migrated to
+  // use Close(). crbug.com/114783
+  if (!context_->async_in_progress()) {
+    context_->CloseSync();
+  } else {
+    AsyncContext* old_ctx = context_;

[willchan no longer on Chromium, 2012/08/27 06:24:06]

s/old_ctx/old_context/

+    context_ = new AsyncContext(bound_net_log_);
+    context_->set_record_uma(old_ctx->record_uma());
+    old_ctx->Destroy();
+  }
 }
 
 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) {

[willchan no longer on Chromium, 2012/08/27 06:24:06]

Since all the cross platform code is in file_stream.cc now (nice!), can you move
all the base::ThreadRestrictions::AssertIOAllowed() calls to the top of the
FooSync() methods? Once that's done, I believe we won't need AssertIOAllowed()
at all in the _posix and _win files, right?

[pivanof, 2012/08/27 08:43:31]

There are still some similarities between _posix and _win which probably are
better to be in one place. And I just got an idea: will it be okay if
FileStream::Context will be declared in _posix.h and _win.h, platform dependent
methods of Context will be defined in _posix.cc and _win.cc but common methods
will be defined in file_stream.cc?
Hm... I agree that it's certainly a good idea to put AssertIOAllowed() in front
of all FooSync() methods here. But if we remove all calls to AssertIOAllowed()
from platform files then we will rely on the fact that certain methods are
called either from FooSync() or from WorkerPool. Is it okay to rely that there
won't be some accidental change later breaking this assumption?

[willchan no longer on Chromium, 2012/09/05 00:03:38]

I see. We generally try to prevent separation of code like this, although we do
do it in base/ in a few places. That's primarily because lots of that code is
not object based. That said, if you think it will make the code cleaner, I'm
open to it.
Please correct me if I'm misunderstanding, but all the entrypoints to the
platform specific files are from FileStream, right? If that's true, and there's
no other way to enter this code, I'd rather have the assertion done at the
obvious entrypoints in FileStream, since it makes it more easily verifiable that
all the sync methods have AssertIOAllowed().

-  return impl_.OpenSync(path, open_flags);
+  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);
+  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);
+  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);
+  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);
+  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