[SPDY] Replace SpdyIOBuffer with new SpdyBuffer class

net/spdy/spdy_session.cc

 // Copyright (c) 2012 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 <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 13 matching lines @@
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "crypto/ec_private_key.h"
 #include "crypto/ec_signature_creator.h"
 #include "net/base/connection_type_histograms.h"
 #include "net/base/net_log.h"
 #include "net/base/net_util.h"
 #include "net/cert/asn1_util.h"
 #include "net/http/http_network_session.h"
 #include "net/http/http_server_properties.h"
+#include "net/spdy/spdy_buffer_producer.h"
 #include "net/spdy/spdy_credential_builder.h"
 #include "net/spdy/spdy_frame_builder.h"
-#include "net/spdy/spdy_frame_producer.h"
 #include "net/spdy/spdy_http_utils.h"
 #include "net/spdy/spdy_protocol.h"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/spdy/spdy_stream.h"
 #include "net/ssl/server_bound_cert_service.h"
 
 namespace net {
 
 namespace {
 
@@ skipping 258 matching lines @@
                          NetLog* net_log)
     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       host_port_proxy_pair_(host_port_proxy_pair),
       spdy_session_pool_(spdy_session_pool),
       http_server_properties_(http_server_properties),
       connection_(new ClientSocketHandle),
       read_buffer_(new IOBuffer(kReadBufferSize)),
       stream_hi_water_mark_(kFirstStreamId),
       write_pending_(false),
       in_flight_write_frame_type_(DATA),
+      in_flight_write_frame_size_(0),
       delayed_write_pending_(false),
       is_secure_(false),
       certificate_error_code_(OK),
       error_(OK),
       state_(STATE_IDLE),
       max_concurrent_streams_(initial_max_concurrent_streams == 0 ?
                               kInitialMaxConcurrentStreams :
                               initial_max_concurrent_streams),
       max_concurrent_streams_limit_(max_concurrent_streams_limit == 0 ?
                                     kMaxConcurrentStreamLimit :
@@ skipping 304 matching lines @@
 }
 
 int SpdySession::GetProtocolVersion() const {
   DCHECK(buffered_spdy_framer_.get());
   return buffered_spdy_framer_->protocol_version();
 }
 
 void SpdySession::EnqueueStreamWrite(
     SpdyStream* stream,
     SpdyFrameType frame_type,
-    scoped_ptr<SpdyFrameProducer> producer) {
+    scoped_ptr<SpdyBufferProducer> producer) {
   DCHECK(frame_type == HEADERS ||
          frame_type == DATA ||
          frame_type == CREDENTIAL ||
          frame_type == SYN_STREAM);
   EnqueueWrite(stream->priority(), frame_type, producer.Pass(), stream);
 }
 
 scoped_ptr<SpdyFrame> SpdySession::CreateSynStream(
     SpdyStreamId stream_id,
     RequestPriority priority,
@@ skipping 81 matching lines @@
     bool fin = flags & CONTROL_FLAG_FIN;
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SEND_HEADERS,
         base::Bind(&NetLogSpdySynCallback,
                    &headers, fin, /*unidirectional=*/false,
                    stream_id, 0));
   }
   return frame.Pass();
 }
 
