Update PpdProvider threading model.

chromeos/printing/ppd_provider.cc

 // Copyright 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 "chromeos/printing/ppd_provider.h"
 
 #include <utility>
 
 #include "base/files/file_util.h"
 #include "base/json/json_parser.h"
 #include "base/memory/ptr_util.h"
 #include "base/sequence_checker.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/threading/sequenced_task_runner_handle.h"
+#include "base/threading/thread_restrictions.h"
 #include "base/time/time.h"
 #include "base/values.h"
 #include "chromeos/printing/ppd_cache.h"
 #include "net/base/load_flags.h"
 #include "net/http/http_status_code.h"
 #include "net/url_request/url_fetcher.h"
 #include "net/url_request/url_fetcher_delegate.h"
 #include "net/url_request/url_request_context_getter.h"
 #include "url/gurl.h"
 
@@ skipping 25 matching lines @@
  private:
   // owner of this delegate.
   PpdProviderImpl* const owner_;
 };
 
 class PpdProviderImpl : public PpdProvider {
  public:
   PpdProviderImpl(
       const std::string& api_key,
       scoped_refptr<net::URLRequestContextGetter> url_context_getter,
+      scoped_refptr<base::SequencedTaskRunner> disk_task_runner,
       std::unique_ptr<PpdCache> cache,
       const PpdProvider::Options& options)
       : api_key_(api_key),
         forwarding_delegate_(this),
         url_context_getter_(url_context_getter),
+        disk_task_runner_(disk_task_runner),
         cache_(std::move(cache)),
-        options_(options) {
+        options_(options),
+        weak_factory_(this) {
     CHECK_GT(options_.max_ppd_contents_size_, 0U);
   }
   ~PpdProviderImpl() override {}
 
   void Resolve(const Printer::PpdReference& ppd_reference,
                const PpdProvider::ResolveCallback& cb) override {
     CHECK(!cb.is_null());
+    CHECK(base::SequencedTaskRunnerHandle::IsSet())
+        << "Resolve must be called from a SequencedTaskRunner context";
+    CHECK(resolve_fetcher_ == nullptr)
+        << "Can't have concurrent PpdProvider Resolve calls";
+    resolve_done_task_runner_ = base::SequencedTaskRunnerHandle::Get();

[stevenjb, 2016/11/11 19:26:43]

There is a lot of complexity here. I am wondering if we can simplify it some
using existing tools.

[Carlson, 2016/11/11 23:06:50]

I appreciate you taking the time to look carefully for stuff like this.  I'm
still finding my way around this codebase and likely doing many things in rather
suboptimal ways.

+    disk_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&PpdProviderImpl::ResolveImpl,
+                              weak_factory_.GetWeakPtr(), ppd_reference, cb));

[stevenjb, 2016/11/11 19:26:43]

Here we would normally use TaskRunner::PostTaskAndReplyWithResult.

[Carlson, 2016/11/11 23:06:50]

I'm not sure that works for what we want here.  Or, rather, I think it doesn't
work, but would be happy to find out otherwise.

If I use PostTaskAndReplyWithResult, then I'm going to end up with the callback
on the disk thread, right?  I thought it was idiomatic in Chrome to have the
callback for an API like this be invoked on the same thread, or at least same
sequence, as the API function.  I don't see how to do that with
PostTaskAndReply*

What am I missing?

[stevenjb, 2016/11/11 23:25:37]

See the documentaiton for TaskRunner::PostTaskAndReply, that's exactly what it
does:

  // Posts |task| on the current TaskRunner.  On completion, |reply|
  // is posted to the thread that called PostTaskAndReply()...

+  }
+
+  // Do the work of Resolve on disk_task_runner
+  void ResolveImpl(const Printer::PpdReference& ppd_reference,
+                   const PpdProvider::ResolveCallback& cb) {
+    base::ThreadRestrictions::AssertIOAllowed();
+    CHECK(!cb.is_null());
     CHECK(resolve_sequence_checker_.CalledOnValidSequence());
     CHECK(base::SequencedTaskRunnerHandle::IsSet())
         << "Resolve must be called from a SequencedTaskRunner context";
     CHECK(resolve_fetcher_ == nullptr)
         << "Can't have concurrent PpdProvider Resolve calls";
 
     base::Optional<base::FilePath> tmp = cache_->Find(ppd_reference);
     if (tmp) {
       // Cache hit.  Schedule the callback now and return.
-      base::SequencedTaskRunnerHandle::Get()->PostTask(
+      resolve_done_task_runner_->PostTask(
           FROM_HERE, base::Bind(cb, PpdProvider::SUCCESS, tmp.value()));

[stevenjb, 2016/11/11 19:26:43]

We would need to pass a unique_ptr<FilePath> to this, set it, and return
PpdProvider::SUCCESS here.

       return;
     }
 
     // We don't have a way to automatically resolve user-supplied PPDs yet.  So
     // if we have one specified, and it's not cached, we fail out rather than
     // fall back to quirks-server based resolution.  The reasoning here is that
     // if the user has specified a PPD when a quirks-server one exists, it
     // probably means the quirks server one doesn't work for some reason, so we
     // shouldn't silently use it.
     if (!ppd_reference.user_supplied_ppd_url.empty()) {
-      base::SequencedTaskRunnerHandle::Get()->PostTask(
+      resolve_done_task_runner_->PostTask(
           FROM_HERE, base::Bind(cb, PpdProvider::NOT_FOUND, base::FilePath()));

[stevenjb, 2016/11/11 19:26:44]

This would return PpdProvider::NOT_FOUND.

       return;
     }
 

[stevenjb, 2016/11/11 19:26:43]

It looks like the rest of this can be safely called on the primary task runner,
so here we could return "success" with an empty file, so the rest of this would
happen in the "Reply" method.

     resolve_reference_ = ppd_reference;
     resolve_done_callback_ = cb;
 
     resolve_fetcher_ =
         net::URLFetcher::Create(GetQuirksServerPpdLookupURL(ppd_reference),
                                 net::URLFetcher::GET, &forwarding_delegate_);

[stevenjb, 2016/11/11 19:26:43]

Here, instead of using a forwarding_delegate_ member, construct a new delegate
for each request and pass |cb| and |ppd_reference| to it. The delegate can store
that task runner it was created on (base::SequencedTaskRunnerHandle::Get) which
would be the same task runner that Resolve() was called on, and
ForwardingURLFetcherDelegate::OnURLFetchComplete() can post
owner_->OnURLFetchComplete() on that task runner, with |cb| and |ppd_reference|
as arguments.

That way, instead of storing state within PpdProviderImpl (which is effectively
a global), we store the state in bindings and ForwardingURLFetcherDelegate. This
will make it easier for other developers to follow what is actually part of the
global state, and what is part of the current request. Even if we never need to
surpport more than one request at a time, this will make the code somewhat
easier to follow.

[Carlson, 2016/11/11 23:06:50]

I spent a while going down this path (because it does seem like a good idea),
but the lifetime of a per-request delegate object is problematic -- I don't see
a good way to ensure that it gets cleaned up.  It can't be self-deleting because
we have no guarantee that OnURLFetchComplete() (or any other delegate method)
will ever be called.  


Did you have an idea here I'm missing?

I can still shift some stuff like the query_task_runner_, query_done_callback_
and similar things into the delegate object, but if we have to keep the
ownership in PpdProvider, we still end up with this sort of pseudo-global state
around the fetch.

I'm happy to do that, just not sure it's as worth it without the other benefits.
 Let me know what you think, or if you have another idea.

[stevenjb, 2016/11/11 23:25:37]

So, for now at least, we can continue to have a single resolve_fetcher_, and can
continue to check above that resolve_fetcher_ is null. Anywhere we reset it
(OnResolveFetchComplete, AbortResolve()), we should also reset the associated
delegate.

We could also just keep the current model but pass the parameters to the single
delegate instance each time.

Either way, I think that wrapping the members associated with the current
request in the delegate still makes sense. It at least makes it a bit more clear
that those members are tied to the request.

     resolve_fetcher_->SetRequestContext(url_context_getter_.get());
     resolve_fetcher_->SetLoadFlags(
         net::LOAD_BYPASS_CACHE | net::LOAD_DISABLE_CACHE |
         net::LOAD_DO_NOT_SAVE_COOKIES | net::LOAD_DO_NOT_SEND_COOKIES |
         net::LOAD_DO_NOT_SEND_AUTH_DATA);
     resolve_fetcher_->Start();
-  };
+  }
 
   void AbortResolve() override {
     // UrlFetcher guarantees that when the object has been destroyed, no further
     // callbacks will occur.
     resolve_fetcher_.reset();
+    resolve_done_task_runner_ = nullptr;
   }
 
   void QueryAvailable(const QueryAvailableCallback& cb) override {
     CHECK(!cb.is_null());
-    CHECK(query_sequence_checker_.CalledOnValidSequence());
+    CHECK(query_fetcher_ == nullptr)
+        << "Can't have concurrent PpdProvider QueryAvailable() calls";
     CHECK(base::SequencedTaskRunnerHandle::IsSet())
         << "QueryAvailable() must be called from a SequencedTaskRunner context";
-    CHECK(query_fetcher_ == nullptr)
-        << "Can't have concurrent PpdProvider QueryAvailable() calls";
+    query_done_task_runner_ = base::SequencedTaskRunnerHandle::Get();
+    disk_task_runner_->PostTask(FROM_HERE,
+                                base::Bind(&PpdProviderImpl::QueryAvailableImpl,
+                                           weak_factory_.GetWeakPtr(), cb));
+  }
+
+  void QueryAvailableImpl(const QueryAvailableCallback& cb) {
+    base::ThreadRestrictions::AssertIOAllowed();
+    CHECK(query_sequence_checker_.CalledOnValidSequence());
 
     base::Optional<PpdProvider::AvailablePrintersMap> result =
         cache_->FindAvailablePrinters();
     if (result) {
       // Satisfy from cache.
-      base::SequencedTaskRunnerHandle::Get()->PostTask(
+      query_done_task_runner_->PostTask(
           FROM_HERE, base::Bind(cb, PpdProvider::SUCCESS, result.value()));
       return;
     }
     // Not in the cache, ask QuirksServer.
     query_done_callback_ = cb;
 
     query_fetcher_ =
         net::URLFetcher::Create(GetQuirksServerPpdListURL(),
                                 net::URLFetcher::GET, &forwarding_delegate_);
     query_fetcher_->SetRequestContext(url_context_getter_.get());
     query_fetcher_->SetLoadFlags(
         net::LOAD_BYPASS_CACHE | net::LOAD_DISABLE_CACHE |
         net::LOAD_DO_NOT_SAVE_COOKIES | net::LOAD_DO_NOT_SEND_COOKIES |
         net::LOAD_DO_NOT_SEND_AUTH_DATA);
     query_fetcher_->Start();
   }
 
   void AbortQueryAvailable() override {
     // UrlFetcher guarantees that when the object has been destroyed, no further
     // callbacks will occur.
     query_fetcher_.reset();
+    query_done_task_runner_ = nullptr;
   }
 
   bool CachePpd(const Printer::PpdReference& ppd_reference,
                 const base::FilePath& ppd_path) override {
     std::string buf;
     if (!base::ReadFileToStringWithMaxSize(ppd_path, &buf,
                                            options_.max_ppd_contents_size_)) {
       return false;
     }
     return static_cast<bool>(cache_->Store(ppd_reference, buf));
   }
 
   // Route to the proper fetch complete handler based on which fetcher caused
   // it.
   void OnURLFetchComplete(const net::URLFetcher* fetcher) {
     if (fetcher == resolve_fetcher_.get()) {
       OnResolveFetchComplete();
     } else if (fetcher == query_fetcher_.get()) {
       OnQueryAvailableFetchComplete();
     } else {
       NOTREACHED() << "Unknown fetcher completed.";
     }
   }
 
  private:
   // Called on the same thread as Resolve() when the fetcher completes its
   // fetch.
   void OnResolveFetchComplete() {
     CHECK(resolve_sequence_checker_.CalledOnValidSequence());
-    // Scope the allocated |resolve_fetcher_| into this function so we clean it
-    // up when we're done here instead of leaving it around until the next
-    // Resolve() call.
-    auto fetcher = std::move(resolve_fetcher_);
+    // Scope the allocated |resolve_fetcher_| and |resolve_done_task_runner_|
+    // into this function so we clean it up when we're done here instead of
+    // leaving it around until the next Resolve() call.
+    auto task_runner(std::move(resolve_done_task_runner_));
+    auto fetcher(std::move(resolve_fetcher_));
     std::string contents;
     if (!ValidateAndGetResponseAsString(*fetcher, &contents)) {
       // Something went wrong with the fetch.
-      resolve_done_callback_.Run(PpdProvider::SERVER_ERROR, base::FilePath());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(resolve_done_callback_,
+                                PpdProvider::SERVER_ERROR, base::FilePath()));
       return;
     }
 
     auto dict = base::DictionaryValue::From(base::JSONReader::Read(contents));
     if (dict == nullptr) {
-      resolve_done_callback_.Run(PpdProvider::SERVER_ERROR, base::FilePath());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(resolve_done_callback_,
+                                PpdProvider::SERVER_ERROR, base::FilePath()));
       return;
     }
     std::string ppd_contents;
     std::string last_updated_time_string;
     uint64_t last_updated_time;
     if (!(dict->GetString(kJSONPPDKey, &ppd_contents) &&
           dict->GetString(kJSONLastUpdatedKey, &last_updated_time_string) &&
           base::StringToUint64(last_updated_time_string, &last_updated_time))) {
       // Malformed response.  TODO(justincarlson) - LOG something here?
-      resolve_done_callback_.Run(PpdProvider::SERVER_ERROR, base::FilePath());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(resolve_done_callback_,
+                                PpdProvider::SERVER_ERROR, base::FilePath()));
       return;
     }
 
     if (ppd_contents.size() > options_.max_ppd_contents_size_) {
       // PPD is too big.
       //
       // Note -- if we ever add shared-ppd-sourcing, e.g. we may serve a ppd to
       // a user that's not from an explicitly trusted source, we should also
       // check *uncompressed* size here to head off zip-bombs (e.g. let's
       // compress 1GBs of zeros into a 900kb file and see what cups does when it
       // tries to expand that...)
       resolve_done_callback_.Run(PpdProvider::SERVER_ERROR, base::FilePath());
       return;
     }
 
     auto ppd_file = cache_->Store(resolve_reference_, ppd_contents);
     if (!ppd_file) {
       // Failed to store.
-      resolve_done_callback_.Run(PpdProvider::INTERNAL_ERROR, base::FilePath());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(resolve_done_callback_,
+                                PpdProvider::INTERNAL_ERROR, base::FilePath()));
       return;
     }
