Support overlapping operations on the SimpleEntryImpl.

net/disk_cache/simple/simple_entry_impl.cc

 // Copyright (c) 2013 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/disk_cache/simple/simple_entry_impl.h"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #include "base/location.h"
@@ skipping 74 matching lines @@
   WorkerPool::PostTask(FROM_HERE,
                        base::Bind(&SimpleSynchronousEntry::DoomEntry, path, key,
                                   MessageLoopProxy::current(), callback),
                        true);
   return net::ERR_IO_PENDING;
 }
 
 void SimpleEntryImpl::Doom() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK(synchronous_entry_);
+  DCHECK(!operation_running_);

[gavinp, 2013/04/17 16:31:37]

I think this is bad; it's always valid to doom an entry.

[felipeg, 2013/04/17 16:54:59]

Done.

 #if defined(OS_POSIX)
   // This call to static SimpleEntryImpl::DoomEntry() will just erase the
   // underlying files. On POSIX, this is fine; the files are still open on the
   // SimpleSynchronousEntry, and operations can even happen on them. The files
   // will be removed from the filesystem when they are closed.
   DoomEntry(index_, path_, key_, CompletionCallback());
 #else
   NOTIMPLEMENTED();
 #endif
 }
 
 void SimpleEntryImpl::Close() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  Release();  // Balanced in CreationOperationCompleted().
+  if (operation_running_) {

[gavinp, 2013/04/17 16:31:37]

Why is the pattern here so different from ReadData / WriteData? Why not create
CloseImpl(), and make this look almost the same as them?

[felipeg, 2013/04/17 16:54:59]

Done.

+    // Postpone close operation.
+    // Push the close operation to the end of the line. This way we run all
+    // operations before we are able close.
+    pending_operations_.push(base::Bind(&SimpleEntryImpl::Close, this));
+    return;
+  } else {
+    DCHECK(pending_operations_.size() == 0);
+    DCHECK(!operation_running_);
+    DCHECK(synchronous_entry_);
+    WorkerPool::PostTask(FROM_HERE,
+                         base::Bind(&SimpleSynchronousEntry::Close,
+                                    base::Unretained(synchronous_entry_)),
+                         true);
+    // Entry::Close() is expected to delete this entry. See disk_cache.h for
+    // details.
+    Release();  // Balanced in CreationOperationCompleted().
+  }
 }
 
 std::string SimpleEntryImpl::GetKey() const {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   return key_;
 }
 
 Time SimpleEntryImpl::GetLastUsed() const {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   return last_used_;
 }
 
 Time SimpleEntryImpl::GetLastModified() const {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   return last_modified_;
 }
 
 int32 SimpleEntryImpl::GetDataSize(int index) const {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   return data_size_[index];
 }
 
 int SimpleEntryImpl::ReadData(int index,
                               int offset,
                               net::IOBuffer* buf,
                               int buf_len,
                               const CompletionCallback& callback) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  // TODO(gavinp): Add support for overlapping reads. The net::HttpCache does
