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 "base/logging.h"
+
+#include "net/base/net_errors.h"
+#include "net/disk_cache/disk_cache.h"
+#include "net/http/http_cache_transaction.h"
+
+namespace net {
+
+HttpCache::Writers::Writers(disk_cache::Entry* disk_entry)
+    : disk_entry_(disk_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 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_);
+  DCHECK(HasTransaction(transaction));
+  active_transaction_ = transaction;
+
+  read_buf_ = std::move(buf);
+  io_buf_len_ = buf_len;
+  next_state_ = State::NETWORK_READ;
+
+  int rv = DoLoop(OK);
+  if (rv == ERR_IO_PENDING)
+    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());
+  DCHECK(network_transaction_ || network_transaction);
+
+  std::pair<TransactionSet::iterator, bool> return_val =
+      all_writers_.insert(transaction);
+  DCHECK_EQ(return_val.second, true);
+
+  if (is_exclusive) {
+    DCHECK_EQ(1u, all_writers_.size());
+    is_exclusive_ = true;
+  }
+
+  if (network_transaction) {
+    DCHECK(!network_transaction_);
+    network_transaction_ = std::move(network_transaction);
+  }
+
+  priority_ = std::max(transaction->priority(), priority_);
+  network_transaction_->SetPriority(priority_);
+}
+
+void HttpCache::Writers::RemoveTransaction(Transaction* transaction) {
+  if (!transaction)
+    return;
+
+  // The transaction should be part of all_writers.
+  auto it = all_writers_.find(transaction);
+  DCHECK(it != all_writers_.end());
+  all_writers_.erase(it);
+
+  if (all_writers_.empty() && next_state_ == State::NONE)
+    ResetStateForEmptyWriters();
+  else
+    UpdatePriority();
+
+  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);
+      // If a waiting transaction existed, there should have been an
+      // active_transaction_.
+      DCHECK(active_transaction_);
+      return;
+    }
+  }
+}
+
+void HttpCache::Writers::UpdatePriority() {
+  // 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_;
+}
+
+HttpCache::TransactionSet HttpCache::Writers::RemoveAllIdleWriters() {
+  // Should be invoked after |waiting_for_read_| transactions and
+  // |active_transaction_| are processed so that all_writers_ only contains idle
+  // writers.
+  DCHECK(ContainsOnlyIdleWriters());
+
+  TransactionSet idle_writers;
+  idle_writers.insert(all_writers_.begin(), all_writers_.end());
+  all_writers_.clear();
+  ResetStateForEmptyWriters();
+  return idle_writers;
+}
+
+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);
+
+  if (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(read_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_NE(State::UNSET, next_state_);
+  DCHECK_NE(State::NONE, next_state_);
+  int rv = result;
+  do {
+    State state = next_state_;
+    next_state_ = State::UNSET;
+    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;
+      case State::UNSET:
+        NOTREACHED() << "bad state";
+        rv = ERR_FAILED;
+        break;
+      case State::NONE:
+        // Do Nothing.
+        break;
+    }

[jkarlin, 2017/07/12 15:05:26]

To enforce that the next state is set:

DCHECK(next_state_ != STATE_UNSET) << "Previous state was " << state;

[shivanisha, 2017/07/13 18:35:50]

Because state is a class enum, getting a compile error while printing it. Rest
of the dcheck should be handled by the case State::UNSET above.

[jkarlin, 2017/07/14 18:30:48]

Ah yes, the unset state does it. Thanks.

+  } while (next_state_ != State::NONE && rv != ERR_IO_PENDING);
+
+  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
+    read_buf_ = NULL;
+    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) {
+    next_state_ = State::NONE;
+    OnNetworkReadFailure(result);
+    return result;
+  }
+
+  next_state_ = State::CACHE_WRITE_DATA;
+  return result;
+}
+
+void HttpCache::Writers::OnNetworkReadFailure(int result) {
+  ProcessFailure(active_transaction_, result);
+
+  active_transaction_ = nullptr;
+
+  // TODO(shivanisha): Invoke DoneWithEntry here while
+  // integrating this class with HttpCache. That will also invoke truncation of
+  // the entry.
+}
+
+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 = disk_entry_->GetDataSize(kResponseContentIndex);
+  CompletionCallback io_callback =
+      base::Bind(&HttpCache::Writers::OnIOComplete, weak_factory_.GetWeakPtr());
+
+  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) {
+    rv = disk_entry_->WriteData(kResponseContentIndex, current_size,
+                                read_buf_.get(), num_bytes, io_callback, true);
+  } else {
+    rv = partial->CacheWrite(disk_entry_, read_buf_.get(), num_bytes,
+                             io_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);
+  }
+  next_state_ = State::NONE;
+  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 disk_entry_->WriteData(kResponseInfoIndex, 0, data.get(), io_buf_len_,
+                                io_callback, true);
+}
+
+int HttpCache::Writers::DoCacheWriteTruncatedResponseComplete(int result) {
+  next_state_ = State::NONE;
+  if (result != io_buf_len_) {
+    DLOG(ERROR) << "failed to write response info to cache";
+
+    // TODO(shivanisha): Invoke DoneWritingToEntry so that this entry is doomed.
+  }
+  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 in OnNetworkReadFailure.
+    int current_size = 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() {
+  DLOG(ERROR) << "failed to write response data to cache";
+
+  // Now writers will only be reading from the network.
+  network_read_only_ = true;
+
+  ProcessFailure(active_transaction_, ERR_CACHE_WRITE_FAILURE);
+
+  active_transaction_ = nullptr;
+
+  // 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 DoneWritingToEntry on integration. Since the
+  // active_transaction_ continues to read from the network, invoke
+  // DoneWritingToEntry with nullptr as transaction so that it is not removed
+  // from |this|.
+}
+
+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, waiting_for_read_
+  // should be empty.
+  DCHECK(waiting_for_read_.empty());
+  for (auto* transaction : all_writers_) {
+    if (transaction == active_transaction_)
+      continue;
+    transaction->SetSharedWritingFailState(result);
+    all_writers_.erase(transaction);
+  }
+}
+
+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