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..a2914873246da1c29fe02b4fdfe67dd374677c02 100644
--- a/net/quic/quic_stream_factory.cc
+++ b/net/quic/quic_stream_factory.cc
@@ -170,15 +170,23 @@ class QuicStreamFactory::Job {
 
   void OnIOComplete(int rv);
 
+  void RunAuxilaryJob();
+
+  void CancelWaitForDataReadyPendingCallback();
+
+  void CancelJob();
+
   void CancelWaitForDataReadyCallback();
 
-  CompletionCallback callback() {
-    return callback_;
-  }
+  CompletionCallback callback() { return callback_; }
 
-  const QuicServerId server_id() const {
-    return server_id_;
-  }
+  const QuicServerId server_id() const { return server_id_; }
+
+  QuicSession* session() { return session_; }
+
+  bool is_cancelled() const { return is_cancelled_; }
+
+  base::WeakPtr<Job> GetWeakPtr() { return weak_factory_.GetWeakPtr(); }
 
  private:
   enum IoState {
@@ -190,15 +198,19 @@ class QuicStreamFactory::Job {
     STATE_CONNECT,
     STATE_RESUME_CONNECT,
     STATE_CONNECT_COMPLETE,
+    STATE_CANCELLED,
   };
   IoState io_state_;
 
   QuicStreamFactory* factory_;
   SingleRequestHostResolver host_resolver_;
   QuicServerId server_id_;
+  base::StringPiece method_;

[Ryan Hamilton, 2015/02/05 20:30:28]

instead of doing this, let's change the Job() constructor to just take bool
is_post

[ramant (doing other things), 2015/02/06 00:59:41]

Done.

   bool is_post_;
   bool was_alternate_protocol_recently_broken_;
   scoped_ptr<QuicServerInfo> server_info_;
+  bool is_cancelled_;
+  bool started_another_job_;
   const BoundNetLog net_log_;
   QuicClientSession* session_;
   CompletionCallback callback_;
@@ -222,13 +234,17 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
       factory_(factory),
       host_resolver_(host_resolver),
       server_id_(host_port_pair, is_https, privacy_mode),
+      method_(method),
       is_post_(method == "POST"),
       was_alternate_protocol_recently_broken_(
           was_alternate_protocol_recently_broken),
       server_info_(server_info),
+      is_cancelled_(false),
+      started_another_job_(false),
       net_log_(net_log),
       session_(nullptr),
-      weak_factory_(this) {}
+      weak_factory_(this) {
+}
 
 QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
                             HostResolver* host_resolver,
@@ -238,11 +254,13 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
       factory_(factory),
       host_resolver_(host_resolver),  // unused
       server_id_(server_id),
-      is_post_(false),  // unused
+      method_("Get"),                                  // unused
+      is_post_(false),                                 // unused
       was_alternate_protocol_recently_broken_(false),  // unused
-      net_log_(session->net_log()),  // unused
+      net_log_(session->net_log()),                    // unused
       session_(session),
-      weak_factory_(this) {}
+      weak_factory_(this) {
+}
 
 QuicStreamFactory::Job::~Job() {
 }
@@ -285,6 +303,8 @@ int QuicStreamFactory::Job::DoLoop(int rv) {
       case STATE_CONNECT_COMPLETE:
         rv = DoConnectComplete(rv);
         break;
+      case STATE_CANCELLED:
+        return OK;  // exit the DoLoop.
       default:
         NOTREACHED() << "io_state_: " << io_state_;
         break;
@@ -310,11 +330,37 @@ void QuicStreamFactory::Job::OnIOComplete(int rv) {
   }
 }
 
+void QuicStreamFactory::Job::RunAuxilaryJob() {
+  started_another_job_ = true;

[Ryan Hamilton, 2015/02/05 20:30:28]

Does started_nother_job_ mean that this job started another job, or that this
job was started by another job, or both? It looks like the only time it's set to
true is here, where |this| is the aux job. But down on 476 is looks like we
expect it to be true in the main job.

[ramant (doing other things), 2015/02/06 00:59:41]

Done.

+  int rv = Run(base::Bind(&QuicStreamFactory::OnJobComplete,
+                          base::Unretained(factory_), this));
+  if (rv == OK) {
+    factory_->OnJobComplete(this, rv);
+  }
+}
+
+void QuicStreamFactory::Job::CancelWaitForDataReadyPendingCallback() {
+  // If we are waiting for WaitForDataReadyCallback, then cancel the pending
+  // callback.
+  if (server_info_ && io_state_ == STATE_LOAD_SERVER_INFO_COMPLETE)
+    server_info_->CancelWaitForDataReadyCallback();
+}
+
+void QuicStreamFactory::Job::CancelJob() {
+  DCHECK(!is_cancelled_);
+  CancelWaitForDataReadyPendingCallback();
+  callback_.Reset();
+  if (session_) {
+    session_->connection()->SendConnectionClose(
+        QUIC_LOAD_SERVER_CONFIG_JOB_CANCELLED);
+    session_ = nullptr;
+  }
+  is_cancelled_ = true;
+  io_state_ = STATE_CANCELLED;
+}
+
 void QuicStreamFactory::Job::CancelWaitForDataReadyCallback() {
-  // If we are waiting for WaitForDataReadyCallback, then cancel the callback.
-  if (io_state_ != STATE_LOAD_SERVER_INFO_COMPLETE)
-    return;
-  server_info_->CancelWaitForDataReadyCallback();
+  CancelWaitForDataReadyPendingCallback();
   OnIOComplete(OK);
 }
 
@@ -332,11 +378,9 @@ int QuicStreamFactory::Job::DoResolveHost() {
   io_state_ = STATE_RESOLVE_HOST_COMPLETE;
   dns_resolution_start_time_ = base::TimeTicks::Now();
   return host_resolver_.Resolve(
-      HostResolver::RequestInfo(server_id_.host_port_pair()),
-      DEFAULT_PRIORITY,
+      HostResolver::RequestInfo(server_id_.host_port_pair()), DEFAULT_PRIORITY,
       &address_list_,
-      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
-                 weak_factory_.GetWeakPtr()),
+      base::Bind(&QuicStreamFactory::Job::OnIOComplete, GetWeakPtr()),
       net_log_);
 }
 
@@ -388,14 +432,19 @@ int QuicStreamFactory::Job::DoLoadServerInfo() {
     factory_->task_runner_->PostDelayedTask(
         FROM_HERE,
         base::Bind(&QuicStreamFactory::Job::CancelWaitForDataReadyCallback,
-                   weak_factory_.GetWeakPtr()),
+                   GetWeakPtr()),
         base::TimeDelta::FromMilliseconds(load_server_info_timeout_ms));
   }
 
   disk_cache_load_start_time_ = base::TimeTicks::Now();
