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. Has constructors that allow it be created on
+//     either thread, but can only be 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
 
@@ skipping 53 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

[eroman, 2017/01/03 20:42:37]

This comment is no longer true.

[estark, 2017/01/05 19:08:39]

Done.

   // lifetime of the AsyncCertNetFetcherImpl.
   explicit AsyncCertNetFetcherImpl(URLRequestContext* context);
 
   // Deletion implicitly cancels any outstanding requests.

[eroman, 2017/01/03 20:42:37]

This comment isn't quite true anymore either.

[estark, 2017/01/05 19:08:39]

Done.

   ~AsyncCertNetFetcherImpl();
 
   // Starts an asynchronous request to fetch the given URL. On completion
-  // |callback| will be invoked.
+  // request->OnJobCompleted() 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.
+  // Completion of the request will never occur synchronously. In other
+  // words it is guaranteed that request->OnJobCompleted() will only be
+  // invoked once the Fetch*() method has returned.
   void Fetch(std::unique_ptr<RequestParams> request_params,
-             RequestCore* request);
+             scoped_refptr<RequestCore> request);
+
+  // Cancels outstanding jobs.
+  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 141 matching lines @@
 // 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);
+  // 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 URLRequest. If StartURLRequest() is invoked subsequently,
+  // it will immediately call OnJobCompleted with an error.
+  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.
@@ skipping 11 matching lines @@
 
   // 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);
 
   // 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_;
 
   // Used to timeout the job when the URLRequest takes too long. This timer is
   // also used for notifying a failure to start the URLRequest.
   base::OneShotTimer timer_;
 
   // Non-owned pointer to the AsyncCertNetFetcherImpl that created this job.
   AsyncCertNetFetcherImpl* parent_;
 
+  // When set, StartURLRequest() will post a task to run
+  // OnJobCompleted() with an error and return immediately.
+  bool cancelled_ = false;
+
   DISALLOW_COPY_AND_ASSIGN(Job);
 };
 
 void RequestCore::Cancel() {
   if (!task_runner_->RunsTasksOnCurrentThread()) {
     task_runner_->PostTask(FROM_HERE, base::Bind(&RequestCore::Cancel, this));
     return;
   }
 
   if (job_) {
     auto* job = job_;
     job_ = nullptr;
     job->DetachRequest(this);
   }
 
   bytes_.clear();
   error_ = ERR_UNEXPECTED;
 }
 
 Job::Job(std::unique_ptr<RequestParams> request_params,
          AsyncCertNetFetcherImpl* parent)
     : request_params_(std::move(request_params)), parent_(parent) {}
 
 Job::~Job() {
   DCHECK(requests_.empty());

[eroman, 2017/01/03 20:42:37]

It looks like Job::Cancel() doesn't clear requests however

[estark, 2017/01/05 19:08:40]

Done. Job::Cancel now calls FailRequest which clears the requests after
signalling them.

   Stop();
 }
 
-void Job::AttachRequest(RequestCore* request) {
+void Job::AttachRequest(scoped_refptr<RequestCore> request) {
   requests_.push_back(request);
   request->AttachedToJob(this);
 }
 
 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) {
+  if (cancelled_) {
+    timer_.Start(
+        FROM_HERE, base::TimeDelta(),
+        base::Bind(&Job::OnJobCompleted, base::Unretained(this), ERR_ABORTED));
+    return;
+  }
   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));
     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() {
+  cancelled_ = true;
+  if (url_request_)
+    url_request_->Cancel();
+}
+
 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 73 matching lines @@
   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);
 
-  for (auto* request : requests_) {
+  for (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_);
   }
 
   return job->AttachRequest(request);
 }
 
+void AsyncCertNetFetcherImpl::Shutdown() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  for (const auto& job : jobs_) {
+    job.first->Cancel();
+  }
+}
+
 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) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
@@ skipping 31 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)) {}
+  CertNetFetcherImpl(scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+                     URLRequestContext* context)
+      : task_runner_(task_runner), context_(context) {}
 
-  ~CertNetFetcherImpl() override { core_->Abandon(); }
+  explicit CertNetFetcherImpl(URLRequestContextGetter* context_getter) {
+    context_getter->GetNetworkTaskRunner()->PostTask(

[eroman, 2017/01/03 20:42:37]

There is a possible race with:

  (1) Create  (on network thread)
  (2) call Shutdown()  on network thread (before the init task has run).

Per comment in header file, I think we should only support creation from the
network thread (initialization can be more direct then).

[estark, 2017/01/05 19:08:40]

Done.

+        FROM_HERE, base::Bind(&CertNetFetcherImpl::InitializeOnNetworkThread,
+                              this, context_getter));
+  }
+
+  void Shutdown() override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
+    base::AutoLock auto_lock(shutdown_lock_);
+    shutdown_ = true;
+    if (impl_) {
+      impl_->Shutdown();
+    }
+    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 {}
+
+  void InitializeOnNetworkThread(URLRequestContextGetter* context_getter) {
+    DCHECK(context_getter->GetNetworkTaskRunner()->RunsTasksOnCurrentThread());
+    task_runner_ = context_getter->GetNetworkTaskRunner();
+    context_ = context_getter->GetURLRequestContext();
+  }
+
+  void DoFetchOnNetworkThread(std::unique_ptr<RequestParams> request_params,
+                              scoped_refptr<RequestCore> request) {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
+
+    if (!context_) {
+      request->Cancel();
+      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));
+    base::AutoLock auto_lock(shutdown_lock_);
+    if (shutdown_)
+      return nullptr;
+
+    if (!task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(&CertNetFetcherImpl::DoFetchOnNetworkThread, this,
+                       base::Passed(&request_params), request_core))) {
+      return nullptr;
+    }
 
     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_;
+  base::Lock shutdown_lock_;
+  bool shutdown_ = false;
 };
 
 }  // namespace
 
-std::unique_ptr<CertNetFetcher> CreateCertNetFetcher(
+scoped_refptr<CertNetFetcher> CreateCertNetFetcher(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    URLRequestContext* context) {
+  return make_scoped_refptr(new CertNetFetcherImpl(task_runner, context));
+}
+
+scoped_refptr<CertNetFetcher> CreateCertNetFetcherOnCallerThread(
     URLRequestContextGetter* context_getter) {
-  return base::MakeUnique<CertNetFetcherImpl>(context_getter);
+  return make_scoped_refptr(new CertNetFetcherImpl(context_getter));
 }
 
 }  // namespace net