Commit 1baa747b authored by rch's avatar rch Committed by Commit bot

QuicChromiumClientSession::StreamRequest to be created by the session

on demand to the client. Make the request hold the stream until the
caller asks for it, and make explicit cancellation unnecessary.

This is in preparation for subsquent cleanups of the session.

Review-Url: https://codereview.chromium.org/2844493002
Cr-Commit-Position: refs/heads/master@{#467247}
parent ea39d1dc
...@@ -73,10 +73,9 @@ void BidirectionalStreamQuicImpl::Start( ...@@ -73,10 +73,9 @@ void BidirectionalStreamQuicImpl::Start(
delegate_ = delegate; delegate_ = delegate;
request_info_ = request_info; request_info_ = request_info;
int rv = stream_request_.StartRequest( stream_request_ = session_->CreateStreamRequest();
session_, &stream_, int rv = stream_request_->StartRequest(base::Bind(
base::Bind(&BidirectionalStreamQuicImpl::OnStreamReady, &BidirectionalStreamQuicImpl::OnStreamReady, weak_factory_.GetWeakPtr()));
weak_factory_.GetWeakPtr()));
if (rv == OK) { if (rv == OK) {
OnStreamReady(rv); OnStreamReady(rv);
} else if (!was_handshake_confirmed_) { } else if (!was_handshake_confirmed_) {
...@@ -306,6 +305,8 @@ void BidirectionalStreamQuicImpl::OnStreamReady(int rv) { ...@@ -306,6 +305,8 @@ void BidirectionalStreamQuicImpl::OnStreamReady(int rv) {
DCHECK_NE(ERR_IO_PENDING, rv); DCHECK_NE(ERR_IO_PENDING, rv);
DCHECK(rv == OK || !stream_); DCHECK(rv == OK || !stream_);
if (rv == OK) { if (rv == OK) {
stream_ = stream_request_->ReleaseStream();
stream_request_.reset();
stream_->SetDelegate(this); stream_->SetDelegate(this);
if (!was_handshake_confirmed_ && request_info_->method == "POST") { if (!was_handshake_confirmed_ && request_info_->method == "POST") {
waiting_for_confirmation_ = true; waiting_for_confirmation_ = true;
......
...@@ -84,7 +84,7 @@ class NET_EXPORT_PRIVATE BidirectionalStreamQuicImpl ...@@ -84,7 +84,7 @@ class NET_EXPORT_PRIVATE BidirectionalStreamQuicImpl
base::WeakPtr<QuicChromiumClientSession> session_; base::WeakPtr<QuicChromiumClientSession> session_;
bool was_handshake_confirmed_; // True if the crypto handshake succeeded. bool was_handshake_confirmed_; // True if the crypto handshake succeeded.
QuicChromiumClientSession::StreamRequest stream_request_; std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request_;
QuicChromiumClientStream* stream_; // Non-owning. QuicChromiumClientStream* stream_; // Non-owning.
const BidirectionalStreamRequestInfo* request_info_; const BidirectionalStreamRequestInfo* request_info_;
......
...@@ -186,37 +186,43 @@ class QuicServerPushHelper : public ServerPushDelegate::ServerPushHelper { ...@@ -186,37 +186,43 @@ class QuicServerPushHelper : public ServerPushDelegate::ServerPushHelper {
} // namespace } // namespace
QuicChromiumClientSession::StreamRequest::StreamRequest() : stream_(nullptr) {} QuicChromiumClientSession::StreamRequest::StreamRequest(
const base::WeakPtr<QuicChromiumClientSession>& session)
: session_(session), stream_(nullptr) {}
QuicChromiumClientSession::StreamRequest::~StreamRequest() { QuicChromiumClientSession::StreamRequest::~StreamRequest() {
CancelRequest(); if (stream_)
stream_->Reset(QUIC_STREAM_CANCELLED);
if (session_)
session_->CancelRequest(this);
} }
int QuicChromiumClientSession::StreamRequest::StartRequest( int QuicChromiumClientSession::StreamRequest::StartRequest(
const base::WeakPtr<QuicChromiumClientSession>& session,
QuicChromiumClientStream** stream,
const CompletionCallback& callback) { const CompletionCallback& callback) {
session_ = session; DCHECK(session_);
stream_ = stream; int rv = session_->TryCreateStream(this);
int rv = session_->TryCreateStream(this, stream_);
if (rv == ERR_IO_PENDING) { if (rv == ERR_IO_PENDING) {
callback_ = callback; callback_ = callback;
} else {
session_.reset();
} }
return rv; return rv;
} }
void QuicChromiumClientSession::StreamRequest::CancelRequest() { QuicChromiumClientStream*
if (session_) QuicChromiumClientSession::StreamRequest::ReleaseStream() {
session_->CancelRequest(this); DCHECK(stream_);
session_.reset(); QuicChromiumClientStream* stream = stream_;
callback_.Reset(); stream_ = nullptr;
return stream;
} }
void QuicChromiumClientSession::StreamRequest::OnRequestCompleteSuccess( void QuicChromiumClientSession::StreamRequest::OnRequestCompleteSuccess(
QuicChromiumClientStream* stream) { QuicChromiumClientStream* stream) {
session_.reset(); session_.reset();
*stream_ = stream; stream_ = stream;
base::ResetAndReturn(&callback_).Run(OK); base::ResetAndReturn(&callback_).Run(OK);
} }
...@@ -456,9 +462,15 @@ void QuicChromiumClientSession::RemoveObserver(Observer* observer) { ...@@ -456,9 +462,15 @@ void QuicChromiumClientSession::RemoveObserver(Observer* observer) {
observers_.erase(observer); observers_.erase(observer);
} }
int QuicChromiumClientSession::TryCreateStream( std::unique_ptr<QuicChromiumClientSession::StreamRequest>
StreamRequest* request, QuicChromiumClientSession::CreateStreamRequest() {
QuicChromiumClientStream** stream) { // base::MakeUnique does not work because the StreamRequest constructor
// is private.
return std::unique_ptr<StreamRequest>(
new StreamRequest(weak_factory_.GetWeakPtr()));
}
int QuicChromiumClientSession::TryCreateStream(StreamRequest* request) {
if (goaway_received()) { if (goaway_received()) {
DVLOG(1) << "Going away."; DVLOG(1) << "Going away.";
return ERR_CONNECTION_CLOSED; return ERR_CONNECTION_CLOSED;
...@@ -475,7 +487,7 @@ int QuicChromiumClientSession::TryCreateStream( ...@@ -475,7 +487,7 @@ int QuicChromiumClientSession::TryCreateStream(
} }
if (GetNumOpenOutgoingStreams() < max_open_outgoing_streams()) { if (GetNumOpenOutgoingStreams() < max_open_outgoing_streams()) {
*stream = CreateOutgoingReliableStreamImpl(); request->stream_ = CreateOutgoingReliableStreamImpl();
return OK; return OK;
} }
......
...@@ -78,24 +78,24 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession ...@@ -78,24 +78,24 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
// A helper class used to manage a request to create a stream. // A helper class used to manage a request to create a stream.
class NET_EXPORT_PRIVATE StreamRequest { class NET_EXPORT_PRIVATE StreamRequest {
public: public:
StreamRequest(); // Cancels any pending stream creation request and resets |stream_| if
// it has not yet been released.
~StreamRequest(); ~StreamRequest();
// Starts a request to create a stream. If OK is returned, then // Starts a request to create a stream. If OK is returned, then
// |stream| will be updated with the newly created stream. If // |stream_| will be updated with the newly created stream. If
// ERR_IO_PENDING is returned, then when the request is eventuallly // ERR_IO_PENDING is returned, then when the request is eventuallly
// complete |callback| will be called. // complete |callback| will be called.
int StartRequest(const base::WeakPtr<QuicChromiumClientSession>& session, int StartRequest(const CompletionCallback& callback);
QuicChromiumClientStream** stream,
const CompletionCallback& callback);
// Cancels any pending stream creation request. May be called // Releases |stream_| to the caller
// repeatedly. QuicChromiumClientStream* ReleaseStream();
void CancelRequest();
private: private:
friend class QuicChromiumClientSession; friend class QuicChromiumClientSession;
StreamRequest(const base::WeakPtr<QuicChromiumClientSession>& session);
// Called by |session_| for an asynchronous request when the stream // Called by |session_| for an asynchronous request when the stream
// request has finished successfully. // request has finished successfully.
void OnRequestCompleteSuccess(QuicChromiumClientStream* stream); void OnRequestCompleteSuccess(QuicChromiumClientStream* stream);
...@@ -107,7 +107,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession ...@@ -107,7 +107,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
base::WeakPtr<QuicChromiumClientSession> session_; base::WeakPtr<QuicChromiumClientSession> session_;
CompletionCallback callback_; CompletionCallback callback_;
QuicChromiumClientStream** stream_; QuicChromiumClientStream* stream_;
// For tracking how much time pending stream requests wait. // For tracking how much time pending stream requests wait.
base::TimeTicks pending_start_time_; base::TimeTicks pending_start_time_;
...@@ -147,6 +147,8 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession ...@@ -147,6 +147,8 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
void AddObserver(Observer* observer); void AddObserver(Observer* observer);
void RemoveObserver(Observer* observer); void RemoveObserver(Observer* observer);
std::unique_ptr<StreamRequest> CreateStreamRequest();
// Attempts to create a new stream. If the stream can be // Attempts to create a new stream. If the stream can be
// created immediately, returns OK. If the open stream limit // created immediately, returns OK. If the open stream limit
// has been reached, returns ERR_IO_PENDING, and |request| // has been reached, returns ERR_IO_PENDING, and |request|
...@@ -154,8 +156,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession ...@@ -154,8 +156,7 @@ class NET_EXPORT_PRIVATE QuicChromiumClientSession
// be completed asynchronously. // be completed asynchronously.
// TODO(rch): remove |stream| from this and use setter on |request| // TODO(rch): remove |stream| from this and use setter on |request|
// and fix in spdy too. // and fix in spdy too.
int TryCreateStream(StreamRequest* request, int TryCreateStream(StreamRequest* request);
QuicChromiumClientStream** stream);
// Cancels the pending stream creation request. // Cancels the pending stream creation request.
void CancelRequest(StreamRequest* request); void CancelRequest(StreamRequest* request);
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "net/log/test_net_log.h" #include "net/log/test_net_log.h"
#include "net/quic/chromium/crypto/proof_verifier_chromium.h" #include "net/quic/chromium/crypto/proof_verifier_chromium.h"
#include "net/quic/chromium/mock_crypto_client_stream_factory.h" #include "net/quic/chromium/mock_crypto_client_stream_factory.h"
#include "net/quic/chromium/mock_quic_data.h"
#include "net/quic/chromium/quic_chromium_alarm_factory.h" #include "net/quic/chromium/quic_chromium_alarm_factory.h"
#include "net/quic/chromium/quic_chromium_client_session_peer.h" #include "net/quic/chromium/quic_chromium_client_session_peer.h"
#include "net/quic/chromium/quic_chromium_connection_helper.h" #include "net/quic/chromium/quic_chromium_connection_helper.h"
...@@ -103,7 +104,8 @@ class QuicChromiumClientSessionTest ...@@ -103,7 +104,8 @@ class QuicChromiumClientSessionTest
} }
void Initialize() { void Initialize() {
socket_factory_.AddSocketDataProvider(socket_data_.get()); if (socket_data_)
socket_factory_.AddSocketDataProvider(socket_data_.get());
std::unique_ptr<DatagramClientSocket> socket = std::unique_ptr<DatagramClientSocket> socket =
socket_factory_.CreateDatagramClientSocket(DatagramSocket::DEFAULT_BIND, socket_factory_.CreateDatagramClientSocket(DatagramSocket::DEFAULT_BIND,
base::Bind(&base::RandInt), base::Bind(&base::RandInt),
...@@ -193,6 +195,129 @@ TEST_P(QuicChromiumClientSessionTest, CryptoConnect) { ...@@ -193,6 +195,129 @@ TEST_P(QuicChromiumClientSessionTest, CryptoConnect) {
CompleteCryptoHandshake(); CompleteCryptoHandshake();
} }
TEST_P(QuicChromiumClientSessionTest, StreamRequest) {
MockQuicData quic_data;
quic_data.AddWrite(client_maker_.MakeInitialSettingsPacket(1, nullptr));
quic_data.AddRead(ASYNC, ERR_IO_PENDING);
quic_data.AddRead(ASYNC, OK); // EOF
quic_data.AddSocketDataToFactory(&socket_factory_);
Initialize();
CompleteCryptoHandshake();
// Request a stream and verify that a stream was created.
std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request =
session_->CreateStreamRequest();
TestCompletionCallback callback;
ASSERT_EQ(OK, stream_request->StartRequest(callback.callback()));
EXPECT_TRUE(stream_request->ReleaseStream() != nullptr);
quic_data.Resume();
EXPECT_TRUE(quic_data.AllReadDataConsumed());
EXPECT_TRUE(quic_data.AllWriteDataConsumed());
}
TEST_P(QuicChromiumClientSessionTest, CancelStreamRequestBeforeRelease) {
MockQuicData quic_data;
quic_data.AddWrite(client_maker_.MakeInitialSettingsPacket(1, nullptr));
quic_data.AddWrite(client_maker_.MakeRstPacket(2, true, kClientDataStreamId1,
QUIC_STREAM_CANCELLED));
quic_data.AddRead(ASYNC, ERR_IO_PENDING);
quic_data.AddRead(ASYNC, OK); // EOF
quic_data.AddSocketDataToFactory(&socket_factory_);
Initialize();
CompleteCryptoHandshake();
// Request a stream and cancel it without releasing the stream.
std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request =
session_->CreateStreamRequest();
TestCompletionCallback callback;
ASSERT_EQ(OK, stream_request->StartRequest(callback.callback()));
stream_request.reset();
quic_data.Resume();
EXPECT_TRUE(quic_data.AllReadDataConsumed());
EXPECT_TRUE(quic_data.AllWriteDataConsumed());
}
TEST_P(QuicChromiumClientSessionTest, AsyncStreamRequest) {
MockQuicData quic_data;
quic_data.AddWrite(client_maker_.MakeInitialSettingsPacket(1, nullptr));
quic_data.AddWrite(client_maker_.MakeRstPacket(2, true, kClientDataStreamId1,
QUIC_RST_ACKNOWLEDGEMENT));
quic_data.AddRead(ASYNC, ERR_IO_PENDING);
quic_data.AddRead(ASYNC, OK); // EOF
quic_data.AddSocketDataToFactory(&socket_factory_);
Initialize();
CompleteCryptoHandshake();
// Open the maximum number of streams so that a subsequent request
// can not proceed immediately.
const size_t kMaxOpenStreams = session_->max_open_outgoing_streams();
for (size_t i = 0; i < kMaxOpenStreams; i++) {
session_->CreateOutgoingDynamicStream(kDefaultPriority);
}
EXPECT_EQ(kMaxOpenStreams, session_->GetNumOpenOutgoingStreams());
// Request a stream and verify that it's pending.
std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request =
session_->CreateStreamRequest();
TestCompletionCallback callback;
ASSERT_EQ(ERR_IO_PENDING, stream_request->StartRequest(callback.callback()));
// Close a stream and ensure the stream request completes.
QuicRstStreamFrame rst(kClientDataStreamId1, QUIC_STREAM_CANCELLED, 0);
session_->OnRstStream(rst);
ASSERT_TRUE(callback.have_result());
EXPECT_THAT(callback.WaitForResult(), IsOk());
EXPECT_TRUE(stream_request->ReleaseStream() != nullptr);
quic_data.Resume();
EXPECT_TRUE(quic_data.AllReadDataConsumed());
EXPECT_TRUE(quic_data.AllWriteDataConsumed());
}
TEST_P(QuicChromiumClientSessionTest, CancelPendingStreamRequest) {
MockQuicData quic_data;
quic_data.AddWrite(client_maker_.MakeInitialSettingsPacket(1, nullptr));
quic_data.AddWrite(client_maker_.MakeRstPacket(2, true, kClientDataStreamId1,
QUIC_RST_ACKNOWLEDGEMENT));
quic_data.AddRead(ASYNC, ERR_IO_PENDING);
quic_data.AddRead(ASYNC, OK); // EOF
quic_data.AddSocketDataToFactory(&socket_factory_);
Initialize();
CompleteCryptoHandshake();
// Open the maximum number of streams so that a subsequent request
// can not proceed immediately.
const size_t kMaxOpenStreams = session_->max_open_outgoing_streams();
for (size_t i = 0; i < kMaxOpenStreams; i++) {
session_->CreateOutgoingDynamicStream(kDefaultPriority);
}
EXPECT_EQ(kMaxOpenStreams, session_->GetNumOpenOutgoingStreams());
// Request a stream and verify that it's pending.
std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request =
session_->CreateStreamRequest();
TestCompletionCallback callback;
ASSERT_EQ(ERR_IO_PENDING, stream_request->StartRequest(callback.callback()));
// Cancel the pending stream request.
stream_request.reset();
// Close a stream and ensure that no new stream is created.
QuicRstStreamFrame rst(kClientDataStreamId1, QUIC_STREAM_CANCELLED, 0);
session_->OnRstStream(rst);
EXPECT_EQ(kMaxOpenStreams - 1, session_->GetNumOpenOutgoingStreams());
quic_data.Resume();
EXPECT_TRUE(quic_data.AllReadDataConsumed());
EXPECT_TRUE(quic_data.AllWriteDataConsumed());
}
TEST_P(QuicChromiumClientSessionTest, MaxNumStreams) { TEST_P(QuicChromiumClientSessionTest, MaxNumStreams) {
MockRead reads[] = {MockRead(SYNCHRONOUS, ERR_IO_PENDING, 0)}; MockRead reads[] = {MockRead(SYNCHRONOUS, ERR_IO_PENDING, 0)};
std::unique_ptr<QuicEncryptedPacket> settings_packet( std::unique_ptr<QuicEncryptedPacket> settings_packet(
...@@ -544,12 +669,10 @@ TEST_P(QuicChromiumClientSessionTest, MaxNumStreamsViaRequest) { ...@@ -544,12 +669,10 @@ TEST_P(QuicChromiumClientSessionTest, MaxNumStreamsViaRequest) {
streams.push_back(stream); streams.push_back(stream);
} }
QuicChromiumClientStream* stream; std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request =
QuicChromiumClientSession::StreamRequest stream_request; session_->CreateStreamRequest();
TestCompletionCallback callback; TestCompletionCallback callback;
ASSERT_EQ(ERR_IO_PENDING, ASSERT_EQ(ERR_IO_PENDING, stream_request->StartRequest(callback.callback()));
stream_request.StartRequest(session_->GetWeakPtr(), &stream,
callback.callback()));
// Close a stream and ensure I can now open a new one. // Close a stream and ensure I can now open a new one.
QuicStreamId stream_id = streams[0]->id(); QuicStreamId stream_id = streams[0]->id();
...@@ -558,7 +681,7 @@ TEST_P(QuicChromiumClientSessionTest, MaxNumStreamsViaRequest) { ...@@ -558,7 +681,7 @@ TEST_P(QuicChromiumClientSessionTest, MaxNumStreamsViaRequest) {
session_->OnRstStream(rst1); session_->OnRstStream(rst1);
ASSERT_TRUE(callback.have_result()); ASSERT_TRUE(callback.have_result());
EXPECT_THAT(callback.WaitForResult(), IsOk()); EXPECT_THAT(callback.WaitForResult(), IsOk());
EXPECT_TRUE(stream != nullptr); EXPECT_TRUE(stream_request->ReleaseStream() != nullptr);
} }
TEST_P(QuicChromiumClientSessionTest, GoAwayReceived) { TEST_P(QuicChromiumClientSessionTest, GoAwayReceived) {
......
...@@ -629,8 +629,8 @@ int QuicHttpStream::DoLoop(int rv) { ...@@ -629,8 +629,8 @@ int QuicHttpStream::DoLoop(int rv) {
int QuicHttpStream::DoRequestStream() { int QuicHttpStream::DoRequestStream() {
next_state_ = STATE_REQUEST_STREAM_COMPLETE; next_state_ = STATE_REQUEST_STREAM_COMPLETE;
return stream_request_.StartRequest( stream_request_ = session_->CreateStreamRequest();
session_, &stream_, return stream_request_->StartRequest(
base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr())); base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
} }
...@@ -641,6 +641,8 @@ int QuicHttpStream::DoRequestStreamComplete(int rv) { ...@@ -641,6 +641,8 @@ int QuicHttpStream::DoRequestStreamComplete(int rv) {
return GetResponseStatus(); return GetResponseStatus();
} }
stream_ = stream_request_->ReleaseStream();
stream_request_.reset();
stream_->SetDelegate(this); stream_->SetDelegate(this);
if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) { if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) {
stream_->DisableConnectionMigration(); stream_->DisableConnectionMigration();
......
...@@ -159,7 +159,7 @@ class NET_EXPORT_PRIVATE QuicHttpStream ...@@ -159,7 +159,7 @@ class NET_EXPORT_PRIVATE QuicHttpStream
QuicVersion quic_version_; QuicVersion quic_version_;
int session_error_; // Error code from the connection shutdown. int session_error_; // Error code from the connection shutdown.
bool was_handshake_confirmed_; // True if the crypto handshake succeeded. bool was_handshake_confirmed_; // True if the crypto handshake succeeded.
QuicChromiumClientSession::StreamRequest stream_request_; std::unique_ptr<QuicChromiumClientSession::StreamRequest> stream_request_;
QuicChromiumClientStream* stream_; // Non-owning. QuicChromiumClientStream* stream_; // Non-owning.
// The following three fields are all owned by the caller and must // The following three fields are all owned by the caller and must
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment