JobController 2: Remove reference between HttpStreamFactoryImpl::Jobs.

net/http/http_stream_factory_impl_job_controller.cc

 // Copyright (c) 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 "net/http/http_stream_factory_impl_job_controller.h"
 
 #include "base/metrics/histogram_macros.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
+#include "base/values.h"
 #include "net/base/host_mapping_rules.h"
 #include "net/http/bidirectional_stream_impl.h"
 #include "net/http/transport_security_state.h"
 #include "net/spdy/spdy_session.h"
 
 namespace net {
 
+// Returns parameters associated with the delay of the HTTP stream job.
+std::unique_ptr<base::Value> NetLogHttpStreamJobDelayCallback(
+    base::TimeDelta delay,
+    NetLogCaptureMode /* capture_mode */) {
+  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  dict->SetInteger("resume_after_ms", static_cast<int>(delay.InMilliseconds()));
+  return std::move(dict);
+}
+
 HttpStreamFactoryImpl::JobController::JobController(
     HttpStreamFactoryImpl* factory,
     HttpStreamRequest::Delegate* delegate,
     HttpNetworkSession* session,
     JobFactory* job_factory)
     : factory_(factory),
       session_(session),
       job_factory_(job_factory),
       request_(nullptr),
       delegate_(delegate),
       is_preconnect_(false),
       job_bound_(false),
-      bound_job_(nullptr) {
+      main_job_is_blocked_(false),
+      bound_job_(nullptr),
+      ptr_factory_(this) {
   DCHECK(factory);
 }
 
 HttpStreamFactoryImpl::JobController::~JobController() {
   main_job_.reset();
   alternative_job_.reset();
   bound_job_ = nullptr;
 }
 
 bool HttpStreamFactoryImpl::JobController::for_websockets() {
@@ skipping 176 matching lines @@
   OnJobSucceeded(job);
   request_->OnWebSocketHandshakeStreamReady(used_ssl_config, used_proxy_info,
                                             stream);
 }
 
 void HttpStreamFactoryImpl::JobController::OnStreamFailed(
     Job* job,
     int status,
     const SSLConfig& used_ssl_config,
     SSLFailureState ssl_failure_state) {
+  MaybeResumeMainJob(job, base::TimeDelta());
+
   if (job_bound_ && bound_job_ != job) {
     // We have bound a job to the associated Request, |job| has been orphaned.
     OnOrphanedJobComplete(job);
     return;
   }
 
   if (!request_)
     return;
   DCHECK_NE(OK, status);
   DCHECK(job);
@@ skipping 19 matching lines @@
   }
 
   request_->OnStreamFailed(status, used_ssl_config, ssl_failure_state);
 }
 
 void HttpStreamFactoryImpl::JobController::OnCertificateError(
     Job* job,
     int status,
     const SSLConfig& used_ssl_config,
     const SSLInfo& ssl_info) {
+  MaybeResumeMainJob(job, base::TimeDelta());
+
   if (job_bound_ && bound_job_ != job) {
     // We have bound a job to the associated Request, |job| has been orphaned.
     OnOrphanedJobComplete(job);
     return;
   }
 
   if (!request_)
     return;
   DCHECK_NE(OK, status);
   if (!bound_job_)
     BindJob(job);
 
   request_->OnCertificateError(status, used_ssl_config, ssl_info);
 }
 
 void HttpStreamFactoryImpl::JobController::OnHttpsProxyTunnelResponse(
     Job* job,
     const HttpResponseInfo& response_info,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpStream* stream) {
+  MaybeResumeMainJob(job, base::TimeDelta());
+
   if (job_bound_ && bound_job_ != job) {
     // We have bound a job to the associated Request, |job| has been orphaned.
     OnOrphanedJobComplete(job);
     return;
   }
 
   if (!bound_job_)
     BindJob(job);
   if (!request_)
     return;
   request_->OnHttpsProxyTunnelResponse(response_info, used_ssl_config,
                                        used_proxy_info, stream);
 }
 
 void HttpStreamFactoryImpl::JobController::OnNeedsClientAuth(
     Job* job,
     const SSLConfig& used_ssl_config,
     SSLCertRequestInfo* cert_info) {
+  MaybeResumeMainJob(job, base::TimeDelta());
+
   if (job_bound_ && bound_job_ != job) {
     // We have bound a job to the associated Request, |job| has been orphaned.
     OnOrphanedJobComplete(job);
     return;
   }
   if (!request_)
     return;
   if (!bound_job_)
     BindJob(job);
 
   request_->OnNeedsClientAuth(used_ssl_config, cert_info);
 }
 
 void HttpStreamFactoryImpl::JobController::OnNeedsProxyAuth(
     Job* job,
     const HttpResponseInfo& proxy_response,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpAuthController* auth_controller) {
+  MaybeResumeMainJob(job, base::TimeDelta());
+
   if (job_bound_ && bound_job_ != job) {
     // We have bound a job to the associated Request, |job| has been orphaned.
     OnOrphanedJobComplete(job);
     return;
   }
 
   if (!request_)
     return;
   if (!bound_job_)
     BindJob(job);
@@ skipping 59 matching lines @@
                                    used_proxy_info, was_npn_negotiated,
                                    protocol_negotiated, using_spdy, net_log);
   }
   if (is_job_orphaned) {
     OnOrphanedJobComplete(job);
   }
 }
 
 void HttpStreamFactoryImpl::JobController::OnPreconnectsComplete(Job* job) {
   DCHECK_EQ(main_job_.get(), job);
+  DCHECK(!bound_job_);
   main_job_.reset();
   factory_->OnPreconnectsCompleteInternal();
   MaybeNotifyFactoryOfCompletion();
 }
 
 void HttpStreamFactoryImpl::JobController::OnOrphanedJobComplete(
     const Job* job) {
   if (job->job_type() == MAIN) {
     DCHECK_EQ(main_job_.get(), job);
     main_job_.reset();
   } else {
     DCHECK_EQ(alternative_job_.get(), job);
     alternative_job_.reset();
   }
 
   MaybeNotifyFactoryOfCompletion();
 }
 
 void HttpStreamFactoryImpl::JobController::AddConnectionAttemptsToRequest(
     Job* job,
     const ConnectionAttempts& attempts) {
   if (is_preconnect_ || (job_bound_ && bound_job_ != job))
     return;
 
   DCHECK(request_);
   request_->AddConnectionAttempts(attempts);
 }
 
