Add HttpStreamFactory Job orphaning semantics.

BUG=54371,69688
TEST=none

Review URL: http://codereview.chromium.org/6591030

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@76389 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 486 insertions, 65 deletions

diff --git a/net/http/http_stream_factory_impl.cc b/net/http/http_stream_factory_impl.cc
index 74942a3..c7b0c93 100644
--- a/net/http/http_stream_factory_impl.cc
+++ b/net/http/http_stream_factory_impl.cc
@@ -35,8 +35,7 @@ HttpStreamRequest* HttpStreamFactoryImpl::RequestStream(
     const BoundNetLog& net_log) {
   Job* job = new Job(this, session_);
   Request* request = new Request(request_info.url, this, delegate, net_log);
-  request_map_[job] = request;
-  request->BindJob(job);
+  request->AttachJob(job);
   job->Start(request, request_info, ssl_config, net_log);
   return request;
 }
@@ -59,27 +58,53 @@ bool HttpStreamFactoryImpl::IsTLSIntolerantServer(const GURL& url) const {
   return ContainsKey(tls_intolerant_servers_, GetHostAndPort(url));
 }
 
-LoadState HttpStreamFactoryImpl::GetLoadState(const Request& request) const {
-  // TODO(willchan): Will break when we do late binding.
-  return request.job()->GetLoadState();
+void HttpStreamFactoryImpl::OrphanJob(Job* job, const Request* request) {
+  DCHECK(ContainsKey(request_map_, job));
+  DCHECK_EQ(request_map_[job], request);
+  DCHECK(!ContainsKey(orphaned_job_set_, job));
+
+  request_map_.erase(job);
+
+  orphaned_job_set_.insert(job);
+  job->Orphan(request);
 }
 
 void HttpStreamFactoryImpl::OnSpdySessionReady(
-    const Job* job,
     scoped_refptr<SpdySession> spdy_session,
-    bool direct) {
-  DCHECK(job->using_spdy());
-  DCHECK(ContainsKey(request_map_, job));
-  Request* request = request_map_[job];
-  request->Complete(job->was_alternate_protocol_available(),
-                    job->was_npn_negotiated(),
-                    job->using_spdy(),
-                    job->net_log().source());
-  bool use_relative_url = direct || request->url().SchemeIs("https");
-  request->OnStreamReady(
-      job->ssl_config(),
-      job->proxy_info(),
-      new SpdyHttpStream(spdy_session, use_relative_url));
+    bool direct,
+    const SSLConfig& used_ssl_config,
+    const ProxyInfo& used_proxy_info,
+    bool was_alternate_protocol_available,
+    bool was_npn_negotiated,
+    bool using_spdy,
+    const NetLog::Source& source) {
+  const HostPortProxyPair& spdy_session_key =
+      spdy_session->host_port_proxy_pair();
+  while (!spdy_session->IsClosed()) {
+    // Each iteration may empty out the RequestSet for |spdy_session_key_ in
+    // |spdy_session_request_map_|. So each time, check for RequestSet and use
+    // the first one.
+    //
+    // TODO(willchan): If it's important, switch RequestSet out for a FIFO
+    // pqueue (Order by priority first, then FIFO within same priority). Unclear
+    // that it matters here.
+    if (!ContainsKey(spdy_session_request_map_, spdy_session_key))
+      break;
+    Request* request = *spdy_session_request_map_[spdy_session_key].begin();
+    request->Complete(was_alternate_protocol_available,
+                      was_npn_negotiated,
+                      using_spdy,
+                      source);
+    bool use_relative_url = direct || request->url().SchemeIs("https");
+    request->OnStreamReady(NULL, used_ssl_config, used_proxy_info,
+                           new SpdyHttpStream(spdy_session, use_relative_url));
+  }
+  // TODO(mbelshe): Alert other valid requests.
+}
+
+void HttpStreamFactoryImpl::OnOrphanedJobComplete(const Job* job) {
+  orphaned_job_set_.erase(job);
+  delete job;
 }
 
 void HttpStreamFactoryImpl::OnPreconnectsComplete(const Job* job) {
diff --git a/net/http/http_stream_factory_impl.h b/net/http/http_stream_factory_impl.h
index 9d6e133..4bdbfca 100644
--- a/net/http/http_stream_factory_impl.h
+++ b/net/http/http_stream_factory_impl.h
@@ -11,6 +11,7 @@
 
 #include "base/ref_counted.h"
 #include "net/http/http_stream_factory.h"
+#include "net/base/net_log.h"
 #include "net/proxy/proxy_server.h"
 
 namespace net {
@@ -44,20 +45,29 @@ class HttpStreamFactoryImpl : public HttpStreamFactory {
   typedef std::set<Request*> RequestSet;
   typedef std::map<HostPortProxyPair, RequestSet> SpdySessionRequestMap;
 
-  LoadState GetLoadState(const Request& request) const;
+  // Detaches |job| from |request|.
+  void OrphanJob(Job* job, const Request* request);
 
   // Called when a SpdySession is ready. It will find appropriate Requests and
   // fulfill them. |direct| indicates whether or not |spdy_session| uses a
   // proxy.
-  void OnSpdySessionReady(const Job* job,
-                          scoped_refptr<SpdySession> spdy_session,
-                          bool direct);
+  void OnSpdySessionReady(scoped_refptr<SpdySession> spdy_session,
+                          bool direct,
+                          const SSLConfig& used_ssl_config,
+                          const ProxyInfo& used_proxy_info,
+                          bool was_alternate_protocol_available,
+                          bool was_npn_negotiated,
+                          bool using_spdy,
+                          const NetLog::Source& source);
 
   // Called when the Job detects that the endpoint indicated by the
   // Alternate-Protocol does not work. Lets the factory update
   // HttpAlternateProtocols with the failure and resets the SPDY session key.
   void OnBrokenAlternateProtocol(const Job*, const HostPortPair& origin);
 
+  // Invoked when an orphaned Job finishes.
+  void OnOrphanedJobComplete(const Job* job);
+
   // Invoked when the Job finishes preconnecting sockets.
   void OnPreconnectsComplete(const Job* job);
 
@@ -75,6 +85,13 @@ class HttpStreamFactoryImpl : public HttpStreamFactory {
 
   SpdySessionRequestMap spdy_session_request_map_;
 
+  // These jobs correspond to jobs orphaned by Requests and now owned by
+  // HttpStreamFactoryImpl. Since they are no longer tied to Requests, they will
+  // not be canceled when Requests are canceled. Therefore, in
+  // ~HttpStreamFactoryImpl, it is possible for some jobs to still exist in this
+  // set. Leftover jobs will be deleted when the factory is destroyed.
+  std::set<const Job*> orphaned_job_set_;
+
   // These jobs correspond to preconnect requests and have no associated Request
   // object. They're owned by HttpStreamFactoryImpl. Leftover jobs will be
   // deleted when the factory is destroyed.
diff --git a/net/http/http_stream_factory_impl_job.cc b/net/http/http_stream_factory_impl_job.cc
index 8ae5f82..f8d8a9e 100644
--- a/net/http/http_stream_factory_impl_job.cc
+++ b/net/http/http_stream_factory_impl_job.cc
@@ -134,6 +134,11 @@ LoadState HttpStreamFactoryImpl::Job::GetLoadState() const {
   }
 }
 
+void HttpStreamFactoryImpl::Job::Orphan(const Request* request) {
+  DCHECK_EQ(request_, request);
+  request_ = NULL;
+}
+
 bool HttpStreamFactoryImpl::Job::was_alternate_protocol_available() const {
   return was_alternate_protocol_available_;
 }
@@ -165,58 +170,99 @@ void HttpStreamFactoryImpl::Job::GetSSLInfo() {
 
 void HttpStreamFactoryImpl::Job::OnStreamReadyCallback() {
   DCHECK(stream_.get());
-  request_->Complete(was_alternate_protocol_available(),
-                    was_npn_negotiated(),
-                    using_spdy(),
-                    net_log_.source());
-  request_->OnStreamReady(ssl_config_, proxy_info_, stream_.release());
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned()) {
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->Complete(was_alternate_protocol_available(),
+                       was_npn_negotiated(),
+                       using_spdy(),
+                       net_log_.source());
+    request_->OnStreamReady(this, ssl_config_, proxy_info_, stream_.release());
+  }
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnSpdySessionReadyCallback() {
   DCHECK(!stream_.get());
+  DCHECK(!IsPreconnecting());
   DCHECK(using_spdy());
   DCHECK(new_spdy_session_);
   scoped_refptr<SpdySession> spdy_session = new_spdy_session_;
   new_spdy_session_ = NULL;
-  stream_factory_->OnSpdySessionReady(this, spdy_session, spdy_session_direct_);
+  if (IsOrphaned()) {
+    stream_factory_->OnSpdySessionReady(
+        spdy_session, spdy_session_direct_, ssl_config_, proxy_info_,
+        was_alternate_protocol_available(), was_npn_negotiated(),
+        using_spdy(), net_log_.source());
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->OnSpdySessionReady(this, spdy_session, spdy_session_direct_);
+  }
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnStreamFailedCallback(int result) {
-  request_->OnStreamFailed(result, ssl_config_);
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned())
+    stream_factory_->OnOrphanedJobComplete(this);
+  else
+    request_->OnStreamFailed(this, result, ssl_config_);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnCertificateErrorCallback(
     int result, const SSLInfo& ssl_info) {
-  request_->OnCertificateError(result, ssl_config_, ssl_info);
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned())
+    stream_factory_->OnOrphanedJobComplete(this);
+  else
+    request_->OnCertificateError(this, result, ssl_config_, ssl_info);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnNeedsProxyAuthCallback(
     const HttpResponseInfo& response,
     HttpAuthController* auth_controller) {
-  request_->OnNeedsProxyAuth(
-      response, ssl_config_, proxy_info_, auth_controller);
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned())
+    stream_factory_->OnOrphanedJobComplete(this);
+  else
+    request_->OnNeedsProxyAuth(
+        this, response, ssl_config_, proxy_info_, auth_controller);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnNeedsClientAuthCallback(
     SSLCertRequestInfo* cert_info) {
-  request_->OnNeedsClientAuth(ssl_config_, cert_info);
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned())
+    stream_factory_->OnOrphanedJobComplete(this);
+  else
+    request_->OnNeedsClientAuth(this, ssl_config_, cert_info);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnHttpsProxyTunnelResponseCallback(
     const HttpResponseInfo& response_info,
     HttpStream* stream) {
-  request_->OnHttpsProxyTunnelResponse(
-      response_info, ssl_config_, proxy_info_, stream);
+  DCHECK(!IsPreconnecting());
+  if (IsOrphaned())
+    stream_factory_->OnOrphanedJobComplete(this);
+  else
+    request_->OnHttpsProxyTunnelResponse(
+        this, response_info, ssl_config_, proxy_info_, stream);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnPreconnectsComplete() {
+  DCHECK(!request_);
+  if (new_spdy_session_) {
+    stream_factory_->OnSpdySessionReady(
+        new_spdy_session_, spdy_session_direct_, ssl_config_,
+        proxy_info_, was_alternate_protocol_available(),
+        was_npn_negotiated(), using_spdy(), net_log_.source());
+  }
   stream_factory_->OnPreconnectsComplete(this);
   // |this| may be deleted after this call.
 }
@@ -522,7 +568,7 @@ int HttpStreamFactoryImpl::Job::DoInitConnection() {
       using_spdy_ = true;
       next_state_ = STATE_CREATE_STREAM;
       return OK;
-    } else if (!IsPreconnecting()) {
+    } else if (request_) {
       // Update the spdy session key for the request that launched this job.
       request_->SetSpdySessionKey(spdy_session_key);
     }
@@ -1134,4 +1180,8 @@ bool HttpStreamFactoryImpl::Job::IsPreconnecting() const {
   return num_streams_ > 0;
 }
 
+bool HttpStreamFactoryImpl::Job::IsOrphaned() const {
+  return !IsPreconnecting() && !request_;
+}
+
 }  // namespace net
diff --git a/net/http/http_stream_factory_impl_job.h b/net/http/http_stream_factory_impl_job.h
index 5a8f029..b466640 100644
--- a/net/http/http_stream_factory_impl_job.h
+++ b/net/http/http_stream_factory_impl_job.h
@@ -57,6 +57,8 @@ class HttpStreamFactoryImpl::Job {
                                  const string16& password);
   LoadState GetLoadState() const;
 
+  void Orphan(const Request* request);
+
   bool was_alternate_protocol_available() const;
   bool was_npn_negotiated() const;
   bool using_spdy() const;
@@ -68,6 +70,9 @@ class HttpStreamFactoryImpl::Job {
   // Indicates whether or not this job is performing a preconnect.
   bool IsPreconnecting() const;
 
+  // Indicates whether or not this Job has been orphaned by a Request.
+  bool IsOrphaned() const;
+
  private:
   enum AlternateProtocolMode {
     kUnspecified,  // Unspecified, check HttpAlternateProtocols
diff --git a/net/http/http_stream_factory_impl_request.cc b/net/http/http_stream_factory_impl_request.cc
index b6e6b9f..b9e0cee 100644
--- a/net/http/http_stream_factory_impl_request.cc
+++ b/net/http/http_stream_factory_impl_request.cc
@@ -7,6 +7,8 @@
 #include "base/logging.h"
 #include "base/stl_util-inl.h"
 #include "net/http/http_stream_factory_impl_job.h"
+#include "net/spdy/spdy_http_stream.h"
+#include "net/spdy/spdy_session.h"
 
 namespace net {
 
@@ -18,7 +20,6 @@ HttpStreamFactoryImpl::Request::Request(const GURL& url,
       factory_(factory),
       delegate_(delegate),
       net_log_(net_log),
-      job_(NULL),
       completed_(false),
       was_alternate_protocol_available_(false),
       was_npn_negotiated_(false),
@@ -30,12 +31,17 @@ HttpStreamFactoryImpl::Request::Request(const GURL& url,
 }
 
 HttpStreamFactoryImpl::Request::~Request() {
+  if (bound_job_.get())
+    DCHECK(jobs_.empty());
+  else
+    DCHECK(!jobs_.empty());
+
   net_log_.EndEvent(NetLog::TYPE_HTTP_STREAM_REQUEST, NULL);
 
-  factory_->request_map_.erase(job_);
+  for (std::set<Job*>::iterator it = jobs_.begin(); it != jobs_.end(); ++it)
+    factory_->request_map_.erase(*it);
 
-  // TODO(willchan): Remove this when we decouple requests and jobs.
-  delete job_;
+  STLDeleteElements(&jobs_);
 
   RemoveRequestFromSpdySessionRequestMap();
 }
@@ -50,10 +56,10 @@ void HttpStreamFactoryImpl::Request::SetSpdySessionKey(
   request_set.insert(this);
 }
 
-void HttpStreamFactoryImpl::Request::BindJob(HttpStreamFactoryImpl::Job* job) {
+void HttpStreamFactoryImpl::Request::AttachJob(Job* job) {
   DCHECK(job);
-  DCHECK(!job_);
-  job_ = job;
+  jobs_.insert(job);
+  factory_->request_map_[job] = this;
 }
 
 void HttpStreamFactoryImpl::Request::Complete(
@@ -73,49 +79,96 @@ void HttpStreamFactoryImpl::Request::Complete(
 }
 
 void HttpStreamFactoryImpl::Request::OnStreamReady(
+    Job* job,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpStream* stream) {
   DCHECK(stream);
   DCHECK(completed_);
+
+  // |job| should only be NULL if we're being serviced by a late bound
+  // SpdySession (one that was not created by a job in our |jobs_| set).
+  if (!job) {
+    DCHECK(!bound_job_.get());
+    DCHECK(!jobs_.empty());
+    // NOTE(willchan): We do *NOT* call OrphanJobs() here. The reason is because
+    // we *WANT* to cancel the unnecessary Jobs from other requests if another
+    // Job completes first.
+    // TODO(mbelshe): Revisit this when we implement ip connection pooling of
+    // SpdySessions. Do we want to orphan the jobs for a different hostname so
+    // they complete? Or do we want to prevent connecting a new SpdySession if
+    // we've already got one available for a different hostname where the ip
+    // address matches up?
+  } else if (!bound_job_.get()) {
+    // We may have other jobs in |jobs_|. For example, if we start multiple jobs
+    // for Alternate-Protocol.
+    OrphanJobsExcept(job);
+  } else {
+    DCHECK(jobs_.empty());
+  }
   delegate_->OnStreamReady(used_ssl_config, used_proxy_info, stream);
 }
 
 void HttpStreamFactoryImpl::Request::OnStreamFailed(
+    Job* job,
     int status,
     const SSLConfig& used_ssl_config) {
   DCHECK_NE(OK, status);
+  if (!bound_job_.get())
+    OrphanJobsExcept(job);
+  else
+    DCHECK(jobs_.empty());
   delegate_->OnStreamFailed(status, used_ssl_config);
 }
 
 void HttpStreamFactoryImpl::Request::OnCertificateError(
+    Job* job,
     int status,
     const SSLConfig& used_ssl_config,
     const SSLInfo& ssl_info) {
   DCHECK_NE(OK, status);
+  if (!bound_job_.get())
+    OrphanJobsExcept(job);
+  else
+    DCHECK(jobs_.empty());
   delegate_->OnCertificateError(status, used_ssl_config, ssl_info);
 }
 
 void HttpStreamFactoryImpl::Request::OnNeedsProxyAuth(
+    Job* job,
     const HttpResponseInfo& proxy_response,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpAuthController* auth_controller) {
+  if (!bound_job_.get())
+    OrphanJobsExcept(job);
+  else
+    DCHECK(jobs_.empty());
   delegate_->OnNeedsProxyAuth(
       proxy_response, used_ssl_config, used_proxy_info, auth_controller);
 }
 
 void HttpStreamFactoryImpl::Request::OnNeedsClientAuth(
+    Job* job,
     const SSLConfig& used_ssl_config,
     SSLCertRequestInfo* cert_info) {
+  if (!bound_job_.get())
+    OrphanJobsExcept(job);
+  else
+    DCHECK(jobs_.empty());
   delegate_->OnNeedsClientAuth(used_ssl_config, cert_info);
 }
 
 void HttpStreamFactoryImpl::Request::OnHttpsProxyTunnelResponse(
+    Job *job,
     const HttpResponseInfo& response_info,
     const SSLConfig& used_ssl_config,
     const ProxyInfo& used_proxy_info,
     HttpStream* stream) {
+  if (!bound_job_.get())
+    OrphanJobsExcept(job);
+  else
+    DCHECK(jobs_.empty());
   delegate_->OnHttpsProxyTunnelResponse(
       response_info, used_ssl_config, used_proxy_info, stream);
 }
@@ -123,15 +176,17 @@ void HttpStreamFactoryImpl::Request::OnHttpsProxyTunnelResponse(
 int HttpStreamFactoryImpl::Request::RestartTunnelWithProxyAuth(
     const string16& username,
     const string16& password) {
-  // We're restarting the job, so ditch the old key. Note that we could actually
-  // keep it around and eliminate the DCHECK in set_spdy_session_key() that
-  // |spdy_session_key_| is NULL, but I prefer to keep the assertion.
-  RemoveRequestFromSpdySessionRequestMap();
-  return job_->RestartTunnelWithProxyAuth(username, password);
+  DCHECK(bound_job_.get());
+  return bound_job_->RestartTunnelWithProxyAuth(username, password);
 }
 
 LoadState HttpStreamFactoryImpl::Request::GetLoadState() const {
-  return factory_->GetLoadState(*this);
+  if (bound_job_.get())
+    return bound_job_->GetLoadState();
+  DCHECK(!jobs_.empty());
+
+  // Just pick the first one.
+  return (*jobs_.begin())->GetLoadState();
 }
 
 bool HttpStreamFactoryImpl::Request::was_alternate_protocol_available() const {
@@ -165,4 +220,68 @@ HttpStreamFactoryImpl::Request::RemoveRequestFromSpdySessionRequestMap() {
   }
 }
 
+void HttpStreamFactoryImpl::Request::OnSpdySessionReady(
+    Job* job,
+    scoped_refptr<SpdySession> spdy_session,
+    bool direct) {
+  DCHECK(job);
+  DCHECK(job->using_spdy());
+
+  // The first case is the usual case.
+  if (!bound_job_.get()) {
+    OrphanJobsExcept(job);
+  } else { // This is the case for HTTPS proxy tunneling.
+    DCHECK_EQ(bound_job_.get(), job);
+    DCHECK(jobs_.empty());
+  }
+
+  // Cache these values in case the job gets deleted.
+  const SSLConfig used_ssl_config = job->ssl_config();
+  const ProxyInfo used_proxy_info = job->proxy_info();
+  const bool was_alternate_protocol_available =
+      job->was_alternate_protocol_available();
+  const bool was_npn_negotiated = job->was_npn_negotiated();
+  const bool using_spdy = job->using_spdy();
+  const NetLog::Source source = job->net_log().source();
+
+  Complete(was_alternate_protocol_available,
+           was_npn_negotiated,
+           using_spdy,
+           source);
+
+  // Cache this so we can still use it if the request is deleted.
+  HttpStreamFactoryImpl* factory = factory_;
+
+  bool use_relative_url = direct || url().SchemeIs("https");
+  delegate_->OnStreamReady(
+      job->ssl_config(),
+      job->proxy_info(),
+      new SpdyHttpStream(spdy_session, use_relative_url));
+  // |this| may be deleted after this point.
+  factory->OnSpdySessionReady(
+      spdy_session, direct, used_ssl_config, used_proxy_info,
+      was_alternate_protocol_available, was_npn_negotiated, using_spdy, source);
+}
+
+void HttpStreamFactoryImpl::Request::OrphanJobsExcept(Job* job) {
+  DCHECK(job);
+  DCHECK(!bound_job_.get());
+  DCHECK(ContainsKey(jobs_, job));
+  bound_job_.reset(job);
+  jobs_.erase(job);
+  factory_->request_map_.erase(job);
+
+  OrphanJobs();
+}
+
+void HttpStreamFactoryImpl::Request::OrphanJobs() {
+  RemoveRequestFromSpdySessionRequestMap();
+
+  std::set<Job*> tmp;
+  tmp.swap(jobs_);
+
+  for (std::set<Job*>::iterator it = tmp.begin(); it != tmp.end(); ++it)
+    factory_->OrphanJob(*it, this);
+}
+
 }  // namespace net
diff --git a/net/http/http_stream_factory_impl_request.h b/net/http/http_stream_factory_impl_request.h
index e7eadbe..3c1b996 100644
--- a/net/http/http_stream_factory_impl_request.h
+++ b/net/http/http_stream_factory_impl_request.h
@@ -5,6 +5,7 @@
 #ifndef NET_HTTP_HTTP_STREAM_FACTORY_IMPL_REQUEST_H_
 #define NET_HTTP_HTTP_STREAM_FACTORY_IMPL_REQUEST_H_
 
+#include <set>
 #include "base/scoped_ptr.h"
 #include "googleurl/src/gurl.h"
 #include "net/base/net_log.h"
@@ -20,9 +21,6 @@ class HttpStreamFactoryImpl::Request : public HttpStreamRequest {
           const BoundNetLog& net_log);
   virtual ~Request();
 
-  // Returns the Job that the Request started up.
-  Job* job() const { return job_; }
-
   // The GURL from the HttpRequestInfo the started the Request.
   const GURL& url() const { return url_; }
 
@@ -32,8 +30,9 @@ class HttpStreamFactoryImpl::Request : public HttpStreamRequest {
   // before knowing if SPDY is available.
   void SetSpdySessionKey(const HostPortProxyPair& spdy_session_key);
 
-  // Binds |job| to this request.
-  void BindJob(HttpStreamFactoryImpl::Job* job);
+  // Attaches |job| to this request. Does not mean that Request will use |job|,
+  // but Request will own |job|.
+  void AttachJob(HttpStreamFactoryImpl::Job* job);
 
   // Marks completion of the request. Must be called before OnStreamReady().
   // |source| is the NetLog::Source generated by the Job that fulfilled this
@@ -47,23 +46,33 @@ class HttpStreamFactoryImpl::Request : public HttpStreamRequest {
   // SpdySessionRequestMap.
   void RemoveRequestFromSpdySessionRequestMap();
 
+  // Called by an attached Job if it sets up a SpdySession.
+  void OnSpdySessionReady(Job* job,
+                          scoped_refptr<SpdySession> spdy_session,
+                          bool direct);
+
   // HttpStreamRequest::Delegate methods which we implement. Note we don't
   // actually subclass HttpStreamRequest::Delegate.
 
-  void OnStreamReady(const SSLConfig& used_ssl_config,
+  void OnStreamReady(Job* job,
+                     const SSLConfig& used_ssl_config,
                      const ProxyInfo& used_proxy_info,
                      HttpStream* stream);
-  void OnStreamFailed(int status, const SSLConfig& used_ssl_config);
-  void OnCertificateError(int status,
+  void OnStreamFailed(Job* job, int status, const SSLConfig& used_ssl_config);
+  void OnCertificateError(Job* job,
+                          int status,
                           const SSLConfig& used_ssl_config,
                           const SSLInfo& ssl_info);
-  void OnNeedsProxyAuth(const HttpResponseInfo& proxy_response,
+  void OnNeedsProxyAuth(Job* job,
+                        const HttpResponseInfo& proxy_response,
                         const SSLConfig& used_ssl_config,
                         const ProxyInfo& used_proxy_info,
                         HttpAuthController* auth_controller);
-  void OnNeedsClientAuth(const SSLConfig& used_ssl_config,
+  void OnNeedsClientAuth(Job* job,
+                         const SSLConfig& used_ssl_config,
                          SSLCertRequestInfo* cert_info);
   void OnHttpsProxyTunnelResponse(
+      Job *job,
       const HttpResponseInfo& response_info,
       const SSLConfig& used_ssl_config,
       const ProxyInfo& used_proxy_info,
@@ -79,13 +88,21 @@ class HttpStreamFactoryImpl::Request : public HttpStreamRequest {
   virtual bool using_spdy() const;
 
  private:
+  // Used to orphan all jobs in |jobs_| other than |job| which becomes "bound"
+  // to the request.
+  void OrphanJobsExcept(Job* job);
+
+  // Used to orphan all jobs in |jobs_|.
+  void OrphanJobs();
+
   const GURL url_;
   HttpStreamFactoryImpl* const factory_;
   HttpStreamRequest::Delegate* const delegate_;
   const BoundNetLog net_log_;
 
-  // The |job_| that this request is tied to.
-  HttpStreamFactoryImpl::Job* job_;
+  // At the point where Job is irrevocably tied to the Request, we set this.
+  scoped_ptr<Job> bound_job_;
+  std::set<HttpStreamFactoryImpl::Job*> jobs_;
   scoped_ptr<const HostPortProxyPair> spdy_session_key_;
 
   bool completed_;
diff --git a/net/spdy/spdy_network_transaction_unittest.cc b/net/spdy/spdy_network_transaction_unittest.cc
index 019cc4c..84e19d6 100644
--- a/net/spdy/spdy_network_transaction_unittest.cc
+++ b/net/spdy/spdy_network_transaction_unittest.cc
@@ -579,9 +579,10 @@ TEST_P(SpdyNetworkTransactionTest, ThreeGets) {
   scoped_ptr<spdy::SpdyFrame> body3(ConstructSpdyBodyFrame(5, false));
   scoped_ptr<spdy::SpdyFrame> fbody3(ConstructSpdyBodyFrame(5, true));
 
-  MockWrite writes[] = { CreateMockWrite(*req),
-                         CreateMockWrite(*req2),
-                         CreateMockWrite(*req3),
+  MockWrite writes[] = {
+    CreateMockWrite(*req),
+    CreateMockWrite(*req2),
+    CreateMockWrite(*req3),
   };
   MockRead reads[] = {
     CreateMockRead(*resp, 1),
@@ -657,6 +658,193 @@ TEST_P(SpdyNetworkTransactionTest, ThreeGets) {
   EXPECT_EQ("hello!hello!", out.response_data);
 }
 
+TEST_P(SpdyNetworkTransactionTest, TwoGetsLateBinding) {
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, false));
+  scoped_ptr<spdy::SpdyFrame> fbody(ConstructSpdyBodyFrame(1, true));
+
+  scoped_ptr<spdy::SpdyFrame> req2(ConstructSpdyGet(NULL, 0, false, 3, LOWEST));
+  scoped_ptr<spdy::SpdyFrame> resp2(ConstructSpdyGetSynReply(NULL, 0, 3));
+  scoped_ptr<spdy::SpdyFrame> body2(ConstructSpdyBodyFrame(3, false));
+  scoped_ptr<spdy::SpdyFrame> fbody2(ConstructSpdyBodyFrame(3, true));
+
+  MockWrite writes[] = {
+    CreateMockWrite(*req),
+    CreateMockWrite(*req2),
+  };
+  MockRead reads[] = {
+    CreateMockRead(*resp, 1),
+    CreateMockRead(*body),
+    CreateMockRead(*resp2, 4),
+    CreateMockRead(*body2),
+    CreateMockRead(*fbody),
+    CreateMockRead(*fbody2),
+    MockRead(true, 0, 0),  // EOF
+  };
+  scoped_refptr<OrderedSocketData> data(
+      new OrderedSocketData(reads, arraysize(reads),
+                            writes, arraysize(writes)));
+
+  MockConnect never_finishing_connect(true, ERR_IO_PENDING);
+
+  scoped_refptr<OrderedSocketData> data_placeholder(
+      new OrderedSocketData(NULL, 0, NULL, 0));
+  data_placeholder->set_connect_data(never_finishing_connect);
+
+  BoundNetLog log;
+  TransactionHelperResult out;
+  NormalSpdyTransactionHelper helper(CreateGetRequest(),
+                                     BoundNetLog(), GetParam());
+  helper.RunPreTestSetup();
+  helper.AddData(data.get());
+  // We require placeholder data because two get requests are sent out, so
+  // there needs to be two sets of SSL connection data.
+  helper.AddData(data_placeholder.get());
+  scoped_ptr<HttpNetworkTransaction> trans1(
+      new HttpNetworkTransaction(helper.session()));
+  scoped_ptr<HttpNetworkTransaction> trans2(
+      new HttpNetworkTransaction(helper.session()));
+
+  TestCompletionCallback callback1;
+  TestCompletionCallback callback2;
+
+  HttpRequestInfo httpreq1 = CreateGetRequest();
+  HttpRequestInfo httpreq2 = CreateGetRequest();
+
+  out.rv = trans1->Start(&httpreq1, &callback1, log);
+  ASSERT_EQ(ERR_IO_PENDING, out.rv);
+  out.rv = trans2->Start(&httpreq2, &callback2, log);
+  ASSERT_EQ(ERR_IO_PENDING, out.rv);
+
+  out.rv = callback1.WaitForResult();
+  ASSERT_EQ(OK, out.rv);
+  out.rv = callback2.WaitForResult();
+  ASSERT_EQ(OK, out.rv);
+
+  const HttpResponseInfo* response1 = trans1->GetResponseInfo();
+  EXPECT_TRUE(response1->headers != NULL);
+  EXPECT_TRUE(response1->was_fetched_via_spdy);
+  out.status_line = response1->headers->GetStatusLine();
+  out.response_info = *response1;
+  out.rv = ReadTransaction(trans1.get(), &out.response_data);
+  EXPECT_EQ(OK, out.rv);
+  EXPECT_EQ("HTTP/1.1 200 OK", out.status_line);
+  EXPECT_EQ("hello!hello!", out.response_data);
+
+  const HttpResponseInfo* response2 = trans2->GetResponseInfo();
+  EXPECT_TRUE(response2->headers != NULL);
+  EXPECT_TRUE(response2->was_fetched_via_spdy);
+  out.status_line = response2->headers->GetStatusLine();
+  out.response_info = *response2;
+  out.rv = ReadTransaction(trans2.get(), &out.response_data);
+  EXPECT_EQ(OK, out.rv);
+  EXPECT_EQ("HTTP/1.1 200 OK", out.status_line);
+  EXPECT_EQ("hello!hello!", out.response_data);
+
+  helper.VerifyDataConsumed();
+}
+
+TEST_P(SpdyNetworkTransactionTest, TwoGetsLateBindingFromPreconnect) {
+  scoped_ptr<spdy::SpdyFrame> req(ConstructSpdyGet(NULL, 0, false, 1, LOWEST));
+  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame> body(ConstructSpdyBodyFrame(1, false));
+  scoped_ptr<spdy::SpdyFrame> fbody(ConstructSpdyBodyFrame(1, true));
+
+  scoped_ptr<spdy::SpdyFrame> req2(ConstructSpdyGet(NULL, 0, false, 3, LOWEST));
+  scoped_ptr<spdy::SpdyFrame> resp2(ConstructSpdyGetSynReply(NULL, 0, 3));
+  scoped_ptr<spdy::SpdyFrame> body2(ConstructSpdyBodyFrame(3, false));
+  scoped_ptr<spdy::SpdyFrame> fbody2(ConstructSpdyBodyFrame(3, true));
+
+  MockWrite writes[] = {
+    CreateMockWrite(*req),
+    CreateMockWrite(*req2),
+  };
+  MockRead reads[] = {
+    CreateMockRead(*resp, 1),
+    CreateMockRead(*body),
+    CreateMockRead(*resp2, 4),
+    CreateMockRead(*body2),
+    CreateMockRead(*fbody),
+    CreateMockRead(*fbody2),
+    MockRead(true, 0, 0),  // EOF
+  };
+  scoped_refptr<OrderedSocketData> preconnect_data(
+      new OrderedSocketData(reads, arraysize(reads),
+                            writes, arraysize(writes)));
+
+  MockConnect never_finishing_connect(true, ERR_IO_PENDING);
+
+  scoped_refptr<OrderedSocketData> data_placeholder(
+      new OrderedSocketData(NULL, 0, NULL, 0));
+  data_placeholder->set_connect_data(never_finishing_connect);
+
+  BoundNetLog log;
+  TransactionHelperResult out;
+  NormalSpdyTransactionHelper helper(CreateGetRequest(),
+                                     BoundNetLog(), GetParam());
+  helper.RunPreTestSetup();
+  helper.AddData(preconnect_data.get());
+  // We require placeholder data because 3 connections are attempted (first is
+  // the preconnect, 2nd and 3rd are the never finished connections.
+  helper.AddData(data_placeholder.get());
+  helper.AddData(data_placeholder.get());
+
+  scoped_ptr<HttpNetworkTransaction> trans1(
+      new HttpNetworkTransaction(helper.session()));
+  scoped_ptr<HttpNetworkTransaction> trans2(
+      new HttpNetworkTransaction(helper.session()));
+
+  TestCompletionCallback callback1;
+  TestCompletionCallback callback2;
+
+  HttpRequestInfo httpreq = CreateGetRequest();
+
+  // Preconnect the first.
+  SSLConfig preconnect_ssl_config;
+  helper.session()->ssl_config_service()->GetSSLConfig(&preconnect_ssl_config);
+  HttpStreamFactory* http_stream_factory =
+      helper.session()->http_stream_factory();
+  if (http_stream_factory->next_protos()) {
+    preconnect_ssl_config.next_protos = *http_stream_factory->next_protos();
+  }
+
+  http_stream_factory->PreconnectStreams(
+      1, httpreq, preconnect_ssl_config, log);
+
+  out.rv = trans1->Start(&httpreq, &callback1, log);
+  ASSERT_EQ(ERR_IO_PENDING, out.rv);
+  out.rv = trans2->Start(&httpreq, &callback2, log);
+  ASSERT_EQ(ERR_IO_PENDING, out.rv);
+
+  out.rv = callback1.WaitForResult();
+  ASSERT_EQ(OK, out.rv);
+  out.rv = callback2.WaitForResult();
+  ASSERT_EQ(OK, out.rv);
+
+  const HttpResponseInfo* response1 = trans1->GetResponseInfo();
+  EXPECT_TRUE(response1->headers != NULL);
+  EXPECT_TRUE(response1->was_fetched_via_spdy);
+  out.status_line = response1->headers->GetStatusLine();
+  out.response_info = *response1;
+  out.rv = ReadTransaction(trans1.get(), &out.response_data);
+  EXPECT_EQ(OK, out.rv);
+  EXPECT_EQ("HTTP/1.1 200 OK", out.status_line);
+  EXPECT_EQ("hello!hello!", out.response_data);
+
+  const HttpResponseInfo* response2 = trans2->GetResponseInfo();
+  EXPECT_TRUE(response2->headers != NULL);
+  EXPECT_TRUE(response2->was_fetched_via_spdy);
+  out.status_line = response2->headers->GetStatusLine();
+  out.response_info = *response2;
+  out.rv = ReadTransaction(trans2.get(), &out.response_data);
+  EXPECT_EQ(OK, out.rv);
+  EXPECT_EQ("HTTP/1.1 200 OK", out.status_line);
+  EXPECT_EQ("hello!hello!", out.response_data);
+
+  helper.VerifyDataConsumed();
+}
+
 // Similar to ThreeGets above, however this test adds a SETTINGS
 // frame.  The SETTINGS frame is read during the IO loop waiting on
 // the first transaction completion, and sets a maximum concurrent