JobController 2: Remove reference between HttpStreamFactoryImpl::Jobs.

net/http/http_stream_factory_impl_job.cc

 // Copyright (c) 2012 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.h"
 
 #include <algorithm>
 #include <string>
 #include <utility>
 
@@ skipping 183 matching lines @@
       io_callback_(base::Bind(&Job::OnIOComplete, base::Unretained(this))),
       connection_(new ClientSocketHandle),
       session_(session),
       next_state_(STATE_NONE),
       pac_request_(NULL),
       destination_(destination),
       origin_url_(origin_url),
       alternative_service_(alternative_service),
       delegate_(delegate),
       job_type_(job_type),
-      blocking_job_(NULL),
-      waiting_job_(NULL),
       using_ssl_(false),
       using_spdy_(false),
       using_quic_(false),
       quic_request_(session_->quic_stream_factory()),
       using_existing_quic_session_(false),
       spdy_certificate_error_(OK),
       establishing_tunnel_(false),
       was_npn_negotiated_(false),
       protocol_negotiated_(kProtoUnknown),
       num_streams_(0),
@@ skipping 65 matching lines @@
     case STATE_RESOLVE_PROXY_COMPLETE:
       return session_->proxy_service()->GetLoadState(pac_request_);
     case STATE_INIT_CONNECTION_COMPLETE:
     case STATE_CREATE_STREAM_COMPLETE:
       return using_quic_ ? LOAD_STATE_CONNECTING : connection_->GetLoadState();
     default:
       return LOAD_STATE_IDLE;
   }
 }
 