-    resolve_done_callback_.Run(PpdProvider::SUCCESS, ppd_file.value());
+    task_runner->PostTask(
+        FROM_HERE, base::Bind(resolve_done_callback_, PpdProvider::SUCCESS,
+                              ppd_file.value()));
   }
 
   // Called on the same thread as QueryAvailable() when the fetcher completes
   // its fetch.
   void OnQueryAvailableFetchComplete() {
     CHECK(query_sequence_checker_.CalledOnValidSequence());
-    // Scope the object fetcher into this function so we clean it up when we're
-    // done here instead of leaving it around until the next QueryAvailable()
-    // call.
-    auto fetcher = std::move(query_fetcher_);
+    // Scope the object fetcher and task_runner into this function so we clean
+    // it up when we're done here instead of leaving it around until the next
+    // QueryAvailable() call.
+    auto task_runner(std::move(query_done_task_runner_));
+    auto fetcher(std::move(query_fetcher_));
     std::string contents;
     if (!ValidateAndGetResponseAsString(*fetcher, &contents)) {
       // Something went wrong with the fetch.
-      query_done_callback_.Run(PpdProvider::SERVER_ERROR,
-                               AvailablePrintersMap());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(query_done_callback_, PpdProvider::SERVER_ERROR,
+                                AvailablePrintersMap()));
       return;
     }
 
     // The server gives us JSON in the form of a list of (manufacturer, list of
     // models) tuples.
     auto top_dict =
         base::DictionaryValue::From(base::JSONReader::Read(contents));
     const base::ListValue* top_list;
     if (top_dict == nullptr || !top_dict->GetList(kJSONTopListKey, &top_list)) {
       LOG(ERROR) << "Malformed response from quirks server";
-      query_done_callback_.Run(PpdProvider::SERVER_ERROR,
-                               PpdProvider::AvailablePrintersMap());
+      task_runner->PostTask(
+          FROM_HERE, base::Bind(query_done_callback_, PpdProvider::SERVER_ERROR,
+                                AvailablePrintersMap()));
       return;
     }
 
     PpdProvider::AvailablePrintersMap result;
     for (const std::unique_ptr<base::Value>& entry : *top_list) {
       base::DictionaryValue* dict;
       std::string manufacturer;
       std::string model;
       base::ListValue* model_list;
       if (!entry->GetAsDictionary(&dict) ||
           !dict->GetString(kJSONManufacturer, &manufacturer) ||
           !dict->GetList(kJSONModelList, &model_list)) {
         LOG(ERROR) << "Unexpected contents in quirks server printer list.";
         // Just skip this entry instead of aborting the whole thing.
         continue;
       }
 
       std::vector<std::string>& dest = result[manufacturer];
       for (const std::unique_ptr<base::Value>& model_value : *model_list) {
         if (model_value->GetAsString(&model)) {
           dest.push_back(model);
         } else {
           LOG(ERROR) << "Skipping unknown model for manufacturer "
                      << manufacturer;
         }
       }
     }
