accept-backoff.patch

From 04b4c8182f598b9c1b1d9cf31a547d3adc860ae1 Mon Sep 17 00:00:00 2001
From: Bert Belder <[email protected]>
Date: Thu, 10 May 2012 16:08:36 +0200
Subject: [PATCH 1/1] Accept backoff

---
 deps/uv/include/uv-private/uv-unix.h |    4 +++-
 deps/uv/include/uv.h                 |   15 +++++++++++++--
 deps/uv/src/unix/internal.h          |    3 ++-
 deps/uv/src/unix/stream.c            |   24 +++++++++++++++++++++++-
 deps/uv/src/unix/tcp.c               |    5 +++++
 lib/child_process.js                 |   16 +++++++---------
 src/tcp_wrap.cc                      |    4 ----
 7 files changed, 53 insertions(+), 18 deletions(-)

diff --git a/deps/uv/include/uv-private/uv-unix.h b/deps/uv/include/uv-private/uv-unix.h
index 1c33684..c805155 100644
--- a/deps/uv/include/uv-private/uv-unix.h
+++ b/deps/uv/include/uv-private/uv-unix.h
@@ -109,7 +109,9 @@ typedef void* uv_lib_t;
   int delayed_error; \
   uv_connection_cb connection_cb; \
   int accepted_fd; \
-  int blocking;
+  int blocking; \
+  /* ABI change! Don't land. */ \
+  ev_idle idle_watcher;
 
 
 /* UV_TCP */
diff --git a/deps/uv/include/uv.h b/deps/uv/include/uv.h
index 340ab11..21c0aa4 100644
--- a/deps/uv/include/uv.h
+++ b/deps/uv/include/uv.h
@@ -506,12 +506,23 @@ UV_EXTERN int uv_tcp_keepalive(uv_tcp_t* handle, int enable,
     unsigned int delay);
 
 /*
- * This setting applies to Windows only.
- * Enable/disable simultaneous asynchronous accept requests that are
+ * TODO: this name is not very appropriate. The function basically controls
+ * aggressive accept() mode, which is enabled by default. It can be disabled to
+ * improve load balancing in a cluster setup.
+ *
+ * TODO: consider merging with uv_pipe_pending_instances()
+ *
+ * Windows: Enable/disable simultaneous asynchronous accept requests that are
  * queued by the operating system when listening for new tcp connections.
  * This setting is used to tune a tcp server for the desired performance.
  * Having simultaneous accepts can significantly improve the rate of
  * accepting connections (which is why it is enabled by default).
+ *
+ * Unix: if enabled, accept connections as fast as possible. If disabled,
+ * temporarily back off from accepting new connections until one loop
+ * iteration passes in which no new events are dequeued. This improves
+ * load balancing in a cluster-like setup at the expense of overall
+ * performance.
  */
 UV_EXTERN int uv_tcp_simultaneous_accepts(uv_tcp_t* handle, int enable);
 
diff --git a/deps/uv/src/unix/internal.h b/deps/uv/src/unix/internal.h
index fc7aad1..e766bb6 100644
--- a/deps/uv/src/unix/internal.h
+++ b/deps/uv/src/unix/internal.h
@@ -153,7 +153,8 @@ enum {
   UV_TCP_NODELAY   = 0x080,  /* Disable Nagle. */
   UV_TCP_KEEPALIVE = 0x100,  /* Turn on keep-alive. */
   UV_TIMER_ACTIVE  = 0x080,
-  UV_TIMER_REPEAT  = 0x100
+  UV_TIMER_REPEAT  = 0x100,
+  UV_STREAM_BALANCED_ACCEPT = 0x200
 };
 
 size_t uv__strlcpy(char* dst, const char* src, size_t size);
diff --git a/deps/uv/src/unix/stream.c b/deps/uv/src/unix/stream.c
index 6f0d137..720ada6 100644
--- a/deps/uv/src/unix/stream.c
+++ b/deps/uv/src/unix/stream.c
@@ -230,7 +230,13 @@ int uv_accept(uv_stream_t* server, uv_stream_t* client) {
     goto out;
   }
 
