Make FileStream::Seek async and add FileStream::SeekSync for sync operation

net/base/file_stream_win.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.
 
 #include "net/base/file_stream.h"
 
 #include <windows.h>
 
 #include "base/file_path.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/metrics/histogram.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/threading/worker_pool.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"
 
 namespace net {
 
 // Ensure that we can just use our Whence values directly.
 COMPILE_ASSERT(FROM_BEGIN == FILE_BEGIN, bad_whence_begin);
 COMPILE_ASSERT(FROM_CURRENT == FILE_CURRENT, bad_whence_current);
 COMPILE_ASSERT(FROM_END == FILE_END, bad_whence_end);
 
-static void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) {
+namespace {
+
+void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) {
   overlapped->Offset = offset.LowPart;
   overlapped->OffsetHigh = offset.HighPart;
 }
 
-static void IncrementOffset(OVERLAPPED* overlapped, DWORD count) {
+void IncrementOffset(OVERLAPPED* overlapped, DWORD count) {
   LARGE_INTEGER offset;
   offset.LowPart = overlapped->Offset;
   offset.HighPart = overlapped->OffsetHigh;
   offset.QuadPart += static_cast<LONGLONG>(count);
   SetOffset(overlapped, offset);
 }
 
-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,
       make_scoped_refptr(
           new FileStreamErrorParameters(GetFileErrorSourceName(source),
@@ skipping 25 matching lines @@
     LOG(WARNING) << "Failed to open file: " << error;
     *result = RecordAndMapError(error,
                                 FILE_ERROR_SOURCE_OPEN,
                                 record_uma,
                                 bound_net_log);
     bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN, NULL);
     return;
   }
 }
 
-// Opens a file using OpenFile() and 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, NULL);
   if (file == base::kInvalidPlatformFileValue)
     return;
 
   CancelIo(file);
 
   if (!base::ClosePlatformFile(file))
     NOTREACHED();
   bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN, NULL);
 }
 
 // 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();
 }
 
+// Invokes a given closure and signals the completion.
+void InvokeAndSignal(const base::Closure& closure,
+                     base::WaitableEvent* on_io_complete) {
+  closure.Run();
+  on_io_complete->Signal();
+}
+
 }  // namespace
 
 // FileStreamWin::AsyncContext ----------------------------------------------
 
 class FileStreamWin::AsyncContext : public MessageLoopForIO::IOHandler {
  public:
   explicit AsyncContext(const net::BoundNetLog& bound_net_log)
       : context_(), is_closing_(false),
         record_uma_(false), bound_net_log_(bound_net_log),
         error_source_(FILE_ERROR_SOURCE_COUNT) {
@@ skipping 131 matching lines @@
       }
     } else {
       CloseSync();
     }
   }
 
   bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE, NULL);
 }
 
 void FileStreamWin::Close(const CompletionCallback& callback) {
-  DCHECK(callback_.is_null());
-  callback_ = 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(&FileStreamWin::OnClosed, weak_ptr_factory_.GetWeakPtr()),
+      base::Bind(&FileStreamWin::OnClosed,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback),
       true /* task_is_slow */);
   DCHECK(posted);
 }
 
 void FileStreamWin::CloseSync() {
   // The logic here is similar to CloseFile() but async_context_.reset() is
   // caled in this function.
 
   // Block until the in-flight open operation is complete.
   // TODO(satorux): Replace this with a DCHECK(open_flags & ASYNC) once this
@@ skipping 16 matching lines @@
   }
 }
 
 int FileStreamWin::Open(const FilePath& path, int open_flags,
                         const CompletionCallback& callback) {
   if (IsOpen()) {
     DLOG(FATAL) << "File is already open!";
     return ERR_UNEXPECTED;
   }
 
-  DCHECK(callback_.is_null());
-  callback_ = callback;
-
   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(&InvokeAndSignal,
+                 base::Bind(&OpenFile, path, open_flags, record_uma_, &file_,
+                            result, bound_net_log_),
+                 on_io_complete_.get()),
       base::Bind(&FileStreamWin::OnOpened,
                  weak_ptr_factory_.GetWeakPtr(),
-                 base::Owned(result)),
+                 callback, base::Owned(result)),
       true /* task_is_slow */);
   DCHECK(posted);
   return ERR_IO_PENDING;
 }
 
 int FileStreamWin::OpenSync(const FilePath& path, int open_flags) {
   if (IsOpen()) {
     DLOG(FATAL) << "File is already open!";
     return ERR_UNEXPECTED;
   }
@@ skipping 15 matching lines @@
                                                    async_context_.get());
   }
 
   return OK;
 }
 
 bool FileStreamWin::IsOpen() const {
   return file_ != base::kInvalidPlatformFileValue;
 }
 
-int64 FileStreamWin::Seek(Whence whence, int64 offset) {
+int FileStreamWin::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());
+
+  int64* result = new int64(-1);
+  on_io_complete_.reset(new base::WaitableEvent(
+      false  /* manual_reset */, false  /* initially_signaled */));
+
+  const bool posted = base::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&InvokeAndSignal,
+                 // Unretained should be fine as we wait signal on

[willchan no longer on Chromium, 2012/04/10 06:48:33]

Nit: s/we wait signal on/we wait for a signal on/

