Attn: Mike Belshe

net/spdy/spdy_session.cc

 // Copyright (c) 2011 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/spdy/spdy_session.h"
 
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop.h"
@@ skipping 228 matching lines @@
       streams_abandoned_count_(0),
       frames_received_(0),
       bytes_received_(0),
       sent_settings_(false),
       received_settings_(false),
       stalled_streams_(0),
       initial_send_window_size_(spdy::kSpdyStreamInitialWindowSize),
       initial_recv_window_size_(spdy::kSpdyStreamInitialWindowSize),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)) {
   DCHECK(HttpStreamFactory::spdy_enabled());
+  // TODO(jtl): Do we need to set kMaxSpdyFrameChunkSize here based on which
+  // transport protocol we're using?
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       make_scoped_refptr(
           new NetLogSpdySessionParameter(host_port_proxy_pair_)));
 
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 
   spdy_framer_.set_visitor(this);
 
   SendSettings();
@@ skipping 498 matching lines @@
           LOG(ERROR) << "SPDY Compression failure";
           CloseSessionOnError(net::ERR_SPDY_PROTOCOL_ERROR, true);
           return;
         }
 
         size = compressed_frame->length() + spdy::SpdyFrame::size();
 
         DCHECK_GT(size, 0u);
 
         // TODO(mbelshe): We have too much copying of data here.
-        IOBufferWithSize* buffer = new IOBufferWithSize(size);
+        // TODO(jtl): Why is this "WithSize"?
+        IOBuffer* buffer = new IOBufferWithSize(size);
         memcpy(buffer->data(), compressed_frame->data(), size);
 
         // Attempt to send the frame.
         in_flight_write_ = SpdyIOBuffer(buffer, size, 0, next_buffer.stream());
       } else {
         size = uncompressed_frame.length() + spdy::SpdyFrame::size();
         in_flight_write_ = next_buffer;
       }
+      // SCTP stream ID assignment is delayed until now, rather than done in
+      // QueueFrame, to be sure the association (connection) has been
+      // established.  SCTP stream ID numbers can't be assigned until
+      // g_max_sctp_streams is set, and that can't happen until the association
+      // is established.
+      // TODO(jtl): In order to support piggy-backing DATA on the COOKIE_ECHO,
+      // as is allowed in FreeBSD and Mac OS X, we need to find a way to allow
+      // at least sctp stream ID 1 to be used prior to g_max_sctp_streams being
+      // set.
+      //
+      if (using_sctp()) {
+        spdy::SpdyFrame frame(in_flight_write_.buffer()->data(), false);
+        in_flight_write_.set_sctp_stream_id(GetSctpStreamID(frame));
+      }
     } else {
       DCHECK(in_flight_write_.buffer()->BytesRemaining());
     }
 
     write_pending_ = true;
+
     int rv = connection_->socket()->Write(in_flight_write_.buffer(),
         in_flight_write_.buffer()->BytesRemaining(), &write_callback_);
     if (rv == net::ERR_IO_PENDING)
       break;
 
     // We sent the frame successfully.
     OnWriteComplete(rv);
 
     // TODO(mbelshe):  Test this error case.  Maybe we should mark the socket
     //                 as in an error state.
@@ skipping 38 matching lines @@
 }
 
 int SpdySession::GetNewStreamId() {
   int id = stream_hi_water_mark_;
   stream_hi_water_mark_ += 2;
   if (stream_hi_water_mark_ > 0x7fff)
     stream_hi_water_mark_ = 1;
   return id;
 }
 