-void HttpStreamFactoryImpl::Job::WaitFor(Job* job) {
-  DCHECK_EQ(STATE_NONE, next_state_);
-  DCHECK_EQ(STATE_NONE, job->next_state_);
-  DCHECK(!blocking_job_);
-  DCHECK(!job->waiting_job_);
+void HttpStreamFactoryImpl::Job::Resume(const base::TimeDelta& delay) {

[Ryan Hamilton, 2016/07/01 00:13:30]

Discussed offline, let's do this:

void HttpStreamFactoryImpl::Job::Resume() {
  DCHECK_EQ(job_type_, MAIN);
  DCHECK_EQ(STATE_WAIT_COMPLETE, next_state_);
  OnIOComplete(OK);
}

Then it's up to the controller to post a task to call Resume() when the Job
should actually resume.

[Zhongyi Shi, 2016/07/11 23:04:56]

Done. Yay!!! Much cleaner.

+  DCHECK_EQ(job_type_, MAIN);
 
-  // Never share connection with other jobs for FTP requests.
-  DCHECK(!request_info_.url.SchemeIs("ftp"));
+  // If |this| job is not past STATE_WAIT_COMPLETE state, then it will
+  // be delayed by the |delay| when it resumes.
+  if (next_state_ == STATE_NONE || next_state_ <= STATE_WAIT_COMPLETE)
+    delegate_->set_wait_time_for_main_job(delay);
 
-  blocking_job_ = job;
-  job->waiting_job_ = this;
-}
-
-void HttpStreamFactoryImpl::Job::ResumeAfterDelay() {
-  DCHECK(!blocking_job_);
-  DCHECK_EQ(STATE_WAIT_FOR_JOB_COMPLETE, next_state_);
-
-  net_log_.AddEvent(NetLog::TYPE_HTTP_STREAM_JOB_DELAYED,
-                    base::Bind(&NetLogHttpStreamJobDelayCallback, wait_time_));
-  base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
-      FROM_HERE, base::Bind(&HttpStreamFactoryImpl::Job::OnIOComplete,
-                            ptr_factory_.GetWeakPtr(), OK),
-      wait_time_);
-}
-
-void HttpStreamFactoryImpl::Job::Resume(Job* job,
-                                        const base::TimeDelta& delay) {
-  DCHECK_EQ(blocking_job_, job);
-  blocking_job_ = NULL;
-
-  // If |this| job is not past STATE_WAIT_FOR_JOB_COMPLETE state, then it will
-  // be delayed by the |wait_time_| when it resumes.
-  if (next_state_ == STATE_NONE || next_state_ <= STATE_WAIT_FOR_JOB_COMPLETE)
-    wait_time_ = delay;
-
-  // We know we're blocked if the next_state_ is STATE_WAIT_FOR_JOB_COMPLETE.
+  // We know we're blocked if the next_state_ is STATE_WAIT_COMPLETE.
   // Unblock |this|.
-  if (next_state_ == STATE_WAIT_FOR_JOB_COMPLETE)
-    ResumeAfterDelay();
+  if (next_state_ == STATE_WAIT_COMPLETE) {
+    net_log_.AddEvent(NetLog::TYPE_HTTP_STREAM_JOB_DELAYED,
+                      base::Bind(&NetLogHttpStreamJobDelayCallback, delay));
+    base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+        FROM_HERE, base::Bind(&HttpStreamFactoryImpl::Job::OnIOComplete,
+                              ptr_factory_.GetWeakPtr(), OK),
+        delegate_->wait_time_for_main_job());
+  }
 }
 
 void HttpStreamFactoryImpl::Job::Orphan() {
   net_log_.AddEvent(NetLog::TYPE_HTTP_STREAM_JOB_ORPHANED);
-  if (blocking_job_) {
-    // We've been orphaned, but there's a job we're blocked on. Don't bother
-    // racing, just cancel ourself.
-    DCHECK(blocking_job_->waiting_job_);
-    blocking_job_->waiting_job_ = NULL;
-    blocking_job_ = NULL;
-    if (delegate_->for_websockets() && connection_ && connection_->socket()) {
-      connection_->socket()->Disconnect();
-    }
-    delegate_->OnOrphanedJobComplete(this);
-  } else if (delegate_->for_websockets()) {
+
+  if (delegate_->for_websockets()) {
     // We cancel this job because a WebSocketHandshakeStream can't be created
     // without a WebSocketHandshakeStreamBase::CreateHelper which is stored in
     // the Request class and isn't retrievable by this job.
     if (connection_ && connection_->socket()) {
       connection_->socket()->Disconnect();
     }
     delegate_->OnOrphanedJobComplete(this);
   }
   // |this| may be deleted after this call.
 }
@@ skipping 196 matching lines @@
 }
 
 int HttpStreamFactoryImpl::Job::RunLoop(int result) {
   TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("net"),
                "HttpStreamFactoryImpl::Job::RunLoop");
   result = DoLoop(result);
 
   if (result == ERR_IO_PENDING)
     return result;
 
