Implement QUIC-based net::BidirectionalStream

net/quic/quic_stream_factory.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/quic/quic_stream_factory.h"
 
 #include <algorithm>
 #include <set>
 #include <utility>
 
@@ skipping 43 matching lines @@
 #include "base/win/windows_version.h"
 #endif
 
 #if defined(USE_OPENSSL)
 #include <openssl/aead.h>
 #include "crypto/openssl_util.h"
 #else
 #include "base/cpu.h"
 #endif
 
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+#include "net/http/bidirectional_stream_job.h"
+#include "net/quic/bidirectional_stream_quic_impl.h"
+#endif
+
 using std::min;
 using NetworkHandle = net::NetworkChangeNotifier::NetworkHandle;
 
 namespace net {
 
 namespace {
 
 enum CreateSessionFailure {
   CREATION_ERROR_CONNECTING_SOCKET,
   CREATION_ERROR_SETTING_RECEIVE_BUFFER,
@@ skipping 417 matching lines @@
     session_ = nullptr;
     return OK;
   }
 
   factory_->ActivateSession(server_id_, session_);
 
   return OK;
 }
 
 QuicStreamRequest::QuicStreamRequest(QuicStreamFactory* factory)
-    : factory_(factory) {}
+    : factory_(factory), for_bidirectional_(false) {}
 
 QuicStreamRequest::~QuicStreamRequest() {
   if (factory_ && !callback_.is_null())
     factory_->CancelRequest(this);
 }
 
 int QuicStreamRequest::Request(const HostPortPair& host_port_pair,
                                PrivacyMode privacy_mode,
                                int cert_verify_flags,
                                const GURL& url,
                                base::StringPiece method,
                                const BoundNetLog& net_log,
-                               const CompletionCallback& callback) {
+                               const CompletionCallback& callback,
+                               bool for_bidirectional) {
   DCHECK(!stream_);
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+  DCHECK(!bidirectional_stream_job_);
+#endif
   DCHECK(callback_.is_null());
   DCHECK(factory_);
   origin_host_ = url.host();
   privacy_mode_ = privacy_mode;
+  for_bidirectional_ = for_bidirectional;
+
   int rv = factory_->Create(host_port_pair, privacy_mode, cert_verify_flags,
                             url, method, net_log, this);
   if (rv == ERR_IO_PENDING) {
     host_port_pair_ = host_port_pair;
     net_log_ = net_log;
     callback_ = callback;
   } else {
     factory_ = nullptr;
   }
-  if (rv == OK)
+  if (rv == OK) {
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+    DCHECK(stream_ || (for_bidirectional_ && bidirectional_stream_job_));

[Ryan Hamilton, 2016/03/03 04:14:57]

This would be true if both stream_ and bidirectional_stream_job_ were both set.
Is that desirable?

[xunjieli, 2016/03/03 16:54:19]

Done.

+#else
     DCHECK(stream_);
+#endif
+  }
   return rv;
 }
 
-void QuicStreamRequest::set_stream(scoped_ptr<QuicHttpStream> stream) {
-  DCHECK(stream);
-  stream_ = std::move(stream);
+void QuicStreamRequest::InitializeStreamFromSession(
+    QuicChromiumClientSession* session) {
+  DCHECK(session);
+  if (for_bidirectional_) {
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+    bidirectional_stream_job_.reset(
+        new BidirectionalStreamQuicImpl(session->GetWeakPtr()));
+#else
+    DCHECK(false);
+#endif
+  } else {
+    stream_.reset(new QuicHttpStream(session->GetWeakPtr()));
+  }

[Ryan Hamilton, 2016/03/03 04:14:58]

How about instead of creating the QuicHttpStream or Bidirectional stream here,
and then storing it in one of two member, how about change ReleaseStream() to
create this object on the fly?

[xunjieli, 2016/03/03 16:54:19]

Done. Great idea! But this would change the semantics of ReleaseStream(), which
now sometimes can return nullptr if session is closed before ReleaseStream() is
invoked. Will that be okay? See quic_stream_factory_test.cc for an example.

[Ryan Hamilton, 2016/03/07 20:12:37]

Good point. I think you'll need to change
HttpStreamFactoryImpl::Job::DoInitConnectionComplete to detect this case. Looks
like it should be pretty simple, though.

[xunjieli, 2016/03/08 15:39:31]

Done.

 }
 
 void QuicStreamRequest::OnRequestComplete(int rv) {
   factory_ = nullptr;
   callback_.Run(rv);
 }
 
 base::TimeDelta QuicStreamRequest::GetTimeDelayForWaitingJob() const {
   if (!factory_)
     return base::TimeDelta();
   return factory_->GetTimeDelayForWaitingJob(
       QuicServerId(host_port_pair_, privacy_mode_));
 }
 
 scoped_ptr<QuicHttpStream> QuicStreamRequest::ReleaseStream() {
   DCHECK(stream_);
+  DCHECK(!for_bidirectional_);
   return std::move(stream_);
 }
 
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+scoped_ptr<BidirectionalStreamJob>
+QuicStreamRequest::ReleaseBidirectionalStreamJob() {
+  DCHECK(bidirectional_stream_job_);
+  DCHECK(for_bidirectional_);
+
+  return std::move(bidirectional_stream_job_);
+}
+#endif
+
 QuicStreamFactory::QuicStreamFactory(
     HostResolver* host_resolver,
     ClientSocketFactory* client_socket_factory,
     base::WeakPtr<HttpServerProperties> http_server_properties,
     CertVerifier* cert_verifier,
     CTPolicyEnforcer* ct_policy_enforcer,
     ChannelIDService* channel_id_service,
     TransportSecurityState* transport_security_state,
     CTVerifier* cert_transparency_verifier,
     SocketPerformanceWatcherFactory* socket_performance_watcher_factory,
@@ skipping 205 matching lines @@
                               const BoundNetLog& net_log,
                               QuicStreamRequest* request) {
   // Enforce session affinity for promised streams.
   QuicClientPromisedInfo* promised =
       push_promise_index_.GetPromised(url.spec());
   if (promised) {
     QuicChromiumClientSession* session =
         static_cast<QuicChromiumClientSession*>(promised->session());
     DCHECK(session);
     if (session->server_id().privacy_mode() == privacy_mode) {
-      request->set_stream(CreateFromSession(session));
+      request->InitializeStreamFromSession(session);
       ++num_push_streams_created_;
       return OK;
     }
     // This should happen extremely rarely (if ever), but if somehow a
     // request comes in with a mismatched privacy mode, consider the
     // promise borked.
     promised->Cancel();
   }
 
   QuicServerId server_id(host_port_pair, privacy_mode);
   // TODO(rtenneti): crbug.com/498823 - delete active_sessions_.empty() checks.
   if (!active_sessions_.empty()) {
     SessionMap::iterator it = active_sessions_.find(server_id);
     if (it != active_sessions_.end()) {
       QuicChromiumClientSession* session = it->second;
       if (!session->CanPool(url.host(), privacy_mode))
         return ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN;
-      request->set_stream(CreateFromSession(session));
+      request->InitializeStreamFromSession(session);
       return OK;
     }
   }
 
   if (HasActiveJob(server_id)) {
     active_requests_[request] = server_id;
     job_requests_map_[server_id].insert(request);
     return ERR_IO_PENDING;
   }
 
@@ skipping 35 matching lines @@
     // related changes.
     if (active_sessions_.empty())
       return ERR_QUIC_PROTOCOL_ERROR;
     SessionMap::iterator it = active_sessions_.find(server_id);
     DCHECK(it != active_sessions_.end());
     if (it == active_sessions_.end())
       return ERR_QUIC_PROTOCOL_ERROR;
     QuicChromiumClientSession* session = it->second;
     if (!session->CanPool(url.host(), privacy_mode))
       return ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN;
-    request->set_stream(CreateFromSession(session));
+    request->InitializeStreamFromSession(session);
   }
   return rv;
 }
 
 void QuicStreamFactory::CreateAuxilaryJob(const QuicServerId server_id,
                                           int cert_verify_flags,
                                           bool server_and_origin_have_same_host,
                                           bool is_post,
                                           const BoundNetLog& net_log) {
   Job* aux_job = new Job(
@@ skipping 57 matching lines @@
         RequestSet::iterator old_request_it = request_it;
         ++request_it;
         // Remove request from containers so that OnRequestComplete() is not
         // called later again on the same request.
         job_requests_map_[server_id].erase(old_request_it);
         active_requests_.erase(request);
         // Notify request of certificate error.
         request->OnRequestComplete(ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN);
         continue;
       }
-      request->set_stream(CreateFromSession(session));
+      request->InitializeStreamFromSession(session);
       ++request_it;
     }
   }
 
   while (!job_requests_map_[server_id].empty()) {
     RequestSet::iterator it = job_requests_map_[server_id].begin();
     QuicStreamRequest* request = *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
@@ skipping 728 matching lines @@
   // Since the session was active, there's no longer an
   // HttpStreamFactoryImpl::Job running which can mark it broken, unless the TCP
   // job also fails. So to avoid not using QUIC when we otherwise could, we mark
   // it as recently broken, which means that 0-RTT will be disabled but we'll
   // still race.
   http_server_properties_->MarkAlternativeServiceRecentlyBroken(
       alternative_service);
 }
 
 }  // namespace net