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 68 matching lines @@
                                const CompletionCallback& callback) {
   if (index)
     index->Remove(key);
   WorkerPool::PostTask(FROM_HERE,
                        base::Bind(&SimpleSynchronousEntry::DoomEntry, path, key,
                                   MessageLoopProxy::current(), callback),
                        true);
   return net::ERR_IO_PENDING;
 }
 
 void SimpleEntryImpl::Doom() {

[gavinp, 2013/04/17 08:12:35]

I think doom is one of the only operations that already can be run completely
concurrently with anything else (including other Dooms).

[felipeg, 2013/04/17 10:06:43]

Done.

+  LOG(INFO) << " DOOM " << key_;
   DCHECK(io_thread_checker_.CalledOnValidThread());
+  if (synchronous_entry_in_use_by_worker_) {
+    // Postpone doom operation.
+    operations_.push(
+        base::Bind(&SimpleEntryImpl::DoomInternal,
+                   weak_ptr_factory_.GetWeakPtr()));
+    return;
+  }
+  DCHECK(operations_.size() == 0);
+  DCHECK(!synchronous_entry_in_use_by_worker_);
+
+  DoomInternal();
+  // TODO(felipeg): Maybe we should have a flag is_doomed_ and do some DCHECKS
+  // around the code, to see if we dont call read in a doomed entry.

[gavinp, 2013/04/17 08:12:35]

I believe it's correct and valid to Read/Write in a doomed entry.

[felipeg, 2013/04/17 10:06:43]

Done.

+  RunNextOperationIfNeeded();
+}
+
+void SimpleEntryImpl::DoomInternal() {
+  LOG(INFO) << " DOOM " << key_;
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  DCHECK(!synchronous_entry_in_use_by_worker_);
 #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() {
+  LOG(INFO) << " CLOSE " << key_;
   DCHECK(io_thread_checker_.CalledOnValidThread());
-  if (!synchronous_entry_in_use_by_worker_) {
-    WorkerPool::PostTask(FROM_HERE,
-                         base::Bind(&SimpleSynchronousEntry::Close,
-                                    base::Unretained(synchronous_entry_)),
-                         true);
+  if (synchronous_entry_in_use_by_worker_) {
+    LOG(INFO) << " CLOSE " << key_;
+    // Postpone close operation.
+    // Push the close operation to the end of the line. This way we run all
+    // operations before we are able close.
+    operations_.push(
+        base::Bind(&SimpleEntryImpl::Close,
+                   weak_ptr_factory_.GetWeakPtr()));
+    return;
   }
+  LOG(INFO) << " CLOSE " << key_;
+  DCHECK(operations_.size() == 0);
+  DCHECK(!synchronous_entry_in_use_by_worker_);
+
+  WorkerPool::PostTask(FROM_HERE,
+                       base::Bind(&SimpleSynchronousEntry::Close,
+                                  base::Unretained(synchronous_entry_)),
+                       true);
   // Entry::Close() is expected to release this entry. See disk_cache.h for
   // details.
   delete this;
 }
 
 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) {
+  LOG(INFO) << " READ " << key_;
   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);
+  if (synchronous_entry_in_use_by_worker_ && !read_only_operation_) {
+    // Postpone read operation.
+    operations_.push(
+        base::Bind(&SimpleEntryImpl::ReadDataInternal,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   index,
+                   offset,
+                   make_scoped_refptr(buf),
+                   buf_len,
+                   callback));
+    return net::ERR_IO_PENDING;
   }
+  DCHECK(operations_.size() == 0);
+  ReadDataInternal(index,
+                   offset,
+                   make_scoped_refptr(buf),
+                   buf_len,
+                   callback);
+  return net::ERR_IO_PENDING;
+}
+
+void SimpleEntryImpl::ReadDataInternal(int index,
+                                       int offset,
+                                       scoped_refptr<net::IOBuffer> buf,
+                                       int buf_len,
+                                       const CompletionCallback& callback) {
+  LOG(INFO) << " READ " << key_;
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  DCHECK(!synchronous_entry_in_use_by_worker_);
+  read_only_operation_ = true;
   synchronous_entry_in_use_by_worker_ = true;
   if (index_)
     index_->UseIfExists(key_);
   SynchronousOperationCallback sync_operation_callback =
       base::Bind(&SimpleEntryImpl::EntryOperationComplete,
                  index_, callback, weak_ptr_factory_.GetWeakPtr(),
                  synchronous_entry_);
   WorkerPool::PostTask(FROM_HERE,
                        base::Bind(&SimpleSynchronousEntry::ReadData,
                                   base::Unretained(synchronous_entry_),
-                                  index, offset, make_scoped_refptr(buf),
+                                  index, offset, buf,

[gavinp, 2013/04/17 08:17:14]

This is really dangerous, calling with the naked buf. The problem is that the
caller can drop their ref to buf, and then buf is deleted under us. The existing
calls used make_scoped_refptr(), which gets copied into the Closure() and again
into the thread when it runs the Closure.

[gavinp, 2013/04/17 09:40:16]

Boy this stuff is subtle.

I was wrong: I failed to notice that ReadDataInternal's argument is a
scoped_refptr<>, so the Closure will copy _THAT_, so it's all good.

                                   buf_len, sync_operation_callback),
                        true);
-  return net::ERR_IO_PENDING;
+}
+
+// TODO(felipeg): Maybe in case of failure from a write operation, we could try
+// to invalidate all operations in the queue of the current entry.
+static void WriteDataNoOp(int noop) {
 }
 
 int SimpleEntryImpl::WriteData(int index,

[gavinp, 2013/04/17 09:23:56]

I think ReadData / WriteData can be made a lot simpler. How about this:

- Always push the Closure onto operations_, regardless of
synchronous_entry_in_use_by_worker_.
- Always call RunNextOperationIfNeeded.

(RunNextOperationIfNeeded should check synchronous_entry_in_use_by_worker_)

[felipeg, 2013/04/17 10:06:43]

Done.

                                int offset,
                                net::IOBuffer* buf,
                                int buf_len,
                                const CompletionCallback& callback,
                                bool truncate) {
+  LOG(INFO) << " WRITE " << key_ << "  buf " << buf << "    buf_len " << buf_len << "    truncate " << truncate;
   DCHECK(io_thread_checker_.CalledOnValidThread());
+
   if (synchronous_entry_in_use_by_worker_) {
-    NOTIMPLEMENTED();
-    CHECK(false);
+    LOG(INFO) << " WRITE " << key_;
+    // Postpone write operation.
+    operations_.push(
+        base::Bind(&SimpleEntryImpl::WriteDataInternal,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   index,
+                   offset,
+                   make_scoped_refptr(buf),
+                   buf_len,
+                   callback,
+                   truncate));
+    // When writting, if there is already a pending operation, we have to return
+    // pending.
+    return net::ERR_IO_PENDING;
   }
+  LOG(INFO) << " WRITE " << key_;
+  DCHECK(operations_.size() == 0);
+  WriteDataInternal(index,
+                    offset,
+                    make_scoped_refptr(buf),
+                    buf_len,
+                    //callback,
+                    base::Bind(&WriteDataNoOp),
+                    truncate);
+  //return net::ERR_IO_PENDING;
+  // Pretend everything goes fine, and make HTTP Cache Transaction happy.
+  return buf_len;
+}
+
+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(!synchronous_entry_in_use_by_worker_);
+  read_only_operation_ = false;
   synchronous_entry_in_use_by_worker_ = true;
   if (index_)
     index_->UseIfExists(key_);
