net/spdy/spdy_stream.cc

   1 // Copyright (c) 2012 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
   5 #include "net/spdy/spdy_stream.h"
   6
   7 #include "base/bind.h"
   8 #include "base/compiler_specific.h"
   9 #include "base/location.h"
  10 #include "base/logging.h"
  11 #include "base/metrics/histogram_macros.h"
  12 #include "base/single_thread_task_runner.h"
  13 #include "base/strings/string_number_conversions.h"
  14 #include "base/strings/stringprintf.h"
  15 #include "base/thread_task_runner_handle.h"
  16 #include "base/values.h"
  17 #include "net/spdy/spdy_buffer_producer.h"
  18 #include "net/spdy/spdy_http_utils.h"
  19 #include "net/spdy/spdy_session.h"
  20
  21 namespace net {
  22
  23 namespace {
  24
  25 scoped_ptr<base::Value> NetLogSpdyStreamErrorCallback(
  26     SpdyStreamId stream_id,
  27     int status,
  28     const std::string* description,
  29     NetLogCaptureMode /* capture_mode */) {
  30   scoped_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
  31   dict->SetInteger("stream_id", static_cast<int>(stream_id));
  32   dict->SetInteger("status", status);
  33   dict->SetString("description", *description);
  34   return dict.Pass();
  35 }
  36
  37 scoped_ptr<base::Value> NetLogSpdyStreamWindowUpdateCallback(
  38     SpdyStreamId stream_id,
  39     int32 delta,
  40     int32 window_size,
  41     NetLogCaptureMode /* capture_mode */) {
  42   scoped_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
  43   dict->SetInteger("stream_id", stream_id);
  44   dict->SetInteger("delta", delta);
  45   dict->SetInteger("window_size", window_size);
  46   return dict.Pass();
  47 }
  48
  49 bool ContainsUppercaseAscii(const std::string& str) {
  50   for (std::string::const_iterator i(str.begin()); i != str.end(); ++i) {
  51     if (*i >= 'A' && *i <= 'Z') {
  52       return true;
  53     }
  54   }
  55   return false;
  56 }
  57
  58 }  // namespace
  59
  60 void SpdyStream::Delegate::OnTrailers(const SpdyHeaderBlock& trailers) {}
  61
  62 // A wrapper around a stream that calls into ProduceSynStreamFrame().
  63 class SpdyStream::SynStreamBufferProducer : public SpdyBufferProducer {
  64  public:
  65   SynStreamBufferProducer(const base::WeakPtr<SpdyStream>& stream)
  66       : stream_(stream) {
  67     DCHECK(stream_.get());
  68   }
  69
  70   ~SynStreamBufferProducer() override {}
  71
  72   scoped_ptr<SpdyBuffer> ProduceBuffer() override {
  73     if (!stream_.get()) {
  74       NOTREACHED();
  75       return scoped_ptr<SpdyBuffer>();
  76     }
  77     DCHECK_GT(stream_->stream_id(), 0u);
  78     return scoped_ptr<SpdyBuffer>(
  79         new SpdyBuffer(stream_->ProduceSynStreamFrame()));
  80   }
  81
  82  private:
  83   const base::WeakPtr<SpdyStream> stream_;
  84 };
  85
  86 SpdyStream::SpdyStream(SpdyStreamType type,
  87                        const base::WeakPtr<SpdySession>& session,
  88                        const GURL& url,
  89                        RequestPriority priority,
  90                        int32 initial_send_window_size,
  91                        int32 max_recv_window_size,
  92                        const BoundNetLog& net_log)
  93     : type_(type),
  94       stream_id_(0),
  95       url_(url),
  96       priority_(priority),
  97       send_stalled_by_flow_control_(false),
  98       send_window_size_(initial_send_window_size),
  99       max_recv_window_size_(max_recv_window_size),
 100       recv_window_size_(max_recv_window_size),
 101       unacked_recv_window_bytes_(0),
 102       session_(session),
 103       delegate_(NULL),
 104       pending_send_status_(MORE_DATA_TO_SEND),
 105       request_time_(base::Time::Now()),
 106       response_headers_status_(RESPONSE_HEADERS_ARE_INCOMPLETE),
 107       io_state_(STATE_IDLE),
 108       response_status_(OK),
 109       net_log_(net_log),
 110       raw_received_bytes_(0),
 111       raw_sent_bytes_(0),
 112       send_bytes_(0),
 113       recv_bytes_(0),
 114       write_handler_guard_(false),
 115       weak_ptr_factory_(this) {
 116   CHECK(type_ == SPDY_BIDIRECTIONAL_STREAM ||
 117         type_ == SPDY_REQUEST_RESPONSE_STREAM ||
 118         type_ == SPDY_PUSH_STREAM);
 119   CHECK_GE(priority_, MINIMUM_PRIORITY);
 120   CHECK_LE(priority_, MAXIMUM_PRIORITY);
 121 }
 122
 123 SpdyStream::~SpdyStream() {
 124   CHECK(!write_handler_guard_);
 125   UpdateHistograms();
 126 }
 127
 128 void SpdyStream::SetDelegate(Delegate* delegate) {
 129   CHECK(!delegate_);
 130   CHECK(delegate);
 131   delegate_ = delegate;
 132
 133   CHECK(io_state_ == STATE_IDLE ||
 134         io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
 135         io_state_ == STATE_RESERVED_REMOTE);
 136
 137   if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
 138     DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 139     base::ThreadTaskRunnerHandle::Get()->PostTask(
 140         FROM_HERE, base::Bind(&SpdyStream::PushedStreamReplay, GetWeakPtr()));
 141   }
 142 }
 143
 144 void SpdyStream::PushedStreamReplay() {
 145   DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 146   DCHECK_NE(stream_id_, 0u);
 147   CHECK_EQ(stream_id_ % 2, 0u);
 148
 149   CHECK_EQ(io_state_, STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
 150   io_state_ = STATE_HALF_CLOSED_LOCAL;
 151
 152   // The delegate methods called below may delete |this|, so use
 153   // |weak_this| to detect that.
 154   base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 155
 156   CHECK(delegate_);
 157   SpdyResponseHeadersStatus status =
 158       delegate_->OnResponseHeadersUpdated(response_headers_);
 159   if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 160     // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
 161     // have been closed. Since we don't have complete headers, assume
 162     // we're waiting for another HEADERS frame, and we had better not
 163     // have any pending data frames.
 164     CHECK(weak_this);
 165     if (!pending_recv_data_.empty()) {
 166       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
 167                      "Data received with incomplete headers.");
 168       session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 169     }
 170     return;
 171   }
 172
 173   // OnResponseHeadersUpdated() may have closed |this|.
 174   if (!weak_this)
 175     return;
 176
 177   response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
 178
 179   while (!pending_recv_data_.empty()) {
 180     // Take ownership of the first element of |pending_recv_data_|.
 181     scoped_ptr<SpdyBuffer> buffer(pending_recv_data_.front());
 182     pending_recv_data_.weak_erase(pending_recv_data_.begin());
 183
 184     bool eof = (buffer == NULL);
 185
 186     CHECK(delegate_);
 187     delegate_->OnDataReceived(buffer.Pass());
 188
 189     // OnDataReceived() may have closed |this|.
 190     if (!weak_this)
 191       return;
 192
 193     if (eof) {
 194       DCHECK(pending_recv_data_.empty());
 195       session_->CloseActiveStream(stream_id_, OK);
 196       DCHECK(!weak_this);
 197       // |pending_recv_data_| is invalid at this point.
 198       break;
 199     }
 200   }
 201 }
 202
 203 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
 204   CHECK_EQ(io_state_, STATE_IDLE);
 205   CHECK(request_headers_);
 206   CHECK_GT(stream_id_, 0u);
 207
 208   SpdyControlFlags flags =
 209       (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
 210       CONTROL_FLAG_FIN : CONTROL_FLAG_NONE;
 211   scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
 212       stream_id_, priority_, flags, *request_headers_));
 213   send_time_ = base::TimeTicks::Now();
 214   return frame.Pass();
 215 }
 216
 217 void SpdyStream::DetachDelegate() {
 218   DCHECK(!IsClosed());
 219   delegate_ = NULL;
 220   Cancel();
 221 }
 222
 223 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size) {
 224   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 225
 226   if (IsClosed())
 227     return;
 228
 229   // Check for wraparound.
 230   if (send_window_size_ > 0) {
 231     DCHECK_LE(delta_window_size, kint32max - send_window_size_);
 232   }
 233   if (send_window_size_ < 0) {
 234     DCHECK_GE(delta_window_size, kint32min - send_window_size_);
 235   }
 236   send_window_size_ += delta_window_size;
 237   PossiblyResumeIfSendStalled();
 238 }
 239
 240 void SpdyStream::OnWriteBufferConsumed(
 241     size_t frame_payload_size,
 242     size_t consume_size,
 243     SpdyBuffer::ConsumeSource consume_source) {
 244   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 245   if (consume_source == SpdyBuffer::DISCARD) {
 246     // If we're discarding a frame or part of it, increase the send
 247     // window by the number of discarded bytes. (Although if we're
 248     // discarding part of a frame, it's probably because of a write
 249     // error and we'll be tearing down the stream soon.)
 250     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
 251     DCHECK_GT(remaining_payload_bytes, 0u);
 252     IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
 253   }
 254   // For consumed bytes, the send window is increased when we receive
 255   // a WINDOW_UPDATE frame.
 256 }
 257
 258 void SpdyStream::IncreaseSendWindowSize(int32 delta_window_size) {
 259   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 260   DCHECK_GE(delta_window_size, 1);
 261
 262   // Ignore late WINDOW_UPDATEs.
 263   if (IsClosed())
 264     return;
 265
 266   if (send_window_size_ > 0) {
 267     // Check for overflow.
 268     int32 max_delta_window_size = kint32max - send_window_size_;
 269     if (delta_window_size > max_delta_window_size) {
 270       std::string desc = base::StringPrintf(
 271           "Received WINDOW_UPDATE [delta: %d] for stream %d overflows "
 272           "send_window_size_ [current: %d]", delta_window_size, stream_id_,
 273           send_window_size_);
 274       session_->ResetStream(stream_id_, RST_STREAM_FLOW_CONTROL_ERROR, desc);
 275       return;
 276     }
 277   }
 278
 279   send_window_size_ += delta_window_size;
 280
 281   net_log_.AddEvent(
 282       NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
 283       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 284                  delta_window_size, send_window_size_));
 285
 286   PossiblyResumeIfSendStalled();
 287 }
 288
 289 void SpdyStream::DecreaseSendWindowSize(int32 delta_window_size) {
 290   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 291
 292   if (IsClosed())
 293     return;
 294
 295   // We only call this method when sending a frame. Therefore,
 296   // |delta_window_size| should be within the valid frame size range.
 297   DCHECK_GE(delta_window_size, 1);
 298   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
 299
 300   // |send_window_size_| should have been at least |delta_window_size| for
 301   // this call to happen.
 302   DCHECK_GE(send_window_size_, delta_window_size);
 303
 304   send_window_size_ -= delta_window_size;
 305
 306   net_log_.AddEvent(
 307       NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
 308       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 309                  -delta_window_size, send_window_size_));
 310 }
 311
 312 void SpdyStream::OnReadBufferConsumed(
 313     size_t consume_size,
 314     SpdyBuffer::ConsumeSource consume_source) {
 315   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 316   DCHECK_GE(consume_size, 1u);
 317   DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
 318   IncreaseRecvWindowSize(static_cast<int32>(consume_size));
 319 }
 320
 321 void SpdyStream::IncreaseRecvWindowSize(int32 delta_window_size) {
 322   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 323
 324   // By the time a read is processed by the delegate, this stream may
 325   // already be inactive.
 326   if (!session_->IsStreamActive(stream_id_))
 327     return;
 328
 329   DCHECK_GE(unacked_recv_window_bytes_, 0);
 330   DCHECK_GE(recv_window_size_, unacked_recv_window_bytes_);
 331   DCHECK_GE(delta_window_size, 1);
 332   // Check for overflow.
 333   DCHECK_LE(delta_window_size, kint32max - recv_window_size_);
 334
 335   recv_window_size_ += delta_window_size;
 336   net_log_.AddEvent(
 337       NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
 338       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 339                  delta_window_size, recv_window_size_));
 340
 341   unacked_recv_window_bytes_ += delta_window_size;
 342   if (unacked_recv_window_bytes_ > max_recv_window_size_ / 2) {
 343     session_->SendStreamWindowUpdate(
 344         stream_id_, static_cast<uint32>(unacked_recv_window_bytes_));
 345     unacked_recv_window_bytes_ = 0;
 346   }
 347 }
 348
 349 void SpdyStream::DecreaseRecvWindowSize(int32 delta_window_size) {
 350   DCHECK(session_->IsStreamActive(stream_id_));
 351   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 352   DCHECK_GE(delta_window_size, 1);
 353
 354   // The receiving window size as the peer knows it is
 355   // |recv_window_size_ - unacked_recv_window_bytes_|, if more data are sent by
 356   // the peer, that means that the receive window is not being respected.
 357   if (delta_window_size > recv_window_size_ - unacked_recv_window_bytes_) {
 358     session_->ResetStream(
 359         stream_id_, RST_STREAM_FLOW_CONTROL_ERROR,
 360         "delta_window_size is " + base::IntToString(delta_window_size) +
 361             " in DecreaseRecvWindowSize, which is larger than the receive " +
 362             "window size of " + base::IntToString(recv_window_size_));
 363     return;
 364   }
 365
 366   recv_window_size_ -= delta_window_size;
 367   net_log_.AddEvent(
 368       NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
 369       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 370                  -delta_window_size, recv_window_size_));
 371 }
 372
 373 int SpdyStream::GetPeerAddress(IPEndPoint* address) const {
 374   return session_->GetPeerAddress(address);
 375 }
 376
 377 int SpdyStream::GetLocalAddress(IPEndPoint* address) const {
 378   return session_->GetLocalAddress(address);
 379 }
 380
 381 bool SpdyStream::WasEverUsed() const {
 382   return session_->WasEverUsed();
 383 }
 384
 385 base::Time SpdyStream::GetRequestTime() const {
 386   return request_time_;
 387 }
 388
 389 void SpdyStream::SetRequestTime(base::Time t) {
 390   request_time_ = t;
 391 }
 392
 393 int SpdyStream::OnInitialResponseHeadersReceived(
 394     const SpdyHeaderBlock& initial_response_headers,
 395     base::Time response_time,
 396     base::TimeTicks recv_first_byte_time) {
 397   // SpdySession guarantees that this is called at most once.
 398   CHECK(response_headers_.empty());
 399
 400   // Check to make sure that we don't receive the response headers
 401   // before we're ready for it.
 402   switch (type_) {
 403     case SPDY_BIDIRECTIONAL_STREAM:
 404       // For a bidirectional stream, we're ready for the response
 405       // headers once we've finished sending the request headers.
 406       if (io_state_ == STATE_IDLE) {
 407         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 408                               "Response received before request sent");
 409         return ERR_SPDY_PROTOCOL_ERROR;
 410       }
 411       break;
 412
 413     case SPDY_REQUEST_RESPONSE_STREAM:
 414       // For a request/response stream, we're ready for the response
 415       // headers once we've finished sending the request headers.
 416       if (io_state_ == STATE_IDLE) {
 417         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 418                               "Response received before request sent");
 419         return ERR_SPDY_PROTOCOL_ERROR;
 420       }
 421       break;
 422
 423     case SPDY_PUSH_STREAM:
 424       // Push streams transition to a locally half-closed state upon headers.
 425       // We must continue to buffer data while waiting for a call to
 426       // SetDelegate() (which may not ever happen).
 427       CHECK_EQ(io_state_, STATE_RESERVED_REMOTE);
 428       if (!delegate_) {
 429         io_state_ = STATE_HALF_CLOSED_LOCAL_UNCLAIMED;
 430       } else {
 431         io_state_ = STATE_HALF_CLOSED_LOCAL;
 432       }
 433       break;
 434   }
 435
 436   DCHECK_NE(io_state_, STATE_IDLE);
 437
 438   response_time_ = response_time;
 439   recv_first_byte_time_ = recv_first_byte_time;
 440   return MergeWithResponseHeaders(initial_response_headers);
 441 }
 442
 443 int SpdyStream::OnAdditionalResponseHeadersReceived(
 444     const SpdyHeaderBlock& additional_response_headers) {
 445   if (type_ == SPDY_REQUEST_RESPONSE_STREAM) {
 446     if (response_headers_status_ != RESPONSE_HEADERS_ARE_COMPLETE) {
 447       session_->ResetStream(
 448           stream_id_, RST_STREAM_PROTOCOL_ERROR,
 449           "Additional headers received for request/response stream");
 450       return ERR_SPDY_PROTOCOL_ERROR;
 451     }
 452     response_headers_status_ = TRAILERS_RECEIVED;
 453     delegate_->OnTrailers(additional_response_headers);
 454     return OK;
 455   }
 456   if (type_ == SPDY_PUSH_STREAM &&
 457       response_headers_status_ == RESPONSE_HEADERS_ARE_COMPLETE) {
 458     session_->ResetStream(
 459         stream_id_, RST_STREAM_PROTOCOL_ERROR,
 460         "Additional headers received for push stream");
 461     return ERR_SPDY_PROTOCOL_ERROR;
 462   }
 463   return MergeWithResponseHeaders(additional_response_headers);
 464 }
 465
 466 void SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
 467   CHECK(!request_headers_.get());
 468   CHECK_EQ(io_state_, STATE_IDLE);
 469   CHECK_EQ(type_, SPDY_PUSH_STREAM);
 470   DCHECK(!delegate_);
 471
 472   io_state_ = STATE_RESERVED_REMOTE;
 473   request_headers_.reset(new SpdyHeaderBlock(headers));
 474 }
 475
 476 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
 477   DCHECK(session_->IsStreamActive(stream_id_));
 478
 479   // If we're still buffering data for a push stream, we will do the
 480   // check for data received with incomplete headers in
 481   // PushedStreamReplayData().
 482   if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
 483     DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 484     // It should be valid for this to happen in the server push case.
 485     // We'll return received data when delegate gets attached to the stream.
 486     if (buffer) {
 487       pending_recv_data_.push_back(buffer.Pass());
 488     } else {
 489       pending_recv_data_.push_back(NULL);
 490       // Note: we leave the stream open in the session until the stream
 491       //       is claimed.
 492     }
 493     return;
 494   }
 495
 496   if (response_headers_status_ == TRAILERS_RECEIVED && buffer) {
 497     // TRAILERS_RECEIVED is only used in SPDY_REQUEST_RESPONSE_STREAM.
 498     DCHECK_EQ(type_, SPDY_REQUEST_RESPONSE_STREAM);
 499     session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 500                           "Data received after trailers");
 501     return;
 502   }
 503
 504   // If we have response headers but the delegate has indicated that
 505   // it's still incomplete, then that's a protocol error.
 506   if (response_headers_status_ == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 507     LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
 508                    "Data received with incomplete headers.");
 509     session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 510     return;
 511   }
 512
 513   CHECK(!IsClosed());
 514
 515   if (!buffer) {
 516     if (io_state_ == STATE_OPEN) {
 517       io_state_ = STATE_HALF_CLOSED_REMOTE;
 518     } else if (io_state_ == STATE_HALF_CLOSED_LOCAL) {
 519       io_state_ = STATE_CLOSED;
 520       // Deletes |this|.
 521       session_->CloseActiveStream(stream_id_, OK);
 522     } else {
 523       NOTREACHED() << io_state_;
 524     }
 525     return;
 526   }
 527
 528   size_t length = buffer->GetRemainingSize();
 529   DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
 530   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 531     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 532     // May close the stream.
 533     DecreaseRecvWindowSize(static_cast<int32>(length));
 534     if (!weak_this)
 535       return;
 536     buffer->AddConsumeCallback(
 537         base::Bind(&SpdyStream::OnReadBufferConsumed, GetWeakPtr()));
 538   }
 539
 540   // Track our bandwidth.
 541   recv_bytes_ += length;
 542   recv_last_byte_time_ = base::TimeTicks::Now();
 543
 544   // May close |this|.
 545   delegate_->OnDataReceived(buffer.Pass());
 546 }
 547
 548 void SpdyStream::OnPaddingConsumed(size_t len) {
 549   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 550     // Decrease window size because padding bytes are received.
 551     // Increase window size because padding bytes are consumed (by discarding).
 552     // Net result: |unacked_recv_window_bytes_| increases by |len|,
 553     // |recv_window_size_| does not change.
 554     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 555     // May close the stream.
 556     DecreaseRecvWindowSize(static_cast<int32>(len));
 557     if (!weak_this)
 558       return;
 559     IncreaseRecvWindowSize(static_cast<int32>(len));
 560   }
 561 }
 562
 563 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
 564                                       size_t frame_size) {
 565   DCHECK_NE(type_, SPDY_PUSH_STREAM);
 566   CHECK(frame_type == SYN_STREAM ||
 567         frame_type == DATA) << frame_type;
 568
 569   int result = (frame_type == SYN_STREAM) ?
 570       OnRequestHeadersSent() : OnDataSent(frame_size);
 571   if (result == ERR_IO_PENDING) {
 572     // The write operation hasn't completed yet.
 573     return;
 574   }
 575
 576   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
 577     if(io_state_ == STATE_OPEN) {
 578       io_state_ = STATE_HALF_CLOSED_LOCAL;
 579     } else if(io_state_ == STATE_HALF_CLOSED_REMOTE) {
 580       io_state_ = STATE_CLOSED;
 581     } else {
 582       NOTREACHED() << io_state_;
 583     }
 584   }
 585   // Notify delegate of write completion. Must not destroy |this|.
 586   CHECK(delegate_);
 587   {
 588     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 589     write_handler_guard_ = true;
 590     if (frame_type == SYN_STREAM) {
 591       delegate_->OnRequestHeadersSent();
 592     } else {
 593       delegate_->OnDataSent();
 594     }
 595     CHECK(weak_this);
 596     write_handler_guard_ = false;
 597   }
 598
 599   if (io_state_ == STATE_CLOSED) {
 600     // Deletes |this|.
 601     session_->CloseActiveStream(stream_id_, OK);
 602   }
 603 }
 604
 605 int SpdyStream::OnRequestHeadersSent() {
 606   CHECK_EQ(io_state_, STATE_IDLE);
 607   CHECK_NE(stream_id_, 0u);
 608
 609   io_state_ = STATE_OPEN;
 610   return OK;
 611 }
 612
 613 int SpdyStream::OnDataSent(size_t frame_size) {
 614   CHECK(io_state_ == STATE_OPEN ||
 615         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 616
 617   size_t frame_payload_size = frame_size - session_->GetDataFrameMinimumSize();
 618
 619   CHECK_GE(frame_size, session_->GetDataFrameMinimumSize());
 620   CHECK_LE(frame_payload_size, session_->GetDataFrameMaximumPayload());
 621
 622   send_bytes_ += frame_payload_size;
 623
 624   // If more data is available to send, dispatch it and
 625   // return that the write operation is still ongoing.
 626   pending_send_data_->DidConsume(frame_payload_size);
 627   if (pending_send_data_->BytesRemaining() > 0) {
 628     QueueNextDataFrame();
 629     return ERR_IO_PENDING;
 630   } else {
 631     pending_send_data_ = NULL;
 632     return OK;
 633   }
 634 }
 635
 636 SpdyMajorVersion SpdyStream::GetProtocolVersion() const {
 637   return session_->GetProtocolVersion();
 638 }
 639
 640 void SpdyStream::LogStreamError(int status, const std::string& description) {
 641   net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_ERROR,
 642                     base::Bind(&NetLogSpdyStreamErrorCallback, stream_id_,
 643                                status, &description));
 644 }
 645
 646 void SpdyStream::OnClose(int status) {
 647   // In most cases, the stream should already be CLOSED. The exception is when a
 648   // SpdySession is shutting down while the stream is in an intermediate state.
 649   io_state_ = STATE_CLOSED;
 650   response_status_ = status;
 651   Delegate* delegate = delegate_;
 652   delegate_ = NULL;
 653   if (delegate)
 654     delegate->OnClose(status);
 655   // Unset |stream_id_| last so that the delegate can look it up.
 656   stream_id_ = 0;
 657 }
 658
 659 void SpdyStream::Cancel() {
 660   // We may be called again from a delegate's OnClose().
 661   if (io_state_ == STATE_CLOSED)
 662     return;
 663
 664   if (stream_id_ != 0) {
 665     session_->ResetStream(stream_id_, RST_STREAM_CANCEL, std::string());
 666   } else {
 667     session_->CloseCreatedStream(GetWeakPtr(), RST_STREAM_CANCEL);
 668   }
 669   // |this| is invalid at this point.
 670 }
 671
 672 void SpdyStream::Close() {
 673   // We may be called again from a delegate's OnClose().
 674   if (io_state_ == STATE_CLOSED)
 675     return;
 676
 677   if (stream_id_ != 0) {
 678     session_->CloseActiveStream(stream_id_, OK);
 679   } else {
 680     session_->CloseCreatedStream(GetWeakPtr(), OK);
 681   }
 682   // |this| is invalid at this point.
 683 }
 684
 685 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
 686   return weak_ptr_factory_.GetWeakPtr();
 687 }
 688
 689 int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> request_headers,
 690                                    SpdySendStatus send_status) {
 691   CHECK_NE(type_, SPDY_PUSH_STREAM);
 692   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
 693   CHECK(!request_headers_);
 694   CHECK(!pending_send_data_.get());
 695   CHECK_EQ(io_state_, STATE_IDLE);
 696   request_headers_ = request_headers.Pass();
 697   pending_send_status_ = send_status;
 698   session_->EnqueueStreamWrite(
 699       GetWeakPtr(), SYN_STREAM,
 700       scoped_ptr<SpdyBufferProducer>(
 701           new SynStreamBufferProducer(GetWeakPtr())));
 702   return ERR_IO_PENDING;
 703 }
 704
 705 void SpdyStream::SendData(IOBuffer* data,
 706                           int length,
 707                           SpdySendStatus send_status) {
 708   CHECK_NE(type_, SPDY_PUSH_STREAM);
 709   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
 710   CHECK(io_state_ == STATE_OPEN ||
 711         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 712   CHECK(!pending_send_data_.get());
 713   pending_send_data_ = new DrainableIOBuffer(data, length);
 714   pending_send_status_ = send_status;
 715   QueueNextDataFrame();
 716 }
 717
 718 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
 719                             bool* was_npn_negotiated,
 720                             NextProto* protocol_negotiated) {
 721   return session_->GetSSLInfo(
 722       ssl_info, was_npn_negotiated, protocol_negotiated);
 723 }
 724
 725 void SpdyStream::PossiblyResumeIfSendStalled() {
 726   if (IsLocallyClosed()) {
 727     return;
 728   }
 729   if (send_stalled_by_flow_control_ && !session_->IsSendStalled() &&
 730       send_window_size_ > 0) {
 731     net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_FLOW_CONTROL_UNSTALLED,
 732                       NetLog::IntegerCallback("stream_id", stream_id_));
 733     send_stalled_by_flow_control_ = false;
 734     QueueNextDataFrame();
 735   }
 736 }
 737
 738 bool SpdyStream::IsClosed() const {
 739   return io_state_ == STATE_CLOSED;
 740 }
 741
 742 bool SpdyStream::IsLocallyClosed() const {
 743   return io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
 744       io_state_ == STATE_HALF_CLOSED_LOCAL ||
 745       io_state_ == STATE_CLOSED;
 746 }
 747
 748 bool SpdyStream::IsIdle() const {
 749   return io_state_ == STATE_IDLE;
 750 }
 751
 752 bool SpdyStream::IsOpen() const {
 753   return io_state_ == STATE_OPEN;
 754 }
 755
 756 bool SpdyStream::IsReservedRemote() const {
 757   return io_state_ == STATE_RESERVED_REMOTE;
 758 }
 759
 760 NextProto SpdyStream::GetProtocol() const {
 761   return session_->protocol();
 762 }
 763
 764 void SpdyStream::AddRawReceivedBytes(size_t received_bytes) {
 765   raw_received_bytes_ += received_bytes;
 766 }
 767
 768 void SpdyStream::AddRawSentBytes(size_t sent_bytes) {
 769   raw_sent_bytes_ += sent_bytes;
 770 }
 771
 772 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
 773   if (stream_id_ == 0)
 774     return false;
 775
 776   return session_->GetLoadTimingInfo(stream_id_, load_timing_info);
 777 }
 778
 779 GURL SpdyStream::GetUrlFromHeaders() const {
 780   if (!request_headers_)
 781     return GURL();
 782
 783   return GetUrlFromHeaderBlock(
 784       *request_headers_, GetProtocolVersion(), type_ == SPDY_PUSH_STREAM);
 785 }
 786
 787 bool SpdyStream::HasUrlFromHeaders() const {
 788   return !GetUrlFromHeaders().is_empty();
 789 }
 790
 791 void SpdyStream::UpdateHistograms() {
 792   // We need at least the receive timers to be filled in, as otherwise
 793   // metrics can be bogus.
 794   if (recv_first_byte_time_.is_null() || recv_last_byte_time_.is_null())
 795     return;
 796
 797   base::TimeTicks effective_send_time;
 798   if (type_ == SPDY_PUSH_STREAM) {
 799     // Push streams shouldn't have |send_time_| filled in.
 800     DCHECK(send_time_.is_null());
 801     effective_send_time = recv_first_byte_time_;
 802   } else {
 803     // For non-push streams, we also need |send_time_| to be filled
 804     // in.
 805     if (send_time_.is_null())
 806       return;
 807     effective_send_time = send_time_;
 808   }
 809
 810   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTimeToFirstByte",
 811                       recv_first_byte_time_ - effective_send_time);
 812   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
 813                       recv_last_byte_time_ - recv_first_byte_time_);
 814   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
 815                       recv_last_byte_time_ - effective_send_time);
 816
 817   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
 818   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 819 }
 820
 821 void SpdyStream::QueueNextDataFrame() {
 822   // Until the request has been completely sent, we cannot be sure
 823   // that our stream_id is correct.
 824   CHECK(io_state_ == STATE_OPEN ||
 825         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 826   CHECK_GT(stream_id_, 0u);
 827   CHECK(pending_send_data_.get());
 828   // Only the final fame may have a length of 0.
 829   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
 830     CHECK_GE(pending_send_data_->BytesRemaining(), 0);
 831   } else {
 832     CHECK_GT(pending_send_data_->BytesRemaining(), 0);
 833   }
 834
 835   SpdyDataFlags flags =
 836       (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
 837       DATA_FLAG_FIN : DATA_FLAG_NONE;
 838   scoped_ptr<SpdyBuffer> data_buffer(
 839       session_->CreateDataBuffer(stream_id_,
 840                                  pending_send_data_.get(),
 841                                  pending_send_data_->BytesRemaining(),
 842                                  flags));
 843   // We'll get called again by PossiblyResumeIfSendStalled().
 844   if (!data_buffer)
 845     return;
 846
 847   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 848     DCHECK_GE(data_buffer->GetRemainingSize(),
 849               session_->GetDataFrameMinimumSize());
 850     size_t payload_size =
 851         data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
 852     DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
 853
 854     // Send window size is based on payload size, so nothing to do if this is
 855     // just a FIN with no payload.
 856     if (payload_size != 0) {
 857       DecreaseSendWindowSize(static_cast<int32>(payload_size));
 858       // This currently isn't strictly needed, since write frames are
 859       // discarded only if the stream is about to be closed. But have it
 860       // here anyway just in case this changes.
 861       data_buffer->AddConsumeCallback(
 862           base::Bind(&SpdyStream::OnWriteBufferConsumed,
 863                      GetWeakPtr(), payload_size));
 864     }
 865   }
 866
 867   session_->EnqueueStreamWrite(
 868       GetWeakPtr(), DATA,
 869       scoped_ptr<SpdyBufferProducer>(
 870           new SimpleBufferProducer(data_buffer.Pass())));
 871 }
 872
 873 int SpdyStream::MergeWithResponseHeaders(
 874     const SpdyHeaderBlock& new_response_headers) {
 875   if (new_response_headers.find("transfer-encoding") !=
 876       new_response_headers.end()) {
 877     session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 878                          "Received transfer-encoding header");
 879     return ERR_SPDY_PROTOCOL_ERROR;
 880   }
 881
 882   for (SpdyHeaderBlock::const_iterator it = new_response_headers.begin();
 883       it != new_response_headers.end(); ++it) {
 884     // Disallow uppercase headers.
 885     if (ContainsUppercaseAscii(it->first)) {
 886       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 887                             "Upper case characters in header: " + it->first);
 888       return ERR_SPDY_PROTOCOL_ERROR;
 889     }
 890
 891     SpdyHeaderBlock::iterator it2 = response_headers_.lower_bound(it->first);
 892     // Disallow duplicate headers.  This is just to be conservative.
 893     if (it2 != response_headers_.end() && it2->first == it->first) {
 894       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 895                             "Duplicate header: " + it->first);
 896       return ERR_SPDY_PROTOCOL_ERROR;
 897     }
 898
 899     response_headers_.insert(it2, *it);
 900   }
 901
 902   // If delegate_ is not yet attached, we'll call
 903   // OnResponseHeadersUpdated() after the delegate gets attached to
 904   // the stream.
 905   if (delegate_) {
 906     // The call to OnResponseHeadersUpdated() below may delete |this|,
 907     // so use |weak_this| to detect that.
 908     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 909
 910     SpdyResponseHeadersStatus status =
 911         delegate_->OnResponseHeadersUpdated(response_headers_);
 912     if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 913       // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
 914       // have been closed.
 915       CHECK(weak_this);
 916       // Incomplete headers are OK only for push streams.
 917       if (type_ != SPDY_PUSH_STREAM) {
 918         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 919                               "Incomplete headers");
 920         return ERR_INCOMPLETE_SPDY_HEADERS;
 921       }
 922     } else if (weak_this) {
 923       response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
 924     }
 925   }
 926
 927   return OK;
 928 }
 929
 930 #define STATE_CASE(s) \
 931   case s: \
 932     description = base::StringPrintf("%s (0x%08X)", #s, s); \
 933     break
 934
 935 std::string SpdyStream::DescribeState(State state) {
 936   std::string description;
 937   switch (state) {
 938     STATE_CASE(STATE_IDLE);
 939     STATE_CASE(STATE_OPEN);
 940     STATE_CASE(STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
 941     STATE_CASE(STATE_HALF_CLOSED_LOCAL);
 942     STATE_CASE(STATE_CLOSED);
 943     default:
 944       description = base::StringPrintf("Unknown state 0x%08X (%u)", state,
 945                                        state);
 946       break;
 947   }
 948   return description;
 949 }
 950
 951 #undef STATE_CASE
 952
 953 }  // namespace net