+void HttpStreamFactoryImpl::JobController::ResumeMainJob() {
+  main_job_->net_log().AddEvent(
+      NetLog::TYPE_HTTP_STREAM_JOB_DELAYED,
+      base::Bind(&NetLogHttpStreamJobDelayCallback, main_job_wait_time_));
+
+  main_job_->Resume();
+  main_job_wait_time_ = base::TimeDelta();
+}
+
+void HttpStreamFactoryImpl::JobController::MaybeResumeMainJob(
+    Job* job,
+    const base::TimeDelta& delay) {
+  DCHECK(job == main_job_.get() || job == alternative_job_.get());
+  if (!main_job_is_blocked_)
+    return;
+
+  if (job == alternative_job_.get() && main_job_) {

[Ryan Hamilton, 2016/07/12 00:21:45]

nit: I recommend reversing the polarity of this check (and possibly adding it to
the previous if:

if (!main_job_is_blocked_ || job != alternative_job_.get() || !main_job_)
  return;

main_job_is_blocked = false;
if (!main_job_->is_waiting())
  return;

...PostTask....()

[Zhongyi Shi, 2016/07/12 23:03:12]

Done.

+    main_job_is_blocked_ = false;
+    if (main_job_->is_waiting()) {
+      base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+          FROM_HERE,
+          base::Bind(&HttpStreamFactoryImpl::JobController::ResumeMainJob,
+                     ptr_factory_.GetWeakPtr()),
+          main_job_wait_time_);
+    }
+  }
+}
+
+void HttpStreamFactoryImpl::JobController::OnConnectionInitialized(Job* job,
+                                                                   int rv) {

[Ryan Hamilton, 2016/07/12 00:21:45]

can we do DCHECK(main_job_is_blocked_)?

[Zhongyi Shi, 2016/07/12 23:03:12]

No, we couldn't. OnConnectionInitialized is called in Job::DoConnection(). If
it's called by the main job, then main_job_is_blocked_ has to be false as main
job must not proceed to create connection if blocked. There're also other cases
where main_job_is_blocked_ is false but this OnConnectionInitialized is called.

+  if (rv != OK && rv != ERR_SPDY_SESSION_ALREADY_EXISTS) {

[Ryan Hamilton, 2016/07/12 00:21:45]

I think perhaps we should move this ERR_SPDY_SESSION_ALREADY_EXISTS check down
into the Job because it's really the job which knows about this error code,
right?

[Zhongyi Shi, 2016/07/12 23:03:12]

Done.

+    // Resume the main job as there's an error raised in connection
+    // initiation except the case where we find an existing SPDY session for
+    // this Job.
+    return MaybeResumeMainJob(job, main_job_wait_time_);
+  }
+}
+
+bool HttpStreamFactoryImpl::JobController::ShouldWait(Job* job) {
+  // The alternative job never waits.
+  if (job == alternative_job_.get())
+    return false;
+
+  if (!main_job_is_blocked_ && main_job_wait_time_.is_zero())
+    return false;
+
+  if (!main_job_wait_time_.is_zero()) {
+    main_job_is_blocked_ = false;
+    base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&HttpStreamFactoryImpl::JobController::ResumeMainJob,
+                   ptr_factory_.GetWeakPtr()),
+        main_job_wait_time_);
+  }

