1 /* $NetBSD: test-ratelim.c,v 1.1.1.1 2013/04/11 16:43:32 christos Exp $ */ 2 /* 3 * Copyright (c) 2009-2012 Niels Provos and Nick Mathewson 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 8 * 1. Redistributions of source code must retain the above copyright 9 * notice, this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * 3. The name of the author may not be used to endorse or promote products 14 * derived from this software without specific prior written permission. 15 * 16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR 17 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 18 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. 19 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, 20 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT 21 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 22 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 23 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 24 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF 25 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 26 */ 27 28 #include <stdio.h> 29 #include <stdlib.h> 30 #include <string.h> 31 #include <assert.h> 32 #include <math.h> 33 34 #ifdef WIN32 35 #include <winsock2.h> 36 #include <ws2tcpip.h> 37 #else 38 #include <sys/socket.h> 39 #include <netinet/in.h> 40 # ifdef _XOPEN_SOURCE_EXTENDED 41 # include <arpa/inet.h> 42 # endif 43 #endif 44 #include <signal.h> 45 46 #include "event2/bufferevent.h" 47 #include "event2/buffer.h" 48 #include "event2/event.h" 49 #include "event2/util.h" 50 #include "event2/listener.h" 51 #include "event2/thread.h" 52 53 #include "../util-internal.h" 54 55 static int cfg_verbose = 0; 56 static int cfg_help = 0; 57 58 static int cfg_n_connections = 30; 59 static int cfg_duration = 5; 60 static int cfg_connlimit = 0; 61 static int cfg_grouplimit = 0; 62 static int cfg_tick_msec = 1000; 63 static int cfg_min_share = -1; 64 65 static int cfg_connlimit_tolerance = -1; 66 static int cfg_grouplimit_tolerance = -1; 67 static int cfg_stddev_tolerance = -1; 68 69 #ifdef _WIN32 70 static int cfg_enable_iocp = 0; 71 #endif 72 73 static struct timeval cfg_tick = { 0, 500*1000 }; 74 75 static struct ev_token_bucket_cfg *conn_bucket_cfg = NULL; 76 static struct ev_token_bucket_cfg *group_bucket_cfg = NULL; 77 struct bufferevent_rate_limit_group *ratelim_group = NULL; 78 static double seconds_per_tick = 0.0; 79 80 struct client_state { 81 size_t queued; 82 ev_uint64_t received; 83 }; 84 85 static int n_echo_conns_open = 0; 86 87 static void 88 loud_writecb(struct bufferevent *bev, void *ctx) 89 { 90 struct client_state *cs = ctx; 91 struct evbuffer *output = bufferevent_get_output(bev); 92 char buf[1024]; 93 #ifdef WIN32 94 int r = rand() % 256; 95 #else 96 int r = random() % 256; 97 #endif 98 memset(buf, r, sizeof(buf)); 99 while (evbuffer_get_length(output) < 8192) { 100 evbuffer_add(output, buf, sizeof(buf)); 101 cs->queued += sizeof(buf); 102 } 103 } 104 105 static void 106 discard_readcb(struct bufferevent *bev, void *ctx) 107 { 108 struct client_state *cs = ctx; 109 struct evbuffer *input = bufferevent_get_input(bev); 110 size_t len = evbuffer_get_length(input); 111 evbuffer_drain(input, len); 112 cs->received += len; 113 } 114 115 static void 116 write_on_connectedcb(struct bufferevent *bev, short what, void *ctx) 117 { 118 if (what & BEV_EVENT_CONNECTED) { 119 loud_writecb(bev, ctx); 120 /* XXXX this shouldn't be needed. */ 121 bufferevent_enable(bev, EV_READ|EV_WRITE); 122 } 123 } 124 125 static void 126 echo_readcb(struct bufferevent *bev, void *ctx) 127 { 128 struct evbuffer *input = bufferevent_get_input(bev); 129 struct evbuffer *output = bufferevent_get_output(bev); 130 131 evbuffer_add_buffer(output, input); 132 if (evbuffer_get_length(output) > 1024000) 133 bufferevent_disable(bev, EV_READ); 134 } 135 136 static void 137 echo_writecb(struct bufferevent *bev, void *ctx) 138 { 139 struct evbuffer *output = bufferevent_get_output(bev); 140 if (evbuffer_get_length(output) < 512000) 141 bufferevent_enable(bev, EV_READ); 142 } 143 144 static void 145 echo_eventcb(struct bufferevent *bev, short what, void *ctx) 146 { 147 if (what & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) { 148 --n_echo_conns_open; 149 bufferevent_free(bev); 150 } 151 } 152 153 static void 154 echo_listenercb(struct evconnlistener *listener, evutil_socket_t newsock, 155 struct sockaddr *sourceaddr, int socklen, void *ctx) 156 { 157 struct event_base *base = ctx; 158 int flags = BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE; 159 struct bufferevent *bev; 160 161 bev = bufferevent_socket_new(base, newsock, flags); 162 bufferevent_setcb(bev, echo_readcb, echo_writecb, echo_eventcb, NULL); 163 if (conn_bucket_cfg) 164 bufferevent_set_rate_limit(bev, conn_bucket_cfg); 165 if (ratelim_group) 166 bufferevent_add_to_rate_limit_group(bev, ratelim_group); 167 ++n_echo_conns_open; 168 bufferevent_enable(bev, EV_READ|EV_WRITE); 169 } 170 171 static int 172 test_ratelimiting(void) 173 { 174 struct event_base *base; 175 struct sockaddr_in sin; 176 struct evconnlistener *listener; 177 178 struct sockaddr_storage ss; 179 ev_socklen_t slen; 180 181 struct bufferevent **bevs; 182 struct client_state *states; 183 struct bufferevent_rate_limit_group *group = NULL; 184 185 int i; 186 187 struct timeval tv; 188 189 ev_uint64_t total_received; 190 double total_sq_persec, total_persec; 191 double variance; 192 double expected_total_persec = -1.0, expected_avg_persec = -1.0; 193 int ok = 1; 194 struct event_config *base_cfg; 195 196 memset(&sin, 0, sizeof(sin)); 197 sin.sin_family = AF_INET; 198 sin.sin_addr.s_addr = htonl(0x7f000001); /* 127.0.0.1 */ 199 sin.sin_port = 0; /* unspecified port */ 200 201 if (0) 202 event_enable_debug_mode(); 203 204 base_cfg = event_config_new(); 205 206 #ifdef _WIN32 207 if (cfg_enable_iocp) { 208 evthread_use_windows_threads(); 209 event_config_set_flag(base_cfg, EVENT_BASE_FLAG_STARTUP_IOCP); 210 } 211 #endif 212 213 base = event_base_new_with_config(base_cfg); 214 event_config_free(base_cfg); 215 216 listener = evconnlistener_new_bind(base, echo_listenercb, base, 217 LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1, 218 (struct sockaddr *)&sin, sizeof(sin)); 219 220 slen = sizeof(ss); 221 if (getsockname(evconnlistener_get_fd(listener), (struct sockaddr *)&ss, 222 &slen) < 0) { 223 perror("getsockname"); 224 return 1; 225 } 226 227 if (cfg_connlimit > 0) { 228 conn_bucket_cfg = ev_token_bucket_cfg_new( 229 cfg_connlimit, cfg_connlimit * 4, 230 cfg_connlimit, cfg_connlimit * 4, 231 &cfg_tick); 232 assert(conn_bucket_cfg); 233 } 234 235 if (cfg_grouplimit > 0) { 236 group_bucket_cfg = ev_token_bucket_cfg_new( 237 cfg_grouplimit, cfg_grouplimit * 4, 238 cfg_grouplimit, cfg_grouplimit * 4, 239 &cfg_tick); 240 group = ratelim_group = bufferevent_rate_limit_group_new( 241 base, group_bucket_cfg); 242 expected_total_persec = cfg_grouplimit; 243 expected_avg_persec = cfg_grouplimit / cfg_n_connections; 244 if (cfg_connlimit > 0 && expected_avg_persec > cfg_connlimit) 245 expected_avg_persec = cfg_connlimit; 246 if (cfg_min_share >= 0) 247 bufferevent_rate_limit_group_set_min_share( 248 ratelim_group, cfg_min_share); 249 } 250 251 if (expected_avg_persec < 0 && cfg_connlimit > 0) 252 expected_avg_persec = cfg_connlimit; 253 254 if (expected_avg_persec > 0) 255 expected_avg_persec /= seconds_per_tick; 256 if (expected_total_persec > 0) 257 expected_total_persec /= seconds_per_tick; 258 259 bevs = calloc(cfg_n_connections, sizeof(struct bufferevent *)); 260 states = calloc(cfg_n_connections, sizeof(struct client_state)); 261 262 for (i = 0; i < cfg_n_connections; ++i) { 263 bevs[i] = bufferevent_socket_new(base, -1, 264 BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE); 265 assert(bevs[i]); 266 bufferevent_setcb(bevs[i], discard_readcb, loud_writecb, 267 write_on_connectedcb, &states[i]); 268 bufferevent_enable(bevs[i], EV_READ|EV_WRITE); 269 bufferevent_socket_connect(bevs[i], (struct sockaddr *)&ss, 270 slen); 271 } 272 273 tv.tv_sec = cfg_duration - 1; 274 tv.tv_usec = 995000; 275 276 event_base_loopexit(base, &tv); 277 278 event_base_dispatch(base); 279 280 ratelim_group = NULL; /* So no more responders get added */ 281 282 for (i = 0; i < cfg_n_connections; ++i) { 283 bufferevent_free(bevs[i]); 284 } 285 evconnlistener_free(listener); 286 287 /* Make sure no new echo_conns get added to the group. */ 288 ratelim_group = NULL; 289 290 /* This should get _everybody_ freed */ 291 while (n_echo_conns_open) { 292 printf("waiting for %d conns\n", n_echo_conns_open); 293 tv.tv_sec = 0; 294 tv.tv_usec = 300000; 295 event_base_loopexit(base, &tv); 296 event_base_dispatch(base); 297 } 298 299 if (group) 300 bufferevent_rate_limit_group_free(group); 301 302 total_received = 0; 303 total_persec = 0.0; 304 total_sq_persec = 0.0; 305 for (i=0; i < cfg_n_connections; ++i) { 306 double persec = states[i].received; 307 persec /= cfg_duration; 308 total_received += states[i].received; 309 total_persec += persec; 310 total_sq_persec += persec*persec; 311 printf("%d: %f per second\n", i+1, persec); 312 } 313 printf(" total: %f per second\n", 314 ((double)total_received)/cfg_duration); 315 if (expected_total_persec > 0) { 316 double diff = expected_total_persec - 317 ((double)total_received/cfg_duration); 318 printf(" [Off by %lf]\n", diff); 319 if (cfg_grouplimit_tolerance > 0 && 320 fabs(diff) > cfg_grouplimit_tolerance) { 321 fprintf(stderr, "Group bandwidth out of bounds\n"); 322 ok = 0; 323 } 324 } 325 326 printf(" average: %f per second\n", 327 (((double)total_received)/cfg_duration)/cfg_n_connections); 328 if (expected_avg_persec > 0) { 329 double diff = expected_avg_persec - (((double)total_received)/cfg_duration)/cfg_n_connections; 330 printf(" [Off by %lf]\n", diff); 331 if (cfg_connlimit_tolerance > 0 && 332 fabs(diff) > cfg_connlimit_tolerance) { 333 fprintf(stderr, "Connection bandwidth out of bounds\n"); 334 ok = 0; 335 } 336 } 337 338 variance = total_sq_persec/cfg_n_connections - total_persec*total_persec/(cfg_n_connections*cfg_n_connections); 339 340 printf(" stddev: %f per second\n", sqrt(variance)); 341 if (cfg_stddev_tolerance > 0 && 342 sqrt(variance) > cfg_stddev_tolerance) { 343 fprintf(stderr, "Connection variance out of bounds\n"); 344 ok = 0; 345 } 346 347 event_base_free(base); 348 free(bevs); 349 free(states); 350 351 return ok ? 0 : 1; 352 } 353 354 static struct option { 355 const char *name; int *ptr; int min; int isbool; 356 } options[] = { 357 { "-v", &cfg_verbose, 0, 1 }, 358 { "-h", &cfg_help, 0, 1 }, 359 { "-n", &cfg_n_connections, 1, 0 }, 360 { "-d", &cfg_duration, 1, 0 }, 361 { "-c", &cfg_connlimit, 0, 0 }, 362 { "-g", &cfg_grouplimit, 0, 0 }, 363 { "-t", &cfg_tick_msec, 10, 0 }, 364 { "--min-share", &cfg_min_share, 0, 0 }, 365 { "--check-connlimit", &cfg_connlimit_tolerance, 0, 0 }, 366 { "--check-grouplimit", &cfg_grouplimit_tolerance, 0, 0 }, 367 { "--check-stddev", &cfg_stddev_tolerance, 0, 0 }, 368 #ifdef _WIN32 369 { "--iocp", &cfg_enable_iocp, 0, 1 }, 370 #endif 371 { NULL, NULL, -1, 0 }, 372 }; 373 374 static int 375 handle_option(int argc, char **argv, int *i, const struct option *opt) 376 { 377 long val; 378 char *endptr = NULL; 379 if (opt->isbool) { 380 *opt->ptr = 1; 381 return 0; 382 } 383 if (*i + 1 == argc) { 384 fprintf(stderr, "Too few arguments to '%s'\n",argv[*i]); 385 return -1; 386 } 387 val = strtol(argv[*i+1], &endptr, 10); 388 if (*argv[*i+1] == '\0' || !endptr || *endptr != '\0') { 389 fprintf(stderr, "Couldn't parse numeric value '%s'\n", 390 argv[*i+1]); 391 return -1; 392 } 393 if (val < opt->min || val > 0x7fffffff) { 394 fprintf(stderr, "Value '%s' is out-of-range'\n", 395 argv[*i+1]); 396 return -1; 397 } 398 *opt->ptr = (int)val; 399 ++*i; 400 return 0; 401 } 402 403 static void 404 usage(void) 405 { 406 fprintf(stderr, 407 "test-ratelim [-v] [-n INT] [-d INT] [-c INT] [-g INT] [-t INT]\n\n" 408 "Pushes bytes through a number of possibly rate-limited connections, and\n" 409 "displays average throughput.\n\n" 410 " -n INT: Number of connections to open (default: 30)\n" 411 " -d INT: Duration of the test in seconds (default: 5 sec)\n"); 412 fprintf(stderr, 413 " -c INT: Connection-rate limit applied to each connection in bytes per second\n" 414 " (default: None.)\n" 415 " -g INT: Group-rate limit applied to sum of all usage in bytes per second\n" 416 " (default: None.)\n" 417 " -t INT: Granularity of timing, in milliseconds (default: 1000 msec)\n"); 418 } 419 420 int 421 main(int argc, char **argv) 422 { 423 int i,j; 424 double ratio; 425 426 #ifdef WIN32 427 WORD wVersionRequested = MAKEWORD(2,2); 428 WSADATA wsaData; 429 430 (void) WSAStartup(wVersionRequested, &wsaData); 431 #endif 432 433 #ifndef WIN32 434 if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) 435 return 1; 436 #endif 437 for (i = 1; i < argc; ++i) { 438 for (j = 0; options[j].name; ++j) { 439 if (!strcmp(argv[i],options[j].name)) { 440 if (handle_option(argc,argv,&i,&options[j])<0) 441 return 1; 442 goto again; 443 } 444 } 445 fprintf(stderr, "Unknown option '%s'\n", argv[i]); 446 usage(); 447 return 1; 448 again: 449 ; 450 } 451 if (cfg_help) { 452 usage(); 453 return 0; 454 } 455 456 cfg_tick.tv_sec = cfg_tick_msec / 1000; 457 cfg_tick.tv_usec = (cfg_tick_msec % 1000)*1000; 458 459 seconds_per_tick = ratio = cfg_tick_msec / 1000.0; 460 461 cfg_connlimit *= ratio; 462 cfg_grouplimit *= ratio; 463 464 { 465 struct timeval tv; 466 evutil_gettimeofday(&tv, NULL); 467 #ifdef WIN32 468 srand(tv.tv_usec); 469 #else 470 srandom(tv.tv_usec); 471 #endif 472 } 473 474 #ifndef _EVENT_DISABLE_THREAD_SUPPORT 475 evthread_enable_lock_debuging(); 476 #endif 477 478 return test_ratelimiting(); 479 } 480