QUIC - Race two connections. One connection which loads data from disk

net/quic/quic_stream_factory.cc

Index: net/quic/quic_stream_factory.cc
diff --git a/net/quic/quic_stream_factory.cc b/net/quic/quic_stream_factory.cc
index da2f56c9e75e89aa8811928c9d31bb12899bbee3..ff3da375016653d921194a4c33c2b44120c0ef0c 100644
--- a/net/quic/quic_stream_factory.cc
+++ b/net/quic/quic_stream_factory.cc
@@ -146,6 +146,7 @@ class QuicStreamFactory::Job {
       PrivacyMode privacy_mode,
       base::StringPiece method,
       QuicServerInfo* server_info,
+      bool is_main_job,
       const BoundNetLog& net_log);
 
   // Creates a new job to handle the resumption of for connecting an
@@ -172,12 +173,18 @@ class QuicStreamFactory::Job {
 
   void CancelWaitForDataReadyCallback();
 
-  CompletionCallback callback() {
-    return callback_;
-  }
+  CompletionCallback callback() { return callback_; }
+
+  const QuicServerId server_id() const { return server_id_; }
+
+  QuicSession* session() { return session_; }
+
+  bool is_main_job() const { return is_main_job_; }
 
-  const QuicServerId server_id() const {
-    return server_id_;
+  bool is_cancelled() const { return is_cancelled_; }

[Ryan Hamilton, 2015/02/03 18:54:36]

nit: newline after

[ramant (doing other things), 2015/02/04 17:40:46]

Done.

+  void CancelJob() {
+    DCHECK(!is_cancelled_);
+    is_cancelled_ = true;
   }
 
  private:
@@ -199,6 +206,8 @@ class QuicStreamFactory::Job {
   bool is_post_;
   bool was_alternate_protocol_recently_broken_;
   scoped_ptr<QuicServerInfo> server_info_;
+  bool is_main_job_;
+  bool is_cancelled_;
   const BoundNetLog net_log_;
   QuicClientSession* session_;
   CompletionCallback callback_;
@@ -217,6 +226,7 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
                             PrivacyMode privacy_mode,
                             base::StringPiece method,
                             QuicServerInfo* server_info,
+                            bool is_main_job,
                             const BoundNetLog& net_log)
     : io_state_(STATE_RESOLVE_HOST),
       factory_(factory),
@@ -226,9 +236,12 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
       was_alternate_protocol_recently_broken_(
           was_alternate_protocol_recently_broken),
       server_info_(server_info),
+      is_main_job_(is_main_job),
+      is_cancelled_(false),
       net_log_(net_log),
       session_(nullptr),
-      weak_factory_(this) {}
+      weak_factory_(this) {
+}
 
 QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
                             HostResolver* host_resolver,
@@ -245,6 +258,10 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
       weak_factory_(this) {}
 
 QuicStreamFactory::Job::~Job() {
+  if (server_info_) {
+    // Cancel any pending callbacks.
+    server_info_->CancelWaitForDataReadyCallback();

[Ryan Hamilton, 2015/02/03 18:54:36]

Out of curiosity, did we need this code before?

[ramant (doing other things), 2015/02/04 17:40:46]

If we had closed the job when it is in the WaitForDataReady (I think we never
did that before).

Deleted this code.

+  }
 }
 
 int QuicStreamFactory::Job::Run(const CompletionCallback& callback) {
@@ -425,6 +442,22 @@ int QuicStreamFactory::Job::DoConnect() {
 
   io_state_ = STATE_CONNECT_COMPLETE;
 
+  // If this is the main job, check if the auxilary job got the server config
+  // from the server, if so discontinue this job. If not, discontinue the
+  // auxilary job.
+  if (factory_->enable_connection_racing()) {
+    if (is_main_job_) {
+      if (!factory_->CryptoConfigCacheIsEmpty(server_id_)) {
+        is_cancelled_ = true;
+        return ERR_CONNECTION_CLOSED;
+      }
+      factory_->CancelAuxilaryJob(this, server_id_);
+    } else if (is_cancelled_) {
+      // This is the auxilary job and it is cancelled.
+      return ERR_CONNECTION_CLOSED;
+    }
+  }

[Ryan Hamilton, 2015/02/03 18:54:36]

This seems to be the only place where a job is cancelled. Does this mean a main
job can not be cancelled? That's surprising to me.

I would have expected that if we start two jobs, the only time we cancel a job
is when on of the jobs finishes. Does that seem plausible or does that actually
cause problems?

[ramant (doing other things), 2015/02/04 17:40:46]

Cancelled the job that loads server config from disk cache (if it looses).
Originally cancelled the job as soon as we have a server config. This avoids two
certificate verifications/proof/channel_id checks happening on the client (to
reduce the work we do on Android/client. Was feeling guilty to use resources on
Android).

Changed the code to race until job completion.

+
   int rv = factory_->CreateSession(server_id_, server_info_.Pass(),
                                    address_list_, net_log_, &session_);
   if (rv != OK) {
@@ -588,6 +621,7 @@ QuicStreamFactory::QuicStreamFactory(
       load_server_info_timeout_srtt_multiplier_(
           load_server_info_timeout_srtt_multiplier),
       enable_truncated_connection_ids_(enable_truncated_connection_ids),
+      enable_connection_racing_(false),
       port_seed_(random_generator_->RandUint64()),
       check_persisted_supports_quic_(true),
       task_runner_(nullptr),
@@ -613,7 +647,11 @@ QuicStreamFactory::~QuicStreamFactory() {
     delete all_sessions_.begin()->first;
     all_sessions_.erase(all_sessions_.begin());
   }
-  STLDeleteValues(&active_jobs_);
+  while (!active_jobs_.empty()) {
+    const QuicServerId server_id = active_jobs_.begin()->first;
+    STLDeleteElements(&(active_jobs_[server_id]));
+    active_jobs_.erase(server_id);
+  }
 }
 
 void QuicStreamFactory::set_require_confirmation(bool require_confirmation) {
@@ -640,12 +678,16 @@ int QuicStreamFactory::Create(const HostPortPair& host_port_pair,
   }
 
   if (HasActiveJob(server_id)) {
-    Job* job = active_jobs_[server_id];
-    active_requests_[request] = job;
-    job_requests_map_[job].insert(request);
+    active_requests_[request] = server_id;
+    job_requests_map_[server_id].insert(request);
     return ERR_IO_PENDING;
   }
 
+  // TODO(rtenneti): |task_runner_| is used by the Job. Initialize task_runner_
+  // in the constructor after WebRequestActionWithThreadsTest.* tests are fixed.
+  if (!task_runner_)
+    task_runner_ = base::MessageLoop::current()->message_loop_proxy().get();
+
   QuicServerInfo* quic_server_info = nullptr;
   if (quic_server_info_factory_) {
     bool load_from_disk_cache = true;
@@ -660,34 +702,50 @@ int QuicStreamFactory::Create(const HostPortPair& host_port_pair,
         load_from_disk_cache = false;
       }
     }
-    if (load_from_disk_cache) {
-      QuicCryptoClientConfig::CachedState* cached =
-          crypto_config_.LookupOrCreate(server_id);
-      DCHECK(cached);
-      if (cached->IsEmpty()) {
-        quic_server_info = quic_server_info_factory_->GetForServer(server_id);
-      }
+    if (load_from_disk_cache && CryptoConfigCacheIsEmpty(server_id)) {
+      quic_server_info = quic_server_info_factory_->GetForServer(server_id);
     }
   }
-  // TODO(rtenneti): Initialize task_runner_ in the constructor after
-  // WebRequestActionWithThreadsTest.* tests are fixed.
-  if (!task_runner_)
-    task_runner_ = base::MessageLoop::current()->message_loop_proxy().get();
 
-  bool was_alternate_protocol_recently_broken =
-      http_server_properties_ &&
-      http_server_properties_->WasAlternateProtocolRecentlyBroken(
-          server_id.host_port_pair());
   scoped_ptr<Job> job(new Job(this, host_resolver_, host_port_pair, is_https,
-                              was_alternate_protocol_recently_broken,
-                              privacy_mode, method, quic_server_info, net_log));
+                              WasAlternateProtocolRecentlyBroken(server_id),
+                              privacy_mode, method, quic_server_info,
+                              true /*is_main_job*/, net_log));
   int rv = job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
                                base::Unretained(this), job.get()));
 
   if (rv == ERR_IO_PENDING) {
-    active_requests_[request] = job.get();
-    job_requests_map_[job.get()].insert(request);
-    active_jobs_[server_id] = job.release();
+    // Start an auxilary job that doesn't require loading of server config
+    // from the disk cache.
+    if (enable_connection_racing_ && quic_server_info &&
+        (CreateAuxilaryJob(server_id, method, net_log, request) == OK)) {
+      return OK;

[Ryan Hamilton, 2015/02/03 18:54:36]

If this returns OK, it looks like we skip the code on like 730. Is that
expected?

[ramant (doing other things), 2015/02/04 17:40:46]

Removed the duplicated code from lines 750-753.

Done.

+    }
+    active_requests_[request] = server_id;
+    job_requests_map_[server_id].insert(request);
+    active_jobs_[server_id].insert(job.release());
+    return rv;
+  }
+  if (rv == OK) {
+    DCHECK(HasActiveSession(server_id));
+    request->set_stream(CreateIfSessionExists(server_id, net_log));
+  }
+  return rv;
+}
+
+int QuicStreamFactory::CreateAuxilaryJob(const QuicServerId server_id,
+                                         base::StringPiece method,
+                                         const BoundNetLog& net_log,
+                                         QuicStreamRequest* request) {
+  scoped_ptr<Job> aux_job(new Job(
+      this, host_resolver_, server_id.host_port_pair(), server_id.is_https(),
+      WasAlternateProtocolRecentlyBroken(server_id), server_id.privacy_mode(),
+      method, nullptr, false /*is_main_job*/, net_log));
+  int rv = aux_job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
+                                   base::Unretained(this), aux_job.get()));
+  if (rv == ERR_IO_PENDING) {
+    active_jobs_[server_id].insert(aux_job.release());
+    return rv;
   }
   if (rv == OK) {
     DCHECK(HasActiveSession(server_id));
@@ -721,6 +779,7 @@ bool QuicStreamFactory::OnResolution(
 }
 
 void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
+  QuicServerId server_id = job->server_id();
   if (rv == OK) {
     // TODO(vadimt): Remove ScopedTracker below once crbug.com/422516 is fixed.
     tracked_objects::ScopedTracker tracking_profile1(
@@ -731,12 +790,13 @@ void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
       set_require_confirmation(false);
 
     // Create all the streams, but do not notify them yet.
-    for (RequestSet::iterator it = job_requests_map_[job].begin();
-         it != job_requests_map_[job].end() ; ++it) {
-      DCHECK(HasActiveSession(job->server_id()));
-      (*it)->set_stream(CreateIfSessionExists(job->server_id(),
-                                              (*it)->net_log()));
+    for (RequestSet::iterator it = job_requests_map_[server_id].begin();
+         it != job_requests_map_[server_id].end(); ++it) {
+      DCHECK(HasActiveSession(server_id));
+      (*it)->set_stream(CreateIfSessionExists(server_id, (*it)->net_log()));
     }
+  } else if (job->is_cancelled() && (!active_jobs_[server_id].empty())) {

[Ryan Hamilton, 2015/02/03 18:54:36]

Can rv == OK if the job was cancelled? In this case is |job| still in
active_jobs_?

[ramant (doing other things), 2015/02/04 17:40:46]

Previously STLDeleteElements was deleting all the jobs. Changed the code to
delete cancelled job right away.

+    return;
   }
 
   // TODO(vadimt): Remove ScopedTracker below once crbug.com/422516 is fixed.
@@ -744,10 +804,10 @@ void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "422516 QuicStreamFactory::OnJobComplete2"));
 
-  while (!job_requests_map_[job].empty()) {
-    RequestSet::iterator it = job_requests_map_[job].begin();
+  while (!job_requests_map_[server_id].empty()) {
+    RequestSet::iterator it = job_requests_map_[server_id].begin();
     QuicStreamRequest* request = *it;
-    job_requests_map_[job].erase(it);
+    job_requests_map_[server_id].erase(it);
     active_requests_.erase(request);
     // Even though we're invoking callbacks here, we don't need to worry
     // about |this| being deleted, because the factory is owned by the
@@ -760,10 +820,9 @@ void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "422516 QuicStreamFactory::OnJobComplete3"));
 
-  active_jobs_.erase(job->server_id());
-  job_requests_map_.erase(job);
-  delete job;
-  return;
+  STLDeleteElements(&(active_jobs_[server_id]));
+  active_jobs_.erase(server_id);
+  job_requests_map_.erase(server_id);
 }
 
 // Returns a newly created QuicHttpStream owned by the caller, if a
