Disk Cache: Replace the worker pool with a sequenced worker pool (posix).

net/disk_cache/file_posix.cc

Index: net/disk_cache/file_posix.cc
diff --git a/net/disk_cache/file_posix.cc b/net/disk_cache/file_posix.cc
index 2ad3db916e329b3f968b2ca4be59111ad41521d5..e817c61787972cc32167365dcb41488de3753aaa 100644
--- a/net/disk_cache/file_posix.cc
+++ b/net/disk_cache/file_posix.cc
@@ -4,165 +4,28 @@
 
 #include "net/disk_cache/file.h"
 
-#include <fcntl.h>
-
 #include "base/bind.h"
+#include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/logging.h"
-#include "base/threading/worker_pool.h"
+#include "base/run_loop.h"
+#include "base/task_runner_util.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "net/base/net_errors.h"
 #include "net/disk_cache/disk_cache.h"
-#include "net/disk_cache/in_flight_io.h"
 
 namespace {
 
-// This class represents a single asynchronous IO operation while it is being
-// bounced between threads.
-class FileBackgroundIO : public disk_cache::BackgroundIO {
- public:
-  // Other than the actual parameters for the IO operation (including the
-  // |callback| that must be notified at the end), we need the controller that
-  // is keeping track of all operations. When done, we notify the controller
-  // (we do NOT invoke the callback), in the worker thead that completed the
-  // operation.
-  FileBackgroundIO(disk_cache::File* file, const void* buf, size_t buf_len,
-                   size_t offset, disk_cache::FileIOCallback* callback,
-                   disk_cache::InFlightIO* controller)
-      : disk_cache::BackgroundIO(controller), callback_(callback), file_(file),
-        buf_(buf), buf_len_(buf_len), offset_(offset) {
-  }
-
-  disk_cache::FileIOCallback* callback() {
-    return callback_;
-  }
-
-  disk_cache::File* file() {
-    return file_;
-  }
-
-  // Read and Write are the operations that can be performed asynchronously.
-  // The actual parameters for the operation are setup in the constructor of
-  // the object. Both methods should be called from a worker thread, by posting
-  // a task to the WorkerPool (they are RunnableMethods). When finished,
-  // controller->OnIOComplete() is called.
-  void Read();
-  void Write();
+// The maximum number of threads for this pool.
+const int kMaxThreads = 5;
 
- private:
-  virtual ~FileBackgroundIO() {}
-
-  disk_cache::FileIOCallback* callback_;
-
-  disk_cache::File* file_;
-  const void* buf_;
-  size_t buf_len_;
-  size_t offset_;
-
-  DISALLOW_COPY_AND_ASSIGN(FileBackgroundIO);
-};
-
-
-// The specialized controller that keeps track of current operations.
-class FileInFlightIO : public disk_cache::InFlightIO {
+class FileWorkerPool : public base::SequencedWorkerPool {
  public:
-  FileInFlightIO() {}
-  virtual ~FileInFlightIO() {}
-
-  // These methods start an asynchronous operation. The arguments have the same
-  // semantics of the File asynchronous operations, with the exception that the
-  // operation never finishes synchronously.
-  void PostRead(disk_cache::File* file, void* buf, size_t buf_len,
-                size_t offset, disk_cache::FileIOCallback* callback);
-  void PostWrite(disk_cache::File* file, const void* buf, size_t buf_len,
-                 size_t offset, disk_cache::FileIOCallback* callback);
-
- protected:
-  // Invokes the users' completion callback at the end of the IO operation.
-  // |cancel| is true if the actual task posted to the thread is still
-  // queued (because we are inside WaitForPendingIO), and false if said task is
-  // the one performing the call.
-  virtual void OnOperationComplete(disk_cache::BackgroundIO* operation,
-                                   bool cancel) OVERRIDE;
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(FileInFlightIO);
+  FileWorkerPool() : base::SequencedWorkerPool(kMaxThreads, "CachePool") {}
 };
 
-// ---------------------------------------------------------------------------
-
-// Runs on a worker thread.
-void FileBackgroundIO::Read() {
-  if (file_->Read(const_cast<void*>(buf_), buf_len_, offset_)) {
-    result_ = static_cast<int>(buf_len_);
-  } else {
-    result_ = net::ERR_CACHE_READ_FAILURE;
-  }
-  NotifyController();
-}
-
-// Runs on a worker thread.
-void FileBackgroundIO::Write() {
-  bool rv = file_->Write(buf_, buf_len_, offset_);
-
-  result_ = rv ? static_cast<int>(buf_len_) : net::ERR_CACHE_WRITE_FAILURE;
-  NotifyController();
-}
-
-// ---------------------------------------------------------------------------
-
-void FileInFlightIO::PostRead(disk_cache::File *file, void* buf, size_t buf_len,
-                          size_t offset, disk_cache::FileIOCallback *callback) {
-  scoped_refptr<FileBackgroundIO> operation(
-      new FileBackgroundIO(file, buf, buf_len, offset, callback, this));
-  file->AddRef();  // Balanced on OnOperationComplete()
-
-  base::WorkerPool::PostTask(FROM_HERE,
-      base::Bind(&FileBackgroundIO::Read, operation.get()), true);
-  OnOperationPosted(operation.get());
-}
-
-void FileInFlightIO::PostWrite(disk_cache::File* file, const void* buf,
-                           size_t buf_len, size_t offset,
-                           disk_cache::FileIOCallback* callback) {
-  scoped_refptr<FileBackgroundIO> operation(
-      new FileBackgroundIO(file, buf, buf_len, offset, callback, this));
-  file->AddRef();  // Balanced on OnOperationComplete()
-
-  base::WorkerPool::PostTask(FROM_HERE,
-      base::Bind(&FileBackgroundIO::Write, operation.get()), true);
-  OnOperationPosted(operation.get());
-}
-
-// Runs on the IO thread.
-void FileInFlightIO::OnOperationComplete(disk_cache::BackgroundIO* operation,
-                                         bool cancel) {
-  FileBackgroundIO* op = static_cast<FileBackgroundIO*>(operation);
-
-  disk_cache::FileIOCallback* callback = op->callback();
-  int bytes = operation->result();
-
-  // Release the references acquired in PostRead / PostWrite.
-  op->file()->Release();
-  callback->OnFileIOComplete(bytes);
-}
-
-// A static object tha will broker all async operations.
-FileInFlightIO* s_file_operations = NULL;
-
-// Returns the current FileInFlightIO.
-FileInFlightIO* GetFileInFlightIO() {
-  if (!s_file_operations) {
-    s_file_operations = new FileInFlightIO;
-  }
-  return s_file_operations;
-}
-
-// Deletes the current FileInFlightIO.
-void DeleteFileInFlightIO() {
-  DCHECK(s_file_operations);
-  delete s_file_operations;
-  s_file_operations = NULL;
-}
+base::LazyInstance<FileWorkerPool>::Leaky s_worker_pool =
+    LAZY_INSTANCE_INITIALIZER;
 
 }  // namespace
 
@@ -205,8 +68,9 @@ bool File::IsValid() const {
 bool File::Read(void* buffer, size_t buffer_len, size_t offset) {

[gavinp, 2013/09/09 21:18:16]

The references to init_ and platform_file_ from these methods sort of scare me,
since they look like they could race. Reading the code it looks safe, since the
platform_file_ is set with init and then left, and then only ever cleared in the
destructor.

What about making these methods static, or putting them in an anonymous
namespace. Pass in the platform_file explicitly, and then the DCHECK can compare
against kInvalidPlatformFileValue, which should be equivalent to checking init?

Of course, this means that the File can't destruct while the operation is
pending (since the file would close before it's used). But that's already
guarded by the use of PostTaskAndReply. So I dunno.

So I guess it's OK as it is, but I just wanted to share that idea in case you
think it's clearer/safer for future coders?

[rvargas (doing something else), 2013/09/09 22:26:57]

I don't think init_ is a problem as it is guarding file open which should be
single thread.

As for the use of platform_file_, I don't really see static methods helping
readability. Object use semantics are not precluding 'races' in the sense of
interactions between multiple operations that the caller wishes to perform
simultaneously, but the object itself should protect about conventional races of
access of its members. Neither one of those properties are affected by jumping
through static methods (I think).

Please correct me if I'm wrong, but the OS should handle simultaneous access to
the same fd from multiple threads without any issues.

[gavinp, 2013/09/10 17:50:15]

I agree.
You're right; I sometimes find the pattern of never, ever accessing members of a
class from another thread can make readability better; when you have that
invariant, it's easier to then read through the accesses on the one thread that
does do it, and figure out if there's a conventional race there. However, in the
case of this class, the use was safe and guarded well enough by well defined
barriers that I think it's fine without my style nit.
The OS should be totally fine on that. Things get tricky when the file is
getting closed, but this code carefully guards against this by closing in a
destructor that runs on the cache thread, and all running worker pools hold a
reference, so it's perfect.

   DCHECK(init_);
   if (buffer_len > static_cast<size_t>(kint32max) ||
-      offset > static_cast<size_t>(kint32max))
+      offset > static_cast<size_t>(kint32max)) {
     return false;
+  }
 
   int ret = base::ReadPlatformFile(platform_file_, offset,
                                    static_cast<char*>(buffer), buffer_len);
@@ -216,8 +80,9 @@ bool File::Read(void* buffer, size_t buffer_len, size_t offset) {
 bool File::Write(const void* buffer, size_t buffer_len, size_t offset) {
   DCHECK(init_);
   if (buffer_len > static_cast<size_t>(kint32max) ||
-      offset > static_cast<size_t>(kint32max))
+      offset > static_cast<size_t>(kint32max)) {
     return false;
+  }
 
   int ret = base::WritePlatformFile(platform_file_, offset,
                                     static_cast<const char*>(buffer),
@@ -225,9 +90,6 @@ bool File::Write(const void* buffer, size_t buffer_len, size_t offset) {
   return (static_cast<size_t>(ret) == buffer_len);
 }
 
-// We have to increase the ref counter of the file before performing the IO to
-// prevent the completion to happen with an invalid handle (if the file is
-// closed while the IO is in flight).
 bool File::Read(void* buffer, size_t buffer_len, size_t offset,
                 FileIOCallback* callback, bool* completed) {
   DCHECK(init_);
@@ -237,10 +99,15 @@ bool File::Read(void* buffer, size_t buffer_len, size_t offset,
     return Read(buffer, buffer_len, offset);
   }
 
-  if (buffer_len > ULONG_MAX || offset > ULONG_MAX)
+  if (buffer_len > static_cast<size_t>(kint32max) ||
+      offset > static_cast<size_t>(kint32max)) {
     return false;
+  }
 
-  GetFileInFlightIO()->PostRead(this, buffer, buffer_len, offset, callback);
+  base::PostTaskAndReplyWithResult(
+      s_worker_pool.Pointer(), FROM_HERE,
+      base::Bind(&File::DoRead, this, buffer, buffer_len, offset),
+      base::Bind(&File::OnOperationComplete, this, callback));
 
   *completed = false;
   return true;
@@ -255,12 +122,23 @@ bool File::Write(const void* buffer, size_t buffer_len, size_t offset,
     return Write(buffer, buffer_len, offset);
   }
 
-  return AsyncWrite(buffer, buffer_len, offset, callback, completed);
+  if (buffer_len > static_cast<size_t>(kint32max) ||

[gavinp, 2013/09/09 21:18:16]

I'd have written this as std::numeric_limits<int32>::max(). But kint32max is in
basictypes. I don't know if either of them should be preferable.

[rvargas (doing something else), 2013/09/09 22:26:57]

Given that both are valid today I have a strong preference for the shorter one.

[gavinp, 2013/09/10 17:50:15]

SGTM. I would probably use numeric_limits<>(), but I have a hard time saying
that's anything other than preference.

+      offset > static_cast<size_t>(kint32max)) {
+    return false;
+  }
+
+  base::PostTaskAndReplyWithResult(
+      s_worker_pool.Pointer(), FROM_HERE,
+      base::Bind(&File::DoWrite, this, buffer, buffer_len, offset),
+      base::Bind(&File::OnOperationComplete, this, callback));
+
+  *completed = false;
+  return true;
 }
 
 bool File::SetLength(size_t length) {
   DCHECK(init_);
-  if (length > ULONG_MAX)
+  if (length > kuint32max)
     return false;
 
   return base::TruncatePlatformFile(platform_file_, length);
@@ -268,24 +146,22 @@ bool File::SetLength(size_t length) {
 
 size_t File::GetLength() {
   DCHECK(init_);
-  off_t ret = lseek(platform_file_, 0, SEEK_END);
-  if (ret < 0)
-    return 0;
-  return ret;
-}
+  int64 len = base::SeekPlatformFile(platform_file_,
+                                     base::PLATFORM_FILE_FROM_END, 0);
 
-// Static.
-void File::WaitForPendingIO(int* num_pending_io) {
-  // We may be running unit tests so we should allow be able to reset the
-  // message loop.
-  GetFileInFlightIO()->WaitForPendingIO();
-  DeleteFileInFlightIO();
+  if (len > static_cast<int64>(kuint32max))
+    return kuint32max;
+
+  return static_cast<size_t>(len);
 }
 
 // Static.
-void File::DropPendingIO() {
-  GetFileInFlightIO()->DropPendingIO();
-  DeleteFileInFlightIO();
+void File::WaitForPendingIO(int* num_pending_io) {
+  // We are running unit tests so we should wait for all callbacks. Sadly, the
+  // worker pool only waits for tasks on the worker pool, not the "Reply" tasks
+  // so we have to let the current message loop to run.
+  s_worker_pool.Get().FlushForTesting();
+  base::RunLoop().RunUntilIdle();
 }
 
 File::~File() {
@@ -293,17 +169,26 @@ File::~File() {
     base::ClosePlatformFile(platform_file_);
 }
 
-bool File::AsyncWrite(const void* buffer, size_t buffer_len, size_t offset,
-                      FileIOCallback* callback, bool* completed) {
-  DCHECK(init_);
-  if (buffer_len > ULONG_MAX || offset > ULONG_MAX)
-    return false;
+// Runs on a worker thread.
+int File::DoRead(void* buffer, size_t buffer_len, size_t offset) {
+  if (Read(const_cast<void*>(buffer), buffer_len, offset))
+    return static_cast<int>(buffer_len);
 
-  GetFileInFlightIO()->PostWrite(this, buffer, buffer_len, offset, callback);
+  return net::ERR_CACHE_READ_FAILURE;
+}
 
-  if (completed)
-    *completed = false;
-  return true;
+// Runs on a worker thread.
+int File::DoWrite(const void* buffer, size_t buffer_len, size_t offset) {
+  if (Write(const_cast<void*>(buffer), buffer_len, offset))
+    return static_cast<int>(buffer_len);
+
+  return net::ERR_CACHE_WRITE_FAILURE;
+}
+
+// This method actually makes sure that the last reference to the file doesn't
+// go away on the worker pool.
+void File::OnOperationComplete(FileIOCallback* callback, int result) {
+  callback->OnFileIOComplete(result);
 }
 
 }  // namespace disk_cache