+
+  // TODO(felipeg):   last_used_  last_modified_
+
+  data_size_[index] = buf_len;
+
   SynchronousOperationCallback sync_operation_callback =
       base::Bind(&SimpleEntryImpl::EntryOperationComplete,
                  index_, callback, weak_ptr_factory_.GetWeakPtr(),
                  synchronous_entry_);
   WorkerPool::PostTask(FROM_HERE,
                        base::Bind(&SimpleSynchronousEntry::WriteData,
                                   base::Unretained(synchronous_entry_),
-                                  index, offset, make_scoped_refptr(buf),
+                                  index, offset, buf,
                                   buf_len, sync_operation_callback, truncate),
                        true);
-  return net::ERR_IO_PENDING;
+  LOG(INFO) << " WRITE " << key_;
 }
 
 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();
   return net::ERR_FAILED;
@@ skipping 39 matching lines @@
 }
 
 SimpleEntryImpl::SimpleEntryImpl(
     SimpleSynchronousEntry* synchronous_entry,
     WeakPtr<SimpleIndex> index)
     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_ptr_factory_(this)),
       path_(synchronous_entry->path()),
       key_(synchronous_entry->key()),
       synchronous_entry_(synchronous_entry),
       synchronous_entry_in_use_by_worker_(false),
+      read_only_operation_(false),
       index_(index) {
   DCHECK(synchronous_entry);
   SetSynchronousData();
 }
 
 SimpleEntryImpl::~SimpleEntryImpl() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
 }
 
 // static
@@ skipping 25 matching lines @@
     int result) {
   DCHECK(sync_entry);
   if (index) {
     if (result >= 0)
       index->UpdateEntrySize(sync_entry->key(), sync_entry->GetFileSize());
     else
       index->Remove(sync_entry->key());
   }
 
   if (entry) {
+    if (entry->read_only_operation_)
+      LOG(INFO) << " READ EntryOperationComplete " << entry->key();
+    else
+      LOG(INFO) << " WRITE EntryOperationComplete " << entry->key();
     DCHECK(entry->synchronous_entry_in_use_by_worker_);
+    entry->read_only_operation_ = false;
     entry->synchronous_entry_in_use_by_worker_ = false;
     entry->SetSynchronousData();
+
+    entry->RunNextOperationIfNeeded();
+
+    completion_callback.Run(result);
   } else {
+
+    LOG(INFO) << " EntryOperationComplete " << entry->key();
     // |entry| must have had Close() called while this operation was in flight.
     // Since the simple cache now only supports one pending entry operation in
     // flight at a time, it's safe to now call Close() on |sync_entry|.
     WorkerPool::PostTask(FROM_HERE,
                          base::Bind(&SimpleSynchronousEntry::Close,
                                     base::Unretained(sync_entry)),
                          true);
+
+    completion_callback.Run(result);
   }
-  completion_callback.Run(result);
+}
+
+void SimpleEntryImpl::RunNextOperationIfNeeded() {
+  LOG(INFO) << " RunNextOperationIfNeeded " << key_;
+  DCHECK(io_thread_checker_.CalledOnValidThread());
+  if (operations_.size() <= 0)
+    return;
+  base::Closure operation = operations_.front();
+  operations_.pop();
+
+  LOG(INFO) << " operation.Run " << key_;
+  // TODO(felipeg): maybe we should post task. But it could shuffle even more
+  // the order of new operations in this entry.

[gavinp, 2013/04/17 08:17:14]

The callers should be totally insensitive to the order of operations. However, I
say no post: this should be cheap anyway right?

[felipeg, 2013/04/17 10:06:43]

No post is better because when debugging our minds won't blow.

+  operation.Run();
 }
 
 void SimpleEntryImpl::SetSynchronousData() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK(!synchronous_entry_in_use_by_worker_);
   // 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