+
     if (!result.empty()) {
       cache_->StoreAvailablePrinters(result);
     } else {
       // An empty map means something is probably wrong; if we cache this map,
       // we'll have an empty map until the cache expires.  So complain and
       // refuse to cache.
       LOG(ERROR) << "Available printers map is unexpectedly empty.  Refusing "
                     "to cache this.";
     }
-    query_done_callback_.Run(PpdProvider::SUCCESS, result);
+    task_runner->PostTask(FROM_HERE, base::Bind(query_done_callback_,
+                                                PpdProvider::SUCCESS, result));
   }
 
   // Generate a url to look up a manufacturer/model from the quirks server
   GURL GetQuirksServerPpdLookupURL(
       const Printer::PpdReference& ppd_reference) const {
     return GURL(base::StringPrintf(
         "https://%s/v2/printer/manufacturers/%s/models/%s?key=%s",
         options_.quirks_server.c_str(),
         ppd_reference.effective_manufacturer.c_str(),
         ppd_reference.effective_model.c_str(), api_key_.c_str()));
@@ skipping 14 matching lines @@
     return ((fetcher.GetStatus().status() == net::URLRequestStatus::SUCCESS) &&
             (fetcher.GetResponseCode() == net::HTTP_OK) &&
             fetcher.GetResponseAsString(contents));
   }
 
   // State held across a Resolve() call.
 
   // Reference we're currently trying to resolve.
   Printer::PpdReference resolve_reference_;
 
