Implement a read-size throttle within the TCP socket.

net/socket/tcp_client_socket_win.cc

 // Copyright (c) 2006-2008 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/socket/tcp_client_socket_win.h"
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/field_trial.h"  // for SlowStart trial
 #include "base/memory_debug.h"
 #include "base/string_util.h"
 #include "base/sys_info.h"
 #include "base/trace_event.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/base/winsock_init.h"
 
 namespace net {
 
@@ skipping 92 matching lines @@
   // |write_overlapped_| is only used for Write();
   OVERLAPPED read_overlapped_;
   OVERLAPPED write_overlapped_;
 
   // The buffers used in Read() and Write().
   WSABUF read_buffer_;
   WSABUF write_buffer_;
   scoped_refptr<IOBuffer> read_iobuffer_;
   scoped_refptr<IOBuffer> write_iobuffer_;
 
+  // Throttle the read size based on our current slow start state.
+  // Returns the throttled read size.
+  int ThrottleReadSize(int size) {

[wtc, 2009/08/25 23:58:16]

This method and the related data members should be in
TCPClientSocketWin.  The Core class should only contain
members that need to be preserved when we cancel a
TCPClientSocket that has an async IO operation in progress.

+    if (!use_slow_start_throttle_)
+      return size;
+
+    if (slow_start_throttle_ < kMaxSlowStartThrottle) {
+      size = std::min(size, slow_start_throttle_);
+      slow_start_throttle_ *= 2;
+    }
+    return size;
+  }
+
  private:
   class ReadDelegate : public base::ObjectWatcher::Delegate {
    public:
     explicit ReadDelegate(Core* core) : core_(core) {}
     virtual ~ReadDelegate() {}
 
     // base::ObjectWatcher::Delegate methods:
     virtual void OnObjectSignaled(HANDLE object);
 
    private:
@@ skipping 18 matching lines @@
   // |reader_| handles the signals from |read_watcher_|.
   ReadDelegate reader_;
   // |writer_| handles the signals from |write_watcher_|.
   WriteDelegate writer_;
 
   // |read_watcher_| watches for events from Connect() and Read().
   base::ObjectWatcher read_watcher_;
   // |write_watcher_| watches for events from Write();
   base::ObjectWatcher write_watcher_;
 
+  // When doing reads from the socket, we try to mirror TCP's slow start.
+  // We do this because otherwise the async IO subsystem artifically delays
+  // returning data to the application.
+  static const int kInitialSlowStartThrottle = 1 * 1024;
+  static const int kMaxSlowStartThrottle = 32 * kInitialSlowStartThrottle;
+  int slow_start_throttle_;
+
+  static bool use_slow_start_throttle_;
+  static bool trial_initialized_;
+
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
+bool TCPClientSocketWin::Core::use_slow_start_throttle_ = true;
+bool TCPClientSocketWin::Core::trial_initialized_ = false;
+
 TCPClientSocketWin::Core::Core(
     TCPClientSocketWin* socket)
     : socket_(socket),
       ALLOW_THIS_IN_INITIALIZER_LIST(reader_(this)),
-      ALLOW_THIS_IN_INITIALIZER_LIST(writer_(this)) {
+      ALLOW_THIS_IN_INITIALIZER_LIST(writer_(this)),
+      slow_start_throttle_(kInitialSlowStartThrottle) {
   memset(&read_overlapped_, 0, sizeof(read_overlapped_));
   memset(&write_overlapped_, 0, sizeof(write_overlapped_));
+
+  // Initialize the AsyncSlowStart FieldTrial.
+  if (!trial_initialized_) {
+    trial_initialized_ = true;
+    scoped_refptr<FieldTrial> trial = new FieldTrial("AsyncSlowStart", 100);
+    int my_group = trial->AppendGroup("_AsyncSlowStart", 50);
+    trial->AppendGroup("_AsyncSlowStart_off", 50);
+
+    // Only use the throttling if the FieldTrial is enabled.
+    use_slow_start_throttle_ = trial->group() == my_group;
+  }
 }
 
 TCPClientSocketWin::Core::~Core() {
   // Make sure the message loop is not watching this object anymore.
   read_watcher_.StopWatching();
   write_watcher_.StopWatching();
 
   WSACloseEvent(read_overlapped_.hEvent);
   memset(&read_overlapped_, 0, sizeof(read_overlapped_));
   WSACloseEvent(write_overlapped_.hEvent);
@@ skipping 180 matching lines @@
 }
 
 int TCPClientSocketWin::Read(IOBuffer* buf,
                              int buf_len,
                              CompletionCallback* callback) {
   DCHECK_NE(socket_, INVALID_SOCKET);
   DCHECK(!waiting_read_);
   DCHECK(!read_callback_);
   DCHECK(!core_->read_iobuffer_);
 
+  buf_len = core_->ThrottleReadSize(buf_len);
+
   core_->read_buffer_.len = buf_len;
   core_->read_buffer_.buf = buf->data();
 
   TRACE_EVENT_BEGIN("socket.read", this, "");
   // TODO(wtc): Remove the CHECK after enough testing.
   CHECK(WaitForSingleObject(core_->read_overlapped_.hEvent, 0) == WAIT_TIMEOUT);
   DWORD num, flags = 0;
   int rv = WSARecv(socket_, &core_->read_buffer_, 1, &num, &flags,
                    &core_->read_overlapped_, NULL);
   if (rv == 0) {
@@ skipping 199 matching lines @@
   BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
                                    &num_bytes, FALSE, &flags);
   WSAResetEvent(core_->write_overlapped_.hEvent);
   TRACE_EVENT_END("socket.write", this, StringPrintf("%d bytes", num_bytes));
   waiting_write_ = false;
   core_->write_iobuffer_ = NULL;
   DoWriteCallback(ok ? num_bytes : MapWinsockError(WSAGetLastError()));
 }
 
 }  // namespace net