Update the FLIP session to use the FlipIOBuffer....

net/flip/flip_session.cc

 // Copyright (c) 2009 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/flip/flip_session.h"
 
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/rand_util.h"
@@ skipping 10 matching lines @@
 #include "net/http/http_response_info.h"
 #include "net/socket/client_socket_factory.h"
 #include "net/socket/ssl_client_socket.h"
 #include "net/tools/dump_cache/url_to_filename_encoder.h"
 
 namespace net {
 
 // static
 scoped_ptr<FlipSessionPool> FlipSession::session_pool_;
 bool FlipSession::use_ssl_ = true;
-int PrioritizedIOBuffer::order_ = 0;
 
 FlipSession* FlipSession::GetFlipSession(
     const HostResolver::RequestInfo& info,
     HttpNetworkSession* session) {
   if (!session_pool_.get())
     session_pool_.reset(new FlipSessionPool());
   return session_pool_->Get(info, session);
 }
 
 FlipSession::FlipSession(std::string host, HttpNetworkSession* session)
@@ skipping 65 matching lines @@
   static const std::string kHttpProtocolVersion("HTTP/1.1");
 
   HttpUtil::HeadersIterator it(info->extra_headers.begin(),
                                info->extra_headers.end(),
                                "\r\n");
   while (it.GetNext()) {
     std::string name = StringToLowerASCII(it.name());
     (*headers)[name] = it.values();
   }
 
-#undef DIVERT_URLS_TO_TEST_SERVER
-#ifdef DIVERT_URLS_TO_TEST_SERVER
-  // TODO(mbelshe): Figure out how to remove this.  This is just for hooking
-  // up to a test server.
-  // For testing content on our test server, we modify the URL.
-  GURL url = info->url;
-  FilePath path(UrlToFilenameEncoder::Encode(url.spec(), FilePath()));
-
-  // We do the unusual conversion from a FilePath::StringType to
-  // an ascii string.  Recognize that StringType is a wstring on windows,
-  // so a failure is technically possible, but this is just used as a test
-  // case, so it's okay.  This code will be deleted.
-  std::string hack_url("/");
-#if defined(OS_WIN)
-  hack_url.append(WideToASCII(path.value()));
-#else
-  hack_url.append(path.value());
-#endif
-
-  // switch backslashes.  HACK
-  std::string::size_type pos(0);
-  while ((pos = hack_url.find('\\', pos)) != std::string::npos) {
-    hack_url.replace(pos, 1, "/");
-    pos += 1;
-  }
-#else
-  std::string hack_url(info->url.PathForRequest());
-#endif  // REWRITE_URLS
-
   (*headers)["method"] = info->method;
-  (*headers)["url"] = hack_url;
+  (*headers)["url"] = info->url.PathForRequest();
   (*headers)["version"] = kHttpProtocolVersion;
   (*headers)["host"] = GetHostAndOptionalPort(info->url);
   if (info->user_agent.length())
     (*headers)["user-agent"] = info->user_agent;
   if (!info->referrer.is_empty())
     (*headers)["referer"] = info->referrer.spec();
 
   // Honor load flags that impact proxy caches.
   if (info->load_flags & LOAD_BYPASS_CACHE) {
     (*headers)["pragma"] = "no-cache";
@@ skipping 69 matching lines @@
     flags = flip::CONTROL_FLAG_FIN;
 
   // Create a SYN_STREAM packet and add to the output queue.
   flip::FlipSynStreamControlFrame* syn_frame =
       flip_framer_.CreateSynStream(stream_id, priority, flags, false, &headers);
   int length = sizeof(flip::FlipFrame) + syn_frame->length();
   IOBufferWithSize* buffer =
       new IOBufferWithSize(length);
   memcpy(buffer->data(), syn_frame, length);
   delete[] syn_frame;
-  queue_.push(PrioritizedIOBuffer(buffer, priority));
+  queue_.push(FlipIOBuffer(buffer, priority, stream));
 
   static StatsCounter flip_requests("flip.requests");
   flip_requests.Increment();
 
   LOG(INFO) << "FETCHING: " << delegate->request()->url.spec();
 
 
   // TODO(mbelshe): Implement POST Data here
 
   // Schedule to write to the socket after we've made it back
@@ skipping 122 matching lines @@
   ReadSocket();
 }
 
 void FlipSession::OnWriteComplete(int result) {
   DCHECK(write_pending_);
   write_pending_ = false;
 
   LOG(INFO) << "Flip write complete (result=" << result << ")";
 
   if (result >= 0) {
+    // TODO(mbelshe) Verify that we wrote ALL the bytes we needed to.
+    //               The code current is broken in the case of a partial write.
+    DCHECK_EQ(static_cast<size_t>(result), in_flight_write_.size());
+
     // Cleanup the write which just completed.
     in_flight_write_.release();
 
     // 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();
   } else {
     // TODO(mbelshe): Deal with result < 0 error case.
     NOTIMPLEMENTED();
@@ skipping 40 matching lines @@
   delayed_write_pending_ = false;
 
   // If the socket isn't connected yet, just wait; we'll get called
   // again when the socket connection completes.
   if (!connection_ready_)
     return;
 
   if (write_pending_)   // Another write is in progress still.
     return;
 
+  // Loop sending frames until we've sent everything or until the write
+  // returns ERR_IO_PENDING.

[wtc, 2009/11/03 23:28:13]

This comment needs updating to match the new code.
I think you can just say "until the write fails".

   while (queue_.size()) {
-    const size_t kMaxSegmentSize = 1430;
-    const size_t kMaxPayload = 4 * kMaxSegmentSize;
-    size_t max_size = std::max(kMaxPayload, queue_.top().size());
+    // Grab the next FlipFrame to send.
+    FlipIOBuffer next_buffer = queue_.top();
+    queue_.pop();
 
-    size_t bytes = 0;
-    // If we have multiple IOs to do, accumulate up to 4 MSS's worth of data
-    // and send them in batch.
-    IOBufferWithSize* buffer = new IOBufferWithSize(max_size);
-    while (queue_.size() && bytes < max_size) {
-      PrioritizedIOBuffer next_buffer = queue_.top();
+    // We've deferred compression until just before we write it to the socket,
+    // which is now.
+    flip::FlipFrame* uncompressed_frame =
+        reinterpret_cast<flip::FlipFrame*>(next_buffer.buffer()->data());
+    scoped_array<flip::FlipFrame> compressed_frame(
+        flip_framer_.CompressFrame(uncompressed_frame));
+    size_t size = compressed_frame.get()->length() + sizeof(flip::FlipFrame);
 
-      // Now that we're outputting the frame, we can finally compress it.
-      flip::FlipFrame* uncompressed_frame =
-          reinterpret_cast<flip::FlipFrame*>(next_buffer.buffer()->data());
-      scoped_array<flip::FlipFrame> compressed_frame(
-          flip_framer_.CompressFrame(uncompressed_frame));
-      size_t size = compressed_frame.get()->length() + sizeof(flip::FlipFrame);
-      if (bytes + size > kMaxPayload)
-        break;
-      memcpy(buffer->data() + bytes, compressed_frame.get(), size);
-      bytes += size;
-      queue_.pop();
-      next_buffer.release();
-    }
-    DCHECK(bytes > 0);
-    in_flight_write_ = PrioritizedIOBuffer(buffer, 0);
+

[wtc, 2009/11/03 23:28:13]

Nit: you added an extra blank line.

+    DCHECK(size > 0);
+
+    // TODO(mbelshe): We have too much copying of data here.
+    IOBufferWithSize* buffer = new IOBufferWithSize(size);
+    memcpy(buffer->data(), compressed_frame.get(), size);
+
+    // Attempt to send the frame.
+    in_flight_write_ = FlipIOBuffer(buffer, 0, next_buffer.stream());
     write_pending_ = true;
-
     int rv = connection_.socket()->Write(in_flight_write_.buffer(),
-                                         bytes, &write_callback_);
+                                         size, &write_callback_);
     if (rv == net::ERR_IO_PENDING)
       break;
+
+    // We sent the frame successfully.
     OnWriteComplete(rv);
+
+    // TODO(mbelshe):  Test this error case.
+    if (rv < 0)

[wtc, 2009/11/03 23:28:13]

I agree with this new code.  We may need to do something
to mark the connection as in an error state and cause the
connection to be closed.

+      break;
   }
 }
 
 void FlipSession::CloseAllStreams(net::Error code) {
   LOG(INFO) << "Closing all FLIP Streams";
 
   static StatsCounter abandoned_streams("flip.abandoned_streams");
   static StatsCounter abandoned_push_streams("flip.abandoned_push_streams");
 
   if (active_streams_.size()) {
@@ skipping 244 matching lines @@
     DeactivateStream(stream_id);
     delete stream;
   }
 }
 
 void FlipSession::OnLameDuck() {
   NOTIMPLEMENTED();
 }
 
 }  // namespace net