-  // If there was an error, we should have already resumed the |waiting_job_|,
-  // if there was one.
-  DCHECK(result == OK || waiting_job_ == NULL);
-
   if (job_type_ == PRECONNECT) {
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE,
         base::Bind(&HttpStreamFactoryImpl::Job::OnPreconnectsComplete,
                    ptr_factory_.GetWeakPtr()));
     return ERR_IO_PENDING;
   }
 
   if (IsCertificateError(result)) {
     // Retrieve SSL information from the socket.
@@ skipping 113 matching lines @@
         DCHECK_EQ(OK, rv);
         rv = DoStart();
         break;
       case STATE_RESOLVE_PROXY:
         DCHECK_EQ(OK, rv);
         rv = DoResolveProxy();
         break;
       case STATE_RESOLVE_PROXY_COMPLETE:
         rv = DoResolveProxyComplete(rv);
         break;
-      case STATE_WAIT_FOR_JOB:
+      case STATE_WAIT:
         DCHECK_EQ(OK, rv);
-        rv = DoWaitForJob();
+        rv = DoWait();
         break;
-      case STATE_WAIT_FOR_JOB_COMPLETE:
-        rv = DoWaitForJobComplete(rv);
+      case STATE_WAIT_COMPLETE:
+        rv = DoWaitComplete(rv);
         break;
       case STATE_INIT_CONNECTION:
         DCHECK_EQ(OK, rv);
         rv = DoInitConnection();
         break;
       case STATE_INIT_CONNECTION_COMPLETE:
         rv = DoInitConnectionComplete(rv);
         break;
       case STATE_WAITING_USER_ACTION:
         rv = DoWaitingUserAction(rv);
@@ skipping 40 matching lines @@
         base::Bind(&NetLogHttpStreamJobCallback, net_log->source(),
                    &request_info_.url, &origin_url_, &alternative_service_,
                    priority_));
     net_log->AddEvent(NetLog::TYPE_HTTP_STREAM_REQUEST_STARTED_JOB,
                       net_log_.source().ToEventParametersCallback());
   }
 
   // Don't connect to restricted ports.
   if (!IsPortAllowedForScheme(destination_.port(),
                               request_info_.url.scheme())) {
-    if (waiting_job_) {
-      waiting_job_->Resume(this, base::TimeDelta());
-      waiting_job_ = NULL;
-    }
     return ERR_UNSAFE_PORT;
   }
 
   next_state_ = STATE_RESOLVE_PROXY;
   return OK;
 }
 
 int HttpStreamFactoryImpl::Job::DoResolveProxy() {
   DCHECK(!pac_request_);
   DCHECK(session_);
@@ skipping 52 matching lines @@
       result = ERR_NO_SUPPORTED_PROXIES;
     } else if (using_quic_ &&
                (!proxy_info_.is_quic() && !proxy_info_.is_direct())) {
       // QUIC can not be spoken to non-QUIC proxies.  This error should not be
       // user visible, because the non-alternative Job should be resumed.
       result = ERR_NO_SUPPORTED_PROXIES;
     }
   }
 
   if (result != OK) {
-    if (waiting_job_) {
-      waiting_job_->Resume(this, base::TimeDelta());
-      waiting_job_ = NULL;
-    }
     return result;
   }
 
-  next_state_ = STATE_WAIT_FOR_JOB;
+  next_state_ = STATE_WAIT;
   return OK;
 }
 
 bool HttpStreamFactoryImpl::Job::ShouldForceQuic() const {
   return session_->params().enable_quic &&
          ContainsKey(session_->params().origins_to_force_quic_on,
                      destination_) &&
          proxy_info_.is_direct() && origin_url_.SchemeIs("https");
 }
 