+                 // on_io_complete_ at the destructor.
+                 base::Bind(&FileStreamWin::SeekFile, base::Unretained(this),
+                            whence, offset, result),
+                 on_io_complete_.get()),
+      base::Bind(&FileStreamWin::OnSeeked,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback, base::Owned(result)),
+      true /* task is slow */);
+  DCHECK(posted);
+  return ERR_IO_PENDING;
+}
+
+int64 FileStreamWin::SeekSync(Whence whence, int64 offset) {
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
   DCHECK(!async_context_.get() || async_context_->callback().is_null());
-
-  LARGE_INTEGER distance, result;
-  distance.QuadPart = offset;
-  DWORD move_method = static_cast<DWORD>(whence);
-  if (!SetFilePointerEx(file_, distance, &result, move_method)) {
-    DWORD error = GetLastError();
-    LOG(WARNING) << "SetFilePointerEx failed: " << error;
-    return RecordAndMapError(error,
-                             FILE_ERROR_SOURCE_SEEK,
-                             record_uma_,
-                             bound_net_log_);
-  }
-  if (async_context_.get()) {
-    async_context_->set_error_source(FILE_ERROR_SOURCE_SEEK);
-    SetOffset(async_context_->overlapped(), result);
-  }
-  return result.QuadPart;
+  int64 result = -1;
+  SeekFile(whence, offset, &result);
+  return result;
 }
 
 int64 FileStreamWin::Available() {
   base::ThreadRestrictions::AssertIOAllowed();
 
   if (!IsOpen())
     return ERR_UNEXPECTED;
 
-  int64 cur_pos = Seek(FROM_CURRENT, 0);
+  int64 cur_pos = SeekSync(FROM_CURRENT, 0);
   if (cur_pos < 0)
     return cur_pos;
 
   LARGE_INTEGER file_size;
   if (!GetFileSizeEx(file_, &file_size)) {
     DWORD error = GetLastError();
     LOG(WARNING) << "GetFileSizeEx failed: " << error;
     return RecordAndMapError(error,
                              FILE_ERROR_SOURCE_GET_SIZE,
                              record_uma_,
@@ skipping 177 matching lines @@
 int64 FileStreamWin::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 = Seek(FROM_BEGIN, bytes);
+  int64 seek_position = SeekSync(FROM_BEGIN, bytes);
   if (seek_position != bytes)
     return ERR_UNEXPECTED;
 
   // And truncate the file.
   BOOL result = SetEndOfFile(file_);
   if (!result) {
     DWORD error = GetLastError();
     LOG(WARNING) << "SetEndOfFile failed: " << error;
     return RecordAndMapError(error,
                              FILE_ERROR_SOURCE_SET_EOF,
@@ skipping 33 matching lines @@
       net::NetLog::TYPE_FILE_STREAM_SOURCE,
       make_scoped_refptr(
           new net::NetLogSourceParameter("source_dependency",
                                          bound_net_log_.source())));
 }
 
 base::PlatformFile FileStreamWin::GetPlatformFileForTesting() {
   return file_;
 }
 
-void FileStreamWin::OnClosed() {
+void FileStreamWin::OnClosed(const CompletionCallback& callback) {
   file_ = base::kInvalidPlatformFileValue;
 
-  CompletionCallback temp = callback_;
-  callback_.Reset();
-
-  // Reset this before Run(). Run() should not issue a new async operation
-  // here, but just to keep it consistent with OnOpened().
-  on_io_complete_.reset();
-  temp.Run(OK);
+  // Reset this before Run() as Run() may issue a new async operation.
+  ResetOnIOComplete();
+  callback.Run(OK);
 }
 
-void FileStreamWin::OnOpened(int* result) {
+void FileStreamWin::SeekFile(Whence whence, int64 offset, int64* result) {
+  LARGE_INTEGER distance, res;
+  distance.QuadPart = offset;
+  DWORD move_method = static_cast<DWORD>(whence);
+  if (!SetFilePointerEx(file_, distance, &res, move_method)) {
+    DWORD error = GetLastError();
+    LOG(WARNING) << "SetFilePointerEx failed: " << error;
+    *result = RecordAndMapError(error,
+                                FILE_ERROR_SOURCE_SEEK,
+                                record_uma_,
+                                bound_net_log_);
+    return;
+  }
+  if (async_context_.get()) {
+    async_context_->set_error_source(FILE_ERROR_SOURCE_SEEK);
+    SetOffset(async_context_->overlapped(), res);
+  }
+  *result = res.QuadPart;
+}
+
+void FileStreamWin::OnOpened(const CompletionCallback& callback, int* result) {
   if (*result == OK) {
     async_context_.reset(new AsyncContext(bound_net_log_));
     if (record_uma_)
       async_context_->EnableErrorStatistics();
     MessageLoopForIO::current()->RegisterIOHandler(file_,
                                                    async_context_.get());
   }
 
-  CompletionCallback temp = callback_;
-  callback_.Reset();
+  // Reset this before Run() as Run() may issue a new async operation.
+  ResetOnIOComplete();
+  callback.Run(*result);
+}
 
+void FileStreamWin::OnSeeked(
+    const Int64CompletionCallback& callback,
+    int64* result) {
   // Reset this before Run() as Run() may issue a new async operation.
+  ResetOnIOComplete();
+  callback.Run(*result);
+}
+
+void FileStreamWin::ResetOnIOComplete() {
   on_io_complete_.reset();
-  temp.Run(*result);
+  weak_ptr_factory_.InvalidateWeakPtrs();
 }
 
 void FileStreamWin::WaitForIOCompletion() {
   if (on_io_complete_.get()) {
     on_io_complete_->Wait();
     on_io_complete_.reset();
   }
 }
 
 }  // namespace net