@@ -842,7 +901,7 @@ void QuicStreamFactory::OnSessionConnectTimeout(
   QuicServerId server_id = *aliases.begin();
   session_aliases_.erase(session);
   Job* job = new Job(this, host_resolver_, session, server_id);
-  active_jobs_[server_id] = job;
+  active_jobs_[server_id].insert(job);
   int rv = job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
                                base::Unretained(this), job));
   DCHECK_EQ(ERR_IO_PENDING, rv);
@@ -850,8 +909,8 @@ void QuicStreamFactory::OnSessionConnectTimeout(
 
 void QuicStreamFactory::CancelRequest(QuicStreamRequest* request) {
   DCHECK(ContainsKey(active_requests_, request));
-  Job* job = active_requests_[request];
-  job_requests_map_[job].erase(request);
+  QuicServerId server_id = active_requests_[request];
+  job_requests_map_[server_id].erase(request);
   active_requests_.erase(request);
 }
 
@@ -921,6 +980,31 @@ bool QuicStreamFactory::HasActiveSession(
   return ContainsKey(active_sessions_, server_id);
 }
 
+bool QuicStreamFactory::HasActiveJob(const QuicServerId& key) const {
+  return ContainsKey(active_jobs_, key);
+}
+
+void QuicStreamFactory::CancelAuxilaryJob(Job* job,
+                                          const QuicServerId& server_id) {
+  Job* aux_job = nullptr;
+  for (JobSet::iterator it = active_jobs_[server_id].begin();
+       it != active_jobs_[server_id].end(); ++it) {
+    if (*it != job) {
+      aux_job = *it;
+      break;
+    }
+  }
+  if (!aux_job) {
+    return;
+  }
+  DCHECK(!aux_job->is_main_job());
+  aux_job->CancelJob();
+  if (aux_job->session()) {
+    aux_job->session()->connection()->SendConnectionClose(
+        QUIC_CONNECTION_IP_POOLED);
+  }
+}
+
 int QuicStreamFactory::CreateSession(
     const QuicServerId& server_id,
     scoped_ptr<QuicServerInfo> server_info,
@@ -1137,10 +1221,6 @@ int QuicStreamFactory::CreateSession(
   return OK;
 }
 
-bool QuicStreamFactory::HasActiveJob(const QuicServerId& key) const {
-  return ContainsKey(active_jobs_, key);
-}
-
 void QuicStreamFactory::ActivateSession(
     const QuicServerId& server_id,
     QuicClientSession* session) {
@@ -1166,6 +1246,20 @@ int64 QuicStreamFactory::GetServerNetworkStatsSmoothedRttInMicroseconds(
   return stats->srtt.InMicroseconds();
 }
 
+bool QuicStreamFactory::WasAlternateProtocolRecentlyBroken(
+    const QuicServerId& server_id) const {
+  return http_server_properties_ &&
+         http_server_properties_->WasAlternateProtocolRecentlyBroken(
+             server_id.host_port_pair());
+}
+
+bool QuicStreamFactory::CryptoConfigCacheIsEmpty(
+    const QuicServerId& server_id) {
+  QuicCryptoClientConfig::CachedState* cached =
+      crypto_config_.LookupOrCreate(server_id);
+  return cached->IsEmpty();
+}
+
 void QuicStreamFactory::InitializeCachedStateInCryptoConfig(
     const QuicServerId& server_id,
     const scoped_ptr<QuicServerInfo>& server_info) {