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 156 matching lines @@
       path_(path),
       entry_hash_(entry_hash),
       use_optimistic_operations_(operations_mode == OPTIMISTIC_OPERATIONS),
       last_used_(Time::Now()),
       last_modified_(last_used_),
       open_count_(0),
       doomed_(false),
       state_(STATE_UNINITIALIZED),
       synchronous_entry_(NULL),
       net_log_(net::BoundNetLog::Make(
-          net_log, net::NetLog::SOURCE_DISK_CACHE_ENTRY)) {
+          net_log, net::NetLog::SOURCE_DISK_CACHE_ENTRY)),
+      stream_0_data_(new net::GrowableIOBuffer()) {
   COMPILE_ASSERT(arraysize(data_size_) == arraysize(crc32s_end_offset_),
                  arrays_should_be_same_size);
   COMPILE_ASSERT(arraysize(data_size_) == arraysize(crc32s_),
                  arrays_should_be_same_size);
   COMPILE_ASSERT(arraysize(data_size_) == arraysize(have_written_),
                  arrays_should_be_same_size);
   COMPILE_ASSERT(arraysize(data_size_) == arraysize(crc_check_state_),
                  arrays_should_be_same_size);
   MakeUninitialized();
   net_log_.BeginEvent(net::NetLog::TYPE_SIMPLE_CACHE_ENTRY,
@@ skipping 142 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 34 matching lines @@
     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;
   }
 
   buf_len = std::min(buf_len, GetDataSize(stream_index) - offset);
 
+  // Since stream 0 data is kept in memory, it is read immediately.
+  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;
+  }
+
   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.
+  if (stream_index == 0) {
+    int ret_value = SetStream0Data(buf, offset, buf_len, truncate);
+    if (!callback.is_null()) {
+      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]
-                                       : crc32(0, Z_NULL, 0);
-    if (buf_len > 0) {
-      crc32s_[stream_index] = crc32(initial_crc,
-                                    reinterpret_cast<const Bytef*>(buf->data()),
-                                    buf_len);
-    }
-    crc32s_end_offset_[stream_index] = offset + buf_len;
-  }
+
+  AdvanceCrc(buf, offset, buf_len, stream_index);
 
   // |entry_stat| needs to be initialized before modifying |data_size_|.
   scoped_ptr<SimpleEntryStat> entry_stat(
       new SimpleEntryStat(last_used_, last_modified_, data_size_));
   if (truncate) {
     data_size_[stream_index] = offset + buf_len;
   } 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.

[pasko, 2013/09/18 16:53:56]

please add: ", the EOF record will need to be written.

[clamy, 2013/09/18 17:20:46]

Done.

+  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;
+  if (in_results->stream_0_data) {
+    stream_0_data_ = in_results->stream_0_data;
+    // The crc was read in SimpleSynchronousEntry.
+    crc_check_state_[0] = CRC_CHECK_DONE;
+    crc32s_[0] = in_results->stream_0_crc32;
+    crc32s_end_offset_[0] = in_results->entry_stat.data_size(0);
+  }
   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 66 matching lines @@
   MakeUninitialized();
   RunNextOperationIfNeeded();
 }
 
 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) {
-    data_size_[i] = entry_stat.data_size[i];
+  last_used_ = entry_stat.last_used();
+  last_modified_ = entry_stat.last_modified();
+  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) {
+  if (buf_len < 0) {
+    RecordReadResult(cache_type_, READ_RESULT_SYNC_READ_FAILURE);
+    return 0;
+  }
+  memcpy(buf->data(), stream_0_data_->data() + offset, buf_len);
+  UpdateDataFromEntryStat(
+      SimpleEntryStat(base::Time::Now(), last_modified_, data_size_));
+  RecordReadResult(cache_type_, READ_RESULT_SUCCESS);
+  return buf_len;
+}
+
+int SimpleEntryImpl::SetStream0Data(net::IOBuffer* buf,
+                                    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
+  // changes of the headers.  Also, support writes to stream 0 that have
+  // different access patterns, as required by the API contract.
+  // All other clients of the Simple Cache are encouraged to use stream 1.
+  have_written_[0] = true;
+  int data_size = GetDataSize(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 needs to be
+    // zero-filled.
+    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)
+      memcpy(stream_0_data_->data() + offset, buf->data(), buf_len);
+    data_size_[0] = buffer_size;
+  }
+  base::Time modification_time = base::Time::Now();
+  AdvanceCrc(buf, offset, buf_len, 0);
+  UpdateDataFromEntryStat(
+      SimpleEntryStat(modification_time, modification_time, data_size_));
+  RecordWriteResult(cache_type_, WRITE_RESULT_SUCCESS);
+  return buf_len;
+}
+
+void SimpleEntryImpl::AdvanceCrc(net::IOBuffer* buffer,
+                                 int offset,
+                                 int length,
+                                 int stream_index) {
+  // 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] : crc32(0, Z_NULL, 0);
+    if (length > 0) {
+      crc32s_[stream_index] = crc32(
+          initial_crc, reinterpret_cast<const Bytef*>(buffer->data()), length);
+    }
+    crc32s_end_offset_[stream_index] = offset + length;
+  }
+}
+
 }  // namespace disk_cache