Adds a new class HttpCache::Writers for multiple cache transactions reading from the network.

net/http/http_cache_writers.cc

+// Copyright (c) 2017 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/http/http_cache_writers.h"
+
+#include <algorithm>
+#include <utility>
+#include "net/disk_cache/disk_cache.h"
+#include "net/http/http_cache_transaction.h"
+
+namespace net {
+
+HttpCache::Writers::Writers(HttpCache* cache, ActiveEntry* entry)
+    : cache_(cache), entry_(entry), weak_factory_(this) {}
+
+HttpCache::Writers::~Writers() {}
+
+int HttpCache::Writers::Read(scoped_refptr<IOBuffer> buf,
+                             int buf_len,
+                             const CompletionCallback& callback,
+                             Transaction* transaction) {
+  DCHECK(buf);
+  DCHECK_GT(buf_len, 0);
+  DCHECK(!callback.is_null());
+  DCHECK(transaction);
+
+  if (next_state_ == State::FAIL_READ) {
+    DCHECK(entry_->doomed);
+    return error_code_;
+  }
+
+  // If another transaction invoked a Read which is currently ongoing, then
+  // this transaction waits for the read to complete and gets its buffer filled
+  // with the data returned from that read.
+  if (next_state_ != State::NONE) {
+    WaitingForRead waiting_transaction(transaction, buf, buf_len, callback);
+    waiting_for_read_.push_back(waiting_transaction);
+    return ERR_IO_PENDING;
+  }
+
+  DCHECK_EQ(next_state_, State::NONE);
+  DCHECK(callback_.is_null());
+  DCHECK_EQ(nullptr, active_transaction_);
+  active_transaction_ = transaction;

[Randy Smith (Not in Mondays), 2017/06/08 17:10:01]

Worth a DCHECK that transaction is in all_writers_?

[shivanisha, 2017/06/12 18:51:14]

done

+
+  read_buf_ = std::move(buf);
+  io_buf_len_ = buf_len;
+  next_state_ = State::NETWORK_READ;
+
+  int rv = DoLoop(OK);
+  if (rv == ERR_IO_PENDING) {

[Randy Smith (Not in Mondays), 2017/06/08 17:10:01]

nit, suggestion: No need for curly braces.

[shivanisha, 2017/06/12 18:51:14]

removed

+    callback_ = callback;
+  }
+
+  return rv;
+}
+
+bool HttpCache::Writers::StopCaching(Transaction* transaction) {
+  // If this is the only transaction in Writers, then stopping will be
+  // successful. If not, then we will not stop caching since there are
+  // other consumers waiting to read from the cache.
+  if (all_writers_.size() == 1) {
+    DCHECK(all_writers_.count(transaction));
+    network_read_only_ = true;
+    return true;
+  }
+  return false;
+}
+
+void HttpCache::Writers::AddTransaction(
+    Transaction* transaction,
+    std::unique_ptr<HttpTransaction> network_transaction,
+    bool is_exclusive) {
+  DCHECK(transaction);
+  DCHECK(CanAddWriters());
+
+  std::pair<TransactionSet::iterator, bool> return_val =
+      all_writers_.insert(transaction);
+  DCHECK_EQ(return_val.second, true);
+
+  if (is_exclusive) {
+    DCHECK_EQ(all_writers_.size(), (size_t)1);

[jkarlin, 2017/06/12 18:30:32]

DCHECK_EQ(1u, all_writers_.size());

DCHECK's crash message is written such that the expected value is the first
argument.

[shivanisha, 2017/06/14 02:33:16]

done

+    is_exclusive_ = true;
+  }
+
+  if (network_transaction) {
+    DCHECK(!network_transaction_);
+    network_transaction_ = std::move(network_transaction);
+  }
+
+  DCHECK(network_transaction_);

[jkarlin, 2017/06/12 18:30:32]

I'd prefer we moved this to the top of the function:

DCHECK(network_transaction_ || network_transaction);

[shivanisha, 2017/06/14 02:33:16]

done

+
+  priority_ = std::max(transaction->priority(), priority_);
+  network_transaction_->SetPriority(priority_);
+}
+
+void HttpCache::Writers::RemoveTransaction(Transaction* transaction) {
+  if (!transaction)

[jkarlin, 2017/06/12 18:30:32]

Can we instead DCHECK(transaction)?

[shivanisha, 2017/06/14 02:33:17]

Explicitly handling nullptr for the following scenario:

active_transaction_ is destroyed mid-read, when buffer is written to cache
DoneWritingToEntry is called with nullptr as the transaction and that invokes
RemoveTransaction(nullptr).

+    return;
+
+  // The transaction should be part of all_writers.
+  auto it = all_writers_.find(transaction);
+  DCHECK(it != all_writers_.end());
+  all_writers_.erase(transaction);
+  if (all_writers_.empty() && next_state_ == State::NONE)
+    ResetStateForEmptyWriters();
+  else
+    PriorityChanged();

[jkarlin, 2017/06/12 18:30:32]

Perhaps rename to UpdatePriority since the priority may or may not have changed?

[shivanisha, 2017/06/14 02:33:17]

done

+
+  if (active_transaction_ == transaction) {
+    active_transaction_ = nullptr;
+    callback_.Reset();
+    return;
+  }
+
+  auto waiting_it = waiting_for_read_.begin();
+  for (; waiting_it != waiting_for_read_.end(); waiting_it++) {
+    if (transaction == waiting_it->transaction) {
+      waiting_for_read_.erase(waiting_it);

[jkarlin, 2017/06/12 18:30:32]

Hmm, is the iterator still safe to increment after calling erase? Do we have a
test covering the case of RemoveTransaction when there are multiple
transactions?

[shivanisha, 2017/06/14 02:33:17]

we do not increment the iterator after erase due to the following return.

+      // If a waiting transaction existed, there should have been an
+      // active_transaction_.
+      DCHECK(active_transaction_);
+      return;
+    }
+  }
+}
+
+void HttpCache::Writers::PriorityChanged() {
+  // Get the current highest priority.
+  RequestPriority current_highest = MINIMUM_PRIORITY;
+  for (auto* transaction : all_writers_)
+    current_highest = std::max(transaction->priority(), current_highest);
+
+  if (priority_ != current_highest) {
+    network_transaction_->SetPriority(current_highest);
+    priority_ = current_highest;
+  }
+}
+
+bool HttpCache::Writers::ContainsOnlyIdleWriters() const {
+  return waiting_for_read_.empty() && !active_transaction_;
+}
+
+void HttpCache::Writers::MoveIdleWritersToReaders() {
+  // Should be invoked after |waiting_for_read_| transactions  and

[jkarlin, 2017/06/12 18:30:32]

extra space between transactions and and

[shivanisha, 2017/06/14 02:33:17]

done

+  // |active_transaction_| are processed so that all_writers_ only contains idle
+  // writers.
+  DCHECK(ContainsOnlyIdleWriters());
+  for (auto* idle_writer : all_writers_) {
+    entry_->readers.insert(idle_writer);

[jkarlin, 2017/06/12 18:30:32]

I think at this point we should treat ActiveEntry as a class instead of a
struct. Directly injecting into readers here means that there are now two files
mutating ActiveEntries. For now, can you add a function to HttpCache for reader
insertion?

[shivanisha, 2017/06/14 02:33:17]

done

+  }
+  all_writers_.clear();
+  ResetStateForEmptyWriters();
+}
+
+bool HttpCache::Writers::CanAddWriters() {
+  if (all_writers_.empty())
+    return true;
+
+  return !is_exclusive_ && !network_read_only_;
+}
+
+void HttpCache::Writers::ProcessFailure(Transaction* transaction, int error) {
+  DCHECK(!transaction || transaction == active_transaction_);
+
+  // Notify waiting_for_read_ of the failure. Tasks will be posted for all the
+  // transactions.
+  ProcessWaitingForReadTransactions(error);
+
+  // Idle readers should fail when Read is invoked on them.
+  SetIdleWritersFailState(error);
+  DCHECK(all_writers_.empty());
+  ResetStateForEmptyWriters();
+}
+
+void HttpCache::Writers::TruncateEntry() {
+  // TODO(shivanisha) On integration, see if the entry really needs to be
+  // truncated on the lines of Transaction::AddTruncatedFlag and then proceed.
+  DCHECK_EQ(next_state_, State::NONE);
+  next_state_ = State::CACHE_WRITE_TRUNCATED_RESPONSE;
+  DoLoop(OK);
+}
+
+LoadState HttpCache::Writers::GetWriterLoadState() {
+  DCHECK(network_transaction_);
+  return network_transaction_->GetLoadState();
+}
+
+HttpCache::Writers::WaitingForRead::WaitingForRead(
+    Transaction* cache_transaction,
+    scoped_refptr<IOBuffer> buf,
+    int len,
+    const CompletionCallback& consumer_callback)
+    : transaction(cache_transaction),
+      read_buf(std::move(buf)),
+      read_buf_len(len),
+      write_len(0),
+      callback(consumer_callback) {
+  DCHECK(cache_transaction);
+  DCHECK(buf);
+  DCHECK_GT(len, 0);
+  DCHECK(!consumer_callback.is_null());
+}
+
+HttpCache::Writers::WaitingForRead::~WaitingForRead() {}
+HttpCache::Writers::WaitingForRead::WaitingForRead(const WaitingForRead&) =
+    default;
+
+int HttpCache::Writers::DoLoop(int result) {
+  DCHECK(next_state_ != State::NONE);
+  int rv = result;
+  do {
+    State state = next_state_;
+    next_state_ = State::NONE;
+    switch (state) {
+      case State::NETWORK_READ:
+        DCHECK_EQ(OK, rv);
+        rv = DoNetworkRead();
+        break;
+      case State::NETWORK_READ_COMPLETE:
+        rv = DoNetworkReadComplete(rv);
+        break;
+      case State::CACHE_WRITE_DATA:
+        rv = DoCacheWriteData(rv);
+        break;
+      case State::CACHE_WRITE_DATA_COMPLETE:
+        rv = DoCacheWriteDataComplete(rv);
+        break;
+      case State::CACHE_WRITE_TRUNCATED_RESPONSE:
+        rv = DoCacheWriteTruncatedResponse();
+        break;
+      case State::CACHE_WRITE_TRUNCATED_RESPONSE_COMPLETE:
+        rv = DoCacheWriteTruncatedResponseComplete(rv);
+        break;
+      default:
+        NOTREACHED() << "bad state";
+        rv = ERR_FAILED;
+        break;
+    }
+  } while (next_state_ != State::NONE && rv != ERR_IO_PENDING);
+
+  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
+    read_buf_ = NULL;  // Release the buffer before invoking the callback.
+    base::ResetAndReturn(&callback_).Run(rv);
+  }
+  return rv;
+}
+
+int HttpCache::Writers::DoNetworkRead() {
+  next_state_ = State::NETWORK_READ_COMPLETE;
+  CompletionCallback io_callback =
+      base::Bind(&HttpCache::Writers::OnIOComplete, weak_factory_.GetWeakPtr());
+  return network_transaction_->Read(read_buf_.get(), io_buf_len_, io_callback);
+}
+
+int HttpCache::Writers::DoNetworkReadComplete(int result) {
+  if (result < 0) {
+    OnNetworkReadFailure(result);
+    return result;
+  }
+
+  next_state_ = State::CACHE_WRITE_DATA;
+  return result;
+}
+
+void HttpCache::Writers::OnNetworkReadFailure(int result) {
+  cache_->DoneWithEntry(entry_, active_transaction_, true);
+  // TODO(shivanisha): Send |result| while invoking ProcessFailure while
+  // integrating this class with HttpCache.
+}
+
+int HttpCache::Writers::DoCacheWriteData(int num_bytes) {
+  next_state_ = State::CACHE_WRITE_DATA_COMPLETE;
+  write_len_ = num_bytes;
+  if (!num_bytes || network_read_only_)
+    return num_bytes;
+
+  int current_size = entry_->disk_entry->GetDataSize(kResponseContentIndex);

[Randy Smith (Not in Mondays), 2017/06/08 17:10:01]

Why is this accurate as the offset to which to write in the cache in a partial
case?  Couldn't a partial read be anywhere in the entry?

[shivanisha, 2017/06/08 17:55:56]

WriteToEntry => partial->CacheWrite which does not take the offset.

[Randy Smith (Not in Mondays), 2017/06/13 17:45:05]

Ah, right.  I'd afraid I'd call that a confusing interface.   Willing to
document (in the headers or before the definition; for private functions I don't
think there's a lot of difference) that WriteToEntry ignores the offset argument
int in the partial case?  Alternatively, you could just hoist the code for
WriteToEntry() up into DoCacheWriteData(); it doesn't look to be called anywhere
else.

[shivanisha, 2017/06/14 02:33:16]

Moved to calling site.

+  CompletionCallback io_callback =
+      base::Bind(&HttpCache::Writers::OnIOComplete, weak_factory_.GetWeakPtr());
+  return WriteToEntry(current_size, read_buf_.get(), num_bytes, io_callback);
+}
+
+int HttpCache::Writers::WriteToEntry(int offset,
+                                     IOBuffer* data,
+                                     int data_len,
+                                     const CompletionCallback& callback) {
+  int rv = 0;
+
+  PartialData* partial = nullptr;
+  // The active transaction must be alive if this is a partial request, as
+  // partial requests are exclusive and hence will always be the active
+  // transaction.
+  // Todo(shivanisha): When partial requests support parallel writing, this
+  // assumption will not be true.
+  if (active_transaction_)
+    partial = active_transaction_->partial();
+
+  if (!partial || !data_len) {
+    rv = entry_->disk_entry->WriteData(kResponseContentIndex, offset, data,
+                                       data_len, callback, true);
+  } else {
+    rv = partial->CacheWrite(entry_->disk_entry, data, data_len, callback);
+  }
+  return rv;
+}
+
+int HttpCache::Writers::DoCacheWriteDataComplete(int result) {
+  if (result != write_len_) {
+    OnCacheWriteFailure();
+
+    // |active_transaction_| can continue reading from the network.
+    result = write_len_;
+  } else {
+    OnDataReceived(result);
+  }
+  return result;
+}
+
+int HttpCache::Writers::DoCacheWriteTruncatedResponse() {
+  next_state_ = State::CACHE_WRITE_TRUNCATED_RESPONSE_COMPLETE;
+  const HttpResponseInfo* response = network_transaction_->GetResponseInfo();
+  scoped_refptr<PickledIOBuffer> data(new PickledIOBuffer());
+  response->Persist(data->pickle(), true /* skip_transient_headers*/, true);
+  data->Done();
+  io_buf_len_ = data->pickle()->size();
+  CompletionCallback io_callback =
+      base::Bind(&HttpCache::Writers::OnIOComplete, weak_factory_.GetWeakPtr());
+  return entry_->disk_entry->WriteData(kResponseInfoIndex, 0, data.get(),
+                                       io_buf_len_, io_callback, true);
+}
+
+int HttpCache::Writers::DoCacheWriteTruncatedResponseComplete(int result) {
+  if (result != io_buf_len_) {
+    DLOG(ERROR) << "failed to write response info to cache";
+    cache_->DoneWritingToEntry(entry_, false);
+  }
+  truncated_ = true;
+  return OK;
+}
+
+void HttpCache::Writers::OnDataReceived(int result) {
+  if (result == 0) {
+    // Check if the response is actually completed or if not, attempt to mark
+    // the entry as truncated.
+    int current_size = entry_->disk_entry->GetDataSize(kResponseContentIndex);
+    const HttpResponseInfo* response_info =
+        network_transaction_->GetResponseInfo();
+    int64_t content_length = response_info->headers->GetContentLength();
+    if (content_length >= 0 && content_length > current_size) {
+      OnNetworkReadFailure(result);
+      return;
+    }
+    // TODO(shivanisha) Invoke cache_->DoneWritingToEntry() with success after
+    // integration with HttpCache layer.
+  }
+
+  // Notify waiting_for_read_. Tasks will be posted for all the
+  // transactions.
+  ProcessWaitingForReadTransactions(write_len_);
+
+  active_transaction_ = nullptr;
+
+  if (all_writers_.empty())
+    ResetStateForEmptyWriters();
+}
+
+void HttpCache::Writers::OnCacheWriteFailure() {
+  network_read_only_ = true;
+
+  // Call the cache_ function here even if |active_transaction_| is alive
+  // because it wouldn't know if this was an error case, since it gets a
+  // positive result back.
+  // TODO(shivanisha) : Invoke ProcessFailure here or through DoneWritingToEntry
+  // on integration.
+  cache_->DoneWritingToEntry(entry_, false);
+}
+
+void HttpCache::Writers::ProcessWaitingForReadTransactions(int result) {
+  for (auto& waiting : waiting_for_read_) {
+    Transaction* transaction = waiting.transaction;
+    int callback_result = result;
+
+    if (result >= 0) {  // success
+      // Save the data in the waiting transaction's read buffer.
+      waiting.write_len = std::min(waiting.read_buf_len, result);
+      memcpy(waiting.read_buf->data(), read_buf_->data(), waiting.write_len);
+      callback_result = waiting.write_len;
+    }
+
+    // If its response completion or failure, this transaction needs to be
+    // removed.
+    if (result <= 0)
+      all_writers_.erase(transaction);
+
+    // Post task to notify transaction.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(waiting.callback, callback_result));
+  }
+
+  waiting_for_read_.clear();
+}
+
+void HttpCache::Writers::SetIdleWritersFailState(int result) {
+  // Since this is only for idle transactions, all waiting_for_read_ and
+  // active_transaction_ should be empty.
+  DCHECK(ContainsOnlyIdleWriters());
+  next_state_ = State::FAIL_READ;
+  error_code_ = result;
+}
+
+void HttpCache::Writers::ResetStateForEmptyWriters() {
+  DCHECK(all_writers_.empty());
+  network_read_only_ = false;
+  network_transaction_.reset();
+}
+
+void HttpCache::Writers::OnIOComplete(int result) {
+  DoLoop(result);
+}
+
+}  // namespace net