bidirectional_stream_spdy_impl.cc 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409
  1. // Copyright 2015 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "net/spdy/bidirectional_stream_spdy_impl.h"
  5. #include <utility>
  6. #include "base/bind.h"
  7. #include "base/location.h"
  8. #include "base/logging.h"
  9. #include "base/threading/thread_task_runner_handle.h"
  10. #include "base/time/time.h"
  11. #include "base/timer/timer.h"
  12. #include "net/http/bidirectional_stream_request_info.h"
  13. #include "net/spdy/spdy_buffer.h"
  14. #include "net/spdy/spdy_http_utils.h"
  15. #include "net/spdy/spdy_stream.h"
  16. #include "net/third_party/quiche/src/quiche/spdy/core/http2_header_block.h"
  17. namespace net {
  18. namespace {
  19. // Time to wait in millisecond to notify |delegate_| of data received.
  20. // Handing small chunks of data to the caller creates measurable overhead.
  21. // So buffer data in short time-spans and send a single read notification.
  22. const int kBufferTimeMs = 1;
  23. } // namespace
  24. BidirectionalStreamSpdyImpl::BidirectionalStreamSpdyImpl(
  25. const base::WeakPtr<SpdySession>& spdy_session,
  26. NetLogSource source_dependency)
  27. : spdy_session_(spdy_session), source_dependency_(source_dependency) {}
  28. BidirectionalStreamSpdyImpl::~BidirectionalStreamSpdyImpl() {
  29. // Sends a RST to the remote if the stream is destroyed before it completes.
  30. ResetStream();
  31. }
  32. void BidirectionalStreamSpdyImpl::Start(
  33. const BidirectionalStreamRequestInfo* request_info,
  34. const NetLogWithSource& net_log,
  35. bool /*send_request_headers_automatically*/,
  36. BidirectionalStreamImpl::Delegate* delegate,
  37. std::unique_ptr<base::OneShotTimer> timer,
  38. const NetworkTrafficAnnotationTag& traffic_annotation) {
  39. DCHECK(!stream_);
  40. DCHECK(timer);
  41. delegate_ = delegate;
  42. timer_ = std::move(timer);
  43. if (!spdy_session_) {
  44. base::ThreadTaskRunnerHandle::Get()->PostTask(
  45. FROM_HERE,
  46. base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError,
  47. weak_factory_.GetWeakPtr(), ERR_CONNECTION_CLOSED));
  48. return;
  49. }
  50. request_info_ = request_info;
  51. int rv = stream_request_.StartRequest(
  52. SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url,
  53. false /* no early data */, request_info_->priority,
  54. request_info_->socket_tag, net_log,
  55. base::BindOnce(&BidirectionalStreamSpdyImpl::OnStreamInitialized,
  56. weak_factory_.GetWeakPtr()),
  57. traffic_annotation, request_info_->detect_broken_connection,
  58. request_info_->heartbeat_interval);
  59. if (rv != ERR_IO_PENDING)
  60. OnStreamInitialized(rv);
  61. }
  62. void BidirectionalStreamSpdyImpl::SendRequestHeaders() {
  63. // Request headers will be sent automatically.
  64. NOTREACHED();
  65. }
  66. int BidirectionalStreamSpdyImpl::ReadData(IOBuffer* buf, int buf_len) {
  67. if (stream_)
  68. DCHECK(!stream_->IsIdle());
  69. DCHECK(buf);
  70. DCHECK(buf_len);
  71. DCHECK(!timer_->IsRunning()) << "There should be only one ReadData in flight";
  72. // If there is data buffered, complete the IO immediately.
  73. if (!read_data_queue_.IsEmpty()) {
  74. return read_data_queue_.Dequeue(buf->data(), buf_len);
  75. } else if (stream_closed_) {
  76. return closed_stream_status_;
  77. }
  78. // Read will complete asynchronously and Delegate::OnReadCompleted will be
  79. // called upon completion.
  80. read_buffer_ = buf;
  81. read_buffer_len_ = buf_len;
  82. return ERR_IO_PENDING;
  83. }
  84. void BidirectionalStreamSpdyImpl::SendvData(
  85. const std::vector<scoped_refptr<IOBuffer>>& buffers,
  86. const std::vector<int>& lengths,
  87. bool end_stream) {
  88. DCHECK_EQ(buffers.size(), lengths.size());
  89. DCHECK(!write_pending_);
  90. if (written_end_of_stream_) {
  91. LOG(ERROR) << "Writing after end of stream is written.";
  92. base::ThreadTaskRunnerHandle::Get()->PostTask(
  93. FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError,
  94. weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
  95. return;
  96. }
  97. write_pending_ = true;
  98. written_end_of_stream_ = end_stream;
  99. if (MaybeHandleStreamClosedInSendData())
  100. return;
  101. DCHECK(!stream_closed_);
  102. int total_len = 0;
  103. for (int len : lengths) {
  104. total_len += len;
  105. }
  106. if (buffers.size() == 1) {
  107. pending_combined_buffer_ = buffers[0];
  108. } else {
  109. pending_combined_buffer_ = base::MakeRefCounted<net::IOBuffer>(total_len);
  110. int len = 0;
  111. // TODO(xunjieli): Get rid of extra copy. Coalesce headers and data frames.
  112. for (size_t i = 0; i < buffers.size(); ++i) {
  113. memcpy(pending_combined_buffer_->data() + len, buffers[i]->data(),
  114. lengths[i]);
  115. len += lengths[i];
  116. }
  117. }
  118. stream_->SendData(pending_combined_buffer_.get(), total_len,
  119. end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
  120. }
  121. NextProto BidirectionalStreamSpdyImpl::GetProtocol() const {
  122. return negotiated_protocol_;
  123. }
  124. int64_t BidirectionalStreamSpdyImpl::GetTotalReceivedBytes() const {
  125. if (stream_closed_)
  126. return closed_stream_received_bytes_;
  127. if (!stream_)
  128. return 0;
  129. return stream_->raw_received_bytes();
  130. }
  131. int64_t BidirectionalStreamSpdyImpl::GetTotalSentBytes() const {
  132. if (stream_closed_)
  133. return closed_stream_sent_bytes_;
  134. if (!stream_)
  135. return 0;
  136. return stream_->raw_sent_bytes();
  137. }
  138. bool BidirectionalStreamSpdyImpl::GetLoadTimingInfo(
  139. LoadTimingInfo* load_timing_info) const {
  140. if (stream_closed_) {
  141. if (!closed_has_load_timing_info_)
  142. return false;
  143. *load_timing_info = closed_load_timing_info_;
  144. return true;
  145. }
  146. // If |stream_| isn't created or has ID 0, return false. This is to match
  147. // the implementation in SpdyHttpStream.
  148. if (!stream_ || stream_->stream_id() == 0)
  149. return false;
  150. return stream_->GetLoadTimingInfo(load_timing_info);
  151. }
  152. void BidirectionalStreamSpdyImpl::PopulateNetErrorDetails(
  153. NetErrorDetails* details) {}
  154. void BidirectionalStreamSpdyImpl::OnHeadersSent() {
  155. DCHECK(stream_);
  156. negotiated_protocol_ = kProtoHTTP2;
  157. if (delegate_)
  158. delegate_->OnStreamReady(/*request_headers_sent=*/true);
  159. }
  160. void BidirectionalStreamSpdyImpl::OnEarlyHintsReceived(
  161. const spdy::Http2HeaderBlock& headers) {
  162. DCHECK(stream_);
  163. // TODO(crbug.com/671310): Plumb Early Hints to `delegate_` if needed.
  164. }
  165. void BidirectionalStreamSpdyImpl::OnHeadersReceived(
  166. const spdy::Http2HeaderBlock& response_headers,
  167. const spdy::Http2HeaderBlock* pushed_request_headers) {
  168. DCHECK(stream_);
  169. if (delegate_)
  170. delegate_->OnHeadersReceived(response_headers);
  171. }
  172. void BidirectionalStreamSpdyImpl::OnDataReceived(
  173. std::unique_ptr<SpdyBuffer> buffer) {
  174. DCHECK(stream_);
  175. DCHECK(!stream_closed_);
  176. // If |buffer| is null, BidirectionalStreamSpdyImpl::OnClose will be invoked
  177. // by SpdyStream to indicate the end of stream.
  178. if (!buffer)
  179. return;
  180. // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust
  181. // recv window size accordingly.
  182. read_data_queue_.Enqueue(std::move(buffer));
  183. if (read_buffer_) {
  184. // Handing small chunks of data to the caller creates measurable overhead.
  185. // So buffer data in short time-spans and send a single read notification.
  186. ScheduleBufferedRead();
  187. }
  188. }
  189. void BidirectionalStreamSpdyImpl::OnDataSent() {
  190. DCHECK(write_pending_);
  191. pending_combined_buffer_ = nullptr;
  192. write_pending_ = false;
  193. if (delegate_)
  194. delegate_->OnDataSent();
  195. }
  196. void BidirectionalStreamSpdyImpl::OnTrailers(
  197. const spdy::Http2HeaderBlock& trailers) {
  198. DCHECK(stream_);
  199. DCHECK(!stream_closed_);
  200. if (delegate_)
  201. delegate_->OnTrailersReceived(trailers);
  202. }
  203. void BidirectionalStreamSpdyImpl::OnClose(int status) {
  204. DCHECK(stream_);
  205. stream_closed_ = true;
  206. closed_stream_status_ = status;
  207. closed_stream_received_bytes_ = stream_->raw_received_bytes();
  208. closed_stream_sent_bytes_ = stream_->raw_sent_bytes();
  209. closed_has_load_timing_info_ =
  210. stream_->GetLoadTimingInfo(&closed_load_timing_info_);
  211. if (status != OK) {
  212. NotifyError(status);
  213. return;
  214. }
  215. ResetStream();
  216. // Complete any remaining read, as all data has been buffered.
  217. // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will
  218. // do nothing.
  219. timer_->Stop();
  220. // |this| might get destroyed after calling into |delegate_| in
  221. // DoBufferedRead().
  222. auto weak_this = weak_factory_.GetWeakPtr();
  223. DoBufferedRead();
  224. if (weak_this.get() && write_pending_)
  225. OnDataSent();
  226. }
  227. bool BidirectionalStreamSpdyImpl::CanGreaseFrameType() const {
  228. return false;
  229. }
  230. NetLogSource BidirectionalStreamSpdyImpl::source_dependency() const {
  231. return source_dependency_;
  232. }
  233. int BidirectionalStreamSpdyImpl::SendRequestHeadersHelper() {
  234. spdy::Http2HeaderBlock headers;
  235. HttpRequestInfo http_request_info;
  236. http_request_info.url = request_info_->url;
  237. http_request_info.method = request_info_->method;
  238. http_request_info.extra_headers = request_info_->extra_headers;
  239. CreateSpdyHeadersFromHttpRequest(http_request_info,
  240. http_request_info.extra_headers, &headers);
  241. written_end_of_stream_ = request_info_->end_stream_on_headers;
  242. return stream_->SendRequestHeaders(std::move(headers),
  243. request_info_->end_stream_on_headers
  244. ? NO_MORE_DATA_TO_SEND
  245. : MORE_DATA_TO_SEND);
  246. }
  247. void BidirectionalStreamSpdyImpl::OnStreamInitialized(int rv) {
  248. DCHECK_NE(ERR_IO_PENDING, rv);
  249. if (rv == OK) {
  250. stream_ = stream_request_.ReleaseStream();
  251. stream_->SetDelegate(this);
  252. rv = SendRequestHeadersHelper();
  253. if (rv == OK) {
  254. OnHeadersSent();
  255. return;
  256. } else if (rv == ERR_IO_PENDING) {
  257. return;
  258. }
  259. }
  260. NotifyError(rv);
  261. }
  262. void BidirectionalStreamSpdyImpl::NotifyError(int rv) {
  263. ResetStream();
  264. write_pending_ = false;
  265. if (delegate_) {
  266. BidirectionalStreamImpl::Delegate* delegate = delegate_;
  267. delegate_ = nullptr;
  268. // Cancel any pending callback.
  269. weak_factory_.InvalidateWeakPtrs();
  270. delegate->OnFailed(rv);
  271. // |this| can be null when returned from delegate.
  272. }
  273. }
  274. void BidirectionalStreamSpdyImpl::ResetStream() {
  275. if (!stream_)
  276. return;
  277. if (!stream_->IsClosed()) {
  278. // This sends a RST to the remote.
  279. stream_->DetachDelegate();
  280. DCHECK(!stream_);
  281. } else {
  282. // Stream is already closed, so it is not legal to call DetachDelegate.
  283. stream_.reset();
  284. }
  285. }
  286. void BidirectionalStreamSpdyImpl::ScheduleBufferedRead() {
  287. // If there is already a scheduled DoBufferedRead, don't issue
  288. // another one. Mark that we have received more data and return.
  289. if (timer_->IsRunning()) {
  290. more_read_data_pending_ = true;
  291. return;
  292. }
  293. more_read_data_pending_ = false;
  294. timer_->Start(FROM_HERE, base::Milliseconds(kBufferTimeMs),
  295. base::BindOnce(&BidirectionalStreamSpdyImpl::DoBufferedRead,
  296. weak_factory_.GetWeakPtr()));
  297. }
  298. void BidirectionalStreamSpdyImpl::DoBufferedRead() {
  299. DCHECK(!timer_->IsRunning());
  300. // Check to see that the stream has not errored out.
  301. DCHECK(stream_ || stream_closed_);
  302. DCHECK(!stream_closed_ || closed_stream_status_ == OK);
  303. // When |more_read_data_pending_| is true, it means that more data has arrived
  304. // since started waiting. Wait a little longer and continue to buffer.
  305. if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) {
  306. ScheduleBufferedRead();
  307. return;
  308. }
  309. int rv = 0;
  310. if (read_buffer_) {
  311. rv = ReadData(read_buffer_.get(), read_buffer_len_);
  312. DCHECK_NE(ERR_IO_PENDING, rv);
  313. read_buffer_ = nullptr;
  314. read_buffer_len_ = 0;
  315. if (delegate_)
  316. delegate_->OnDataRead(rv);
  317. }
  318. }
  319. bool BidirectionalStreamSpdyImpl::ShouldWaitForMoreBufferedData() const {
  320. if (stream_closed_)
  321. return false;
  322. DCHECK_GT(read_buffer_len_, 0);
  323. return read_data_queue_.GetTotalSize() <
  324. static_cast<size_t>(read_buffer_len_);
  325. }
  326. bool BidirectionalStreamSpdyImpl::MaybeHandleStreamClosedInSendData() {
  327. if (stream_)
  328. return false;
  329. // If |stream_| is closed without an error before client half closes,
  330. // blackhole any pending write data. crbug.com/650438.
  331. if (stream_closed_ && closed_stream_status_ == OK) {
  332. base::ThreadTaskRunnerHandle::Get()->PostTask(
  333. FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::OnDataSent,
  334. weak_factory_.GetWeakPtr()));
  335. return true;
  336. }
  337. LOG(ERROR) << "Trying to send data after stream has been destroyed.";
  338. base::ThreadTaskRunnerHandle::Get()->PostTask(
  339. FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError,
  340. weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
  341. return true;
  342. }
  343. } // namespace net