SimpleCache: merge the first and second stream in one file

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 <algorithm>
 #include <cstring>
 #include <vector>
 
@@ skipping 319 matching lines @@
                               int buf_len,
                               const CompletionCallback& callback) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
 
   if (net_log_.IsLoggingAllEvents()) {
     net_log_.AddEvent(net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_READ_CALL,
         CreateNetLogReadWriteDataCallback(stream_index, offset, buf_len,
                                           false));
   }
 
-  if (stream_index < 0 || stream_index >= kSimpleEntryFileCount ||
+  if (stream_index < 0 || stream_index >= kSimpleEntryStreamCount ||
       buf_len < 0) {
     if (net_log_.IsLoggingAllEvents()) {
       net_log_.AddEvent(net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_READ_END,
           CreateNetLogReadWriteCompleteCallback(net::ERR_INVALID_ARGUMENT));
     }
 
     RecordReadResult(cache_type_, READ_RESULT_INVALID_ARGUMENT);
     return net::ERR_INVALID_ARGUMENT;
   }
   if (pending_operations_.empty() && (offset >= GetDataSize(stream_index) ||
@@ skipping 25 matching lines @@
                                bool truncate) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
 
   if (net_log_.IsLoggingAllEvents()) {
     net_log_.AddEvent(
         net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_WRITE_CALL,
         CreateNetLogReadWriteDataCallback(stream_index, offset, buf_len,
                                           truncate));
   }
 
-  if (stream_index < 0 || stream_index >= kSimpleEntryFileCount || offset < 0 ||
-      buf_len < 0) {
+  if (stream_index < 0 || stream_index >= kSimpleEntryStreamCount ||
+      offset < 0 || buf_len < 0) {
     if (net_log_.IsLoggingAllEvents()) {
       net_log_.AddEvent(
           net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_WRITE_END,
           CreateNetLogReadWriteCompleteCallback(net::ERR_INVALID_ARGUMENT));
     }
     RecordWriteResult(cache_type_, WRITE_RESULT_INVALID_ARGUMENT);
     return net::ERR_INVALID_ARGUMENT;
   }
   if (backend_.get() && offset + buf_len > backend_->GetMaxFileSize()) {
     if (net_log_.IsLoggingAllEvents()) {
       net_log_.AddEvent(
           net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_WRITE_END,
           CreateNetLogReadWriteCompleteCallback(net::ERR_FAILED));
     }
     RecordWriteResult(cache_type_, WRITE_RESULT_OVER_MAX_SIZE);
     return net::ERR_FAILED;
   }
   ScopedOperationRunner operation_runner(this);
 
-  // Currently, Simple Cache is only used for HTTP, which stores the headers in
-  // stream 0 and always writes them with a single, truncating write.  Detect
-  // these writes and record the size and size changes of the headers.  Also,
-  // note writes to stream 0 that violate those assumptions.
-  if (stream_index == 0) {
-    if (offset == 0 && truncate)
-      RecordHeaderSizeChange(cache_type_, data_size_[0], buf_len);
-    else
-      RecordUnexpectedStream0Write(cache_type_);
-  }
-
   // We can only do optimistic Write if there is no pending operations, so
   // that we are sure that the next call to RunNextOperationIfNeeded will
   // actually run the write operation that sets the stream size. It also
   // prevents from previous possibly-conflicting writes that could be stacked
   // in the |pending_operations_|. We could optimize this for when we have
   // only read operations enqueued.
   const bool optimistic =
       (use_optimistic_operations_ && state_ == STATE_READY &&
        pending_operations_.size() == 0);
   CompletionCallback op_callback;
@@ skipping 259 matching lines @@
   DCHECK_EQ(STATE_UNINITIALIZED, state_);
   DCHECK(!synchronous_entry_);
 
   state_ = STATE_IO_PENDING;
 
   // Since we don't know the correct values for |last_used_| and
   // |last_modified_| yet, we make this approximation.
   last_used_ = last_modified_ = base::Time::Now();
 
   // If creation succeeds, we should mark all streams to be saved on close.
-  for (int i = 0; i < kSimpleEntryFileCount; ++i)
+  for (int i = 0; i < kSimpleEntryStreamCount; ++i)
     have_written_[i] = true;
 
   const base::TimeTicks start_time = base::TimeTicks::Now();
   scoped_ptr<SimpleEntryCreationResults> results(
       new SimpleEntryCreationResults(
           SimpleEntryStat(last_used_, last_modified_, data_size_)));
   Closure task = base::Bind(&SimpleSynchronousEntry::CreateEntry,
                             cache_type_,
                             path_,
                             key_,
@@ skipping 14 matching lines @@
   DCHECK(io_thread_checker_.CalledOnValidThread());
   typedef SimpleSynchronousEntry::CRCRecord CRCRecord;
   scoped_ptr<std::vector<CRCRecord> >
       crc32s_to_write(new std::vector<CRCRecord>());
 
   net_log_.AddEvent(net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_CLOSE_BEGIN);
 
   if (state_ == STATE_READY) {
     DCHECK(synchronous_entry_);
     state_ = STATE_IO_PENDING;
-    for (int i = 0; i < kSimpleEntryFileCount; ++i) {
+    for (int i = 0; i < kSimpleEntryStreamCount; ++i) {
       if (have_written_[i]) {
         if (GetDataSize(i) == crc32s_end_offset_[i]) {
           int32 crc = GetDataSize(i) == 0 ? crc32(0, Z_NULL, 0) : crc32s_[i];
           crc32s_to_write->push_back(CRCRecord(i, true, crc));
         } else {
           crc32s_to_write->push_back(CRCRecord(i, false, 0));
         }
       }
     }
   } else {
     DCHECK(STATE_UNINITIALIZED == state_ || STATE_FAILURE == state_);
   }
 
   if (synchronous_entry_) {
     Closure task =
         base::Bind(&SimpleSynchronousEntry::Close,
                    base::Unretained(synchronous_entry_),
                    SimpleEntryStat(last_used_, last_modified_, data_size_),
-                   base::Passed(&crc32s_to_write));
+                   base::Passed(&crc32s_to_write),
+                   stream_0_data_);
     Closure reply = base::Bind(&SimpleEntryImpl::CloseOperationComplete, this);
     synchronous_entry_ = NULL;
     worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 
-    for (int i = 0; i < kSimpleEntryFileCount; ++i) {
+    for (int i = 0; i < kSimpleEntryStreamCount; ++i) {
       if (!have_written_[i]) {
         SIMPLE_CACHE_UMA(ENUMERATION,
                          "CheckCRCResult", cache_type_,
                          crc_check_state_[i], CRC_CHECK_MAX);
       }
     }
   } else {
     CloseOperationComplete();
   }
 }
@@ skipping 32 matching lines @@
   DCHECK_EQ(STATE_READY, state_);
   if (offset >= GetDataSize(stream_index) || offset < 0 || !buf_len) {
     RecordReadResult(cache_type_, READ_RESULT_FAST_EMPTY_RETURN);
     // If there is nothing to read, we bail out before setting state_ to
     // STATE_IO_PENDING.
     if (!callback.is_null())
       MessageLoopProxy::current()->PostTask(FROM_HERE, base::Bind(callback, 0));
     return;
   }
 
+  // Since stream 0 data is kept in memory, it is read immediately.

[gavinp, 2013/09/17 15:07:43]

Why not catch this case in ReadData, so we can return synchronously there? Saves
a posttask for the callback which seems like a performance win.

Same for WriteData?

[clamy, 2013/09/18 16:17:15]

Returning with net::OK in ReadData causes some unit tests to fail, since they
expect net::ERR_IO_PENDING. Similarly, clients may not expect an immediate
return.

For WriteData, it was there initially and Egor asked me to take it out in order
to make the code more readable. 

[gavinp, 2013/09/18 17:20:35]

It makes me sad that we're slowing down our code just for the unit tests. But
clients should all expect an immediate return is possible; after all, the in
memory backend returns immediately on everything.

Can we at least move this code to ReadData/WriteData and add a TODO to be
synchronous? If you look again at the unit tests, you might see that they
already special case the in memory backend, we could certainly use that for
stream 0 in simple cache, too?

[pasko, 2013/09/18 17:35:27]

I'm sorry, I just did not like to have an optimization in WriteData() when it
was not strictly necessary. Now I think the optimization itself is not a
complicated piece of code and can stay for WriteData().

With ReadData() I am less confident, we may investigate later what we can do for
tests that are expecting ERR_IO_PENDING, and whether it is worth it.

Put a TODO here?

[clamy, 2013/09/18 18:29:38]

Done.

+  if (stream_index == 0) {
+    int ret_value = ReadStream0Data(buf, offset, buf_len);
+    if (!callback.is_null()) {
+      MessageLoopProxy::current()->PostTask(FROM_HERE,
+                                            base::Bind(callback, ret_value));
+    }
+    return;
+  }
+
   buf_len = std::min(buf_len, GetDataSize(stream_index) - offset);
 
   state_ = STATE_IO_PENDING;
   if (!doomed_ && backend_.get())
     backend_->index()->UseIfExists(entry_hash_);
 
   scoped_ptr<uint32> read_crc32(new uint32());
   scoped_ptr<int> result(new int());
-  scoped_ptr<base::Time> last_used(new base::Time());
+  scoped_ptr<SimpleEntryStat> entry_stat(
+      new SimpleEntryStat(last_used_, last_modified_, data_size_));
   Closure task = base::Bind(
       &SimpleSynchronousEntry::ReadData,
       base::Unretained(synchronous_entry_),
       SimpleSynchronousEntry::EntryOperationData(stream_index, offset, buf_len),
       make_scoped_refptr(buf),
       read_crc32.get(),
-      last_used.get(),
+      entry_stat.get(),
       result.get());
   Closure reply = base::Bind(&SimpleEntryImpl::ReadOperationComplete,
                              this,
                              stream_index,
                              offset,
                              callback,
                              base::Passed(&read_crc32),
-                             base::Passed(&last_used),
+                             base::Passed(&entry_stat),
                              base::Passed(&result));
   worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 void SimpleEntryImpl::WriteDataInternal(int stream_index,
                                        int offset,
                                        net::IOBuffer* buf,
                                        int buf_len,
                                        const CompletionCallback& callback,
                                        bool truncate) {
@@ skipping 16 matching lines @@
     }
     if (!callback.is_null()) {
       MessageLoopProxy::current()->PostTask(
           FROM_HERE, base::Bind(callback, net::ERR_FAILED));
     }
     // |this| may be destroyed after return here.
     return;
   }
 
   DCHECK_EQ(STATE_READY, state_);
+  // Since stream 0 data is kept in memory, it will be written immediatly.

[pasko, 2013/09/17 16:40:52]

add extra empty line above please, I just like when a block gets commented and
preceded by an empty line. Moving DCHECK below the comment is also possible

[clamy, 2013/09/18 16:17:15]

Done.

+  if (stream_index == 0) {
+    int ret_value = CopyStream0Data(buf, offset, buf_len, truncate);
+    if (!callback.is_null()) {
+      // PostTask prevents creating a loop when calling the callback directly.
+      MessageLoopProxy::current()->PostTask(FROM_HERE,
+                                            base::Bind(callback, ret_value));
+    }
+    return;
+  }
+
   state_ = STATE_IO_PENDING;
   if (!doomed_ && backend_.get())
     backend_->index()->UseIfExists(entry_hash_);
   // It is easy to incrementally compute the CRC from [0 .. |offset + buf_len|)
   // if |offset == 0| or we have already computed the CRC for [0 .. offset).
   // We rely on most write operations being sequential, start to end to compute
   // the crc of the data. When we write to an entry and close without having
   // done a sequential write, we don't check the CRC on read.
   if (offset == 0 || crc32s_end_offset_[stream_index] == offset) {
     uint32 initial_crc = (offset != 0) ? crc32s_[stream_index]
@@ skipping 14 matching lines @@
   } else {
     data_size_[stream_index] = std::max(offset + buf_len,
                                         GetDataSize(stream_index));
   }
 
   // Since we don't know the correct values for |last_used_| and
   // |last_modified_| yet, we make this approximation.
   last_used_ = last_modified_ = base::Time::Now();
 
   have_written_[stream_index] = true;
+  // Writing on stream 1 affects the placement of stream 0 in the file.
+  if (stream_index == 1)
+    have_written_[0] = true;
 
   scoped_ptr<int> result(new int());
   Closure task = base::Bind(&SimpleSynchronousEntry::WriteData,
                             base::Unretained(synchronous_entry_),
                             SimpleSynchronousEntry::EntryOperationData(
                                 stream_index, offset, buf_len, truncate),
                             make_scoped_refptr(buf),
                             entry_stat.get(),
                             result.get());
   Closure reply = base::Bind(&SimpleEntryImpl::WriteOperationComplete,
@@ skipping 36 matching lines @@
     MakeUninitialized();
     return;
   }
   // If out_entry is NULL, it means we already called ReturnEntryToCaller from
   // the optimistic Create case.
   if (out_entry)
     ReturnEntryToCaller(out_entry);
 
   state_ = STATE_READY;
   synchronous_entry_ = in_results->sync_entry;
+  stream_0_data_ = in_results->stream_0_data;
+  // The crc was read in SimpleSynchronousEntry.
+  crc_check_state_[0] = CRC_CHECK_DONE;
   if (key_.empty()) {
     SetKey(synchronous_entry_->key());
   } else {
     // This should only be triggered when creating an entry. The key check in
     // the open case is handled in SimpleBackendImpl.
     DCHECK_EQ(key_, synchronous_entry_->key());
   }
   UpdateDataFromEntryStat(in_results->entry_stat);
   SIMPLE_CACHE_UMA(TIMES,
                    "EntryCreationTime", cache_type_,
@@ skipping 27 matching lines @@
         completion_callback, *result));
   }
   RunNextOperationIfNeeded();
 }
 
 void SimpleEntryImpl::ReadOperationComplete(
     int stream_index,
     int offset,
     const CompletionCallback& completion_callback,
     scoped_ptr<uint32> read_crc32,
-    scoped_ptr<base::Time> last_used,
+    scoped_ptr<SimpleEntryStat> entry_stat,
     scoped_ptr<int> result) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK(synchronous_entry_);
   DCHECK_EQ(STATE_IO_PENDING, state_);
   DCHECK(read_crc32);
   DCHECK(result);
 
   if (*result > 0 &&
       crc_check_state_[stream_index] == CRC_CHECK_NEVER_READ_AT_ALL) {
     crc_check_state_[stream_index] = CRC_CHECK_NEVER_READ_TO_END;
@@ skipping 15 matching lines @@
       // entry, one reader can be behind the other. In this case we compute
       // the crc as the most advanced reader progresses, and check it for
       // both readers as they read the last byte.
 
       net_log_.AddEvent(net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_CHECKSUM_BEGIN);
 
       scoped_ptr<int> new_result(new int());
       Closure task = base::Bind(&SimpleSynchronousEntry::CheckEOFRecord,
                                 base::Unretained(synchronous_entry_),
                                 stream_index,
-                                data_size_[stream_index],
+                                *entry_stat,
                                 crc32s_[stream_index],
                                 new_result.get());
       Closure reply = base::Bind(&SimpleEntryImpl::ChecksumOperationComplete,
                                  this, *result, stream_index,
                                  completion_callback,
                                  base::Passed(&new_result));
       worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
       crc_check_state_[stream_index] = CRC_CHECK_DONE;
       return;
     }
   }
 
   if (*result < 0) {
     RecordReadResult(cache_type_, READ_RESULT_SYNC_READ_FAILURE);
   } else {
     RecordReadResult(cache_type_, READ_RESULT_SUCCESS);
     if (crc_check_state_[stream_index] == CRC_CHECK_NEVER_READ_TO_END &&
         offset + *result == GetDataSize(stream_index)) {
       crc_check_state_[stream_index] = CRC_CHECK_NOT_DONE;
     }
   }
   if (net_log_.IsLoggingAllEvents()) {
     net_log_.AddEvent(
         net::NetLog::TYPE_SIMPLE_CACHE_ENTRY_READ_END,
         CreateNetLogReadWriteCompleteCallback(*result));
   }
 
   EntryOperationComplete(
-      stream_index,
-      completion_callback,
-      SimpleEntryStat(*last_used, last_modified_, data_size_),
-      result.Pass());
+      stream_index, completion_callback, *entry_stat, result.Pass());
 }
 
 void SimpleEntryImpl::WriteOperationComplete(
     int stream_index,
     const CompletionCallback& completion_callback,
     scoped_ptr<SimpleEntryStat> entry_stat,
     scoped_ptr<int> result) {
   if (*result >= 0)
     RecordWriteResult(cache_type_, WRITE_RESULT_SUCCESS);
   else
@@ skipping 68 matching lines @@
 }
 
 void SimpleEntryImpl::UpdateDataFromEntryStat(
     const SimpleEntryStat& entry_stat) {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK(synchronous_entry_);
   DCHECK_EQ(STATE_READY, state_);
 
   last_used_ = entry_stat.last_used;
   last_modified_ = entry_stat.last_modified;
-  for (int i = 0; i < kSimpleEntryFileCount; ++i) {
+  for (int i = 0; i < kSimpleEntryStreamCount; ++i) {
     data_size_[i] = entry_stat.data_size[i];
   }
   if (!doomed_ && backend_.get())
     backend_->index()->UpdateEntrySize(entry_hash_, GetDiskUsage());
 }
 
 int64 SimpleEntryImpl::GetDiskUsage() const {
   int64 file_size = 0;
-  for (int i = 0; i < kSimpleEntryFileCount; ++i) {
+  for (int i = 0; i < kSimpleEntryStreamCount; ++i) {
     file_size +=
         simple_util::GetFileSizeFromKeyAndDataSize(key_, data_size_[i]);
   }
   return file_size;
 }
 
 void SimpleEntryImpl::RecordReadIsParallelizable(
     const SimpleEntryOperation& operation) const {
   if (!executing_operation_)
     return;
@@ skipping 57 matching lines @@
     } else {
       type = conflicting ? WRITE_FOLLOWS_CONFLICTING_WRITE
                          : WRITE_FOLLOWS_NON_CONFLICTING_WRITE;
     }
   }
   SIMPLE_CACHE_UMA(ENUMERATION,
                    "WriteDependencyType", cache_type_,
                    type, WRITE_DEPENDENCY_TYPE_MAX);
 }
 
+int SimpleEntryImpl::ReadStream0Data(net::IOBuffer* buf,
+                                     int offset,
+                                     int buf_len) {
+  int read_size = std::min(data_size_[0] - offset, buf_len);
+  if (read_size < 0) {
+    RecordReadResult(cache_type_, READ_RESULT_SYNC_READ_FAILURE);
+    return 0;
+  }
+  memcpy(buf->data(), stream_0_data_->data() + offset, read_size);

[pasko, 2013/09/17 16:40:52]

Seems like a null deref with stream_0_data_ if a Read is issued immediately
after an optimistic Create. Why should we be having the uninitialized state with
stream_0_data_ at all? It would be safer to just always have a growable buffer
in place. Destroyed when the last refcount goes away.

[clamy, 2013/09/18 16:17:15]

This code is called from ReadInternal, which is called by the operation queue.
The operation queue only calls the next operation when creates returns.

But added the growable buffer nonetheless.

+  UpdateDataFromEntryStat(
+      SimpleEntryStat(base::Time::Now(), last_modified_, data_size_));
+  RecordReadResult(cache_type_, READ_RESULT_SUCCESS);
+  return read_size;
+}
+
+int SimpleEntryImpl::CopyStream0Data(net::IOBuffer* buf,

[pasko, 2013/09/17 16:40:52]

hm, I dunno, it seems it would be a tiny bit less confusing to me to have it
named as SetStream0Data() or MutateStream0Data(). Copy is a bit generic and
raises questions like "From where and to where?" and "At which circumstance? Is
it on reading or writing?".

I
 am
   picky
        on
          naming.
Sorry.

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

I'd be OK with these being "ReadDataFromBuffer" and "WriteDataToBuffer", and
even better if the buffer was an argument, and they were in the anonymous
namespace?

[pasko, 2013/09/18 08:14:10]

I don't think it is a good idea since the method uses a few more object members:
have_written_, cache_type_, data_size_, crc32s_, crc32s_end_offset_.

[clamy, 2013/09/18 16:17:15]

I went with SetStream0Data, and as Egor pointed out, given the number of class
members used, I can hardly put it in an anonymous namespace.

+                                     int offset,
+                                     int buf_len,
+                                     bool truncate) {
+  // Currently, stream 0 is only used for HTTP headers, and always writes them
+  // with a single, truncating write.  Detect these writes and record the size
+  // and size changes of the headers.  Also, supports writes to stream 0 that
+  // violate those assumptions. All other clients of the Simple Cache are
+  // encouraged to use stream 1.
+  if (!stream_0_data_)
+    stream_0_data_ = new net::GrowableIOBuffer();
+  have_written_[0] = true;
+  int data_size = data_size_[0];
+  if (offset == 0 && truncate) {
+    RecordHeaderSizeChange(cache_type_, data_size, buf_len);
+    stream_0_data_->SetCapacity(buf_len);
+    memcpy(stream_0_data_->data(), buf->data(), buf_len);
+    data_size_[0] = buf_len;
+  } else {
+    RecordUnexpectedStream0Write(cache_type_);
+    const int buffer_size =
+        truncate ? offset + buf_len : std::max(offset + buf_len, data_size);
+    stream_0_data_->SetCapacity(buffer_size);
+    // If |stream_0_data_| was extended. the extension until offset need to be

[pasko, 2013/09/17 16:40:52]

s/./,/
s/need/needs/
s/zeroed/zero-filled/

In other words: "If |stream_0_data_| was extended, the extension until offset
needs to be zero-filled."

[clamy, 2013/09/18 16:17:15]

Done.

+    // zeroed.
+    const int fill_size = offset <= data_size ? 0 : offset - data_size;
+    if (fill_size > 0)
+      memset(stream_0_data_->data() + data_size, 0, fill_size);
+    if (buf)

[pasko, 2013/09/17 16:40:52]

why this check? does it happen in practice? I don't see this guard at line 1296
on similar occasion.

[clamy, 2013/09/18 16:17:15]

Some unit tests like 0-length truncating writes with a null buffer as a
parameter.

+      memcpy(stream_0_data_->data() + offset, buf->data(), buf_len);
+    data_size_[0] = buffer_size;
+  }
+  base::Time modification_time = base::Time::Now();
+  UpdateDataFromEntryStat(
+      SimpleEntryStat(modification_time, modification_time, data_size_));
+  if (stream_0_data_) {
+    crc32s_[0] = crc32(crc32(0L, Z_NULL, 0),
+                       reinterpret_cast<const Bytef*>(stream_0_data_->data()),
+                       data_size_[0]);

[pasko, 2013/09/17 16:40:52]

this needs updating crc32s_end_offset_, I think it makes sense to reuse the code
that does it:
AdvanceCrc(buffer, length, stream_index) // maybe offset in the buffer as well,
but seems not needed now

this should update crc32s_[stream_index] and crc32s_end_offset_[stream_index]

[clamy, 2013/09/18 16:17:15]

Done.

+  } else {
+    crc32s_[0] = crc32(0L, Z_NULL, 0);
+  }
+  RecordWriteResult(cache_type_, WRITE_RESULT_SUCCESS);
+  return buf_len;
+}
+
 }  // namespace disk_cache