-  ev_io_start(streamServer->loop->ev, &streamServer->read_watcher);
+  if (!(handle->flags & UV_STREAM_BALANCED_ACCEPT)) {
+    ev_io_start(streamServer->loop->ev, &streamServer->read_watcher);
+  } else if (!ev_active(&streamServer->idle_watcher)) {
+    ev_idle_start(streamServer->loop->ev, &streamServer->idle_watcher);
+    uv_unref(streamServer->loop);
+  }
+
   streamServer->accepted_fd = -1;
   status = 0;
 
@@ -240,7 +246,23 @@ out:
 }
 
 
+void uv__server_accept_restart(EV_P_ ev_idle* idle, int revents) {
+  uv_stream_t* stream = idle->data;
+
+  assert(revents & EV_IDLE);
+
+  ev_idle_stop(stream->loop->ev, idle);
+  uv_ref(stream->loop);
+
+  if (stream->accepted_fd < 0);
+    ev_io_start(stream->loop->ev, &stream->read_watcher);
+}
+
+
 int uv_listen(uv_stream_t* stream, int backlog, uv_connection_cb cb) {
+  ev_idle_init(&stream->idle_watcher, uv__server_accept_restart);
+  stream->idle_watcher.data = stream;
+
   switch (stream->type) {
     case UV_TCP:
       return uv_tcp_listen((uv_tcp_t*)stream, backlog, cb);
diff --git a/deps/uv/src/unix/tcp.c b/deps/uv/src/unix/tcp.c
index ee94ab3..fba352c 100644
--- a/deps/uv/src/unix/tcp.c
+++ b/deps/uv/src/unix/tcp.c
@@ -322,5 +322,10 @@ int uv_tcp_keepalive(uv_tcp_t* handle, int enable, unsigned int delay) {
 
 
 int uv_tcp_simultaneous_accepts(uv_tcp_t* handle, int enable) {
+  if (enable) {
+    handle->flags &= ~UV_STREAM_BALANCED_ACCEPT;
+  } else {
+    handle->flags |= UV_STREAM_BALANCED_ACCEPT;
+  }
   return 0;
 }
diff --git a/lib/child_process.js b/lib/child_process.js
index f3a5bdd..ab9808b 100644
--- a/lib/child_process.js
+++ b/lib/child_process.js
@@ -73,15 +73,13 @@ function setupChannel(target, channel) {
 
   var jsonBuffer = '';
 
-  if (isWindows) {
-    var setSimultaneousAccepts = function(handle) {
-      var simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
-              process.env.NODE_MANY_ACCEPTS != '0') ? true : false;
-
-      if (handle._simultaneousAccepts != simultaneousAccepts) {
-        handle.setSimultaneousAccepts(simultaneousAccepts);
-        handle._simultaneousAccepts = simultaneousAccepts;
-      }
+  var setSimultaneousAccepts = function(handle) {
+    var simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
+            process.env.NODE_MANY_ACCEPTS != '0') ? true : false;
+
+    if (handle._simultaneousAccepts != simultaneousAccepts) {
+      handle.setSimultaneousAccepts(simultaneousAccepts);
+      handle._simultaneousAccepts = simultaneousAccepts;
     }
   }
 
diff --git a/src/tcp_wrap.cc b/src/tcp_wrap.cc
index 14e6d3e..c7b0d59 100644
--- a/src/tcp_wrap.cc
+++ b/src/tcp_wrap.cc
@@ -121,9 +121,7 @@ void TCPWrap::Initialize(Handle<Object> target) {
   NODE_SET_PROTOTYPE_METHOD(t, "setNoDelay", SetNoDelay);
   NODE_SET_PROTOTYPE_METHOD(t, "setKeepAlive", SetKeepAlive);
 
-#ifdef _WIN32
   NODE_SET_PROTOTYPE_METHOD(t, "setSimultaneousAccepts", SetSimultaneousAccepts);
-#endif
 
   tcpConstructor = Persistent<Function>::New(t->GetFunction());
 
@@ -276,7 +274,6 @@ Handle<Value> TCPWrap::SetKeepAlive(const Arguments& args) {
 }
 
 
-#ifdef _WIN32
 Handle<Value> TCPWrap::SetSimultaneousAccepts(const Arguments& args) {
   HandleScope scope;
 
@@ -290,7 +287,6 @@ Handle<Value> TCPWrap::SetSimultaneousAccepts(const Arguments& args) {
 
   return Undefined();
 }
-#endif
 
 
 Handle<Value> TCPWrap::Bind(const Arguments& args) {
-- 
1.7.10.msysgit.1