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..85e0fb1d3ea7941a4549d676f5b60614ac1a7668 100644
--- a/net/quic/quic_stream_factory.cc
+++ b/net/quic/quic_stream_factory.cc
@@ -170,15 +170,19 @@ class QuicStreamFactory::Job {
 
   void OnIOComplete(int rv);
 
+  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_; }
 
  private:
   enum IoState {
@@ -199,6 +203,8 @@ class QuicStreamFactory::Job {
   bool is_post_;
   bool was_alternate_protocol_recently_broken_;
   scoped_ptr<QuicServerInfo> server_info_;
+  bool waiting_to_load_server_config_;
+  bool is_cancelled_;
   const BoundNetLog net_log_;
   QuicClientSession* session_;
   CompletionCallback callback_;
@@ -226,9 +232,12 @@ QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
       was_alternate_protocol_recently_broken_(
           was_alternate_protocol_recently_broken),
       server_info_(server_info),
+      waiting_to_load_server_config_(false),
+      is_cancelled_(false),
       net_log_(net_log),
       session_(nullptr),
-      weak_factory_(this) {}
+      weak_factory_(this) {
+}
 
 QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
                             HostResolver* host_resolver,
@@ -310,11 +319,22 @@ void QuicStreamFactory::Job::OnIOComplete(int rv) {
   }
 }
 
-void QuicStreamFactory::Job::CancelWaitForDataReadyCallback() {
-  // If we are waiting for WaitForDataReadyCallback, then cancel the callback.
-  if (io_state_ != STATE_LOAD_SERVER_INFO_COMPLETE)
+void QuicStreamFactory::Job::CancelWaitForDataReadyPendingCallback() {
+  if (!server_info_ || io_state_ != STATE_LOAD_SERVER_INFO_COMPLETE)
     return;
+  // If we are waiting for WaitForDataReadyCallback, then cancel the pending
+  // callback.

[Ryan Hamilton, 2015/02/04 20:10:29]

nit: I think I'd reverse the order here:
 
  // If we are waiting for WaitForDataReadyCallback, then cancel the pending
  // callback.
  if (server_info_ && io_state_ 0= STATE_LOAD_SERVER_INFO_COMPLETE)
    server_info_->CancelWaitForDataReadyCallback();
  }

This makes it one line shorter and makes it clear what the "if we are waiting
for..." part of the comment applies to (namely the io_state_).

[ramant (doing other things), 2015/02/05 03:39:41]

Done.

   server_info_->CancelWaitForDataReadyCallback();
+}
+
+void QuicStreamFactory::Job::CancelJob() {
+  DCHECK(!is_cancelled_);
+  CancelWaitForDataReadyPendingCallback();
+  is_cancelled_ = true;
+}
+
+void QuicStreamFactory::Job::CancelWaitForDataReadyCallback() {
+  CancelWaitForDataReadyPendingCallback();
   OnIOComplete(OK);
 }
 
@@ -393,6 +413,7 @@ int QuicStreamFactory::Job::DoLoadServerInfo() {
   }
 
   disk_cache_load_start_time_ = base::TimeTicks::Now();
+  waiting_to_load_server_config_ = true;
   return server_info_->WaitForDataReady(
       base::Bind(&QuicStreamFactory::Job::OnIOComplete,
                  weak_factory_.GetWeakPtr()));
@@ -425,6 +446,21 @@ int QuicStreamFactory::Job::DoConnect() {
 
   io_state_ = STATE_CONNECT_COMPLETE;
 
+  if (factory_->enable_connection_racing()) {
+    if (is_cancelled_) {
+      return ERR_CONNECTION_CLOSED;
+    }

[Ryan Hamilton, 2015/02/04 20:10:29]

If this is the right place to check this value, I'd do it outside of the "if
(enable_racing())" check. That being said, I'm concerned that this is not the
only place we need to check this value. Should we check this in DoLoop? Perhaps
there should be a new state STATE_CANCELLED which Cancel() sets io_state_ to?

[ramant (doing other things), 2015/02/05 03:39:41]

The only time, we cancel the job is when we wait (thus start another job) and
the other job got the server config from disk.

This job will be deleted when the other job finishes (we delete the jobs when
OnJobComplete finishes. Did this, as we had talked offline to reduce the number
of places where we delete the object.).

Done.

+    if (waiting_to_load_server_config_ &&
+        (server_info_->state().server_config.empty() ||
+         !factory_->CryptoConfigCacheIsEmpty(server_id_))) {
+      // 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();
+      return ERR_CONNECTION_CLOSED;
+    }
+  }
+
   int rv = factory_->CreateSession(server_id_, server_info_.Pass(),
                                    address_list_, net_log_, &session_);
   if (rv != OK) {
@@ -588,6 +624,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 +650,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 +681,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 +705,32 @@ 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));
+  // Start an auxilary job that loads server config from the disk cache.
+  if (enable_connection_racing_ && quic_server_info &&
+      (CreateAuxilaryJob(quic_server_info, server_id, method, net_log,
+                         request) == OK)) {
+    DCHECK(HasActiveSession(server_id));
+    request->set_stream(CreateIfSessionExists(server_id, net_log));
+    return OK;
+  }

[Ryan Hamilton, 2015/02/04 20:10:29]

Do you think it would make sense to delay starting the "no waiting" job until
the "wait for disk cache job" has resolved the host?

[ramant (doing other things), 2015/02/05 03:39:41]

Done.

+
+  scoped_ptr<Job> job(new Job(
+      this, host_resolver_, host_port_pair, is_https,
+      WasAlternateProtocolRecentlyBroken(server_id), privacy_mode, method,
+      enable_connection_racing_ ? nullptr : quic_server_info, 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();
+    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 +739,22 @@ int QuicStreamFactory::Create(const HostPortPair& host_port_pair,
   return rv;
 }
 
+int QuicStreamFactory::CreateAuxilaryJob(QuicServerInfo* quic_server_info,
+                                         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, quic_server_info, 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;
+}
+
 bool QuicStreamFactory::OnResolution(
     const QuicServerId& server_id,
     const AddressList& address_list) {
@@ -721,6 +780,14 @@ bool QuicStreamFactory::OnResolution(
 }
 
 void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
+  QuicServerId server_id = job->server_id();
+  if (job->is_cancelled()) {
+    JobSet* jobs = &(active_jobs_[server_id]);
+    if (jobs && (jobs->erase(job) == 1))
+      delete job;
+    DCHECK_LT(0u, jobs->size());  // Verify there is atleast one job.

[Ryan Hamilton, 2015/02/04 20:10:29]

It looks like jobs might be null. in that case, this DCHECK would be ... bad :> 
What's the invariant about when cancelled jobs must be removed from active_jobs?
Is it as soon as they're cancelled?

[ramant (doing other things), 2015/02/05 03:39:41]

In the previous implementation, removed the jobs as soon as they are cancelled
and OnJobComplete was called for those jobs.

If there is more than one job, we cancel that job (and all the callbacks) and
delete the job.

[ramant (doing other things), 2015/02/05 20:06:28]

Checked that there is at least one other job to complete the request.

The job that finishes the request, deletes all the jobs.

+    return;
+  }
   if (rv == OK) {
     // TODO(vadimt): Remove ScopedTracker below once crbug.com/422516 is fixed.
     tracked_objects::ScopedTracker tracking_profile1(
@@ -731,11 +798,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 +810,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 +826,10 @@ 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;
+  CancelOtherJob(job, server_id);                 // Cancel any pending job.
+  STLDeleteElements(&(active_jobs_[server_id]));  // Delete all jobs.

[Ryan Hamilton, 2015/02/04 20:10:29]

nit: I think this two comments are pretty redundant with the names of the
methods. I'd probably nuke 'em

[ramant (doing other things), 2015/02/05 03:39:41]

Done.

+  active_jobs_.erase(server_id);
+  job_requests_map_.erase(server_id);
 }
 
 // Returns a newly created QuicHttpStream owned by the caller, if a
@@ -842,7 +908,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 +916,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 +987,31 @@ bool QuicStreamFactory::HasActiveSession(
   return ContainsKey(active_sessions_, server_id);
 }
 
+bool QuicStreamFactory::HasActiveJob(const QuicServerId& key) const {
+  return ContainsKey(active_jobs_, key);
+}
+
+void QuicStreamFactory::CancelOtherJob(Job* job,
+                                       const QuicServerId& server_id) {
+  Job* other_job = nullptr;
+  for (JobSet::iterator it = active_jobs_[server_id].begin();

[Ryan Hamilton, 2015/02/04 20:10:29]

nit: for (Job* job : active_jobs_[server_id]) {
...
}

[ramant (doing other things), 2015/02/05 03:39:41]

Done.

+       it != active_jobs_[server_id].end(); ++it) {
+    if (*it != job) {
+      other_job = *it;

[Ryan Hamilton, 2015/02/04 20:10:29]

I think I'd be inclined to do this:

CancelOtherJobs(...) {
  for (Job* other_job : active_jobs_[server_id]) {
    if (other_job == job || other_job->is_cancelled())
      continue;
    other_job->Cancel();
}

I think I would move the SendConnectionClose() logic into the Cancel() method.
Also, consider a new error code for this condition. Not required but might be a
good idea.

[ramant (doing other things), 2015/02/05 03:39:41]

Done.

+      break;
+    }
+  }
+  if (!other_job)
+    return;
+  if (other_job->is_cancelled())
+    return;
+  other_job->CancelJob();

[Ryan Hamilton, 2015/02/04 20:10:29]

It looks like CancelJob does not invoke the callback to notify the factory that
the job is complete. Will something notify the factory about this job?

[ramant (doing other things), 2015/02/05 03:39:41]

Deleted this method. Factory is cancelling the job and deleting the job.

In the other case, when we call CancelJob, we call Factory's OnJobComplete.

+  if (other_job->session()) {
+    other_job->session()->connection()->SendConnectionClose(
+        QUIC_CONNECTION_IP_POOLED);
+  }
+}
+
 int QuicStreamFactory::CreateSession(
     const QuicServerId& server_id,
     scoped_ptr<QuicServerInfo> server_info,
@@ -1137,10 +1228,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 +1253,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) {