-scoped_ptr<SpdyFrame> SpdySession::CreateDataFrame(SpdyStreamId stream_id,
-                                                   net::IOBuffer* data,
-                                                   int len,
-                                                   SpdyDataFlags flags) {
-  // Find our stream
+scoped_ptr<SpdyBuffer> SpdySession::CreateDataBuffer(SpdyStreamId stream_id,
+                                                     net::IOBuffer* data,
+                                                     int len,
+                                                     SpdyDataFlags flags) {
+  // Find our stream.
   CHECK(IsStreamActive(stream_id));
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
 
   if (len < 0) {
     NOTREACHED();
-    return scoped_ptr<SpdyFrame>();
+    return scoped_ptr<SpdyBuffer>();
   }
 
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
     flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
   }
 
   // Obey send window size of the stream (and session, if applicable)
   // if flow control is enabled.
   if (flow_control_state_ >= FLOW_CONTROL_STREAM) {
     int32 effective_window_size = stream->send_window_size();
     if (effective_window_size <= 0) {
       // Because we queue frames onto the session, it is possible that
       // a stream was not flow controlled at the time it attempted the
       // write, but when we go to fulfill the write, it is now flow
       // controlled.  This is why we need the session to mark the stream
       // as stalled - because only the session knows for sure when the
       // stall occurs.
       stream->set_send_stalled_by_flow_control(true);
       net_log().AddEvent(
           NetLog::TYPE_SPDY_SESSION_STREAM_STALLED_ON_STREAM_SEND_WINDOW,
           NetLog::IntegerCallback("stream_id", stream_id));
-      return scoped_ptr<SpdyFrame>();
+      return scoped_ptr<SpdyBuffer>();
     }
     if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
       effective_window_size =
           std::min(effective_window_size, session_send_window_size_);
       if (effective_window_size <= 0) {
         DCHECK(IsSendStalled());
         stream->set_send_stalled_by_flow_control(true);
         QueueSendStalledStream(stream);
         net_log().AddEvent(
             NetLog::TYPE_SPDY_SESSION_STREAM_STALLED_ON_SESSION_SEND_WINDOW,
             NetLog::IntegerCallback("stream_id", stream_id));
-        return scoped_ptr<SpdyFrame>();
+        return scoped_ptr<SpdyBuffer>();
       }
     }
 
     int new_len = std::min(len, effective_window_size);
     if (new_len < len) {
       len = new_len;
       flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
     }
     if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION)
       DecreaseSendWindowSize(static_cast<int32>(len));
@@ skipping 10 matching lines @@
   // Send PrefacePing for DATA_FRAMEs with nonzero payload size.
   if (len > 0)
     SendPrefacePingIfNoneInFlight();
 
   // TODO(mbelshe): reduce memory copies here.
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyFrame> frame(
       buffered_spdy_framer_->CreateDataFrame(
           stream_id, data->data(), static_cast<uint32>(len), flags));
 
-  return frame.Pass();
+  return scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass()));
 }
 
 void SpdySession::CloseStream(SpdyStreamId stream_id, int status) {
   DCHECK_NE(0u, stream_id);
   // TODO(mbelshe): We should send a RST_STREAM control frame here
   //                so that the server can cancel a large send.
 
   DeleteStream(stream_id, status);
 }
 
