Allow CertNetFetcher to be shutdown from the network thread

net/cert_net/cert_net_fetcher_impl.cc

 // Copyright 2015 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.
 //
 // Overview
 //
 // The main entry point is CertNetFetcherImpl. This is an implementation of
 // CertNetFetcher that provides a service for fetching network requests.
 //
 // The interface for CertNetFetcher is synchronous, however allows
 // overlapping requests. When starting a request CertNetFetcherImpl
 // returns a CertNetFetcher::Request (CertNetFetcherImpl) that the
 // caller can use to cancel the fetch, or wait for it to complete
 // (blocking).
 //
+// The CertNetFetcherImpl is shared between a network thread and a
+// caller thread that waits for fetches to happen on the network thread.
+//
 // The classes are mainly organized based on their thread affinity:
 //
 // ---------------
-// Lives on caller thread
+// Straddles caller thread and network thread
 // ---------------
 //
 // CertNetFetcherImpl (implements CertNetFetcher)
-//   * Main entry point
-//   * Provides a service to start/cancel/wait for URL fetches
+//   * Main entry point. Must be created and shutdown from the network thread.
+//   * Provides a service to start/cancel/wait for URL fetches, to be
+//     used on the caller thread.
 //   * Returns callers a CertNetFetcher::Request as a handle
 //   * Requests can run in parallel, however will block the current thread when
 //     reading results.
 //   * Posts tasks to network thread to coordinate actual work
 //
-// CertNetFetcherRequestImpl (implements CertNetFetcher::Request)
-//   * Wrapper for cancelling events, or waiting for a request to complete
-//   * Waits on a WaitableEvent to complete requests.
-//
-// ---------------
-// Straddles caller thread and network thread
-// ---------------
-//
-// CertNetFetcherCore
-//   * Reference-counted bridge between CertNetFetcherImpl and the dependencies
-//     on network thread.
-//   * Small wrapper to holds the state that is conceptually owned by
-//     CertNetFetcherImpl, but belongs on the network thread.
-//
 // RequestCore
 //   * Reference-counted bridge between CertNetFetcherRequestImpl and the
 //     dependencies on the network thread
 //   * Holds the result of the request, a WaitableEvent for signaling
 //     completion, and pointers for canceling work on network thread.
 //
 // ---------------
+// Lives on caller thread
+// ---------------
+//
+// CertNetFetcherRequestImpl (implements CertNetFetcher::Request)
+//   * Wrapper for cancelling events, or waiting for a request to complete
+//   * Waits on a WaitableEvent to complete requests.
+//
+// ---------------
 // Lives on network thread
 // ---------------
 //
 // AsyncCertNetFetcherImpl
 //   * Asyncronous manager for outstanding requests. Handles de-duplication,
 //     timeouts, and actual integration with network stack. This is where the
 //     majority of the logic lives.
 //   * Signals completion of requests through RequestCore's WaitableEvent.
 //   * Attaches requests to Jobs for the purpose of de-duplication
 
 #include "net/cert_net/cert_net_fetcher_impl.h"
 
 #include <tuple>
 #include <utility>
 
 #include "base/callback_helpers.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/numerics/safe_math.h"
 #include "base/synchronization/waitable_event.h"
+#include "base/threading/thread_task_runner_handle.h"
 #include "base/timer/timer.h"
 #include "net/base/load_flags.h"
 #include "net/cert/cert_net_fetcher.h"
 #include "net/url_request/redirect_info.h"
 #include "net/url_request/url_request_context.h"