-  return server_info_->WaitForDataReady(
-      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
-                 weak_factory_.GetWeakPtr()));
+  int rv = server_info_->WaitForDataReady(
+      base::Bind(&QuicStreamFactory::Job::OnIOComplete, GetWeakPtr()));
+  if (rv == ERR_IO_PENDING && factory_->enable_connection_racing()) {
+    // If we are waiting to load server config from the disk cache, then start
+    // another job.
+    factory_->CreateAuxilaryJob(server_id_, method_, net_log_);
+  }
+  return rv;
 }
 
 int QuicStreamFactory::Job::DoLoadServerInfoComplete(int rv) {
@@ -409,9 +458,8 @@ int QuicStreamFactory::Job::DoLoadServerInfoComplete(int rv) {
                         base::TimeTicks::Now() - disk_cache_load_start_time_);
   }
 
-  if (rv != OK) {
+  if (rv != OK)
     server_info_.reset();
-  }
 
   io_state_ = STATE_CONNECT;
   return OK;
@@ -425,6 +473,16 @@ int QuicStreamFactory::Job::DoConnect() {
 
   io_state_ = STATE_CONNECT_COMPLETE;
 
+  if (started_another_job_ &&
+      ((server_info_ && server_info_->state().server_config.empty()) ||
+       !factory_->CryptoConfigCacheIsEmpty(server_id_))) {

[Ryan Hamilton, 2015/02/05 20:30:28]

Instead of tracking this here, would it make more sense to do this is
DoLoadServerInfoComplete()?

I think that in DoResolveHostComplete, we could check if we have server_info_
and if not, skip straight to STATE_CONNECTION. This would skip DoLoadServerInfo
and DoLoadServerInfoComplete when we have no server info.

[ramant (doing other things), 2015/02/06 00:59:41]

Done.

+    // If we have started another job and if we didn't load the server config
+    // from the disk cache or if we have received a new server config from the
+    // server, then cancel the current job.
+    CancelJob();

[Ryan Hamilton, 2015/02/05 20:30:28]

Instead of invoking CancelJob() can we simply return with the ERROR below and
rely on OnJobComplete() to do the right cleanup?

[ramant (doing other things), 2015/02/06 00:59:41]

Done.

+    return ERR_CONNECTION_CLOSED;
+  }
+
   int rv = factory_->CreateSession(server_id_, server_info_.Pass(),
                                    address_list_, net_log_, &session_);
   if (rv != OK) {
@@ -588,6 +646,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 +672,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 +703,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 +727,22 @@ 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,
+                              WasAlternateProtocolRecentlyBroken(server_id),
                               privacy_mode, method, quic_server_info, net_log));
   int rv = job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
                                base::Unretained(this), job.get()));
