Instead of enqueueing SPDY frames, enqueue SPDY streams that are ready to produce data.

net/spdy/spdy_session.h

 // 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.
 
 #ifndef NET_SPDY_SPDY_SESSION_H_
 #define NET_SPDY_SPDY_SESSION_H_
 
 #include <algorithm>
 #include <list>
 #include <map>
 #include <queue>
 #include <string>
 
 #include "base/gtest_prod_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
 #include "base/time.h"
 #include "net/base/io_buffer.h"
 #include "net/base/load_states.h"
 #include "net/base/net_errors.h"
 #include "net/base/request_priority.h"
+#include "net/base/ssl_client_cert_type.h"
 #include "net/base/ssl_config_service.h"
 #include "net/base/upload_data_stream.h"
 #include "net/socket/client_socket_handle.h"
 #include "net/socket/ssl_client_socket.h"
 #include "net/socket/stream_socket.h"
 #include "net/spdy/buffered_spdy_framer.h"
 #include "net/spdy/spdy_credential_state.h"
 #include "net/spdy/spdy_io_buffer.h"
 #include "net/spdy/spdy_protocol.h"
 #include "net/spdy/spdy_session_pool.h"
@@ skipping 54 matching lines @@
                SpdyProtocolErrorDetails_SpdyErrors_mismatch);
 
 COMPILE_ASSERT(PROTOCOL_ERROR_UNEXPECTED_PING ==
                static_cast<SpdyProtocolErrorDetails>(NUM_STATUS_CODES +
                                                      STATUS_CODE_INVALID),
                SpdyProtocolErrorDetails_SpdyErrors_mismatch);
 
 class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
                                public BufferedSpdyFramerVisitorInterface {
  public:
+  // Defines an interface for producing SpdyIOBuffers.
+  class NET_EXPORT_PRIVATE SpdyIOBufferProducer {
+   public:
+    SpdyIOBufferProducer() {}
+
+    // Returns a newly created SpdyIOBuffer, owned by the caller, or NULL
+    // if not buffer is ready to be produced.
+    virtual SpdyIOBuffer* ProduceNextBuffer(SpdySession* session) = 0;
+
+    virtual RequestPriority GetPriority() const = 0;
+
+    virtual ~SpdyIOBufferProducer() {}
+
+   protected:
+    // Activates |spdy_stream| in |spdy_session|.
+    static void ActivateStream(SpdySession* spdy_session,
+                               SpdyStream* spdy_stream);
+
+    static SpdyIOBuffer* CreateIOBuffer(SpdyFrame* frame,
+                                        RequestPriority priority,
+                                        SpdyStream* spdy_stream);
+
+   private:
+    DISALLOW_COPY_AND_ASSIGN(SpdyIOBufferProducer);
+  };
+
   // Create a new SpdySession.
   // |host_port_proxy_pair| is the host/port that this session connects to, and
   // the proxy configuration settings that it's using.
   // |spdy_session_pool| is the SpdySessionPool that owns us.  Its lifetime must
   // strictly be greater than |this|.
   // |session| is the HttpNetworkSession.  |net_log| is the NetLog that we log
   // network events to.
   SpdySession(const HostPortProxyPair& host_port_proxy_pair,
               SpdySessionPool* spdy_session_pool,
               HttpServerProperties* http_server_properties,
@@ skipping 44 matching lines @@
   // For SSL-based sessions, verifies that the server certificate in use by
   // this session provides authentication for the domain and no client
   // certificate or channel ID was sent to the original server during the SSL
   // handshake.  NOTE:  This function can have false negatives on some
   // platforms.
   // TODO(wtc): rename this function and the Net.SpdyIPPoolDomainMatch
   // histogram because this function does more than verifying domain
   // authentication now.
   bool VerifyDomainAuthentication(const std::string& domain);
 
+  // Records that |stream| has a write available from |producer|.
+  // |producer| will be owned by this SpdySession.
+  void SetStreamHasWriteAvailable(SpdyStream* stream,
+                                  SpdyIOBufferProducer* producer);
+
   // Send the SYN frame for |stream_id|. This also sends PING message to check
   // the status of the connection.
-  int WriteSynStream(
+  SpdySynStreamControlFrame* CreateSynStream(
       SpdyStreamId stream_id,
       RequestPriority priority,
       uint8 credential_slot,
       SpdyControlFlags flags,
       const SpdyHeaderBlock& headers);
 
   // Write a CREDENTIAL frame to the session.
-  int WriteCredentialFrame(const std::string& origin,
-                           SSLClientCertType type,
-                           const std::string& key,
-                           const std::string& cert,
-                           RequestPriority priority);
+  SpdyCredentialControlFrame* CreateCredentialFrame(const std::string& origin,
+                                                    SSLClientCertType type,
+                                                    const std::string& key,
+                                                    const std::string& cert,
+                                                    RequestPriority priority);
 
   // Write a data frame to the stream.
   // Used to create and queue a data frame for the given stream.
-  int WriteStreamData(SpdyStreamId stream_id, net::IOBuffer* data,
-                      int len,
-                      SpdyDataFlags flags);
+  SpdyDataFrame* CreateDataFrame(SpdyStreamId stream_id,
+                                 net::IOBuffer* data, int len,
+                                 SpdyDataFlags flags);
 
   // Close a stream.
   void CloseStream(SpdyStreamId stream_id, int status);
 
+  // Close a stream that has been created but is not yet active.
+  void CloseCreatedStream(SpdyStream* stream, int status);
+
   // Reset a stream by sending a RST_STREAM frame with given status code.
   // Also closes the stream.  Was not piggybacked to CloseStream since not
   // all of the calls to CloseStream necessitate sending a RST_STREAM.
   void ResetStream(SpdyStreamId stream_id,
                    SpdyStatusCodes status,
                    const std::string& description);
 
   // Check if a stream is active.
   bool IsStreamActive(SpdyStreamId stream_id) const;
 
@@ skipping 66 matching lines @@
   bool WasEverUsed() const {
     return connection_->socket()->WasEverUsed();
   }
 
   void set_spdy_session_pool(SpdySessionPool* pool) {
     spdy_session_pool_ = NULL;
   }
 
   // Returns true if session is not currently active
   bool is_active() const {
-    return !active_streams_.empty();
+    return !active_streams_.empty() || !created_streams_.empty();
   }
 
   // Access to the number of active and pending streams.  These are primarily
   // available for testing and diagnostics.
   size_t num_active_streams() const { return active_streams_.size(); }
   size_t num_unclaimed_pushed_streams() const {
       return unclaimed_pushed_streams_.size();
   }
+  size_t num_created_streams() const { return created_streams_.size(); }
 
   // Returns true if flow control is enabled for the session.
   bool is_flow_control_enabled() const {
     return flow_control_;
   }
 
   // Returns the current |initial_send_window_size_|.
   int32 initial_send_window_size() const {
     return initial_send_window_size_;
   }
@@ skipping 59 matching lines @@
     const BoundNetLog* stream_net_log;
     CompletionCallback callback;
   };
   typedef std::queue<PendingCreateStream, std::list<PendingCreateStream> >
       PendingCreateStreamQueue;
   typedef std::map<int, scoped_refptr<SpdyStream> > ActiveStreamMap;
   typedef std::map<std::string,
       std::pair<scoped_refptr<SpdyStream>, base::TimeTicks> > PushedStreamMap;
   typedef std::priority_queue<SpdyIOBuffer> OutputQueue;
 
+  typedef std::set<scoped_refptr<SpdyStream> > CreatedStreamSet;
+  typedef std::map<SpdyIOBufferProducer*, SpdyStream*> StreamProducerMap;
+  class SpdyIOBufferProducerCompare {
+   public:
+    bool operator() (const SpdyIOBufferProducer* lhs,
+                     const SpdyIOBufferProducer* rhs) const {
+      return lhs->GetPriority() < rhs->GetPriority();
+    }
+  };
+  typedef std::priority_queue<SpdyIOBufferProducer*,
+                              std::vector<SpdyIOBufferProducer*>,
+                              SpdyIOBufferProducerCompare> WriteQueue;
+
   struct CallbackResultPair {
     CallbackResultPair(const CompletionCallback& callback_in, int result_in)
         : callback(callback_in), result(result_in) {}
     ~CallbackResultPair();
 
     CompletionCallback callback;
     int result;
   };
 
   typedef std::map<const scoped_refptr<SpdyStream>*, CallbackResultPair>
@@ skipping 58 matching lines @@
   // Write current data to the socket.
   void WriteSocketLater();
   void WriteSocket();
 
   // Get a new stream id.
   int GetNewStreamId();
 
   // Queue a frame for sending.
   // |frame| is the frame to send.
   // |priority| is the priority for insertion into the queue.
-  // |stream| is the stream which this IO is associated with (or NULL).
-  void QueueFrame(SpdyFrame* frame, RequestPriority priority,
-                  SpdyStream* stream);
+  void QueueFrame(SpdyFrame* frame, RequestPriority priority);
 
   // Track active streams in the active stream list.
   void ActivateStream(SpdyStream* stream);
   void DeleteStream(SpdyStreamId id, int status);
 
   // Removes this session from the session pool.
   void RemoveFromPool();
 
   // Check if we have a pending pushed-stream for this url
   // Returns the stream if found (and returns it from the pending
   // list), returns NULL otherwise.
   scoped_refptr<SpdyStream> GetActivePushStream(const std::string& url);
 
   // Calls OnResponseReceived().
   // Returns true if successful.
   bool Respond(const SpdyHeaderBlock& headers,
                const scoped_refptr<SpdyStream> stream);
 
   void RecordPingRTTHistogram(base::TimeDelta duration);
   void RecordHistograms();
   void RecordProtocolErrorHistogram(SpdyProtocolErrorDetails details);
 
   // Closes all streams.  Used as part of shutdown.
   void CloseAllStreams(net::Error status);
 
+  void LogAbandonedStream(const scoped_refptr<SpdyStream>& stream,
+                          net::Error status);
+
   // Invokes a user callback for stream creation.  We provide this method so it
   // can be deferred to the MessageLoop, so we avoid re-entrancy problems.
   void InvokeUserStreamCreationCallback(scoped_refptr<SpdyStream>* stream);
 
   // Remove old unclaimed pushed streams.
   void DeleteExpiredPushedStreams();
 
   // BufferedSpdyFramerVisitorInterface:
   virtual void OnError(SpdyFramer::SpdyError error_code) OVERRIDE;
   virtual void OnStreamError(SpdyStreamId stream_id,
@@ skipping 98 matching lines @@
   // the server to start pushing the stream]) or there are still network events
   // incoming even though the consumer has already gone away (cancellation).
   // TODO(willchan): Perhaps we should separate out cancelled streams and move
   // them into a separate ActiveStreamMap, and not deliver network events to
   // them?
   ActiveStreamMap active_streams_;
   // Map of all the streams that have already started to be pushed by the
   // server, but do not have consumers yet.
   PushedStreamMap unclaimed_pushed_streams_;
 
-  // As we gather data to be sent, we put it into the output queue.
-  OutputQueue queue_;
+  // Set of all created streams but that have not yet sent any frames.
+  CreatedStreamSet created_streams_;
+
+  // As streams have data to be sent, we put them into the write queue.
+  WriteQueue write_queue_;
+
+  // Mapping from SpdyIOBufferProducers to their corresponding SpdyStream
+  // so that when a stream is destroyed, we can remove the corresponding
+  // producer from |write_queue_|.
+  StreamProducerMap stream_producers_;
 
   // The packet we are currently sending.
   bool write_pending_;            // Will be true when a write is in progress.
   SpdyIOBuffer in_flight_write_;  // This is the write buffer in progress.
 
   // Flag if we have a pending message scheduled for WriteSocket.
   bool delayed_write_pending_;
 
   // Flag if we're using an SSL connection for this SpdySession.
   bool is_secure_;
@@ skipping 89 matching lines @@
   base::TimeDelta hung_interval_;
 
   // This SPDY proxy is allowed to push resources from origins that are
   // different from those of their associated streams.
   HostPortPair trusted_spdy_proxy_;
 };
 
 }  // namespace net
 
 #endif  // NET_SPDY_SPDY_SESSION_H_