-#include "net/url_request/url_request_context_getter.h"
 
 // TODO(eroman): Add support for POST parameters.
 // TODO(eroman): Add controls for bypassing the cache.
 // TODO(eroman): Add a maximum number of in-flight jobs/requests.
 // TODO(eroman): Add NetLog integration.
 
 namespace net {
 
 namespace {
 
@@ skipping 26 matching lines @@
 
 // AsyncCertNetFetcherImpl manages URLRequests in an async fashion on the
 // URLRequestContexts's task runner thread.
 //
 //  * Schedules
 //  * De-duplicates requests
 //  * Handles timeouts
 class AsyncCertNetFetcherImpl {
  public:
   // Initializes AsyncCertNetFetcherImpl using the specified URLRequestContext
-  // for issuing requests. |context| must remain valid for the entire
-  // lifetime of the AsyncCertNetFetcherImpl.
+  // for issuing requests. |context| must remain valid until Shutdown() is
+  // called or the AsyncCertNetFetcherImpl is destroyed.
   explicit AsyncCertNetFetcherImpl(URLRequestContext* context);
 
-  // Deletion implicitly cancels any outstanding requests.
+  // The AsyncCertNetFetcherImpl is expected to be kept alive until all
+  // requests have completed or Shutdown() is called.
   ~AsyncCertNetFetcherImpl();
 
   // Starts an asynchronous request to fetch the given URL. On completion
-  // |callback| will be invoked.
-  //
-  // Completion of the request will never occur synchronously. In other words it
-  // is guaranteed that |callback| will only be invoked once the Fetch*() method
-  // has returned.
+  // request->OnJobCompleted() will be invoked.
   void Fetch(std::unique_ptr<RequestParams> request_params,
-             RequestCore* request);
+             scoped_refptr<RequestCore> request);
+
+  // Cancels outstanding jobs, which stops network requests and signals the
+  // corresponding RequestCores that the requests have completed.
+  void Shutdown();
 
  private:
   friend class Job;
 
   // Finds a job with a matching RequestPararms or returns nullptr if there was
   // no match.
   Job* FindJob(const RequestParams& params);
 
   // Removes |job| from the in progress jobs and transfers ownership to the
   // caller.
@@ skipping 50 matching lines @@
 class RequestCore : public base::RefCountedThreadSafe<RequestCore> {
  public:
   explicit RequestCore(scoped_refptr<base::SingleThreadTaskRunner> task_runner)
       : completion_event_(base::WaitableEvent::ResetPolicy::MANUAL,
                           base::WaitableEvent::InitialState::NOT_SIGNALED),
         task_runner_(std::move(task_runner)) {}
 
   void AttachedToJob(Job* job) {
     DCHECK(task_runner_->RunsTasksOnCurrentThread());
     DCHECK(!job_);
+    // Requests should not be attached to jobs after they have been signalled
+    // with a cancellation error (which happens via either Cancel() or
+    // SignalImmediateError()).
+    DCHECK_NE(error_, ERR_ABORTED);
     job_ = job;
   }
 
   void OnJobCompleted(Job* job,
                       Error error,
                       const std::vector<uint8_t>& response_body) {
     DCHECK(task_runner_->RunsTasksOnCurrentThread());
 
     DCHECK_EQ(job_, job);
     job_ = nullptr;
 
     error_ = error;
     bytes_ = response_body;
     completion_event_.Signal();
   }
 
-  // Can be called from any thread.
+  // Can be called from any thread. Detaches this request from its job (if it is
+  // attached to any), but does not signal completion.
   void Cancel();
 
+  // Can be used to signal that an error was encountered before the request was
+  // attached to a job. Can be called from any thread.
+  void SignalImmediateError();
+
   // Should only be called once.
   void WaitForResult(Error* error, std::vector<uint8_t>* bytes) {
     DCHECK(!task_runner_->RunsTasksOnCurrentThread());
 
     completion_event_.Wait();
     *bytes = std::move(bytes_);
     *error = error_;
 
     error_ = ERR_UNEXPECTED;
   }
 
  private:
   friend class base::RefCountedThreadSafe<RequestCore>;
 
   ~RequestCore() {
     // Requests should have been cancelled prior to destruction.
     DCHECK(!job_);
   }
 
   // A non-owned pointer to the job that is executing the request.
   Job* job_ = nullptr;
 
-  // May be written to from network thread.
+  // May be written to from network thread, or from the caller thread only when
+  // there is no work that will be done on the network thread (e.g. when the
+  // network thread has been shutdown before the request begins). See comment in
+  // SignalImmediateError.
   Error error_;
   std::vector<uint8_t> bytes_;
 
   // Indicates when |error_| and |bytes_| have been written to.
   base::WaitableEvent completion_event_;
 
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   DISALLOW_COPY_AND_ASSIGN(RequestCore);
 };
@@ skipping 28 matching lines @@
 // Job tracks an outstanding URLRequest as well as all of the pending requests
 // for it.
 class Job : public URLRequest::Delegate {
  public:
   Job(std::unique_ptr<RequestParams> request_params,
       AsyncCertNetFetcherImpl* parent);
   ~Job() override;
 
   const RequestParams& request_params() const { return *request_params_; }
 
-  // Create a request and attaches it to the job. When the job completes it will
-  // notify the request of completion through OnJobCompleted. Note that the Job
-  // does NOT own the request.
-  void AttachRequest(RequestCore* request);
+  // Creates a request and attaches it to the job. When the job completes it
+  // will notify the request of completion through OnJobCompleted.
+  void AttachRequest(scoped_refptr<RequestCore> request);
 
   // Removes |request| from the job.
   void DetachRequest(RequestCore* request);
 
   // Creates and starts a URLRequest for the job. After the URLRequest has
   // completed, OnJobCompleted() will be invoked and all the registered requests
   // notified of completion.
   void StartURLRequest(URLRequestContext* context);
 
+  // Cancels the request with an ERR_ABORTED error and invokes
+  // RequestCore::OnJobCompleted() to notify the registered requests of the
+  // cancellation. The job is *not* removed from the AsyncCertNetFetcherImpl.
+  void Cancel();
+
  private:
   // Implementation of URLRequest::Delegate
   void OnReceivedRedirect(URLRequest* request,
                           const RedirectInfo& redirect_info,
                           bool* defer_redirect) override;
   void OnResponseStarted(URLRequest* request, int net_error) override;
   void OnReadCompleted(URLRequest* request, int bytes_read) override;
 
   // Clears the URLRequest and timer. Helper for doing work common to
   // cancellation and job completion.
   void Stop();
 
   // Reads as much data as available from |request|.
   void ReadBody(URLRequest* request);
 
   // Helper to copy the partial bytes read from the read IOBuffer to an
   // aggregated buffer.
   bool ConsumeBytesRead(URLRequest* request, int num_bytes);
 
   // Called when the URLRequest has completed (either success or failure).
   void OnUrlRequestCompleted(int net_error);
 
   // Called when the Job has completed. The job may finish in response to a
   // timeout, an invalid URL, or the URLRequest completing. By the time this
   // method is called, the |response_body_| variable have been assigned.
   void OnJobCompleted(Error error);
 
+  // Calls r->OnJobCompleted() for each RequestCore |r| currently attached
+  // to this job, and then clears |requests_|.
+  void CompleteAndClearRequests(Error error);
+
   // Cancels a request with a specified error code and calls
   // OnUrlRequestCompleted().
   void FailRequest(Error error);
 
-  // The requests attached to this job (non-owned).
-  std::vector<RequestCore*> requests_;
+  // The requests attached to this job.
+  std::vector<scoped_refptr<RequestCore>> requests_;
 
   // The input parameters for starting a URLRequest.
   std::unique_ptr<RequestParams> request_params_;
 
   // The URLRequest response information.
   std::vector<uint8_t> response_body_;
 
   std::unique_ptr<URLRequest> url_request_;
   scoped_refptr<IOBuffer> read_buffer_;
 
@@ skipping 13 matching lines @@
     return;
   }
 
   if (job_) {
     auto* job = job_;
     job_ = nullptr;
     job->DetachRequest(this);
   }
 
   bytes_.clear();
-  error_ = ERR_UNEXPECTED;
+  error_ = ERR_ABORTED;
+}
+
+void RequestCore::SignalImmediateError() {
+  // Writing to |error_| is safe here from either thread, because
+  // SignalImmediateError is only to be called before this request is attached
+  // to a job. In particular, if called from the caller thread, no work will be
+  // done on the network thread for this request, so |error_| will only be
+  // written and read on the caller thread. If called from the network thread,
+  // |error_| will only be written to on the network thread and will not be read
+  // on the caller thread until |completion_event_| is signalled (after which it
+  // will be not be written on the network thread again).
+  error_ = ERR_ABORTED;
+  completion_event_.Signal();
 }
 
 Job::Job(std::unique_ptr<RequestParams> request_params,
          AsyncCertNetFetcherImpl* parent)
     : request_params_(std::move(request_params)), parent_(parent) {}
 
 Job::~Job() {
   DCHECK(requests_.empty());
   Stop();
 }
 
-void Job::AttachRequest(RequestCore* request) {
-  requests_.push_back(request);
+void Job::AttachRequest(scoped_refptr<RequestCore> request) {
   request->AttachedToJob(this);
+  requests_.push_back(std::move(request));

[estark, 2017/01/12 01:06:31]

Switched the order of these lines so I could use std::move as suggested

 }
 
 void Job::DetachRequest(RequestCore* request) {
   std::unique_ptr<Job> delete_this;
 
   auto it = std::find(requests_.begin(), requests_.end(), request);
   DCHECK(it != requests_.end());
   requests_.erase(it);
 
   // If there are no longer any requests attached to the job then
   // cancel and delete it.
   if (requests_.empty())
     delete_this = parent_->RemoveJob(this);
 }
 
 void Job::StartURLRequest(URLRequestContext* context) {
   Error error = CanFetchUrl(request_params_->url);
   if (error != OK) {
-    // TODO(eroman): Don't post a task for this case.
-    timer_.Start(
-        FROM_HERE, base::TimeDelta(),
-        base::Bind(&Job::OnJobCompleted, base::Unretained(this), error));
+    OnJobCompleted(error);
     return;
   }
 
   // Start the URLRequest.
   read_buffer_ = new IOBuffer(kReadBufferSizeInBytes);
   url_request_ =
       context->CreateRequest(request_params_->url, DEFAULT_PRIORITY, this);
   if (request_params_->http_method == HTTP_METHOD_POST)
     url_request_->set_method("POST");
   url_request_->SetLoadFlags(LOAD_DO_NOT_SAVE_COOKIES |
                              LOAD_DO_NOT_SEND_COOKIES);
   url_request_->Start();
 
   // Start a timer to limit how long the job runs for.
   if (request_params_->timeout > base::TimeDelta())
     timer_.Start(
         FROM_HERE, request_params_->timeout,
         base::Bind(&Job::FailRequest, base::Unretained(this), ERR_TIMED_OUT));
 }
 
+void Job::Cancel() {
+  // Stop the timer and clear the URLRequest.
+  Stop();
+  // Signal attached requests that they've been completed.
+  CompleteAndClearRequests(static_cast<Error>(ERR_ABORTED));
+}
+
 void Job::OnReceivedRedirect(URLRequest* request,
                              const RedirectInfo& redirect_info,
                              bool* defer_redirect) {
   DCHECK_EQ(url_request_.get(), request);
 
   // Ensure that the new URL matches the policy.
   Error error = CanFetchUrl(redirect_info.new_url);
   if (error != OK) {
     FailRequest(error);
     return;
@@ skipping 72 matching lines @@
   Error result = static_cast<Error>(net_error);
   OnJobCompleted(result);
 }
 
 void Job::OnJobCompleted(Error error) {
   DCHECK_NE(ERR_IO_PENDING, error);
   // Stop the timer and clear the URLRequest.
   Stop();
 
   std::unique_ptr<Job> delete_this = parent_->RemoveJob(this);
+  CompleteAndClearRequests(error);
+}
 
-  for (auto* request : requests_) {
+void Job::CompleteAndClearRequests(Error error) {
+  for (const auto& request : requests_) {
     request->OnJobCompleted(this, error, response_body_);
   }
 
   requests_.clear();
 }
 
 void Job::FailRequest(Error error) {
   DCHECK_NE(ERR_IO_PENDING, error);
   int result = url_request_->CancelWithError(error);
   OnUrlRequestCompleted(result);
 }
 
 AsyncCertNetFetcherImpl::AsyncCertNetFetcherImpl(URLRequestContext* context)
     : context_(context) {
   // Allow creation to happen from another thread.
   thread_checker_.DetachFromThread();
 }
 
 AsyncCertNetFetcherImpl::~AsyncCertNetFetcherImpl() {
   DCHECK(thread_checker_.CalledOnValidThread());
   jobs_.clear();
 }
 
 bool JobComparator::operator()(const Job* job1, const Job* job2) const {
   return job1->request_params() < job2->request_params();
 }
 
 void AsyncCertNetFetcherImpl::Fetch(
     std::unique_ptr<RequestParams> request_params,
-    RequestCore* request) {
+    scoped_refptr<RequestCore> request) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   // If there is an in-progress job that matches the request parameters use it.
   // Otherwise start a new job.
   Job* job = FindJob(*request_params);
-
-  if (!job) {
-    job = new Job(std::move(request_params), this);
-    jobs_[job] = base::WrapUnique(job);
-    job->StartURLRequest(context_);
+  if (job) {
+    job->AttachRequest(request);
+    return;
   }
 
-  return job->AttachRequest(request);
+  job = new Job(std::move(request_params), this);
+  jobs_[job] = base::WrapUnique(job);
+  // Attach the request before calling StartURLRequest; this ensures that the
+  // request will get signalled if StartURLRequest completes the job
+  // synchronously.
+  job->AttachRequest(request);
+  job->StartURLRequest(context_);

[estark, 2017/01/12 01:06:31]

This little reshuffling was to accommodate the fact that StartURLRequest can now
call OnJobCompleted synchronously.

+}
+
+void AsyncCertNetFetcherImpl::Shutdown() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  for (const auto& job : jobs_) {
+    job.first->Cancel();
+  }
+  jobs_.clear();
 }
 
 struct JobToRequestParamsComparator {
   bool operator()(const JobSet::value_type& job,
                   const RequestParams& value) const {
     return job.first->request_params() < value;
   }
 };
 
 Job* AsyncCertNetFetcherImpl::FindJob(const RequestParams& params) {
@@ skipping 33 matching lines @@
 
   ~CertNetFetcherRequestImpl() override {
     if (core_)
       core_->Cancel();
   }
 
  private:
   scoped_refptr<RequestCore> core_;
 };
 
-class CertNetFetcherCore
-    : public base::RefCountedThreadSafe<CertNetFetcherCore> {
- public:
-  explicit CertNetFetcherCore(URLRequestContextGetter* context_getter)
-      : context_getter_(context_getter) {}
-
-  void Abandon() {
-    GetNetworkTaskRunner()->PostTask(
-        FROM_HERE,
-        base::Bind(&CertNetFetcherCore::DoAbandonOnNetworkThread, this));
-  }
-
-  scoped_refptr<base::SingleThreadTaskRunner> GetNetworkTaskRunner() {
-    return context_getter_->GetNetworkTaskRunner();
-  }
-
-  void DoFetchOnNetworkThread(std::unique_ptr<RequestParams> request_params,
-                              scoped_refptr<RequestCore> request) {
-    DCHECK(GetNetworkTaskRunner()->RunsTasksOnCurrentThread());
-
-    if (!impl_) {
-      impl_.reset(
-          new AsyncCertNetFetcherImpl(context_getter_->GetURLRequestContext()));
-    }
-
-    // Don't need to retain a reference to |request| because consume is
-    // expected to keep it alive.
-    impl_->Fetch(std::move(request_params), request.get());
-  }
-
- private:
-  friend class base::RefCountedThreadSafe<CertNetFetcherCore>;
-
-  void DoAbandonOnNetworkThread() {
-    DCHECK(GetNetworkTaskRunner()->RunsTasksOnCurrentThread());
-    impl_.reset();
-  }
-
-  ~CertNetFetcherCore() { DCHECK(!impl_); }
-
-  scoped_refptr<URLRequestContextGetter> context_getter_;
-
-  std::unique_ptr<AsyncCertNetFetcherImpl> impl_;
-
-  DISALLOW_COPY_AND_ASSIGN(CertNetFetcherCore);
-};
-
 class CertNetFetcherImpl : public CertNetFetcher {
  public:
-  explicit CertNetFetcherImpl(URLRequestContextGetter* context_getter)
-      : core_(new CertNetFetcherCore(context_getter)) {}
+  explicit CertNetFetcherImpl(URLRequestContext* context)
+      : task_runner_(base::ThreadTaskRunnerHandle::Get()), context_(context) {}
 
-  ~CertNetFetcherImpl() override { core_->Abandon(); }
+  void Shutdown() override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
+    if (impl_) {
+      impl_->Shutdown();
+      impl_.reset();
+    }
+    context_ = nullptr;
+  }
 
   std::unique_ptr<Request> FetchCaIssuers(const GURL& url,
                                           int timeout_milliseconds,
                                           int max_response_bytes) override {
     std::unique_ptr<RequestParams> request_params(new RequestParams);
 
     request_params->url = url;
     request_params->http_method = HTTP_METHOD_GET;
     request_params->timeout = GetTimeout(timeout_milliseconds);
     request_params->max_response_bytes =
@@ skipping 25 matching lines @@
     request_params->url = url;
     request_params->http_method = HTTP_METHOD_GET;
     request_params->timeout = GetTimeout(timeout_milliseconds);
     request_params->max_response_bytes =
         GetMaxResponseBytes(max_response_bytes, kMaxResponseSizeInBytesForAia);
 
     return DoFetch(std::move(request_params));
   }
 
  private:
+  ~CertNetFetcherImpl() override {
+    // The fetcher must be shutdown (at which point |context_| will be set to
+    // null) before destruction.
+    DCHECK(!context_);
+  }
+
+  void DoFetchOnNetworkThread(std::unique_ptr<RequestParams> request_params,
+                              scoped_refptr<RequestCore> request) {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
+
+    if (!context_) {
+      // The fetcher might have been shutdown between when this task was posted
+      // and when it is running. In this case, signal the request and do not
+      // start a network request.
+      request->SignalImmediateError();
+      return;
+    }
+
+    if (!impl_) {
+      impl_.reset(new AsyncCertNetFetcherImpl(context_));
+    }
+
+    impl_->Fetch(std::move(request_params), request);
+  }
+
   std::unique_ptr<Request> DoFetch(
       std::unique_ptr<RequestParams> request_params) {
-    auto task_runner = core_->GetNetworkTaskRunner();
-    scoped_refptr<RequestCore> request_core = new RequestCore(task_runner);
+    scoped_refptr<RequestCore> request_core = new RequestCore(task_runner_);
 
-    task_runner->PostTask(
-        FROM_HERE,
-        base::Bind(&CertNetFetcherCore::DoFetchOnNetworkThread, core_,
-                   base::Passed(&request_params), request_core));
+    // If the fetcher has already been shutdown, DoFetchOnNetworkThread will
+    // signal the request with an error. However, if the fetcher shuts down
+    // before DoFetchOnNetworkThread runs and PostTask still returns true, then
+    // the request will hang (that is, WaitForResult will not return).
+    if (!task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(&CertNetFetcherImpl::DoFetchOnNetworkThread, this,
+                       base::Passed(&request_params), request_core))) {
+      request_core->SignalImmediateError();
+    }
 
     return base::MakeUnique<CertNetFetcherRequestImpl>(std::move(request_core));
   }
 
  private:
-  scoped_refptr<CertNetFetcherCore> core_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  // Not owned. |context_| must stay valid until Shutdown() is called.
+  URLRequestContext* context_ = nullptr;
+  std::unique_ptr<AsyncCertNetFetcherImpl> impl_;
 };
 
 }  // namespace
 
-std::unique_ptr<CertNetFetcher> CreateCertNetFetcher(
-    URLRequestContextGetter* context_getter) {
-  return base::MakeUnique<CertNetFetcherImpl>(context_getter);
+scoped_refptr<CertNetFetcher> CreateCertNetFetcher(URLRequestContext* context) {
+  return make_scoped_refptr(new CertNetFetcherImpl(context));
 }
 
 }  // namespace net