@@ skipping 146 matching lines @@
   return OK;
 }
 
 void SpdySession::OnWriteComplete(int result) {
   // Releasing the in-flight write can have a side-effect of dropping
   // the last reference to |this|.  Hold a reference through this
   // function.
   scoped_refptr<SpdySession> self(this);
 
   DCHECK(write_pending_);
-  DCHECK_GT(in_flight_write_.buffer()->BytesRemaining(), 0);
+  DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
 
   last_activity_time_ = base::TimeTicks::Now();
   write_pending_ = false;
 
   if (result < 0) {
-    in_flight_write_.Release();
+    in_flight_write_.reset();
     in_flight_write_frame_type_ = DATA;
-    CloseSessionOnError(static_cast<net::Error>(result), true, "Write error");
+    in_flight_write_frame_size_ = 0;
+    in_flight_write_stream_ = NULL;
+    CloseSessionOnError(static_cast<Error>(result), true, "Write error");
     return;
   }
 
   // It should not be possible to have written more bytes than our
   // in_flight_write_.
-  DCHECK_LE(result, in_flight_write_.buffer()->BytesRemaining());
+  DCHECK_LE(static_cast<size_t>(result),
+            in_flight_write_->GetRemainingSize());
 
-  in_flight_write_.buffer()->DidConsume(result);
+  if (result > 0) {
+    in_flight_write_->Consume(static_cast<size_t>(result));
 
-  // We only notify the stream when we've fully written the pending frame.
-  if (in_flight_write_.buffer()->BytesRemaining() == 0) {
-    DCHECK_GT(result, 0);
+    // We only notify the stream when we've fully written the pending frame.
+    if (in_flight_write_->GetRemainingSize() == 0) {
+        // It is possible that the stream was cancelled while we were
+        // writing to the socket.
+      if (in_flight_write_stream_ && !in_flight_write_stream_->cancelled()) {
+        DCHECK_GT(in_flight_write_frame_size_, 0u);
+        in_flight_write_stream_->OnFrameWriteComplete(
+            in_flight_write_frame_type_,
+            in_flight_write_frame_size_);
+      }
 
-    scoped_refptr<SpdyStream> stream = in_flight_write_.stream();
-
-    // It is possible that the stream was cancelled while we were writing
-    // to the socket.
-    if (stream && !stream->cancelled()) {
-      DCHECK_GT(in_flight_write_.buffer()->size(), 0);
-      stream->OnFrameWriteComplete(
-          in_flight_write_frame_type_,
-          static_cast<size_t>(in_flight_write_.buffer()->size()));
+      // Cleanup the write which just completed.
+      in_flight_write_.reset();
+      in_flight_write_frame_type_ = DATA;
+      in_flight_write_frame_size_ = 0;
+      in_flight_write_stream_ = NULL;
     }
-
-    // Cleanup the write which just completed.
-    in_flight_write_.Release();
-    in_flight_write_frame_type_ = DATA;
   }
 
   // Write more data.  We're already in a continuation, so we can go
   // ahead and write it immediately (without going back to the message
   // loop).
   WriteSocketLater();
 }
 
 void SpdySession::WriteSocketLater() {
   if (delayed_write_pending_)
@@ skipping 19 matching lines @@
   if (!IsConnected())
     return;
 
   if (write_pending_)   // Another write is in progress still.
     return;
 
   // Loop sending frames until we've sent everything or until the write
   // returns error (or ERR_IO_PENDING).
   DCHECK(buffered_spdy_framer_.get());
   while (true) {
-    if (in_flight_write_.buffer()) {
-      DCHECK_GT(in_flight_write_.buffer()->BytesRemaining(), 0);
+    if (in_flight_write_) {
+      DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
     } else {
       // Grab the next frame to send.
       SpdyFrameType frame_type = DATA;
-      scoped_ptr<SpdyFrameProducer> producer;
+      scoped_ptr<SpdyBufferProducer> producer;
       scoped_refptr<SpdyStream> stream;
       if (!write_queue_.Dequeue(&frame_type, &producer, &stream))
         break;
 
       // It is possible that a stream had data to write, but a
       // WINDOW_UPDATE frame has been received which made that
       // stream no longer writable.
       // TODO(rch): consider handling that case by removing the
       // stream from the writable queue?
       if (stream.get() && stream->cancelled())
         continue;
 
       // Activate the stream only when sending the SYN_STREAM frame to
       // guarantee monotonically-increasing stream IDs.
       if (frame_type == SYN_STREAM) {
         if (stream.get() && stream->stream_id() == 0) {
           ActivateStream(stream);
         } else {
           NOTREACHED();
           continue;
         }
       }
 
-      scoped_ptr<SpdyFrame> frame = producer->ProduceFrame();
-      if (!frame) {
+      in_flight_write_ = producer->ProduceBuffer();
+      if (!in_flight_write_) {
         NOTREACHED();
         continue;
       }
-      DCHECK_GT(frame->size(), 0u);
-
-      // TODO(mbelshe): We have too much copying of data here.
-      scoped_refptr<IOBufferWithSize> buffer =
-          new IOBufferWithSize(frame->size());
-      memcpy(buffer->data(), frame->data(), frame->size());
-      in_flight_write_ = SpdyIOBuffer(buffer, frame->size(), stream);
       in_flight_write_frame_type_ = frame_type;
+      in_flight_write_frame_size_ = in_flight_write_->GetRemainingSize();
+      DCHECK_GE(in_flight_write_frame_size_,
+                buffered_spdy_framer_->GetFrameMinimumSize());
+      in_flight_write_stream_ = stream;
     }
 
     write_pending_ = true;
+    // We keep |in_flight_write_| alive until OnWriteComplete(), so
+    // it's okay to use GetIOBufferForRemainingData() since the socket
+    // doesn't use the IOBuffer past OnWriteComplete().
     int rv = connection_->socket()->Write(
-        in_flight_write_.buffer(),
-        in_flight_write_.buffer()->BytesRemaining(),
+        in_flight_write_->GetIOBufferForRemainingData(),
+        in_flight_write_->GetRemainingSize(),
         base::Bind(&SpdySession::OnWriteComplete, weak_factory_.GetWeakPtr()));
     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.
     if (rv < 0)
@@ skipping 169 matching lines @@
 
 void SpdySession::EnqueueSessionWrite(RequestPriority priority,
                                       SpdyFrameType frame_type,
                                       scoped_ptr<SpdyFrame> frame) {
   DCHECK(frame_type == RST_STREAM ||
          frame_type == SETTINGS ||
          frame_type == WINDOW_UPDATE ||
          frame_type == PING);
   EnqueueWrite(
       priority, frame_type,
-      scoped_ptr<SpdyFrameProducer>(new SimpleFrameProducer(frame.Pass())),
+      scoped_ptr<SpdyBufferProducer>(
+          new SimpleBufferProducer(
+              scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass())))),
       NULL);
 }
 
 void SpdySession::EnqueueWrite(RequestPriority priority,
                                SpdyFrameType frame_type,
-                               scoped_ptr<SpdyFrameProducer> producer,
+                               scoped_ptr<SpdyBufferProducer> producer,
                                const scoped_refptr<SpdyStream>& stream) {
   write_queue_.Enqueue(priority, frame_type, producer.Pass(), stream);
   WriteSocketLater();
 }
 
 void SpdySession::ActivateStream(SpdyStream* stream) {
   if (stream->stream_id() == 0) {
     stream->set_stream_id(GetNewStreamId());
     created_streams_.erase(scoped_refptr<SpdyStream>(stream));
   }
@@ skipping 97 matching lines @@
                                     const char* data,
                                     size_t len,
                                     bool fin) {
   DCHECK_LT(len, 1u << 24);
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_DATA,
         base::Bind(&NetLogSpdyDataCallback, stream_id, len, fin));
   }
 
+  ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
+
   // By the time data comes in, the stream may already be inactive.
-  if (!IsStreamActive(stream_id))
+  if (it == active_streams_.end())
     return;
 
   // Only decrease the window size for data for active streams.
   if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION && len > 0)
     DecreaseRecvWindowSize(static_cast<int32>(len));
 
-  scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
-  stream->OnDataReceived(data, len);
+  scoped_ptr<SpdyBuffer> buffer;
+  if (data) {
+    DCHECK_GT(len, 0u);
+    buffer.reset(new SpdyBuffer(data, len));
+  } else {
+    DCHECK_EQ(len, 0u);
+  }
+  it->second->OnDataReceived(buffer.Pass());
 }
 
 void SpdySession::OnSetting(SpdySettingsIds id,
                             uint8 flags,
                             uint32 value) {
   HandleSetting(id, value);
   http_server_properties_->SetSpdySetting(
       host_port_pair(),
       id,
       static_cast<SpdySettingsFlags>(flags),
@@ skipping 248 matching lines @@
   if (!IsStreamActive(stream_id)) {
     // NOTE:  it may just be that the stream was cancelled.
     LOG(WARNING) << "Received RST for invalid stream" << stream_id;
     return;
   }
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
   CHECK(!stream->cancelled());
 
   if (status == 0) {
-    stream->OnDataReceived(NULL, 0);
+    stream->OnDataReceived(scoped_ptr<SpdyBuffer>());
   } else if (status == RST_STREAM_REFUSED_STREAM) {
     DeleteStream(stream_id, ERR_SPDY_SERVER_REFUSED_STREAM);
   } else {
     RecordProtocolErrorHistogram(
         PROTOCOL_ERROR_RST_STREAM_FOR_NON_ACTIVE_STREAM);
     stream->LogStreamError(
         ERR_SPDY_PROTOCOL_ERROR,
         base::StringPrintf("SPDY stream closed with status: %d", status));
     // TODO(mbelshe): Map from Spdy-protocol errors to something sensical.
     //                For now, it doesn't matter much - it is a protocol error.
@@ skipping 600 matching lines @@
   }
 
   session_recv_window_size_ -= delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_RECV_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_recv_window_size_));
 }
 
 }  // namespace net