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 176 matching lines @@
                                 AlternativeService alternative_service,
                                 NetLog* net_log)
     : request_info_(request_info),
       priority_(priority),
       server_ssl_config_(server_ssl_config),
       proxy_ssl_config_(proxy_ssl_config),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_HTTP_STREAM_JOB)),
       io_callback_(base::Bind(&Job::OnIOComplete, base::Unretained(this))),
       connection_(new ClientSocketHandle),
       session_(session),
+      state_(STATE_NONE),
       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_);
-
-  // Never share connection with other jobs for FTP requests.
-  DCHECK(!request_info_.url.SchemeIs("ftp"));
-
-  blocking_job_ = job;
-  job->waiting_job_ = this;
-}
-
-void HttpStreamFactoryImpl::Job::ResumeAfterDelay() {
-  DCHECK(!blocking_job_);
-  DCHECK_EQ(STATE_WAIT_FOR_JOB_COMPLETE, next_state_);
+void HttpStreamFactoryImpl::Job::ResumeAfterDelay(
+    const base::TimeDelta& delay) {
+  DCHECK((job_type_ == ALTERNATIVE || !delegate_->blocking()));
 
   net_log_.AddEvent(NetLog::TYPE_HTTP_STREAM_JOB_DELAYED,
-                    base::Bind(&NetLogHttpStreamJobDelayCallback, wait_time_));
+                    base::Bind(&NetLogHttpStreamJobDelayCallback, delay));
   base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
       FROM_HERE, base::Bind(&HttpStreamFactoryImpl::Job::OnIOComplete,
                             ptr_factory_.GetWeakPtr(), OK),
-      wait_time_);
+      delay);
 }
 
-void HttpStreamFactoryImpl::Job::Resume(Job* job,
-                                        const base::TimeDelta& delay) {
-  DCHECK_EQ(blocking_job_, job);
-  blocking_job_ = NULL;
+void HttpStreamFactoryImpl::Job::Resume(const base::TimeDelta& delay) {
+  DCHECK_EQ(job_type_, MAIN);
+  // 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);
 
-  // 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)
+    ResumeAfterDelay(delegate_->wait_time_for_main_job());

[Ryan Hamilton, 2016/06/29 23:12:28]

Can the delegate (the controller) simply call Resume() on this job when this job
should resume? In other words, can the delay time be moved up?

[Zhongyi Shi, 2016/06/30 22:53:39]

Done. Yay!!

 }
 
 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 193 matching lines @@
   TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("net"),
                "HttpStreamFactoryImpl::Job::OnIOComplete");
   RunLoop(result);
 }
 
 int HttpStreamFactoryImpl::Job::RunLoop(int result) {
   TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("net"),
                "HttpStreamFactoryImpl::Job::RunLoop");
   result = DoLoop(result);
 
-  if (result == ERR_IO_PENDING)
+  if (result == ERR_IO_PENDING) {
+    if (state_ == STATE_INIT_CONNECTION) {
+      delegate_->OnInitConnectionNotSuccessful(this, delay_);
+      delay_ = base::TimeDelta();
+    }
     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)) {
@@ skipping 100 matching lines @@
           FROM_HERE, base::Bind(&Job::OnStreamFailedCallback,
                                 ptr_factory_.GetWeakPtr(), result));
       return ERR_IO_PENDING;
   }
 }
 
 int HttpStreamFactoryImpl::Job::DoLoop(int result) {
   DCHECK_NE(next_state_, STATE_NONE);
   int rv = result;
   do {
-    State state = next_state_;
+    state_ = next_state_;
     next_state_ = STATE_NONE;
-    switch (state) {
+    switch (state_) {
       case STATE_START:
         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() {
+  if (delegate_->ShoudWait(this)) {
+    next_state_ = STATE_WAIT_COMPLETE;
+    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;
-}
-
-int HttpStreamFactoryImpl::Job::DoWaitForJobComplete(int result) {
-  DCHECK(!blocking_job_);
-  DCHECK_EQ(OK, result);
-  wait_time_ = base::TimeDelta();
   next_state_ = STATE_INIT_CONNECTION;
   return OK;
 }
+
+int HttpStreamFactoryImpl::Job::DoWaitComplete(int result) {
+  DCHECK((job_type_ == ALTERNATIVE || !delegate_->blocking()));
+  DCHECK_EQ(OK, result);
+  delegate_->set_wait_time_for_main_job(base::TimeDelta());
+  next_state_ = STATE_INIT_CONNECTION;
+  return OK;
+}
 
 int HttpStreamFactoryImpl::Job::DoInitConnection() {
   // 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_->blocking()));
   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) {
+        // Start the waiting job after the delay returned by
+        // GetTimeDelayForWaitingJob().
+        //
+        // If QUIC request fails during handshake, then
+        // DoInitConnectionComplete() will start the waiting job.
+        delay_ = 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;
   }
 
+  // TODO(willchan): Make this a bit more exact. Maybe there are recoverable
+  // errors, such as ignoring certificate errors for Alternate-Protocol.
+  // Report failure to delegate as there's error in DoInitConnection. Delegate
+  // might be able to resume the other job.
+  if (result < 0) {
+    delegate_->OnInitConnectionNotSuccessful(this, delay_);
+    delay_ = base::TimeDelta();
+  }
+
   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