+  // TaskRunner on which to invoke the Resolve callback;
+  scoped_refptr<base::SequencedTaskRunner> resolve_done_task_runner_;
+
   // Callback to invoke on completion.
   PpdProvider::ResolveCallback resolve_done_callback_;
 
   // Check that Resolve() and its callback are sequenced appropriately.
   base::SequenceChecker resolve_sequence_checker_;
 
-  // Fetcher for the current call, if any.
+  // Fetcher for the current call, if any.  resolve_fetcher_ is null if there
+  // is no running Resolve().
   std::unique_ptr<net::URLFetcher> resolve_fetcher_;
 
   // State held across a QueryAvailable() call.
 
   // Callback to invoke on completion.
   PpdProvider::QueryAvailableCallback query_done_callback_;
 
+  // TaskRunner on which to invoke the Query Available callback;
+  scoped_refptr<base::SequencedTaskRunner> query_done_task_runner_;
+
   // Check that QueryAvailable() and its callback are sequenced appropriately.
   base::SequenceChecker query_sequence_checker_;
 
   // Fetcher for the current call, if any.
   std::unique_ptr<net::URLFetcher> query_fetcher_;
 
   // Common state.
 
   // API key for accessing quirks server.
   const std::string api_key_;
 
   ForwardingURLFetcherDelegate forwarding_delegate_;
   scoped_refptr<net::URLRequestContextGetter> url_context_getter_;
+  scoped_refptr<base::SequencedTaskRunner> disk_task_runner_;
   std::unique_ptr<PpdCache> cache_;
 
   // Construction-time options, immutable.
   const PpdProvider::Options options_;
+
+  base::WeakPtrFactory<PpdProviderImpl> weak_factory_;
 };
 
 void ForwardingURLFetcherDelegate::OnURLFetchComplete(
     const net::URLFetcher* source) {
   owner_->OnURLFetchComplete(source);
 }
 
 }  // namespace
 
 // static
 std::unique_ptr<PpdProvider> PpdProvider::Create(
     const std::string& api_key,
     scoped_refptr<net::URLRequestContextGetter> url_context_getter,
+    scoped_refptr<base::SequencedTaskRunner> disk_task_runner,
     std::unique_ptr<PpdCache> cache,
     const PpdProvider::Options& options) {
-  return base::MakeUnique<PpdProviderImpl>(api_key, url_context_getter,
-                                           std::move(cache), options);
+  return base::MakeUnique<PpdProviderImpl>(
+      api_key, url_context_getter, disk_task_runner, std::move(cache), options);
 }
 
 }  // namespace printing
 }  // namespace chromeos