[Ryan Hamilton, 2016/07/12 00:21:45]

Should this be:

if (main_job_is_blocked_)
  return true;

if (main_job_wait_time_.is_zero())
  return false;

PostTask();
return true; 

With the code as written, If the main job is blocked, and the wait time is
non-zero, we'll post a task. But I don't think we should do that while the main
job is blocked. (Admittedly, I don't think we can get into this state, but I
think it's cleaner to simply return true and do nothing else if the main job is
blocked.)

[Zhongyi Shi, 2016/07/12 23:03:12]

Done!

Hmmm, you are right. We should never run into the case where the wait_time is
nonzero but main_job_is_still_blocked. As we only set the nonzero wait time for
delayed TCP and returns
ERR_IO_PENDING to call OnConnectionInitialzed which then will ResumeMainJob to
set the
main_job_is_blocked to false.

But that does reminds me to only update wait time when main job is blocked in
SetWaitTimeForMainJob.

+  return true;
+}
+
 void HttpStreamFactoryImpl::JobController::SetSpdySessionKey(
     Job* job,
     const SpdySessionKey& spdy_session_key) {
   if (is_preconnect_ || (job_bound_ && bound_job_ != job))
     return;
 
   DCHECK(request_);
   if (!request_->HasSpdySessionKey()) {
     RequestSet& request_set =
         factory_->spdy_session_request_map_[spdy_session_key];
@@ skipping 29 matching lines @@
 }
 
 const BoundNetLog* HttpStreamFactoryImpl::JobController::GetNetLog(
     Job* job) const {
   if (is_preconnect_ || (job_bound_ && bound_job_ != job))
     return nullptr;
   DCHECK(request_);
   return &request_->net_log();
 }
 
+void HttpStreamFactoryImpl::JobController::SetWaitTimeForMainJob(
+    const base::TimeDelta& delay) {
+  main_job_wait_time_ = delay;

[Ryan Hamilton, 2016/07/12 00:21:45]

DCHECK(main_job_is_blocked);

[Zhongyi Shi, 2016/07/12 23:03:12]

Nooo. This method is called by the Job without knowing whether there's main job
blocking.There's no guarantee that main job is blocked. 

In fact, there is a case that we force quic with only one job, then the main job
is not blocked, but it will try call SetWaitTimeForMainJob, which if correctly
written, should do nothing. 

+}
+
 WebSocketHandshakeStreamBase::CreateHelper* HttpStreamFactoryImpl::
     JobController::websocket_handshake_stream_create_helper() {
   DCHECK(request_);
   return request_->websocket_handshake_stream_create_helper();
 }
 
 void HttpStreamFactoryImpl::JobController::CreateJobs(
     const HttpRequestInfo& request_info,
     RequestPriority priority,
     const SSLConfig& server_ssl_config,
@@ skipping 25 matching lines @@
     HostPortPair alternative_destination(alternative_service.host_port_pair());
     ignore_result(
         ApplyHostMappingRules(request_info.url, &alternative_destination));
 
     alternative_job_.reset(job_factory_->CreateJob(
         this, ALTERNATIVE, session_, request_info, priority, server_ssl_config,
         proxy_ssl_config, alternative_destination, origin_url,
         alternative_service, net_log.net_log()));
     AttachJob(alternative_job_.get());
 
-    main_job_->WaitFor(alternative_job_.get());
-    // Make sure to wait until we call WaitFor(), before starting
-    // |alternative_job|, otherwise |alternative_job| will not notify |job|
-    // appropriately.
+    main_job_is_blocked_ = true;
     alternative_job_->Start(request_->stream_type());
   }
   // Even if |alternative_job| has already finished, it will not have notified
   // the request yet, since we defer that to the next iteration of the
   // MessageLoop, so starting |main_job_| is always safe.
   main_job_->Start(request_->stream_type());
 }
 
 void HttpStreamFactoryImpl::JobController::AttachJob(Job* job) {
   DCHECK(job);
@@ skipping 274 matching lines @@
       first_alternative_service = alternative_service;
   }
 
   // Ask delegate to mark QUIC as broken for the origin.
   if (quic_advertised && quic_all_broken && delegate != nullptr)
     delegate->OnQuicBroken();
 
   return first_alternative_service;
 }
 }