-int HttpStreamFactoryImpl::Job::DoWaitForJob() {
-  if (!blocking_job_ && wait_time_.is_zero()) {
-    // There is no |blocking_job_| and there is no |wait_time_|.
-    next_state_ = STATE_INIT_CONNECTION;
-    return OK;
-  }
+int HttpStreamFactoryImpl::Job::DoWait() {
+  next_state_ = STATE_WAIT_COMPLETE;
+  if (delegate_->ShouldWait(this))
+    return ERR_IO_PENDING;
 
-  next_state_ = STATE_WAIT_FOR_JOB_COMPLETE;
-  if (!wait_time_.is_zero()) {
-    // If there is a waiting_time, then resume the job after the wait_time_.
-    DCHECK(!blocking_job_);
-    ResumeAfterDelay();
-  }
-
-  return ERR_IO_PENDING;
+  next_state_ = STATE_INIT_CONNECTION;
+  return OK;
 }
 
-int HttpStreamFactoryImpl::Job::DoWaitForJobComplete(int result) {
-  DCHECK(!blocking_job_);
+int HttpStreamFactoryImpl::Job::DoWaitComplete(int result) {
+  DCHECK((job_type_ == ALTERNATIVE || !delegate_->main_job_is_blocked()));
   DCHECK_EQ(OK, result);
-  wait_time_ = base::TimeDelta();
+  delegate_->set_wait_time_for_main_job(base::TimeDelta());

[Ryan Hamilton, 2016/07/01 00:13:30]

Why does this happen here? I would have thought this method would only be called
by the alternative job when the alternative job finally knows how long the main
job should wait for.

This reminds me that we should make sure we have test coverage for the case
where:
alt job starts
alt job returns ERR_IO_PENDING from DoResolveProxy
main job starts
main job gets OK from DoResolveProxy
main job Waits 

I suspect we have this coverage, but if not we should make sure we add it.

[Zhongyi Shi, 2016/07/11 23:04:56]

Added DoNotResumeMainJobWhenAltJobPending covers this.

[Zhongyi Shi, 2016/07/11 23:04:56]

This SetWaitTimeForMainJob is to clear the main_job_wait_time_ in controller,
but you're right, we are able to move this up to controller to clear the wait
time.

   next_state_ = STATE_INIT_CONNECTION;
   return OK;
 }
 
 int HttpStreamFactoryImpl::Job::DoInitConnection() {
+  int result = DoInitConnectionImpl();
+  // Report to delegate that this job has initialized the connection.

[Ryan Hamilton, 2016/07/01 00:13:30]

nit: I'd just drop this since your method is pretty readable.

[Zhongyi Shi, 2016/07/11 23:04:56]

Done.

+  delegate_->OnConnectionInitialized(this, result);
+
+  return result;
+}
+
+int HttpStreamFactoryImpl::Job::DoInitConnectionImpl() {
   // TODO(pkasting): Remove ScopedTracker below once crbug.com/462812 is fixed.
   tracked_objects::ScopedTracker tracking_profile(
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "462812 HttpStreamFactoryImpl::Job::DoInitConnection"));
-  DCHECK(!blocking_job_);
+  DCHECK((job_type_ == ALTERNATIVE || !delegate_->main_job_is_blocked()));
   DCHECK(!connection_->is_initialized());
   DCHECK(proxy_info_.proxy_server().is_valid());
   next_state_ = STATE_INIT_CONNECTION_COMPLETE;
 
   using_ssl_ = origin_url_.SchemeIs("https") || origin_url_.SchemeIs("wss") ||
                IsSpdyAlternative();
   using_spdy_ = false;
 
   if (ShouldForceQuic())
     using_quic_ = true;
@@ skipping 51 matching lines @@
       destination = destination_;
       ssl_config = &server_ssl_config_;
     }
     int rv =
         quic_request_.Request(destination, request_info_.privacy_mode,
                               ssl_config->GetCertVerifyFlags(), url,
                               request_info_.method, net_log_, io_callback_);
     if (rv == OK) {
       using_existing_quic_session_ = true;
     } else {
-      // OK, there's no available QUIC session. Let |waiting_job_| resume
-      // if it's paused.
-      if (waiting_job_) {
-        if (rv == ERR_IO_PENDING) {
-          // Start the |waiting_job_| after the delay returned by
-          // GetTimeDelayForWaitingJob().
-          //
-          // If QUIC request fails during handshake, then
-          // DoInitConnectionComplete() will start the |waiting_job_|.
-          waiting_job_->Resume(this, quic_request_.GetTimeDelayForWaitingJob());
-        } else {
-          // QUIC request has failed, resume the |waiting_job_|.
-          waiting_job_->Resume(this, base::TimeDelta());
-        }
-        waiting_job_ = NULL;
+      base::TimeDelta delay;
+      // OK, there's no available QUIC session. Let JobController to resume
+      // the waiting job if it's paused.
+      if (rv == ERR_IO_PENDING) {
+        // Set the wait time for main job if it is delayed TCP.
+        delegate_->set_wait_time_for_main_job(

[Ryan Hamilton, 2016/07/01 00:13:30]

Yeah, this is the only place I'd expect this method to be called.

+            quic_request_.GetTimeDelayForWaitingJob());
       }
     }
     return rv;
   }
 
   SpdySessionKey spdy_session_key = GetSpdySessionKey();
 
   // Check first if we have a spdy session for this group.  If so, then go
   // straight to using that.
   if (CanUseExistingSpdySession()) {
@@ skipping 12 matching lines @@
       existing_spdy_session_ = spdy_session;
       return OK;
     }
   }
   if (using_ssl_) {
     // Ask |delegate_delegate_| to update the spdy session key for the request
     // that launched this job.
     delegate_->SetSpdySessionKey(this, spdy_session_key);
   }
 