+uint16 SpdySession::GetSctpStreamID(const spdy::SpdyFrame& frame)  {
+  uint32 stream_id = 0;
+  if (!frame.is_control_frame()) {
+    stream_id = reinterpret_cast<const spdy::SpdyDataFrame&>(frame).stream_id();
+  } else if (!using_sctp_control_stream()) {
+    spdy::SpdyControlType type =
+        reinterpret_cast<const spdy::SpdyControlFrame&>(frame).type();
+    // n.b. When not using an SCTP control stream, SETTINGS, NOOP, PING
+    // and GOAWAY control frames are still sent on SCTP stream 0 because
+    // they are not associated with a specific SPDY stream.
+    switch (type) {
+      case spdy::SYN_STREAM:
+        stream_id = static_cast<const spdy::SpdySynStreamControlFrame&>
+                        (frame).stream_id();
+        break;
+      case spdy::SYN_REPLY:
+        stream_id = static_cast<const spdy::SpdySynReplyControlFrame&>
+                        (frame).stream_id();
+        break;
+      case spdy::RST_STREAM:
+        stream_id = static_cast<const spdy::SpdyRstStreamControlFrame&>
+                        (frame).stream_id();
+        break;
+      case spdy::HEADERS:
+        stream_id = static_cast<const spdy::SpdyHeadersControlFrame&>
+                        (frame).stream_id();
+        break;
+      case spdy::WINDOW_UPDATE:
+        stream_id = static_cast<const spdy::SpdyWindowUpdateControlFrame&>
+                        (frame).stream_id();
+        break;
+      default:
+        break;
+    }
+  }
+
+  return stream_id ? MapSpdyToSctp(stream_id) : 0;
+}
+
 void SpdySession::QueueFrame(spdy::SpdyFrame* frame,
                              spdy::SpdyPriority priority,
                              SpdyStream* stream) {
   int length = spdy::SpdyFrame::size() + frame->length();
-  IOBuffer* buffer = new IOBuffer(length);
+  IOBuffer* buffer;
+  buffer = new IOBuffer(length);
   memcpy(buffer->data(), frame->data(), length);
+  // can't assign IOBuffer::sctp_stream_id_ here because the association may not
+  // be established yet.  See comments in SpdySession::WriteSocket()
   queue_.push(SpdyIOBuffer(buffer, length, priority, stream));
 
   WriteSocketLater();
 }
 
 void SpdySession::CloseSessionOnError(net::Error err, bool remove_from_pool) {
   // Closing all streams can have a side-effect of dropping the last reference
   // to |this|.  Hold a reference through this function.
   scoped_refptr<SpdySession> self(this);
 
@@ skipping 262 matching lines @@
   spdy::SpdyStreamId stream_id = frame.stream_id();
 
   bool valid_stream = IsStreamActive(stream_id);
   if (!valid_stream) {
     // NOTE:  it may just be that the stream was cancelled.
     LOG(WARNING) << "Received SYN_REPLY for invalid stream " << stream_id;
     return;
   }
 
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
+  if (using_sctp() && using_sctp_control_stream())
+    stream->set_syn_reply_received();
   CHECK_EQ(stream->stream_id(), stream_id);
   CHECK(!stream->cancelled());
 
   if (stream->response_received()) {
     LOG(WARNING) << "Received duplicate SYN_REPLY for stream " << stream_id;
     CloseStream(stream->stream_id(), ERR_SPDY_PROTOCOL_ERROR);
     return;
   }
   stream->set_response_received();
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
         make_scoped_refptr(new NetLogSpdySynParameter(
             headers, static_cast<spdy::SpdyControlFlags>(frame.flags()),
             stream_id, 0)));
   }
 
   Respond(*headers, stream);
+
+  // When using SCTP with a control stream, loss can result in the SYN_REPLY
+  // arriving after all the data.  In this case close_pending() will be true
+  // and we need to call OnDataReceived to shutdown the metrics and close the
+  // stream.
+  if (using_sctp() && using_sctp_control_stream() &&
+      stream->close_pending())
+    stream->OnDataReceived(NULL, 0);
 }
 
 void SpdySession::OnHeaders(const spdy::SpdyHeadersControlFrame& frame,
                             const linked_ptr<spdy::SpdyHeaderBlock>& headers) {
   spdy::SpdyStreamId stream_id = frame.stream_id();
 
   bool valid_stream = IsStreamActive(stream_id);
   if (!valid_stream) {
     // NOTE:  it may just be that the stream was cancelled.
     LOG(WARNING) << "Received HEADERS for invalid stream " << stream_id;
@@ skipping 341 matching lines @@
   if (it == pending_callback_map_.end())
     return;
 
   CompletionCallback* callback = it->second.callback;
   int result = it->second.result;
   pending_callback_map_.erase(it);
   callback->Run(result);
 }
 
 }  // namespace net