/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to
* deal in the Software without restriction, including without limitation the
* rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
* sell copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
* IN THE SOFTWARE.
*/
#include "task.h"
#include "uv.h"
#include <math.h>
#include <stdio.h>
static int TARGET_CONNECTIONS;
#define WRITE_BUFFER_SIZE 8192
#define MAX_SIMULTANEOUS_CONNECTS 100
#define PRINT_STATS 0
#define STATS_INTERVAL 1000 /* msec */
#define STATS_COUNT 5
static void do_write(uv_stream_t*);
static void maybe_connect_some(void);
static uv_req_t* req_alloc(void);
static void req_free(uv_req_t* uv_req);
static void buf_alloc(uv_handle_t* handle, size_t size, uv_buf_t* buf);
static void buf_free(const uv_buf_t* buf);
static uv_loop_t* loop;
static uv_tcp_t tcpServer;
static uv_pipe_t pipeServer;
static uv_stream_t* server;
static struct sockaddr_in listen_addr;
static struct sockaddr_in connect_addr;
static int64_t start_time;
static int max_connect_socket = 0;
static int max_read_sockets = 0;
static int read_sockets = 0;
static int write_sockets = 0;
static int64_t nrecv = 0;
static int64_t nrecv_total = 0;
static int64_t nsent = 0;
static int64_t nsent_total = 0;
static int stats_left = 0;
static char write_buffer[WRITE_BUFFER_SIZE];
/* Make this as large as you need. */
#define MAX_WRITE_HANDLES 1000
static stream_type type;
static uv_tcp_t tcp_write_handles[MAX_WRITE_HANDLES];
static uv_pipe_t pipe_write_handles[MAX_WRITE_HANDLES];
static uv_timer_t timer_handle;
static double gbit(int64_t bytes, int64_t passed_ms) {
double gbits = ((double)bytes / (1024 * 1024 * 1024)) * 8;
return gbits / ((double)passed_ms / 1000);
}
static void show_stats(uv_timer_t* handle) {
int64_t diff;
int i;
#if PRINT_STATS
fprintf(stderr, "connections: %d, write: %.1f gbit/s\n",
write_sockets,
gbit(nsent, STATS_INTERVAL));
fflush(stderr);
#endif
/* Exit if the show is over */
if (!--stats_left) {
uv_update_time(loop);
diff = uv_now(loop) - start_time;
fprintf(stderr, "%s_pump%d_client: %.1f gbit/s\n",
type == TCP ? "tcp" : "pipe",
write_sockets,
gbit(nsent_total, diff));
fflush(stderr);
for (i = 0; i < write_sockets; i++) {
if (type == TCP)
uv_close((uv_handle_t*) &tcp_write_handles[i], NULL);
else
uv_close((uv_handle_t*) &pipe_write_handles[i], NULL);
}
exit(0);
}
/* Reset read and write counters */
nrecv = 0;
nsent = 0;
}
static void read_show_stats(void) {
int64_t diff;
uv_update_time(loop);
diff = uv_now(loop) - start_time;
fprintf(stderr, "%s_pump%d_server: %.1f gbit/s\n",
type == TCP ? "tcp" : "pipe",
max_read_sockets,
gbit(nrecv_total, diff));
fflush(stderr);
}
static void read_sockets_close_cb(uv_handle_t* handle) {
free(handle);
read_sockets--;
/* If it's past the first second and everyone has closed their connection
* Then print stats.
*/
if (uv_now(loop) - start_time > 1000 && read_sockets == 0) {
read_show_stats();
uv_close((uv_handle_t*)server, NULL);
}
}
static void start_stats_collection(void) {
int r;
/* Show-stats timer */
stats_left = STATS_COUNT;
r = uv_timer_init(loop, &timer_handle);
ASSERT(r == 0);
r = uv_timer_start(&timer_handle, show_stats, STATS_INTERVAL, STATS_INTERVAL);
ASSERT(r == 0);
uv_update_time(loop);
start_time = uv_now(loop);
}
static void read_cb(uv_stream_t* stream, ssize_t bytes, const uv_buf_t* buf) {
if (nrecv_total == 0) {
ASSERT(start_time == 0);
uv_update_time(loop);
start_time = uv_now(loop);
}
if (bytes < 0) {
uv_close((uv_handle_t*)stream, read_sockets_close_cb);
return;
}
buf_free(buf);
nrecv += bytes;
nrecv_total += bytes;
}
static void write_cb(uv_write_t* req, int status) {
ASSERT(status == 0);
req_free((uv_req_t*) req);
nsent += sizeof write_buffer;
nsent_total += sizeof write_buffer;
do_write((uv_stream_t*) req->handle);
}
static void do_write(uv_stream_t* stream) {
uv_write_t* req;
uv_buf_t buf;
int r;
buf.base = (char*) &write_buffer;
buf.len = sizeof write_buffer;
req = (uv_write_t*) req_alloc();
r = uv_write(req, stream, &buf, 1, write_cb);
ASSERT(r == 0);
}
static void connect_cb(uv_connect_t* req, int status) {
int i;
if (status) {
fprintf(stderr, "%s", uv_strerror(status));
fflush(stderr);
}
ASSERT(status == 0);
write_sockets++;
req_free((uv_req_t*) req);
maybe_connect_some();
if (write_sockets == TARGET_CONNECTIONS) {
start_stats_collection();
/* Yay! start writing */
for (i = 0; i < write_sockets; i++) {
if (type == TCP)
do_write((uv_stream_t*) &tcp_write_handles[i]);
else
do_write((uv_stream_t*) &pipe_write_handles[i]);
}
}
}
static void maybe_connect_some(void) {
uv_connect_t* req;
uv_tcp_t* tcp;
uv_pipe_t* pipe;
int r;
while (max_connect_socket < TARGET_CONNECTIONS &&
max_connect_socket < write_sockets + MAX_SIMULTANEOUS_CONNECTS) {
if (type == TCP) {
tcp = &tcp_write_handles[max_connect_socket++];
r = uv_tcp_init(loop, tcp);
ASSERT(r == 0);
req = (uv_connect_t*) req_alloc();
r = uv_tcp_connect(req,
tcp,
(const struct sockaddr*) &connect_addr,
connect_cb);
ASSERT(r == 0);
} else {
pipe = &pipe_write_handles[max_connect_socket++];
r = uv_pipe_init(loop, pipe, 0);
ASSERT(r == 0);
req = (uv_connect_t*) req_alloc();
uv_pipe_connect(req, pipe, TEST_PIPENAME, connect_cb);
}
}
}
static void connection_cb(uv_stream_t* s, int status) {
uv_stream_t* stream;
int r;
ASSERT(server == s);
ASSERT(status == 0);
if (type == TCP) {
stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
r = uv_tcp_init(loop, (uv_tcp_t*)stream);
ASSERT(r == 0);
} else {
stream = (uv_stream_t*)malloc(sizeof(uv_pipe_t));
r = uv_pipe_init(loop, (uv_pipe_t*)stream, 0);
ASSERT(r == 0);
}
r = uv_accept(s, stream);
ASSERT(r == 0);
r = uv_read_start(stream, buf_alloc, read_cb);
ASSERT(r == 0);
read_sockets++;
max_read_sockets++;
}
/*
* Request allocator
*/
typedef struct req_list_s {
union uv_any_req uv_req;
struct req_list_s* next;
} req_list_t;
static req_list_t* req_freelist = NULL;
static uv_req_t* req_alloc(void) {
req_list_t* req;
req = req_freelist;
if (req != NULL) {
req_freelist = req->next;
return (uv_req_t*) req;
}
req = (req_list_t*) malloc(sizeof *req);
return (uv_req_t*) req;
}
static void req_free(uv_req_t* uv_req) {
req_list_t* req = (req_list_t*) uv_req;
req->next = req_freelist;
req_freelist = req;
}
/*
* Buffer allocator
*/
typedef struct buf_list_s {
uv_buf_t uv_buf_t;
struct buf_list_s* next;
} buf_list_t;
static buf_list_t* buf_freelist = NULL;
static void buf_alloc(uv_handle_t* handle, size_t size, uv_buf_t* buf) {
buf_list_t* ab;
ab = buf_freelist;
if (ab != NULL)
buf_freelist = ab->next;
else {
ab = malloc(size + sizeof(*ab));
ab->uv_buf_t.len = size;
ab->uv_buf_t.base = (char*) (ab + 1);
}
*buf = ab->uv_buf_t;
}
static void buf_free(const uv_buf_t* buf) {
buf_list_t* ab = (buf_list_t*) buf->base - 1;
ab->next = buf_freelist;
buf_freelist = ab;
}
HELPER_IMPL(tcp_pump_server) {
int r;
type = TCP;
loop = uv_default_loop();
ASSERT(0 == uv_ip4_addr("0.0.0.0", TEST_PORT, &listen_addr));
/* Server */
server = (uv_stream_t*)&tcpServer;
r = uv_tcp_init(loop, &tcpServer);
ASSERT(r == 0);
r = uv_tcp_bind(&tcpServer, (const struct sockaddr*) &listen_addr, 0);
ASSERT(r == 0);
r = uv_listen((uv_stream_t*)&tcpServer, MAX_WRITE_HANDLES, connection_cb);
ASSERT(r == 0);
uv_run(loop, UV_RUN_DEFAULT);
return 0;
}
HELPER_IMPL(pipe_pump_server) {
int r;
type = PIPE;
loop = uv_default_loop();
/* Server */
server = (uv_stream_t*)&pipeServer;
r = uv_pipe_init(loop, &pipeServer, 0);
ASSERT(r == 0);
r = uv_pipe_bind(&pipeServer, TEST_PIPENAME);
ASSERT(r == 0);
r = uv_listen((uv_stream_t*)&pipeServer, MAX_WRITE_HANDLES, connection_cb);
ASSERT(r == 0);
uv_run(loop, UV_RUN_DEFAULT);
MAKE_VALGRIND_HAPPY();
return 0;
}
static void tcp_pump(int n) {
ASSERT(n <= MAX_WRITE_HANDLES);
TARGET_CONNECTIONS = n;
type = TCP;
loop = uv_default_loop();
ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &connect_addr));
/* Start making connections */
maybe_connect_some();
uv_run(loop, UV_RUN_DEFAULT);
MAKE_VALGRIND_HAPPY();
}
static void pipe_pump(int n) {
ASSERT(n <= MAX_WRITE_HANDLES);
TARGET_CONNECTIONS = n;
type = PIPE;
loop = uv_default_loop();
/* Start making connections */
maybe_connect_some();
uv_run(loop, UV_RUN_DEFAULT);
MAKE_VALGRIND_HAPPY();
}
BENCHMARK_IMPL(tcp_pump100_client) {
tcp_pump(100);
return 0;
}
BENCHMARK_IMPL(tcp_pump1_client) {
tcp_pump(1);
return 0;
}
BENCHMARK_IMPL(pipe_pump100_client) {
pipe_pump(100);
return 0;
}
BENCHMARK_IMPL(pipe_pump1_client) {
pipe_pump(1);
return 0;
}