xref: /aosp_15_r20/external/cronet/net/spdy/spdy_stream.cc (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b)
1 // Copyright 2012 The Chromium Authors
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4 
5 #include "net/spdy/spdy_stream.h"
6 
7 #include <algorithm>
8 #include <limits>
9 #include <string_view>
10 #include <utility>
11 
12 #include "base/check_op.h"
13 #include "base/compiler_specific.h"
14 #include "base/functional/bind.h"
15 #include "base/location.h"
16 #include "base/metrics/histogram_functions.h"
17 #include "base/metrics/histogram_macros.h"
18 #include "base/notreached.h"
19 #include "base/strings/string_number_conversions.h"
20 #include "base/strings/stringprintf.h"
21 #include "base/task/single_thread_task_runner.h"
22 #include "base/trace_event/memory_usage_estimator.h"
23 #include "base/values.h"
24 #include "net/base/load_timing_info.h"
25 #include "net/http/http_status_code.h"
26 #include "net/log/net_log.h"
27 #include "net/log/net_log_capture_mode.h"
28 #include "net/log/net_log_event_type.h"
29 #include "net/spdy/spdy_buffer_producer.h"
30 #include "net/spdy/spdy_http_utils.h"
31 #include "net/spdy/spdy_log_util.h"
32 #include "net/spdy/spdy_session.h"
33 
34 namespace net {
35 
36 namespace {
37 
NetLogSpdyStreamErrorParams(spdy::SpdyStreamId stream_id,int net_error,std::string_view description)38 base::Value::Dict NetLogSpdyStreamErrorParams(spdy::SpdyStreamId stream_id,
39                                               int net_error,
40                                               std::string_view description) {
41   return base::Value::Dict()
42       .Set("stream_id", static_cast<int>(stream_id))
43       .Set("net_error", ErrorToShortString(net_error))
44       .Set("description", description);
45 }
46 
NetLogSpdyStreamWindowUpdateParams(spdy::SpdyStreamId stream_id,int32_t delta,int32_t window_size)47 base::Value::Dict NetLogSpdyStreamWindowUpdateParams(
48     spdy::SpdyStreamId stream_id,
49     int32_t delta,
50     int32_t window_size) {
51   return base::Value::Dict()
52       .Set("stream_id", static_cast<int>(stream_id))
53       .Set("delta", delta)
54       .Set("window_size", window_size);
55 }
56 
NetLogSpdyDataParams(spdy::SpdyStreamId stream_id,int size,bool fin)57 base::Value::Dict NetLogSpdyDataParams(spdy::SpdyStreamId stream_id,
58                                        int size,
59                                        bool fin) {
60   return base::Value::Dict()
61       .Set("stream_id", static_cast<int>(stream_id))
62       .Set("size", size)
63       .Set("fin", fin);
64 }
65 
66 }  // namespace
67 
68 // A wrapper around a stream that calls into ProduceHeadersFrame().
69 class SpdyStream::HeadersBufferProducer : public SpdyBufferProducer {
70  public:
HeadersBufferProducer(const base::WeakPtr<SpdyStream> & stream)71   explicit HeadersBufferProducer(const base::WeakPtr<SpdyStream>& stream)
72       : stream_(stream) {
73     DCHECK(stream_.get());
74   }
75 
76   ~HeadersBufferProducer() override = default;
77 
ProduceBuffer()78   std::unique_ptr<SpdyBuffer> ProduceBuffer() override {
79     if (!stream_.get()) {
80       NOTREACHED();
81       return nullptr;
82     }
83     DCHECK_GT(stream_->stream_id(), 0u);
84     return std::make_unique<SpdyBuffer>(stream_->ProduceHeadersFrame());
85   }
86 
87  private:
88   const base::WeakPtr<SpdyStream> stream_;
89 };
90 
SpdyStream(SpdyStreamType type,const base::WeakPtr<SpdySession> & session,const GURL & url,RequestPriority priority,int32_t initial_send_window_size,int32_t max_recv_window_size,const NetLogWithSource & net_log,const NetworkTrafficAnnotationTag & traffic_annotation,bool detect_broken_connection)91 SpdyStream::SpdyStream(SpdyStreamType type,
92                        const base::WeakPtr<SpdySession>& session,
93                        const GURL& url,
94                        RequestPriority priority,
95                        int32_t initial_send_window_size,
96                        int32_t max_recv_window_size,
97                        const NetLogWithSource& net_log,
98                        const NetworkTrafficAnnotationTag& traffic_annotation,
99                        bool detect_broken_connection)
100     : type_(type),
101       url_(url),
102       priority_(priority),
103       send_window_size_(initial_send_window_size),
104       max_recv_window_size_(max_recv_window_size),
105       recv_window_size_(max_recv_window_size),
106       last_recv_window_update_(base::TimeTicks::Now()),
107       session_(session),
108       request_time_(base::Time::Now()),
109       net_log_(net_log),
110       traffic_annotation_(traffic_annotation),
111       detect_broken_connection_(detect_broken_connection) {
112   CHECK(type_ == SPDY_BIDIRECTIONAL_STREAM ||
113         type_ == SPDY_REQUEST_RESPONSE_STREAM);
114   CHECK_GE(priority_, MINIMUM_PRIORITY);
115   CHECK_LE(priority_, MAXIMUM_PRIORITY);
116 }
117 
~SpdyStream()118 SpdyStream::~SpdyStream() {
119   CHECK(!write_handler_guard_);
120 }
121 
SetDelegate(Delegate * delegate)122 void SpdyStream::SetDelegate(Delegate* delegate) {
123   CHECK(!delegate_);
124   CHECK(delegate);
125   delegate_ = delegate;
126 
127   CHECK(io_state_ == STATE_IDLE || io_state_ == STATE_RESERVED_REMOTE);
128 }
129 
ProduceHeadersFrame()130 std::unique_ptr<spdy::SpdySerializedFrame> SpdyStream::ProduceHeadersFrame() {
131   CHECK_EQ(io_state_, STATE_IDLE);
132   CHECK(request_headers_valid_);
133   CHECK_GT(stream_id_, 0u);
134 
135   spdy::SpdyControlFlags flags = (pending_send_status_ == NO_MORE_DATA_TO_SEND)
136                                      ? spdy::CONTROL_FLAG_FIN
137                                      : spdy::CONTROL_FLAG_NONE;
138   std::unique_ptr<spdy::SpdySerializedFrame> frame(session_->CreateHeaders(
139       stream_id_, priority_, flags, std::move(request_headers_),
140       delegate_->source_dependency()));
141   request_headers_valid_ = false;
142   send_time_ = base::TimeTicks::Now();
143   return frame;
144 }
145 
DetachDelegate()146 void SpdyStream::DetachDelegate() {
147   DCHECK(!IsClosed());
148   delegate_ = nullptr;
149   Cancel(ERR_ABORTED);
150 }
151 
SetPriority(RequestPriority priority)152 void SpdyStream::SetPriority(RequestPriority priority) {
153   if (priority_ == priority) {
154     return;
155   }
156 
157   session_->UpdateStreamPriority(this, /* old_priority = */ priority_,
158                                  /* new_priority = */ priority);
159 
160   priority_ = priority;
161 }
162 
AdjustSendWindowSize(int32_t delta_window_size)163 bool SpdyStream::AdjustSendWindowSize(int32_t delta_window_size) {
164   if (IsClosed())
165     return true;
166 
167   if (delta_window_size > 0) {
168     if (send_window_size_ >
169         std::numeric_limits<int32_t>::max() - delta_window_size) {
170       return false;
171     }
172   } else {
173     // Minimum allowed value for spdy::SETTINGS_INITIAL_WINDOW_SIZE is 0 and
174     // maximum is 2^31-1.  Data are not sent when |send_window_size_ < 0|, that
175     // is, |send_window_size_ | can only decrease by a change in
176     // spdy::SETTINGS_INITIAL_WINDOW_SIZE.  Therefore |send_window_size_| should
177     // never be able to become less than -(2^31-1).
178     DCHECK_LE(std::numeric_limits<int32_t>::min() - delta_window_size,
179               send_window_size_);
180   }
181 
182   send_window_size_ += delta_window_size;
183 
184   net_log_.AddEvent(NetLogEventType::HTTP2_STREAM_UPDATE_SEND_WINDOW, [&] {
185     return NetLogSpdyStreamWindowUpdateParams(stream_id_, delta_window_size,
186                                               send_window_size_);
187   });
188 
189   PossiblyResumeIfSendStalled();
190   return true;
191 }
192 
OnWriteBufferConsumed(size_t frame_payload_size,size_t consume_size,SpdyBuffer::ConsumeSource consume_source)193 void SpdyStream::OnWriteBufferConsumed(
194     size_t frame_payload_size,
195     size_t consume_size,
196     SpdyBuffer::ConsumeSource consume_source) {
197   if (consume_source == SpdyBuffer::DISCARD) {
198     // If we're discarding a frame or part of it, increase the send
199     // window by the number of discarded bytes. (Although if we're
200     // discarding part of a frame, it's probably because of a write
201     // error and we'll be tearing down the stream soon.)
202     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
203     DCHECK_GT(remaining_payload_bytes, 0u);
204     IncreaseSendWindowSize(static_cast<int32_t>(remaining_payload_bytes));
205   }
206   // For consumed bytes, the send window is increased when we receive
207   // a WINDOW_UPDATE frame.
208 }
209 
IncreaseSendWindowSize(int32_t delta_window_size)210 void SpdyStream::IncreaseSendWindowSize(int32_t delta_window_size) {
211   DCHECK_GE(delta_window_size, 1);
212 
213   if (!AdjustSendWindowSize(delta_window_size)) {
214     std::string desc = base::StringPrintf(
215         "Received WINDOW_UPDATE [delta: %d] for stream %d overflows "
216         "send_window_size_ [current: %d]",
217         delta_window_size, stream_id_, send_window_size_);
218     session_->ResetStream(stream_id_, ERR_HTTP2_FLOW_CONTROL_ERROR, desc);
219   }
220 }
221 
DecreaseSendWindowSize(int32_t delta_window_size)222 void SpdyStream::DecreaseSendWindowSize(int32_t delta_window_size) {
223   if (IsClosed())
224     return;
225 
226   // We only call this method when sending a frame. Therefore,
227   // |delta_window_size| should be within the valid frame size range.
228   DCHECK_GE(delta_window_size, 1);
229   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
230 
231   // |send_window_size_| should have been at least |delta_window_size| for
232   // this call to happen.
233   DCHECK_GE(send_window_size_, delta_window_size);
234 
235   send_window_size_ -= delta_window_size;
236 
237   net_log_.AddEvent(NetLogEventType::HTTP2_STREAM_UPDATE_SEND_WINDOW, [&] {
238     return NetLogSpdyStreamWindowUpdateParams(stream_id_, -delta_window_size,
239                                               send_window_size_);
240   });
241 }
242 
OnReadBufferConsumed(size_t consume_size,SpdyBuffer::ConsumeSource consume_source)243 void SpdyStream::OnReadBufferConsumed(
244     size_t consume_size,
245     SpdyBuffer::ConsumeSource consume_source) {
246   DCHECK_GE(consume_size, 1u);
247   DCHECK_LE(consume_size,
248             static_cast<size_t>(std::numeric_limits<int32_t>::max()));
249   IncreaseRecvWindowSize(static_cast<int32_t>(consume_size));
250 }
251 
IncreaseRecvWindowSize(int32_t delta_window_size)252 void SpdyStream::IncreaseRecvWindowSize(int32_t delta_window_size) {
253   // By the time a read is processed by the delegate, this stream may
254   // already be inactive.
255   if (!session_->IsStreamActive(stream_id_))
256     return;
257 
258   DCHECK_GE(unacked_recv_window_bytes_, 0);
259   DCHECK_GE(recv_window_size_, unacked_recv_window_bytes_);
260   DCHECK_GE(delta_window_size, 1);
261   // Check for overflow.
262   DCHECK_LE(delta_window_size,
263             std::numeric_limits<int32_t>::max() - recv_window_size_);
264 
265   recv_window_size_ += delta_window_size;
266   net_log_.AddEvent(NetLogEventType::HTTP2_STREAM_UPDATE_RECV_WINDOW, [&] {
267     return NetLogSpdyStreamWindowUpdateParams(stream_id_, delta_window_size,
268                                               recv_window_size_);
269   });
270 
271   // Update the receive window once half of the buffer is ready to be acked
272   // to prevent excessive window updates on fast downloads. Also send an update
273   // if too much time has elapsed since the last update to deal with
274   // slow-reading clients so the server doesn't think the stream is idle.
275   unacked_recv_window_bytes_ += delta_window_size;
276   const base::TimeDelta elapsed =
277       base::TimeTicks::Now() - last_recv_window_update_;
278   if (unacked_recv_window_bytes_ > max_recv_window_size_ / 2 ||
279       elapsed >= session_->TimeToBufferSmallWindowUpdates()) {
280     last_recv_window_update_ = base::TimeTicks::Now();
281     session_->SendStreamWindowUpdate(
282         stream_id_, static_cast<uint32_t>(unacked_recv_window_bytes_));
283     unacked_recv_window_bytes_ = 0;
284   }
285 }
286 
DecreaseRecvWindowSize(int32_t delta_window_size)287 void SpdyStream::DecreaseRecvWindowSize(int32_t delta_window_size) {
288   DCHECK(session_->IsStreamActive(stream_id_));
289   DCHECK_GE(delta_window_size, 1);
290 
291   // The receiving window size as the peer knows it is
292   // |recv_window_size_ - unacked_recv_window_bytes_|, if more data are sent by
293   // the peer, that means that the receive window is not being respected.
294   if (delta_window_size > recv_window_size_ - unacked_recv_window_bytes_) {
295     session_->ResetStream(
296         stream_id_, ERR_HTTP2_FLOW_CONTROL_ERROR,
297         "delta_window_size is " + base::NumberToString(delta_window_size) +
298             " in DecreaseRecvWindowSize, which is larger than the receive " +
299             "window size of " + base::NumberToString(recv_window_size_));
300     return;
301   }
302 
303   recv_window_size_ -= delta_window_size;
304   net_log_.AddEvent(NetLogEventType::HTTP2_STREAM_UPDATE_RECV_WINDOW, [&] {
305     return NetLogSpdyStreamWindowUpdateParams(stream_id_, -delta_window_size,
306                                               recv_window_size_);
307   });
308 }
309 
GetPeerAddress(IPEndPoint * address) const310 int SpdyStream::GetPeerAddress(IPEndPoint* address) const {
311   return session_->GetPeerAddress(address);
312 }
313 
GetLocalAddress(IPEndPoint * address) const314 int SpdyStream::GetLocalAddress(IPEndPoint* address) const {
315   return session_->GetLocalAddress(address);
316 }
317 
WasEverUsed() const318 bool SpdyStream::WasEverUsed() const {
319   return session_->WasEverUsed();
320 }
321 
GetRequestTime() const322 base::Time SpdyStream::GetRequestTime() const {
323   return request_time_;
324 }
325 
SetRequestTime(base::Time t)326 void SpdyStream::SetRequestTime(base::Time t) {
327   request_time_ = t;
328 }
329 
OnHeadersReceived(const spdy::Http2HeaderBlock & response_headers,base::Time response_time,base::TimeTicks recv_first_byte_time)330 void SpdyStream::OnHeadersReceived(
331     const spdy::Http2HeaderBlock& response_headers,
332     base::Time response_time,
333     base::TimeTicks recv_first_byte_time) {
334   switch (response_state_) {
335     case READY_FOR_HEADERS: {
336       // No header block has been received yet.
337       DCHECK(response_headers_.empty());
338 
339       spdy::Http2HeaderBlock::const_iterator it =
340           response_headers.find(spdy::kHttp2StatusHeader);
341       if (it == response_headers.end()) {
342         const std::string error("Response headers do not include :status.");
343         LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
344         session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
345         return;
346       }
347 
348       int status;
349       if (!base::StringToInt(it->second, &status)) {
350         const std::string error("Cannot parse :status.");
351         LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
352         session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
353         return;
354       }
355 
356       base::UmaHistogramSparse("Net.SpdyResponseCode", status);
357 
358       // Include informational responses (1xx) in the TTFB as per the resource
359       // timing spec for responseStart.
360       if (recv_first_byte_time_.is_null())
361         recv_first_byte_time_ = recv_first_byte_time;
362       // Also record the TTFB of non-informational responses.
363       if (status / 100 != 1) {
364         DCHECK(recv_first_byte_time_for_non_informational_response_.is_null());
365         recv_first_byte_time_for_non_informational_response_ =
366             recv_first_byte_time;
367       }
368 
369       // Handle informational responses (1xx):
370       // * Pass through 101 Switching Protocols, because broken servers might
371       //   send this as a response to a WebSocket request, in which case it
372       //   needs to pass through so that the WebSocket layer can signal an
373       //   error.
374       // * Plumb 103 Early Hints to the delegate.
375       // * Ignore other informational responses.
376       if (status / 100 == 1 && status != HTTP_SWITCHING_PROTOCOLS) {
377         if (status == HTTP_EARLY_HINTS)
378           OnEarlyHintsReceived(response_headers, recv_first_byte_time);
379         return;
380       }
381 
382       response_state_ = READY_FOR_DATA_OR_TRAILERS;
383 
384       switch (type_) {
385         case SPDY_BIDIRECTIONAL_STREAM:
386         case SPDY_REQUEST_RESPONSE_STREAM:
387           // A bidirectional stream or a request/response stream is ready for
388           // the response headers only after request headers are sent.
389           if (io_state_ == STATE_IDLE) {
390             const std::string error("Response received before request sent.");
391             LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
392             session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
393             return;
394           }
395           break;
396       }
397 
398       DCHECK_NE(io_state_, STATE_IDLE);
399 
400       response_time_ = response_time;
401       SaveResponseHeaders(response_headers, status);
402 
403       break;
404     }
405     case READY_FOR_DATA_OR_TRAILERS:
406       // Second header block is trailers.
407       response_state_ = TRAILERS_RECEIVED;
408       delegate_->OnTrailers(response_headers);
409       break;
410 
411     case TRAILERS_RECEIVED:
412       // No further header blocks are allowed after trailers.
413       const std::string error("Header block received after trailers.");
414       LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
415       session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
416       break;
417   }
418 }
419 
OnDataReceived(std::unique_ptr<SpdyBuffer> buffer)420 void SpdyStream::OnDataReceived(std::unique_ptr<SpdyBuffer> buffer) {
421   DCHECK(session_->IsStreamActive(stream_id_));
422 
423   if (response_state_ == READY_FOR_HEADERS) {
424     const std::string error("DATA received before headers.");
425     LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
426     session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
427     return;
428   }
429 
430   if (response_state_ == TRAILERS_RECEIVED && buffer) {
431     const std::string error("DATA received after trailers.");
432     LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
433     session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
434     return;
435   }
436 
437   if (io_state_ == STATE_HALF_CLOSED_REMOTE) {
438     const std::string error("DATA received on half-closed (remove) stream.");
439     LogStreamError(ERR_HTTP2_STREAM_CLOSED, error);
440     session_->ResetStream(stream_id_, ERR_HTTP2_STREAM_CLOSED, error);
441     return;
442   }
443 
444   // Track our bandwidth.
445   recv_bytes_ += buffer ? buffer->GetRemainingSize() : 0;
446   recv_last_byte_time_ = base::TimeTicks::Now();
447 
448   CHECK(!IsClosed());
449 
450   if (!buffer) {
451     if (io_state_ == STATE_OPEN) {
452       io_state_ = STATE_HALF_CLOSED_REMOTE;
453       // Inform the delegate of EOF. This may delete |this|.
454       delegate_->OnDataReceived(nullptr);
455     } else if (io_state_ == STATE_HALF_CLOSED_LOCAL) {
456       io_state_ = STATE_CLOSED;
457       // Deletes |this|.
458       session_->CloseActiveStream(stream_id_, OK);
459     } else {
460       NOTREACHED() << io_state_;
461     }
462     return;
463   }
464 
465   size_t length = buffer->GetRemainingSize();
466   DCHECK_LE(length, spdy::kHttp2DefaultFramePayloadLimit);
467   base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
468   // May close the stream.
469   DecreaseRecvWindowSize(static_cast<int32_t>(length));
470   if (!weak_this)
471     return;
472   buffer->AddConsumeCallback(
473       base::BindRepeating(&SpdyStream::OnReadBufferConsumed, GetWeakPtr()));
474 
475   // May close |this|.
476   delegate_->OnDataReceived(std::move(buffer));
477 }
478 
OnPaddingConsumed(size_t len)479 void SpdyStream::OnPaddingConsumed(size_t len) {
480   // Decrease window size because padding bytes are received.
481   // Increase window size because padding bytes are consumed (by discarding).
482   // Net result: |unacked_recv_window_bytes_| increases by |len|,
483   // |recv_window_size_| does not change.
484   base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
485   // May close the stream.
486   DecreaseRecvWindowSize(static_cast<int32_t>(len));
487   if (!weak_this)
488     return;
489   IncreaseRecvWindowSize(static_cast<int32_t>(len));
490 }
491 
OnFrameWriteComplete(spdy::SpdyFrameType frame_type,size_t frame_size)492 void SpdyStream::OnFrameWriteComplete(spdy::SpdyFrameType frame_type,
493                                       size_t frame_size) {
494   if (frame_type != spdy::SpdyFrameType::HEADERS &&
495       frame_type != spdy::SpdyFrameType::DATA) {
496     return;
497   }
498 
499   int result = (frame_type == spdy::SpdyFrameType::HEADERS)
500                    ? OnHeadersSent()
501                    : OnDataSent(frame_size);
502   if (result == ERR_IO_PENDING) {
503     // The write operation hasn't completed yet.
504     return;
505   }
506 
507   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
508     if (io_state_ == STATE_OPEN) {
509       io_state_ = STATE_HALF_CLOSED_LOCAL;
510     } else if (io_state_ == STATE_HALF_CLOSED_REMOTE) {
511       io_state_ = STATE_CLOSED;
512     } else {
513       NOTREACHED() << io_state_;
514     }
515   }
516   // Notify delegate of write completion. Must not destroy |this|.
517   CHECK(delegate_);
518   {
519     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
520     write_handler_guard_ = true;
521     if (frame_type == spdy::SpdyFrameType::HEADERS) {
522       delegate_->OnHeadersSent();
523     } else {
524       delegate_->OnDataSent();
525     }
526     CHECK(weak_this);
527     write_handler_guard_ = false;
528   }
529 
530   if (io_state_ == STATE_CLOSED) {
531     // Deletes |this|.
532     session_->CloseActiveStream(stream_id_, OK);
533   }
534 }
535 
OnHeadersSent()536 int SpdyStream::OnHeadersSent() {
537   CHECK_EQ(io_state_, STATE_IDLE);
538   CHECK_NE(stream_id_, 0u);
539 
540   io_state_ = STATE_OPEN;
541   return OK;
542 }
543 
OnDataSent(size_t frame_size)544 int SpdyStream::OnDataSent(size_t frame_size) {
545   CHECK(io_state_ == STATE_OPEN ||
546         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
547 
548   size_t frame_payload_size = frame_size - spdy::kDataFrameMinimumSize;
549 
550   CHECK_GE(frame_size, spdy::kDataFrameMinimumSize);
551   CHECK_LE(frame_payload_size, spdy::kHttp2DefaultFramePayloadLimit);
552 
553   // If more data is available to send, dispatch it and
554   // return that the write operation is still ongoing.
555   pending_send_data_->DidConsume(frame_payload_size);
556   if (pending_send_data_->BytesRemaining() > 0) {
557     QueueNextDataFrame();
558     return ERR_IO_PENDING;
559   } else {
560     pending_send_data_ = nullptr;
561     return OK;
562   }
563 }
564 
LogStreamError(int error,std::string_view description)565 void SpdyStream::LogStreamError(int error, std::string_view description) {
566   net_log_.AddEvent(NetLogEventType::HTTP2_STREAM_ERROR, [&] {
567     return NetLogSpdyStreamErrorParams(stream_id_, error, description);
568   });
569 }
570 
OnClose(int status)571 void SpdyStream::OnClose(int status) {
572   // In most cases, the stream should already be CLOSED. The exception is when a
573   // SpdySession is shutting down while the stream is in an intermediate state.
574   io_state_ = STATE_CLOSED;
575   if (status == ERR_HTTP2_RST_STREAM_NO_ERROR_RECEIVED) {
576     if (response_state_ == READY_FOR_HEADERS) {
577       status = ERR_HTTP2_PROTOCOL_ERROR;
578     } else {
579       status = OK;
580     }
581   }
582   Delegate* delegate = delegate_;
583   delegate_ = nullptr;
584   if (delegate)
585     delegate->OnClose(status);
586   // Unset |stream_id_| last so that the delegate can look it up.
587   stream_id_ = 0;
588 }
589 
Cancel(int error)590 void SpdyStream::Cancel(int error) {
591   // We may be called again from a delegate's OnClose().
592   if (io_state_ == STATE_CLOSED)
593     return;
594 
595   if (stream_id_ != 0) {
596     session_->ResetStream(stream_id_, error, std::string());
597   } else {
598     session_->CloseCreatedStream(GetWeakPtr(), error);
599   }
600   // |this| is invalid at this point.
601 }
602 
Close()603 void SpdyStream::Close() {
604   // We may be called again from a delegate's OnClose().
605   if (io_state_ == STATE_CLOSED)
606     return;
607 
608   if (stream_id_ != 0) {
609     session_->CloseActiveStream(stream_id_, OK);
610   } else {
611     session_->CloseCreatedStream(GetWeakPtr(), OK);
612   }
613   // |this| is invalid at this point.
614 }
615 
GetWeakPtr()616 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
617   return weak_ptr_factory_.GetWeakPtr();
618 }
619 
SendRequestHeaders(spdy::Http2HeaderBlock request_headers,SpdySendStatus send_status)620 int SpdyStream::SendRequestHeaders(spdy::Http2HeaderBlock request_headers,
621                                    SpdySendStatus send_status) {
622   net_log_.AddEvent(
623       NetLogEventType::HTTP_TRANSACTION_HTTP2_SEND_REQUEST_HEADERS,
624       [&](NetLogCaptureMode capture_mode) {
625         return Http2HeaderBlockNetLogParams(&request_headers, capture_mode);
626       });
627   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
628   CHECK(!request_headers_valid_);
629   CHECK(!pending_send_data_.get());
630   CHECK_EQ(io_state_, STATE_IDLE);
631   request_headers_ = std::move(request_headers);
632   request_headers_valid_ = true;
633   pending_send_status_ = send_status;
634   session_->EnqueueStreamWrite(
635       GetWeakPtr(), spdy::SpdyFrameType::HEADERS,
636       std::make_unique<HeadersBufferProducer>(GetWeakPtr()));
637   return ERR_IO_PENDING;
638 }
639 
SendData(IOBuffer * data,int length,SpdySendStatus send_status)640 void SpdyStream::SendData(IOBuffer* data,
641                           int length,
642                           SpdySendStatus send_status) {
643   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
644   CHECK(io_state_ == STATE_OPEN ||
645         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
646   CHECK(!pending_send_data_.get());
647   pending_send_data_ = base::MakeRefCounted<DrainableIOBuffer>(data, length);
648   pending_send_status_ = send_status;
649   QueueNextDataFrame();
650 }
651 
GetSSLInfo(SSLInfo * ssl_info) const652 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info) const {
653   return session_->GetSSLInfo(ssl_info);
654 }
655 
GetNegotiatedProtocol() const656 NextProto SpdyStream::GetNegotiatedProtocol() const {
657   return session_->GetNegotiatedProtocol();
658 }
659 
PossiblyResumeIfSendStalled()660 SpdyStream::ShouldRequeueStream SpdyStream::PossiblyResumeIfSendStalled() {
661   if (IsLocallyClosed() || !send_stalled_by_flow_control_)
662     return DoNotRequeue;
663   if (session_->IsSendStalled() || send_window_size_ <= 0) {
664     return Requeue;
665   }
666   net_log_.AddEventWithIntParams(
667       NetLogEventType::HTTP2_STREAM_FLOW_CONTROL_UNSTALLED, "stream_id",
668       stream_id_);
669   send_stalled_by_flow_control_ = false;
670   QueueNextDataFrame();
671   return DoNotRequeue;
672 }
673 
IsClosed() const674 bool SpdyStream::IsClosed() const {
675   return io_state_ == STATE_CLOSED;
676 }
677 
IsLocallyClosed() const678 bool SpdyStream::IsLocallyClosed() const {
679   return io_state_ == STATE_HALF_CLOSED_LOCAL || io_state_ == STATE_CLOSED;
680 }
681 
IsIdle() const682 bool SpdyStream::IsIdle() const {
683   return io_state_ == STATE_IDLE;
684 }
685 
IsOpen() const686 bool SpdyStream::IsOpen() const {
687   return io_state_ == STATE_OPEN;
688 }
689 
IsReservedRemote() const690 bool SpdyStream::IsReservedRemote() const {
691   return io_state_ == STATE_RESERVED_REMOTE;
692 }
693 
AddRawReceivedBytes(size_t received_bytes)694 void SpdyStream::AddRawReceivedBytes(size_t received_bytes) {
695   raw_received_bytes_ += received_bytes;
696 }
697 
AddRawSentBytes(size_t sent_bytes)698 void SpdyStream::AddRawSentBytes(size_t sent_bytes) {
699   raw_sent_bytes_ += sent_bytes;
700 }
701 
GetLoadTimingInfo(LoadTimingInfo * load_timing_info) const702 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
703   if (stream_id_ == 0)
704     return false;
705   bool result = session_->GetLoadTimingInfo(stream_id_, load_timing_info);
706   // TODO(acomminos): recv_first_byte_time_ is actually the time after all
707   // headers have been parsed. We should add support for reporting the time the
708   // first bytes of the HEADERS frame were received to BufferedSpdyFramer
709   // (https://crbug.com/568024).
710   load_timing_info->receive_headers_start = recv_first_byte_time_;
711   load_timing_info->receive_non_informational_headers_start =
712       recv_first_byte_time_for_non_informational_response_;
713   load_timing_info->first_early_hints_time = first_early_hints_time_;
714   return result;
715 }
716 
QueueNextDataFrame()717 void SpdyStream::QueueNextDataFrame() {
718   // Until the request has been completely sent, we cannot be sure
719   // that our stream_id is correct.
720   CHECK(io_state_ == STATE_OPEN ||
721         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
722   CHECK_GT(stream_id_, 0u);
723   CHECK(pending_send_data_.get());
724   // Only the final fame may have a length of 0.
725   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
726     CHECK_GE(pending_send_data_->BytesRemaining(), 0);
727   } else {
728     CHECK_GT(pending_send_data_->BytesRemaining(), 0);
729   }
730 
731   spdy::SpdyDataFlags flags = (pending_send_status_ == NO_MORE_DATA_TO_SEND)
732                                   ? spdy::DATA_FLAG_FIN
733                                   : spdy::DATA_FLAG_NONE;
734   int effective_len;
735   bool end_stream;
736   std::unique_ptr<SpdyBuffer> data_buffer(
737       session_->CreateDataBuffer(stream_id_, pending_send_data_.get(),
738                                  pending_send_data_->BytesRemaining(), flags,
739                                  &effective_len, &end_stream));
740   // We'll get called again by PossiblyResumeIfSendStalled().
741   if (!data_buffer)
742     return;
743 
744   DCHECK_GE(data_buffer->GetRemainingSize(), spdy::kDataFrameMinimumSize);
745   size_t payload_size =
746       data_buffer->GetRemainingSize() - spdy::kDataFrameMinimumSize;
747   DCHECK_LE(payload_size, spdy::kHttp2DefaultFramePayloadLimit);
748 
749   // Send window size is based on payload size, so nothing to do if this is
750   // just a FIN with no payload.
751   if (payload_size != 0) {
752     DecreaseSendWindowSize(static_cast<int32_t>(payload_size));
753     // This currently isn't strictly needed, since write frames are
754     // discarded only if the stream is about to be closed. But have it
755     // here anyway just in case this changes.
756     data_buffer->AddConsumeCallback(base::BindRepeating(
757         &SpdyStream::OnWriteBufferConsumed, GetWeakPtr(), payload_size));
758   }
759 
760   if (session_->GreasedFramesEnabled() && delegate_ &&
761       delegate_->CanGreaseFrameType()) {
762     session_->EnqueueGreasedFrame(GetWeakPtr());
763   }
764 
765   session_->net_log().AddEvent(NetLogEventType::HTTP2_SESSION_SEND_DATA, [&] {
766     return NetLogSpdyDataParams(stream_id_, effective_len, end_stream);
767   });
768 
769   session_->EnqueueStreamWrite(
770       GetWeakPtr(), spdy::SpdyFrameType::DATA,
771       std::make_unique<SimpleBufferProducer>(std::move(data_buffer)));
772 }
773 
OnEarlyHintsReceived(const spdy::Http2HeaderBlock & response_headers,base::TimeTicks recv_first_byte_time)774 void SpdyStream::OnEarlyHintsReceived(
775     const spdy::Http2HeaderBlock& response_headers,
776     base::TimeTicks recv_first_byte_time) {
777   // Record the timing of the 103 Early Hints response for the experiment
778   // (https://crbug.com/1093693).
779   if (first_early_hints_time_.is_null())
780     first_early_hints_time_ = recv_first_byte_time;
781 
782   // Transfer-encoding is a connection specific header.
783   if (response_headers.find("transfer-encoding") != response_headers.end()) {
784     const char error[] = "Received transfer-encoding header";
785     LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
786     session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
787     return;
788   }
789 
790   if (type_ != SPDY_REQUEST_RESPONSE_STREAM || io_state_ == STATE_IDLE) {
791     const char error[] = "Early Hints received before request sent.";
792     LogStreamError(ERR_HTTP2_PROTOCOL_ERROR, error);
793     session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR, error);
794     return;
795   }
796 
797   // `delegate_` must be attached at this point when `type_` is
798   // SPDY_REQUEST_RESPONSE_STREAM.
799   CHECK(delegate_);
800   delegate_->OnEarlyHintsReceived(response_headers);
801 }
802 
SaveResponseHeaders(const spdy::Http2HeaderBlock & response_headers,int status)803 void SpdyStream::SaveResponseHeaders(
804     const spdy::Http2HeaderBlock& response_headers,
805     int status) {
806   if (response_headers.contains("transfer-encoding")) {
807     session_->ResetStream(stream_id_, ERR_HTTP2_PROTOCOL_ERROR,
808                           "Received transfer-encoding header");
809     return;
810   }
811 
812   DCHECK(response_headers_.empty());
813   response_headers_ = response_headers.Clone();
814 
815   // If delegate is not yet attached, OnHeadersReceived() will be called after
816   // the delegate gets attached to the stream.
817   if (!delegate_)
818     return;
819 
820   delegate_->OnHeadersReceived(response_headers_);
821 }
822 
823 #define STATE_CASE(s)                                       \
824   case s:                                                   \
825     description = base::StringPrintf("%s (0x%08X)", #s, s); \
826     break
827 
DescribeState(State state)828 std::string SpdyStream::DescribeState(State state) {
829   std::string description;
830   switch (state) {
831     STATE_CASE(STATE_IDLE);
832     STATE_CASE(STATE_OPEN);
833     STATE_CASE(STATE_HALF_CLOSED_LOCAL);
834     STATE_CASE(STATE_CLOSED);
835     default:
836       description =
837           base::StringPrintf("Unknown state 0x%08X (%u)", state, state);
838       break;
839   }
840   return description;
841 }
842 
843 #undef STATE_CASE
844 
845 }  // namespace net
846