-  // make overlapping read requests when multiple transactions access the same
-  // entry as read only. This might make calling SimpleSynchronousEntry::Close()
-  // correctly more tricky (see SimpleEntryImpl::EntryOperationComplete).
-  if (synchronous_entry_in_use_by_worker_) {
-    NOTIMPLEMENTED();
-    CHECK(false);
-  }
-  synchronous_entry_in_use_by_worker_ = true;
-  index_->UseIfExists(key_);
-  SynchronousOperationCallback sync_operation_callback =
-      base::Bind(&SimpleEntryImpl::EntryOperationComplete,
-                 this, callback);
-  WorkerPool::PostTask(FROM_HERE,
-                       base::Bind(&SimpleSynchronousEntry::ReadData,
-                                  base::Unretained(synchronous_entry_),
-                                  index, offset, make_scoped_refptr(buf),
-                                  buf_len, sync_operation_callback),
-                       true);
+  if (index < 0 || index >= kSimpleEntryFileCount || buf_len < 0)
+    return net::ERR_INVALID_ARGUMENT;
+  if (offset >= data_size_[index] || offset < 0 || !buf_len)
+    return 0;
+  // TODO(felipeg): Optimization: Add support for truly parallel read
+  // operations.
+  pending_operations_.push(
+      base::Bind(&SimpleEntryImpl::ReadDataInternal,
+                 this,
+                 index,
+                 offset,
+                 make_scoped_refptr(buf),
+                 buf_len,
+                 callback));
+  RunNextOperationIfNeeded();
   return net::ERR_IO_PENDING;
 }
 
 int SimpleEntryImpl::WriteData(int index,
                                int offset,
                                net::IOBuffer* buf,
                                int buf_len,
                                const CompletionCallback& callback,
                                bool truncate) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  if (synchronous_entry_in_use_by_worker_) {
-    NOTIMPLEMENTED();
-    CHECK(false);
-  }
-  synchronous_entry_in_use_by_worker_ = true;
-  index_->UseIfExists(key_);
-  SynchronousOperationCallback sync_operation_callback =
-      base::Bind(&SimpleEntryImpl::EntryOperationComplete,
-                 this, callback);
-  WorkerPool::PostTask(FROM_HERE,
-                       base::Bind(&SimpleSynchronousEntry::WriteData,
-                                  base::Unretained(synchronous_entry_),
-                                  index, offset, make_scoped_refptr(buf),
-                                  buf_len, sync_operation_callback, truncate),
-                       true);
+  if (index < 0 || index >= kSimpleEntryFileCount || offset < 0 || buf_len < 0)
+    return net::ERR_INVALID_ARGUMENT;
+  pending_operations_.push(
+      base::Bind(&SimpleEntryImpl::WriteDataInternal,
+                 this,
+                 index,
+                 offset,
+                 make_scoped_refptr(buf),
+                 buf_len,
+                 callback,
+                 truncate));
+  RunNextOperationIfNeeded();
+  // TODO(felipeg): Optimization: Add support for optimistic writes, quickly
+  // returning net::OK here.
   return net::ERR_IO_PENDING;
 }
 
 int SimpleEntryImpl::ReadSparseData(int64 offset,
                                     net::IOBuffer* buf,
                                     int buf_len,
                                     const CompletionCallback& callback) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   // TODO(gavinp): Determine if the simple backend should support sparse data.
   NOTIMPLEMENTED();
@@ skipping 35 matching lines @@
 int SimpleEntryImpl::ReadyForSparseIO(const CompletionCallback& callback) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   // TODO(gavinp): Determine if the simple backend should support sparse data.
   NOTIMPLEMENTED();
   return net::ERR_FAILED;
 }
 
 SimpleEntryImpl::SimpleEntryImpl(const scoped_refptr<SimpleIndex>& index,
                                  const base::FilePath& path,
                                  const std::string& key)
-    : constructor_thread_(base::MessageLoopProxy::current()),
-      index_(index),
+    : index_(index),
       path_(path),
       key_(key),
       synchronous_entry_(NULL),
-      synchronous_entry_in_use_by_worker_(false) {
+      operation_running_(false) {
+  DCHECK(index_.get());
 }
 
 SimpleEntryImpl::~SimpleEntryImpl() {
-  if (synchronous_entry_) {
-    base::Closure close_sync_entry =
-        base::Bind(&SimpleSynchronousEntry::Close,
-                   base::Unretained(synchronous_entry_));
-    // We aren't guaranteed to be able to run IO on our constructor thread, but
-    // we are also not guaranteed to be allowed to run WorkerPool::PostTask on
-    // our other threads.
-    if (constructor_thread_->BelongsToCurrentThread())
-      WorkerPool::PostTask(FROM_HERE, close_sync_entry, true);
-    else
-      close_sync_entry.Run();
-  }
+  DCHECK(pending_operations_.size() == 0);

[gavinp, 2013/04/17 16:31:37]

DCHECK_NE(0, pending_operations_.size());

[felipeg, 2013/04/17 16:54:59]

You mean:
  DCHECK_EQ(0U, pending_operations_.size());

+  DCHECK(!operation_running_);

[gavinp, 2013/04/17 16:31:37]

+ DCHECK(!synchronous_entry);

[felipeg, 2013/04/17 16:54:59]

you mean:
DCHECK(synchronous_entry);

[gavinp, 2013/04/18 05:39:42]

Hrrrm. I think I meant DCHECK(!synchronous_entry), but now I'm not sure that's
thread safe. If CloseInternal() sets synchronous_entry_ = NULL; the odds are
very good that the destructor will run on the IO thread, since we're only
running CloseInternal() because of the PostTask from the worker pool.

But now I'm not so sure: if we Release() in CloseInternal() but the worker pool
loop still holds a ref, we could delete on the IO thread and there may not have
been a barrier.

So now I think we'd have to do:

synchronous_entry_ = NULL;
subtle::MemoryBarrier();
Release();

in CloseInternal(). That's... sketchy.

So I'm now down to saying we can't assert on this.

+}
+
+bool SimpleEntryImpl::RunNextOperationIfNeeded() {
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  if (pending_operations_.size() <= 0 || operation_running_)
+    return false;
+  base::Closure operation = pending_operations_.front();
+  pending_operations_.pop();
+  operation.Run();
+  return true;
+}
+
+void SimpleEntryImpl::ReadDataInternal(int index,
+                                       int offset,
+                                       scoped_refptr<net::IOBuffer> buf,
+                                       int buf_len,
+                                       const CompletionCallback& callback) {
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  DCHECK(!operation_running_);
+  operation_running_ = true;
+  index_->UseIfExists(key_);
+  SynchronousOperationCallback sync_operation_callback =
+      base::Bind(&SimpleEntryImpl::EntryOperationComplete,
+                 this, callback);
+  WorkerPool::PostTask(FROM_HERE,
+                       base::Bind(&SimpleSynchronousEntry::ReadData,
+                                  base::Unretained(synchronous_entry_),
+                                  index, offset, buf,
+                                  buf_len, sync_operation_callback),
+                       true);
+}
+
+void SimpleEntryImpl::WriteDataInternal(int index,
+                                       int offset,
+                                       scoped_refptr<net::IOBuffer> buf,
+                                       int buf_len,
+                                       const CompletionCallback& callback,
+                                       bool truncate) {
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  DCHECK(!operation_running_);
+  operation_running_ = true;
+  index_->UseIfExists(key_);
+
+  last_used_ = base::Time::Now();
+  last_modified_ = base::Time::Now();
+  data_size_[index] = buf_len;

[gavinp, 2013/04/17 16:31:37]

This isn't right; I think this is code leftover from the "fast writes" idea,
which should be a separate CL.

[felipeg, 2013/04/17 16:54:59]

Independently if this is a leftover from fast writes, I still believe that here
is the place where it should be set.
But I am removing for now
we put it back when we need

[gavinp, 2013/04/18 05:39:42]

Yes, definitely. But data_size_[index] = buf_len won't be right either; more
like

if (truncate)
  data_size_[index] = offset + buf_len;
else
  data_size_[index] = std::max(offset + buf_len, data_size_[index]);

But that can wait for another CL... :D

+
+  SynchronousOperationCallback sync_operation_callback =
+      base::Bind(&SimpleEntryImpl::EntryOperationComplete,
+                 this, callback);
+  WorkerPool::PostTask(FROM_HERE,
+                       base::Bind(&SimpleSynchronousEntry::WriteData,
+                                  base::Unretained(synchronous_entry_),
+                                  index, offset, buf,
+                                  buf_len, sync_operation_callback, truncate),
+                       true);
 }
 
 void SimpleEntryImpl::CreationOperationComplete(
     Entry** out_entry,
     const CompletionCallback& completion_callback,
     SimpleSynchronousEntry* sync_entry) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   if (!sync_entry) {
     completion_callback.Run(net::ERR_FAILED);
     // If OpenEntry failed, we must remove it from our index.
@@ skipping 10 matching lines @@
   index_->Insert(key_);
   *out_entry = this;
   completion_callback.Run(net::OK);
 }
 
 void SimpleEntryImpl::EntryOperationComplete(
     const CompletionCallback& completion_callback,
     int result) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK(synchronous_entry_);
-  DCHECK(synchronous_entry_in_use_by_worker_);
-  synchronous_entry_in_use_by_worker_ = false;
+  DCHECK(operation_running_);
+
+  operation_running_ = false;
   SetSynchronousData();
   if (result >= 0) {
     index_->UpdateEntrySize(synchronous_entry_->key(),
                             synchronous_entry_->GetFileSize());
   } else {
     index_->Remove(synchronous_entry_->key());
   }
   completion_callback.Run(result);
+  RunNextOperationIfNeeded();
 }
 
 void SimpleEntryImpl::SetSynchronousData() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  DCHECK(!synchronous_entry_in_use_by_worker_);
+  DCHECK(!operation_running_);
   // TODO(felipeg): These copies to avoid data races are not optimal. While
   // adding an IO thread index (for fast misses etc...), we can store this data
   // in that structure. This also solves problems with last_used() on ext4
   // filesystems not being accurate.
   last_used_ = synchronous_entry_->last_used();
   last_modified_ = synchronous_entry_->last_modified();
   for (int i = 0; i < kSimpleEntryFileCount; ++i)
     data_size_[i] = synchronous_entry_->data_size(i);
 }
 
 }  // namespace disk_cache