Improve pound benchmark by reconnecting in close_cb

test/benchmark-pound.c

@@ -28,22 +28,27 @@
 #undef NANOSEC
 #define NANOSEC ((uint64_t)10e8)
 
+#define DEBUG 0
+
 /* Base class for tcp_conn_rec and pipe_conn_rec.
  * The ordering of fields matters!
  */
 typedef struct {
+  int i;
   uv_connect_t conn_req;
   uv_write_t write_req;
   uv_stream_t stream;
 } conn_rec;
 
 typedef struct {
+  int i;
   uv_connect_t conn_req;
   uv_write_t write_req;
   uv_tcp_t stream;
 } tcp_conn_rec;
 
 typedef struct {
+  int i;
   uv_connect_t conn_req;
   uv_write_t write_req;
   uv_pipe_t stream;
@@ -54,8 +59,7 @@ static char buffer[] = "QS";
 static tcp_conn_rec tcp_conns[MAX_CONNS];
 static pipe_conn_rec pipe_conns[MAX_CONNS];
 
-static uint64_t start_time;
-static uint64_t end_time;
+static uint64_t start; /* in ms  */
 static int closed_streams;
 static int conns_failed;
 
@@ -77,12 +81,23 @@ static uv_buf_t alloc_cb(uv_stream_t* stream, size_t suggested_size) {
 }
 
 
+static void after_write(uv_write_t* req, int status) {
+  if (status != 0) {
+    fprintf(stderr, "write error %s\n", uv_err_name(uv_last_error()));
+    uv_close((uv_handle_t*)req->handle, close_cb);
+    conns_failed++;
+    return;
+  }
+}
+
+
 static void connect_cb(uv_connect_t* req, int status) {
   conn_rec* conn;
   uv_buf_t buf;
   int r;
 
   if (status != 0) {
+    fprintf(stderr, "connect error %s\n", uv_err_name(uv_last_error()));
     uv_close((uv_handle_t*)req->handle, close_cb);
     conns_failed++;
     return;
@@ -94,38 +109,92 @@ static void connect_cb(uv_connect_t* req, int status) {
   conn = req->data;
   ASSERT(conn != NULL);
 
+#if DEBUG
+  printf("connect_cb %d\n", conn->i);
+#endif
+
   r = uv_read_start(&conn->stream, alloc_cb, read_cb);
   ASSERT(r == 0);
 
   buf.base = buffer;
   buf.len = sizeof(buffer) - 1;
 
-  r = uv_write(&conn->write_req, &conn->stream, &buf, 1, NULL);
+  r = uv_write(&conn->write_req, &conn->stream, &buf, 1, after_write);
   ASSERT(r == 0);
 }
 
 
 static void read_cb(uv_stream_t* stream, ssize_t nread, uv_buf_t buf) {
+  conn_rec* p = stream->data;
+  uv_err_t err = uv_last_error();
+
   ASSERT(stream != NULL);
-  ASSERT(nread == -1 && uv_last_error().code == UV_EOF);
+
+#if DEBUG
+  printf("read_cb %d\n", p->i);
+#endif
+
   uv_close((uv_handle_t*)stream, close_cb);
+
+  if (nread == -1) {
+    if (err.code == UV_EOF) {
+      ;
+    } else if (err.code == UV_ECONNRESET) {
+      conns_failed++;
+    } else {
+      fprintf(stderr, "read error %s\n", uv_err_name(uv_last_error()));
+      ASSERT(0);
+    }
+  }
+}
+
+static void make_connect(tcp_conn_rec* p) {
+  struct sockaddr_in addr;
+  int r;
+
+  r = uv_tcp_init(&p->stream);
+  ASSERT(r == 0);
+
+  addr = uv_ip4_addr("127.0.0.1", TEST_PORT);
+
+  r = uv_tcp_connect(&p->conn_req, &p->stream, addr, connect_cb);
+  if (r) {
+    fprintf(stderr, "uv_tcp_connect error %s\n",
+        uv_err_name(uv_last_error()));
+    ASSERT(0);
+  }
+
+#if DEBUG
+  printf("make connect %d\n", p->i);
+#endif
+
+  p->conn_req.data = p;
+  p->write_req.data = p;
+  p->stream.data = p;
 }
 
 
 static void close_cb(uv_handle_t* handle) {
+  tcp_conn_rec* p = handle->data;
+
   ASSERT(handle != NULL);
   closed_streams++;
+
+#if DEBUG
+  printf("close_cb %d\n", p->i);
+#endif
+
+  if (uv_now() - start < 10000) {
+    make_connect(p);
+  }
 }
 
 
 static void* tcp_do_setup(int num, void* arg) {
-  tcp_conn_rec* pe;
-  tcp_conn_rec* p;
-  int r;
+  int i;
 
-  for (p = tcp_conns, pe = p + num; p < pe; p++) {
-    r = uv_tcp_init(&p->stream);
-    ASSERT(r == 0);
+  for (i = 0; i < num; i++) {
+    tcp_conns[i].i = i;
   }
 
   return tcp_conns;
@@ -147,17 +216,14 @@ static void* pipe_do_setup(int num, void* arg) {
 
 
 static int tcp_do_connect(int num, void* conns, void* arg) {
-  struct sockaddr_in addr;
   tcp_conn_rec* pe;
   tcp_conn_rec* p;
   int r;
+  int i;
 
-  addr = uv_ip4_addr("127.0.0.1", TEST_PORT);
-  for (p = tcp_conns, pe = p + num; p < pe; p++) {
-    r = uv_tcp_connect(&p->conn_req, &p->stream, addr, connect_cb);
-    ASSERT(r == 0);
 
-    p->conn_req.data = p;
+  for (i = 0; i < num; i++) {
+    make_connect(&tcp_conns[i]);
   }
 
   return 0;
@@ -188,23 +254,26 @@ static int pound_it(int concurrency,
   double secs;
   void* state;
   int r;
+  uint64_t start_time; /* in ns */
+  uint64_t end_time;
 
   uv_init();
 
+  uv_update_time();
+  start = uv_now();
+
   /* Run benchmark for at least five seconds. */
   start_time = uv_hrtime();
-  do {
-    state = do_setup(concurrency, arg);
-    ASSERT(state != NULL);
 
-    r = do_connect(concurrency, state, arg);
-    ASSERT(!r);
+  state = do_setup(concurrency, arg);
+  ASSERT(state != NULL);
 
-    uv_run();
+  r = do_connect(concurrency, state, arg);
+  ASSERT(!r);
 
-    end_time = uv_hrtime();
-  }
-  while ((end_time - start_time) < 5 * NANOSEC);
+  uv_run();
+
+  end_time = uv_hrtime();
 
   /* Number of fractional seconds it took to run the benchmark. */
   secs = (double)(end_time - start_time) / NANOSEC;

test/echo-server.c

@@ -195,7 +195,7 @@ static int tcp4_echo_start(int port) {
   r = uv_listen((uv_stream_t*)&tcpServer, SOMAXCONN, on_connection);
   if (r) {
     /* TODO: Error codes */
-    fprintf(stderr, "Listen error\n");
+    fprintf(stderr, "Listen error %s\n", uv_err_name(uv_last_error()));
     return 1;
   }