-  // OK, there's no available SPDY session. Let |waiting_job_| resume if it's
-  // paused.
-  if (waiting_job_) {
-    waiting_job_->Resume(this, base::TimeDelta());
-    waiting_job_ = NULL;
-  }
-
   if (proxy_info_.is_http() || proxy_info_.is_https())
     establishing_tunnel_ = using_ssl_;
 
   const bool expect_spdy = IsSpdyAlternative();
 
   base::WeakPtr<HttpServerProperties> http_server_properties =
       session_->http_server_properties();
   if (http_server_properties) {
     http_server_properties->MaybeForceHTTP11(destination_, &server_ssl_config_);
     if (proxy_info_.is_http() || proxy_info_.is_https()) {
@@ skipping 32 matching lines @@
   }
 
   return InitSocketHandleForHttpRequest(
       GetSocketGroup(), destination_, request_info_.extra_headers,
       request_info_.load_flags, priority_, session_, proxy_info_, expect_spdy,
       server_ssl_config_, proxy_ssl_config_, request_info_.privacy_mode,
       net_log_, connection_.get(), resolution_callback, io_callback_);
 }
 
 int HttpStreamFactoryImpl::Job::DoInitConnectionComplete(int result) {
-  if (using_quic_ && result < 0 && waiting_job_) {
-    waiting_job_->Resume(this, base::TimeDelta());
-    waiting_job_ = NULL;
-  }
   if (job_type_ == PRECONNECT) {
     if (using_quic_)
       return result;
     DCHECK_EQ(OK, result);
     return OK;
   }
 
   if (result == ERR_SPDY_SESSION_ALREADY_EXISTS) {
+    DCHECK(!using_quic_);
     // We found a SPDY connection after resolving the host.  This is
     // probably an IP pooled connection.
     SpdySessionKey spdy_session_key = GetSpdySessionKey();
     existing_spdy_session_ =
         session_->spdy_session_pool()->FindAvailableSession(
             spdy_session_key, origin_url_, net_log_);
     if (existing_spdy_session_) {
       using_spdy_ = true;
       next_state_ = STATE_CREATE_STREAM;
     } else {
       // It is possible that the spdy session no longer exists.
       ReturnToStateInitConnection(true /* close connection */);
     }
     return OK;
   }
 
   if (proxy_info_.is_quic() && using_quic_) {
     // Mark QUIC proxy as bad if QUIC got disabled on the destination port.
     // Underlying QUIC layer would have closed the connection.
     HostPortPair destination = proxy_info_.proxy_server().host_port_pair();
     if (session_->quic_stream_factory()->IsQuicDisabled(destination.port())) {
       using_quic_ = false;
       return ReconsiderProxyAfterError(ERR_QUIC_PROTOCOL_ERROR);
     }
   }
 
-  // TODO(willchan): Make this a bit more exact. Maybe there are recoverable
-  // errors, such as ignoring certificate errors for Alternate-Protocol.
-  if (result < 0 && waiting_job_) {
-    waiting_job_->Resume(this, base::TimeDelta());
-    waiting_job_ = NULL;
-  }
-
   // |result| may be the result of any of the stacked pools. The following
   // logic is used when determining how to interpret an error.
   // If |result| < 0:
   //   and connection_->socket() != NULL, then the SSL handshake ran and it
   //     is a potentially recoverable error.
   //   and connection_->socket == NULL and connection_->is_ssl_error() is true,
   //     then the SSL handshake ran with an unrecoverable error.
   //   otherwise, the error came from one of the other pools.
   bool ssl_started = using_ssl_ && (result == OK || connection_->socket() ||
                                     connection_->is_ssl_error());
@@ skipping 610 matching lines @@
 
   ConnectionAttempts socket_attempts = connection_->connection_attempts();
   if (connection_->socket()) {
     connection_->socket()->GetConnectionAttempts(&socket_attempts);
   }
 
   delegate_->AddConnectionAttemptsToRequest(this, socket_attempts);
 }
 
 }  // namespace net