Fixes the cache lock issue.

net/http/http_cache_shared_writers.cc

+// Copyright (c) 2016 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 "build/build_config.h"  // For OS_POSIX
+
+#if defined(OS_POSIX)
+#include <unistd.h>
+#endif
+
+#include <memory>
+#include <utility>
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/compiler_specific.h"
+#include "base/format_macros.h"
+#include "base/location.h"
+#include "base/macros.h"
+#include "base/single_thread_task_runner.h"
+#include "net/disk_cache/disk_cache.h"
+#include "net/http/http_cache_shared_writers.h"
+#include "net/http/http_cache_transaction.h"
+
+namespace net {
+
+HttpCache::SharedWriters::SharedWriters(
+    HttpCache* cache,
+    ActiveEntry* entry,
+    Transaction* cache_transaction,
+    RequestPriority priority,
+    std::unique_ptr<HttpTransaction> network_transaction)
+    : cache_(cache->GetWeakPtr()),
+      entry_(entry),
+      priority_(priority),
+      weak_factory_(this) {
+  cache_transaction->SetShared();
+  all_writers_.insert(cache_transaction);
+  network_transaction_ = std::move(network_transaction);
+  io_callback_ = base::Bind(&HttpCache::SharedWriters::OnIOComplete,
+                            weak_factory_.GetWeakPtr());
+}
+
+HttpCache::SharedWriters::~SharedWriters() {}
+
+bool HttpCache::SharedWriters::AddTransaction(Transaction* transaction) {
+  transaction->SetShared();
+
+  if (!validating_transaction_) {
+    validating_transaction_ = transaction;
+    return true;
+  }
+
+  waiting_for_validation_.push_back(transaction);
+  return false;
+}
+
+bool HttpCache::SharedWriters::empty() {
+  int count = all_writers_.size() + waiting_for_validation_.size() +
+              (validating_transaction_ ? 1 : 0);
+  return count ? false : true;
+}
+
+int HttpCache::SharedWriters::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_DONE && rv != ERR_IO_PENDING &&
+           next_state_ != STATE_NONE);
+
+  // Save the callback as this object may be destroyed in the cache callback.
+  bool destroyed = false;
+  CompletionCallback callback = callback_;
+  if (next_state_ == STATE_DONE && cache_callback_) {
+    base::ResetAndReturn(&cache_callback_).Run(&destroyed);
+  }
+
+  if (rv != ERR_IO_PENDING && !callback.is_null()) {
+    if (!destroyed) {
+      read_buf_ = NULL;  // Release the buffer before invoking the callback.
+      base::ResetAndReturn(&callback_).Run(rv);
+    } else {
+      base::ResetAndReturn(&callback).Run(rv);
+    }
+  }
+  // This object may have been destroyed in the callback or cache_callback_.
+
+  return rv;
+}
+
+int HttpCache::SharedWriters::Read(scoped_refptr<IOBuffer> buf,
+                                   int buf_len,
+                                   const CompletionCallback& callback,
+                                   Transaction* transaction,
+                                   bool* read_in_progress) {
+  DCHECK(buf);
+  DCHECK_GT(buf_len, 0);
+  DCHECK(!callback.is_null());
+
+  // If another transaction is already reading from the network, then this
+  // transaction waits for the read to complete and gets its buffer filled
+  // with the data returned from that read.
+  if (active_transaction_) {
+    WaitingForRead waiting_transaction(transaction, buf, buf_len, callback);
+    waiting_for_read_.push_back(waiting_transaction);
+    *read_in_progress = true;
+    return ERR_IO_PENDING;
+  }
+
+  DCHECK_EQ(next_state_, STATE_NONE);
+  DCHECK(callback_.is_null());
+
+  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) {
+    DCHECK(callback_.is_null());
+    callback_ = callback;
+  }
+  return rv;
+}
+
+int HttpCache::SharedWriters::DoNetworkRead() {
+  next_state_ = STATE_NETWORK_READ_COMPLETE;
+  return network_transaction_->Read(read_buf_.get(), io_buf_len_, io_callback_);
+}
+
+int HttpCache::SharedWriters::DoNetworkReadComplete(int result) {
+  // Remember at this point active_transaction_ may or may not be alive.
+  if (result < 0) {
+    // Empty SharedWriters of all transactions.
+    OnNetworkReadFailure(result);
+    return result;
+  }
+
+  if (result == 0) {
+    // Check if the response is actually completed or if not, attempt to mark
+    // the entry as truncated.
+    if (cache_->IsResponseCompleted(entry_,
+                                    network_transaction_->GetResponseInfo())) {
+      ProcessWaitingForReadTransactions(result);
+      ResponseDataComplete();
+    } else {
+      OnNetworkReadFailure(result);
+    }
+    return result;
+  }
+
+  // Successful non zero response.
+
+  // if no consumer exists, then invoke cache write itself.
+  if (!active_transaction_)
+    next_state_ = STATE_CACHE_WRITE_DATA;
+
+  return result;
+}
+
+void HttpCache::SharedWriters::OnNetworkReadFailure(int result) {
+  FailureCleanup(result, false);
+  if (AttemptTruncation()) {
+    rv_post_truncation_ = result;
+  } else {
+    next_state_ = STATE_DONE;
+    cache_callback_ = base::Bind(&HttpCache::ResponseDoneSharedWriters,
+                                 cache_->GetWeakPtr(), entry_, false);
+  }
+}
+
+bool HttpCache::SharedWriters::AttemptTruncation() {
+  if (cache_->CanResumeEntry(true, "GET",
+                             network_transaction_->GetResponseInfo(), entry_)) {
+    next_state_ = STATE_CACHE_WRITE_TRUNCATED_RESPONSE;
+    return true;
+  } else {
+    return false;
+  }
+}
+
+int HttpCache::SharedWriters::DoCacheWriteTruncatedResponse() {
+  next_state_ = STATE_CACHE_WRITE_TRUNCATED_RESPONSE_COMPLETE;
+  return cache_->WriteResponseInfo(entry_,
+                                   network_transaction_->GetResponseInfo(),
+                                   io_callback_, true, &io_buf_len_);
+}
+
+int HttpCache::SharedWriters::DoCacheWriteTruncatedResponseComplete(
+    int result) {
+  bool success = true;
+  if (result != io_buf_len_) {
+    DLOG(ERROR) << "failed to write response info to cache";
+    success = false;
+  }
+
+  next_state_ = STATE_DONE;
+  cache_callback_ = base::Bind(&HttpCache::ResponseDoneSharedWriters,
+                               cache_->GetWeakPtr(), entry_, success);
+
+  // If consumer exists, return the saved value.
+  int rv = 0;
+  if (callback_) {
+    rv = rv_post_truncation_;
+    rv_post_truncation_ = 0;
+  } else {
+    rv = result;
+  }
+  return rv;
+}
+
+int HttpCache::SharedWriters::CacheWrite(scoped_refptr<IOBuffer> buf,
+                                         int write_len,
+                                         const CompletionCallback& callback,
+                                         Transaction* transaction) {
+  DCHECK_EQ(next_state_, STATE_NONE);
+  DCHECK(buf);
+  DCHECK_GE(write_len, 0);
+  DCHECK(callback_.is_null());
+  DCHECK(!callback.is_null());
+  DCHECK_EQ(active_transaction_, transaction);
+
+  read_buf_ = std::move(buf);
+  next_state_ = STATE_CACHE_WRITE_DATA;
+  int rv = DoLoop(write_len);
+
+  if (rv == ERR_IO_PENDING) {
+    DCHECK(callback_.is_null());
+    callback_ = callback;
+  }
+
+  return rv;
+}
+
+int HttpCache::SharedWriters::DoCacheWriteData(int num_bytes) {
+  next_state_ = STATE_CACHE_WRITE_DATA_COMPLETE;
+  write_len_ = num_bytes;
+  int current_size = entry_->disk_entry->GetDataSize(kResponseContentIndex);
+  return entry_->disk_entry->WriteData(kResponseContentIndex, current_size,
+                                       read_buf_.get(), num_bytes, io_callback_,
+                                       true);
+}
+
+int HttpCache::SharedWriters::DoCacheWriteDataComplete(int result) {
+  if (result != write_len_) {
+    // Need to take care of all the transactions in SharedWriters and
+    // delete SharedWriters as without the cache, we cannot continue the shared
+    // logic.
+    OnCacheWriteFailure();
+  } else {
+    OnCacheWriteSuccess(result);
+  }
+
+  return result;
+}
+
+void HttpCache::SharedWriters::OnCacheWriteSuccess(int result) {
+  // Save the data in all the waiting transactions' read buffers.
+  for (auto it = waiting_for_read_.begin(); it != waiting_for_read_.end();
+       it++) {
+    it->write_len = std::min(it->read_buf_len, result);
+    memcpy(it->read_buf->data(), read_buf_->data(), it->write_len);
+  }
+  // Notify waiting_for_read_. Tasks will be posted for all the
+  // transactions.
+  ProcessWaitingForReadTransactions(write_len_);
+
+  if (result > 0) {  // not the end of response
+    active_transaction_ = nullptr;
+    return;
+  }
+
+  DCHECK_EQ(result, 0);
+
+  ResponseDataComplete();
+}
+
+void HttpCache::SharedWriters::ResponseDataComplete() {
+  ResetActiveTransaction();
+
+  // If there is a transaction validating currently, return.
+  if (validating_transaction_)
+    return;
+
+  // Else empty the SharedWriters object.
+  MoveIdleWritersToReaders();
+  DCHECK(all_writers_.empty());
+
+  MoveToPendingQueue();
+
+  // Inform cache_ so it can take care of entry_.
+  next_state_ = STATE_DONE;
+  cache_callback_ = base::Bind(&HttpCache::ResponseDoneSharedWriters,
+                               cache_->GetWeakPtr(), entry_, true);
+}
+
+void HttpCache::SharedWriters::OnCacheWriteFailure() {
+  Transaction* current_writer = active_transaction_;
+
+  // Needs network_transaction_ to be valid so needs to be invoked before
+  // ContinueWithoutSharedWriting.
+  FailureCleanup(net::ERR_CACHE_WRITE_FAILURE, true);
+
+  if (current_writer)  // If the transaction is still alive in this callback.
+    current_writer->ContinueWithoutSharedWriting(
+        std::move(network_transaction_), false);
+
+  // Inform cache_ so it can take care of entry_.
+  next_state_ = STATE_DONE;
+  cache_callback_ = base::Bind(&HttpCache::ResponseDoneSharedWriters,
+                               cache_->GetWeakPtr(), entry_, false);
+}
+
+void HttpCache::SharedWriters::MoveIdleWritersToReaders() {
+  // Should be invoked after waiting_for_read_ are all processed so that
+  // all_writers_ only contains the idle writers.
+  DCHECK(waiting_for_read_.empty());
+  DCHECK(!active_transaction_);
+  for (auto idle_writer : all_writers_) {
+    entry_->readers.insert(idle_writer);
+    idle_writer->ResetShared(false, true);
+  }
+  all_writers_.clear();
+}
+
+void HttpCache::SharedWriters::ProcessWaitingForReadTransactions(int result) {
+  for (auto it = waiting_for_read_.begin(); it != waiting_for_read_.end();
+       it++) {
+    Transaction* transaction = it->transaction;
+
+    if (result > 0) {  // success
+      // Fill result with the length of buffer filled for this transaction which
+      // may be different from the transaction that actually wrote to the cache
+      // based on the buffer size.
+      result = it->write_len;
+    } else {
+      // If its response completion or failure, this transaction needs to be
+      // removed.
+      transaction->ResetShared();
+      all_writers_.erase(transaction);
+    }
+
+    // Post task to notify transaction.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(it->callback, result));
+  }
+
+  waiting_for_read_.clear();
+}
+
+void HttpCache::SharedWriters::FailureCleanup(int error,
+                                              bool continue_network_reading) {
+  ResetActiveTransaction(continue_network_reading);
+
+  // Notify waiting_for_read_ of the failure. Tasks will be posted for all the
+  // transactions.
+  ProcessWaitingForReadTransactions(error);
+
+  // Idle readers should know to fail when Read is invoked by their consumers.
+  SetIdleWritersFailState(error);
+  DCHECK(all_writers_.empty());
+
+  // If there exists a validating_transaction_, it may be waiting
+  // to read response headers from the cache or waiting for receiving
+  // validation response from the network. In both scenarios, it should be safe
+  // to fail.
+  if (validating_transaction_) {
+    validating_transaction_->SetSharedWritingFailState(error);
+    validating_transaction_->ResetShared(true);
+    validating_transaction_ = nullptr;
+  }
+
+  MoveToPendingQueue();
+}
+
+void HttpCache::SharedWriters::StopCaching(Transaction* transaction) {
+  // If this is the only transaction in SharedWriters either in validation or
+  // reading stage, then stopping will be successful. If not, then we will not
+  // stop caching since there are other consumers waiting to read from the
+  // cache.
+  bool result = false;
+  if (transaction == validating_transaction_) {
+    if (all_writers_.empty()) {
+      result = true;
+      validating_transaction_ = nullptr;
+    }
+  } else if (all_writers_.size() == 1 && all_writers_.count(transaction) &&
+             !validating_transaction_) {
+    if (active_transaction_ == transaction) {
+      active_transaction_ = nullptr;
+    }
+    all_writers_.erase(transaction);
+    result = true;
+  }
+  if (result) {
+    transaction->ContinueWithoutSharedWriting(std::move(network_transaction_),
+                                              true);
+    entry_->writer = transaction;
+    MoveToPendingQueue();
+  }
+  // If stopped, let the cache_ destroy |this|.
+  if (result)
+    cache_->ResetSharedWriters(entry_);
+}
+
+void HttpCache::SharedWriters::ResetActiveTransaction(
+    bool continue_network_reading) {
+  // If active_transaction_ is already destroyed, return.
+  if (!active_transaction_)
+    return;
+  active_transaction_->ResetShared(continue_network_reading);
+  all_writers_.erase(active_transaction_);
+  active_transaction_ = nullptr;
+  PriorityChanged();
+}
+
+void HttpCache::SharedWriters::SetIdleWritersFailState(int result) {
+  // Since this is only for idle transactions, all waiting_for_read_ and
+  // active_transaction_ should be empty.
+  DCHECK(waiting_for_read_.empty());
+  DCHECK(!active_transaction_);
+
+  for (auto transaction : all_writers_) {
+    transaction->SetSharedWritingFailState(result);
+    transaction->ResetShared();
+  }
+
+  all_writers_.clear();
+}
+
+void HttpCache::SharedWriters::OnValidationMatch(Transaction* transaction,
+                                                 RequestPriority priority) {
+  DCHECK_EQ(validating_transaction_, transaction);
+  ValidationDoneContinue(transaction, priority);
+}
+
+void HttpCache::SharedWriters::ValidationDoneContinue(
+    Transaction* transaction,
+    RequestPriority priority) {
+  validating_transaction_ = nullptr;
+  if (priority > priority_) {
+    network_transaction_->SetPriority(priority);
+    priority_ = priority;
+  }
+  all_writers_.insert(transaction);
+  ProcessFirstWaitingValidation();
+}
+
+void HttpCache::SharedWriters::PriorityChanged() {
+  RequestPriority current_highest = getCurrentHighestPriority();
+  if (priority_ != current_highest) {
+    network_transaction_->SetPriority(current_highest);
+    priority_ = current_highest;
+  }
+}
+
+RequestPriority HttpCache::SharedWriters::getCurrentHighestPriority() {
+  RequestPriority priority = MINIMUM_PRIORITY;
+  for (auto transaction : all_writers_)
+    priority = std::max(transaction->priority(), priority);
+  return priority;
+}
+
+void HttpCache::SharedWriters::ProcessFirstWaitingValidation() {
+  if (!waiting_for_validation_.empty() || validating_transaction_)
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::Bind(&HttpCache::SharedWriters::OnProcessFirstWaitingValidation,
+                   weak_factory_.GetWeakPtr()));
+}
+
+void HttpCache::SharedWriters::OnProcessFirstWaitingValidation() {
+  if (waiting_for_validation_.empty() && !validating_transaction_)
+    return;
+
+  Transaction* transaction = nullptr;
+  if (validating_transaction_) {
+    transaction = validating_transaction_;
+  } else {
+    transaction = waiting_for_validation_.front();
+    waiting_for_validation_.erase(waiting_for_validation_.begin());
+    validating_transaction_ = transaction;
+  }
+  transaction->io_callback().Run(OK);
+}
+
+std::unique_ptr<HttpTransaction> HttpCache::SharedWriters::OnValidationNoMatch(
+    const std::string& key,
+    Transaction* transaction,
+    std::unique_ptr<HttpTransaction> network_transaction,
+    RequestPriority priority) {
+  DCHECK_EQ(validating_transaction_, transaction);
+  // If there is no transaction in all_writers_, its ok to rewrite the entry
+  // response.
+  if (all_writers_.empty()) {
+    network_transaction_ = std::move(network_transaction);
+    ValidationDoneContinue(transaction, priority);
+    return std::unique_ptr<HttpTransaction>();
+  }
+
+  transaction->ResetShared();
+  validating_transaction_ = nullptr;
+  MoveToPendingQueue();
+  cache_->DoomEntryRestartPendingQueue(key, entry_);
+  return network_transaction;
+}
+
+void HttpCache::SharedWriters::DoneReading(Transaction* transaction) {
+  // Should only be invoked when the transaction is not currently reading.
+  DCHECK_NE(transaction, active_transaction_);
+  auto it = waiting_for_read_.begin();
+  for (; it != waiting_for_read_.end(); it++)
+    DCHECK_NE(transaction, it->transaction);
+
+  // The transaction should be part of all_writers.
+  size_t result = all_writers_.erase(transaction);
+  DCHECK_EQ(result, (size_t)1);
+  transaction->ResetShared();
+
+  // If active_transaction_ is set, then wait for active_transaction_ to detect
+  // the end
+  // of stream.
+  if (active_transaction_) {
+    return;
+  }
+  DCHECK(waiting_for_read_.empty());
+  // If there is a transaction validating currently, return.
+  if (validating_transaction_) {
+    return;
+  }
+
+  // Else empty the SharedWriters object.
+  MoveIdleWritersToReaders();
+  MoveToPendingQueue();
+  DCHECK(empty());
+  cache_->ResetSharedWritersProcessPendingQueue(entry_);
+}
+
+void HttpCache::SharedWriters::RemoveIdleTransaction(Transaction* transaction) {
+  // The transaction should be part of all_writers.
+  auto it = all_writers_.find(transaction);
+  DCHECK(it != all_writers_.end());
+  all_writers_.erase(transaction);
+  transaction->ResetShared();
+  PriorityChanged();
+  // If the response is not complete and there are no more consumers for this,
+  // attempt to mark it as truncated. If |this| is empty, it will also be
+  // destroyed in this call.
+  if (empty())
+    cache_->RemovedSharedWriterTransaction(transaction, entry_);
+}
+
+void HttpCache::SharedWriters::RemoveWaitingForReadTransaction(
+    Transaction* transaction) {
+  auto it = waiting_for_read_.begin();
+  for (; it != waiting_for_read_.end(); it++) {
+    if (transaction == it->transaction) {
+      waiting_for_read_.erase(it);
+      all_writers_.erase(transaction);
+      transaction->ResetShared();
+      PriorityChanged();
+      // If a waiting transaction existed, there should have been an
+      // active_transaction_.
+      DCHECK(active_transaction_);
+      break;
+    }
+  }
+}
+
+void HttpCache::SharedWriters::RemoveValidatingTransaction(
+    Transaction* transaction) {
+  DCHECK_EQ(validating_transaction_, transaction);
+  validating_transaction_ = nullptr;
+  transaction->ResetShared();
+  ProcessFirstWaitingValidation();
+  if (empty())
+    cache_->ResetSharedWritersProcessPendingQueue(entry_);
+}
+
+bool HttpCache::SharedWriters::RemoveWaitingTransaction(
+    Transaction* transaction) {
+  auto it = std::find(waiting_for_validation_.begin(),
+                      waiting_for_validation_.end(), transaction);
+  if (it != waiting_for_validation_.end()) {
+    transaction->ResetShared();
+    waiting_for_validation_.erase(it);
+    return true;
+  }
+  return false;
+}
+
+void HttpCache::SharedWriters::RemoveActiveTransaction(
+    Transaction* transaction) {
+  DCHECK_EQ(active_transaction_, transaction);
+  ResetActiveTransaction();
+  callback_.Reset();
+  // If the response is not complete and there are no more consumers for this,
+  // attempt to mark it as truncated. If |this| is empty, it will also be
+  // destroyed in this call.
+  if (empty())
+    cache_->RemovedSharedWriterTransaction(transaction, entry_);
+}
+
+void HttpCache::SharedWriters::MoveFromPendingQueue() {
+  auto it = entry_->pending_queue.begin();
+  while (it != entry_->pending_queue.end()) {
+    Transaction* transaction = *it;
+    if (transaction->IsEligibleForSharedWriting()) {
+      transaction->SetShared();
+      waiting_for_validation_.push_back(transaction);
+      it = entry_->pending_queue.erase(it);
+    } else {
+      ++it;
+    }
+  }
+}
+
+void HttpCache::SharedWriters::MoveToPendingQueue() {
+  // For maintaining the order of the transactions as they arrived, append
+  // these to the front of the pending_queue. Note that the order is preserved
+  // only among the transactions that are eligible for sharing. For others, they
+  // may have arrived earlier but may be processed later which is fair since
+  // they have to anyways wait till the entry is written to the cache.
+  while (!waiting_for_validation_.empty()) {
+    Transaction* transaction = waiting_for_validation_.back();
+    transaction->ResetShared(true);
+    entry_->pending_queue.push_front(transaction);
+    waiting_for_validation_.pop_back();
+  }
+}
+
+bool HttpCache::SharedWriters::CanAddNewTransaction() {
+  if (next_state_ == STATE_CACHE_WRITE_TRUNCATED_RESPONSE_COMPLETE)
+    return false;
+  return true;
+}
+
+HttpCache::SharedWriters::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) {}
+
+HttpCache::SharedWriters::WaitingForRead::~WaitingForRead() {}
+
+HttpCache::SharedWriters::WaitingForRead::WaitingForRead(
+    const WaitingForRead&) = default;
+
+void HttpCache::SharedWriters::OnIOComplete(int result) {
+  DoLoop(result);
+}
+
+}  // namespace net