-
+  DCHECK(!job->is_cancelled());
   if (rv == ERR_IO_PENDING) {
-    active_requests_[request] = job.get();
-    job_requests_map_[job.get()].insert(request);
-    active_jobs_[server_id] = job.release();
+    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));
@@ -696,6 +751,19 @@ int QuicStreamFactory::Create(const HostPortPair& host_port_pair,
   return rv;
 }
 
+void QuicStreamFactory::CreateAuxilaryJob(const QuicServerId server_id,
+                                          base::StringPiece method,
+                                          const BoundNetLog& net_log) {
+  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, net_log);
+  active_jobs_[server_id].insert(aux_job);
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&QuicStreamFactory::Job::RunAuxilaryJob,
+                                    aux_job->GetWeakPtr()));
+}
+
 bool QuicStreamFactory::OnResolution(
     const QuicServerId& server_id,
     const AddressList& address_list) {
@@ -721,6 +789,12 @@ bool QuicStreamFactory::OnResolution(
 }
 
 void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
+  QuicServerId server_id = job->server_id();
+  if (job->is_cancelled()) {
+    // Verify there is atleast one other job to handle the requests.
+    DCHECK_LT(1u, active_jobs_[server_id].size());
+    return;

[Ryan Hamilton, 2015/02/05 20:30:28]

Do we need to move this job from active_jobs_?

[ramant (doing other things), 2015/02/06 00:59:42]

Done.

+  }
   if (rv == OK) {
     // TODO(vadimt): Remove ScopedTracker below once crbug.com/422516 is fixed.
     tracked_objects::ScopedTracker tracking_profile1(
@@ -731,11 +805,10 @@ 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()));
     }
   }
 
@@ -744,10 +817,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 +833,15 @@ 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;
+  for (Job* other_job : active_jobs_[server_id]) {
+    if (other_job == job || other_job->is_cancelled())
+      continue;
+    other_job->CancelJob();
+  }
+
+  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 +920,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 +928,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 +999,10 @@ bool QuicStreamFactory::HasActiveSession(
   return ContainsKey(active_sessions_, server_id);
 }
 
+bool QuicStreamFactory::HasActiveJob(const QuicServerId& key) const {
+  return ContainsKey(active_jobs_, key);
+}
+
 int QuicStreamFactory::CreateSession(
     const QuicServerId& server_id,
     scoped_ptr<QuicServerInfo> server_info,
@@ -1137,10 +1219,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 +1244,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) {