1 /* 2 * Copyright (c) 2009-2012 Niels Provos and Nick Mathewson 3 * 4 * Redistribution and use in source and binary forms, with or without 5 * modification, are permitted provided that the following conditions 6 * are met: 7 * 1. Redistributions of source code must retain the above copyright 8 * notice, this list of conditions and the following disclaimer. 9 * 2. Redistributions in binary form must reproduce the above copyright 10 * notice, this list of conditions and the following disclaimer in the 11 * documentation and/or other materials provided with the distribution. 12 * 3. The name of the author may not be used to endorse or promote products 13 * derived from this software without specific prior written permission. 14 * 15 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR 16 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 17 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. 18 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, 19 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT 20 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 21 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 22 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 23 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF 24 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 25 */ 26 #include "../util-internal.h" 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 static struct evutil_weakrand_state weakrand_state; 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 static int cfg_group_drain = 0; 65 66 static int cfg_connlimit_tolerance = -1; 67 static int cfg_grouplimit_tolerance = -1; 68 static int cfg_stddev_tolerance = -1; 69 70 #ifdef _WIN32 71 static int cfg_enable_iocp = 0; 72 #endif 73 74 static struct timeval cfg_tick = { 0, 500*1000 }; 75 76 static struct ev_token_bucket_cfg *conn_bucket_cfg = NULL; 77 static struct ev_token_bucket_cfg *group_bucket_cfg = NULL; 78 struct bufferevent_rate_limit_group *ratelim_group = NULL; 79 static double seconds_per_tick = 0.0; 80 81 struct client_state { 82 size_t queued; 83 ev_uint64_t received; 84 85 }; 86 static const struct timeval *ms100_common=NULL; 87 88 /* info from check_bucket_levels_cb */ 89 static int total_n_bev_checks = 0; 90 static ev_int64_t total_rbucket_level=0; 91 static ev_int64_t total_wbucket_level=0; 92 static ev_int64_t total_max_to_read=0; 93 static ev_int64_t total_max_to_write=0; 94 static ev_int64_t max_bucket_level=EV_INT64_MIN; 95 static ev_int64_t min_bucket_level=EV_INT64_MAX; 96 97 /* from check_group_bucket_levels_cb */ 98 static int total_n_group_bev_checks = 0; 99 static ev_int64_t total_group_rbucket_level = 0; 100 static ev_int64_t total_group_wbucket_level = 0; 101 102 static int n_echo_conns_open = 0; 103 104 /* Info on the open connections */ 105 struct bufferevent **bevs; 106 struct client_state *states; 107 struct bufferevent_rate_limit_group *group = NULL; 108 109 static void check_bucket_levels_cb(evutil_socket_t fd, short events, void *arg); 110 111 static void 112 loud_writecb(struct bufferevent *bev, void *ctx) 113 { 114 struct client_state *cs = ctx; 115 struct evbuffer *output = bufferevent_get_output(bev); 116 char buf[1024]; 117 int r = evutil_weakrand_(&weakrand_state); 118 memset(buf, r, sizeof(buf)); 119 while (evbuffer_get_length(output) < 8192) { 120 evbuffer_add(output, buf, sizeof(buf)); 121 cs->queued += sizeof(buf); 122 } 123 } 124 125 static void 126 discard_readcb(struct bufferevent *bev, void *ctx) 127 { 128 struct client_state *cs = ctx; 129 struct evbuffer *input = bufferevent_get_input(bev); 130 size_t len = evbuffer_get_length(input); 131 evbuffer_drain(input, len); 132 cs->received += len; 133 } 134 135 static void 136 write_on_connectedcb(struct bufferevent *bev, short what, void *ctx) 137 { 138 if (what & BEV_EVENT_CONNECTED) { 139 loud_writecb(bev, ctx); 140 /* XXXX this shouldn't be needed. */ 141 bufferevent_enable(bev, EV_READ|EV_WRITE); 142 } 143 } 144 145 static void 146 echo_readcb(struct bufferevent *bev, void *ctx) 147 { 148 struct evbuffer *input = bufferevent_get_input(bev); 149 struct evbuffer *output = bufferevent_get_output(bev); 150 151 evbuffer_add_buffer(output, input); 152 if (evbuffer_get_length(output) > 1024000) 153 bufferevent_disable(bev, EV_READ); 154 } 155 156 static void 157 echo_writecb(struct bufferevent *bev, void *ctx) 158 { 159 struct evbuffer *output = bufferevent_get_output(bev); 160 if (evbuffer_get_length(output) < 512000) 161 bufferevent_enable(bev, EV_READ); 162 } 163 164 static void 165 echo_eventcb(struct bufferevent *bev, short what, void *ctx) 166 { 167 if (what & (BEV_EVENT_EOF|BEV_EVENT_ERROR)) { 168 --n_echo_conns_open; 169 bufferevent_free(bev); 170 } 171 } 172 173 static void 174 echo_listenercb(struct evconnlistener *listener, evutil_socket_t newsock, 175 struct sockaddr *sourceaddr, int socklen, void *ctx) 176 { 177 struct event_base *base = ctx; 178 int flags = BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE; 179 struct bufferevent *bev; 180 181 bev = bufferevent_socket_new(base, newsock, flags); 182 bufferevent_setcb(bev, echo_readcb, echo_writecb, echo_eventcb, NULL); 183 if (conn_bucket_cfg) { 184 struct event *check_event = 185 event_new(base, -1, EV_PERSIST, check_bucket_levels_cb, bev); 186 bufferevent_set_rate_limit(bev, conn_bucket_cfg); 187 188 assert(bufferevent_get_token_bucket_cfg(bev) != NULL); 189 event_add(check_event, ms100_common); 190 } 191 if (ratelim_group) 192 bufferevent_add_to_rate_limit_group(bev, ratelim_group); 193 ++n_echo_conns_open; 194 bufferevent_enable(bev, EV_READ|EV_WRITE); 195 } 196 197 /* Called periodically to check up on how full the buckets are */ 198 static void 199 check_bucket_levels_cb(evutil_socket_t fd, short events, void *arg) 200 { 201 struct bufferevent *bev = arg; 202 203 ev_ssize_t r = bufferevent_get_read_limit(bev); 204 ev_ssize_t w = bufferevent_get_write_limit(bev); 205 ev_ssize_t rm = bufferevent_get_max_to_read(bev); 206 ev_ssize_t wm = bufferevent_get_max_to_write(bev); 207 /* XXXX check that no value is above the cofigured burst 208 * limit */ 209 total_rbucket_level += r; 210 total_wbucket_level += w; 211 total_max_to_read += rm; 212 total_max_to_write += wm; 213 #define B(x) \ 214 if ((x) > max_bucket_level) \ 215 max_bucket_level = (x); \ 216 if ((x) < min_bucket_level) \ 217 min_bucket_level = (x) 218 B(r); 219 B(w); 220 #undef B 221 222 total_n_bev_checks++; 223 if (total_n_bev_checks >= .8 * ((double)cfg_duration / cfg_tick_msec) * cfg_n_connections) { 224 event_free(event_base_get_running_event(bufferevent_get_base(bev))); 225 } 226 } 227 228 static void 229 check_group_bucket_levels_cb(evutil_socket_t fd, short events, void *arg) 230 { 231 if (ratelim_group) { 232 ev_ssize_t r = bufferevent_rate_limit_group_get_read_limit(ratelim_group); 233 ev_ssize_t w = bufferevent_rate_limit_group_get_write_limit(ratelim_group); 234 total_group_rbucket_level += r; 235 total_group_wbucket_level += w; 236 } 237 ++total_n_group_bev_checks; 238 } 239 240 static void 241 group_drain_cb(evutil_socket_t fd, short events, void *arg) 242 { 243 bufferevent_rate_limit_group_decrement_read(ratelim_group, cfg_group_drain); 244 bufferevent_rate_limit_group_decrement_write(ratelim_group, cfg_group_drain); 245 } 246 247 static int 248 test_ratelimiting(void) 249 { 250 struct event_base *base; 251 struct sockaddr_in sin; 252 struct evconnlistener *listener; 253 254 struct sockaddr_storage ss; 255 ev_socklen_t slen; 256 257 int i; 258 259 struct timeval tv; 260 261 ev_uint64_t total_received; 262 double total_sq_persec, total_persec; 263 double variance; 264 double expected_total_persec = -1.0, expected_avg_persec = -1.0; 265 int ok = 1; 266 struct event_config *base_cfg; 267 struct event *periodic_level_check; 268 struct event *group_drain_event=NULL; 269 270 memset(&sin, 0, sizeof(sin)); 271 sin.sin_family = AF_INET; 272 sin.sin_addr.s_addr = htonl(0x7f000001); /* 127.0.0.1 */ 273 sin.sin_port = 0; /* unspecified port */ 274 275 if (0) 276 event_enable_debug_mode(); 277 278 base_cfg = event_config_new(); 279 280 #ifdef _WIN32 281 if (cfg_enable_iocp) { 282 evthread_use_windows_threads(); 283 event_config_set_flag(base_cfg, EVENT_BASE_FLAG_STARTUP_IOCP); 284 } 285 #endif 286 287 base = event_base_new_with_config(base_cfg); 288 event_config_free(base_cfg); 289 if (! base) { 290 fprintf(stderr, "Couldn't create event_base"); 291 return 1; 292 } 293 294 listener = evconnlistener_new_bind(base, echo_listenercb, base, 295 LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1, 296 (struct sockaddr *)&sin, sizeof(sin)); 297 if (! listener) { 298 fprintf(stderr, "Couldn't create listener"); 299 return 1; 300 } 301 302 slen = sizeof(ss); 303 if (getsockname(evconnlistener_get_fd(listener), (struct sockaddr *)&ss, 304 &slen) < 0) { 305 perror("getsockname"); 306 return 1; 307 } 308 309 if (cfg_connlimit > 0) { 310 conn_bucket_cfg = ev_token_bucket_cfg_new( 311 cfg_connlimit, cfg_connlimit * 4, 312 cfg_connlimit, cfg_connlimit * 4, 313 &cfg_tick); 314 assert(conn_bucket_cfg); 315 } 316 317 if (cfg_grouplimit > 0) { 318 group_bucket_cfg = ev_token_bucket_cfg_new( 319 cfg_grouplimit, cfg_grouplimit * 4, 320 cfg_grouplimit, cfg_grouplimit * 4, 321 &cfg_tick); 322 group = ratelim_group = bufferevent_rate_limit_group_new( 323 base, group_bucket_cfg); 324 expected_total_persec = cfg_grouplimit - (cfg_group_drain / seconds_per_tick); 325 expected_avg_persec = cfg_grouplimit / cfg_n_connections; 326 if (cfg_connlimit > 0 && expected_avg_persec > cfg_connlimit) 327 expected_avg_persec = cfg_connlimit; 328 if (cfg_min_share >= 0) 329 bufferevent_rate_limit_group_set_min_share( 330 ratelim_group, cfg_min_share); 331 } 332 333 if (expected_avg_persec < 0 && cfg_connlimit > 0) 334 expected_avg_persec = cfg_connlimit; 335 336 if (expected_avg_persec > 0) 337 expected_avg_persec /= seconds_per_tick; 338 if (expected_total_persec > 0) 339 expected_total_persec /= seconds_per_tick; 340 341 bevs = calloc(cfg_n_connections, sizeof(struct bufferevent *)); 342 states = calloc(cfg_n_connections, sizeof(struct client_state)); 343 if (bevs == NULL || states == NULL) { 344 printf("Unable to allocate memory...\n"); 345 return 1; 346 } 347 348 for (i = 0; i < cfg_n_connections; ++i) { 349 bevs[i] = bufferevent_socket_new(base, -1, 350 BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE); 351 assert(bevs[i]); 352 bufferevent_setcb(bevs[i], discard_readcb, loud_writecb, 353 write_on_connectedcb, &states[i]); 354 bufferevent_enable(bevs[i], EV_READ|EV_WRITE); 355 bufferevent_socket_connect(bevs[i], (struct sockaddr *)&ss, 356 slen); 357 } 358 359 tv.tv_sec = cfg_duration - 1; 360 tv.tv_usec = 995000; 361 362 event_base_loopexit(base, &tv); 363 364 tv.tv_sec = 0; 365 tv.tv_usec = 100*1000; 366 ms100_common = event_base_init_common_timeout(base, &tv); 367 368 periodic_level_check = event_new(base, -1, EV_PERSIST, check_group_bucket_levels_cb, NULL); 369 event_add(periodic_level_check, ms100_common); 370 371 if (cfg_group_drain && ratelim_group) { 372 group_drain_event = event_new(base, -1, EV_PERSIST, group_drain_cb, NULL); 373 event_add(group_drain_event, &cfg_tick); 374 } 375 376 event_base_dispatch(base); 377 378 ratelim_group = NULL; /* So no more responders get added */ 379 event_free(periodic_level_check); 380 if (group_drain_event) 381 event_del(group_drain_event); 382 383 for (i = 0; i < cfg_n_connections; ++i) { 384 bufferevent_free(bevs[i]); 385 } 386 evconnlistener_free(listener); 387 388 /* Make sure no new echo_conns get added to the group. */ 389 ratelim_group = NULL; 390 391 /* This should get _everybody_ freed */ 392 while (n_echo_conns_open) { 393 printf("waiting for %d conns\n", n_echo_conns_open); 394 tv.tv_sec = 0; 395 tv.tv_usec = 300000; 396 event_base_loopexit(base, &tv); 397 event_base_dispatch(base); 398 } 399 400 if (group) 401 bufferevent_rate_limit_group_free(group); 402 403 if (total_n_bev_checks) { 404 printf("Average read bucket level: %f\n", 405 (double)total_rbucket_level/total_n_bev_checks); 406 printf("Average write bucket level: %f\n", 407 (double)total_wbucket_level/total_n_bev_checks); 408 printf("Highest read bucket level: %f\n", 409 (double)max_bucket_level); 410 printf("Highest write bucket level: %f\n", 411 (double)min_bucket_level); 412 printf("Average max-to-read: %f\n", 413 ((double)total_max_to_read)/total_n_bev_checks); 414 printf("Average max-to-write: %f\n", 415 ((double)total_max_to_write)/total_n_bev_checks); 416 } 417 if (total_n_group_bev_checks) { 418 printf("Average group read bucket level: %f\n", 419 ((double)total_group_rbucket_level)/total_n_group_bev_checks); 420 printf("Average group write bucket level: %f\n", 421 ((double)total_group_wbucket_level)/total_n_group_bev_checks); 422 } 423 424 total_received = 0; 425 total_persec = 0.0; 426 total_sq_persec = 0.0; 427 for (i=0; i < cfg_n_connections; ++i) { 428 double persec = states[i].received; 429 persec /= cfg_duration; 430 total_received += states[i].received; 431 total_persec += persec; 432 total_sq_persec += persec*persec; 433 printf("%d: %f per second\n", i+1, persec); 434 } 435 printf(" total: %f per second\n", 436 ((double)total_received)/cfg_duration); 437 if (expected_total_persec > 0) { 438 double diff = expected_total_persec - 439 ((double)total_received/cfg_duration); 440 printf(" [Off by %lf]\n", diff); 441 if (cfg_grouplimit_tolerance > 0 && 442 fabs(diff) > cfg_grouplimit_tolerance) { 443 fprintf(stderr, "Group bandwidth out of bounds\n"); 444 ok = 0; 445 } 446 } 447 448 printf(" average: %f per second\n", 449 (((double)total_received)/cfg_duration)/cfg_n_connections); 450 if (expected_avg_persec > 0) { 451 double diff = expected_avg_persec - (((double)total_received)/cfg_duration)/cfg_n_connections; 452 printf(" [Off by %lf]\n", diff); 453 if (cfg_connlimit_tolerance > 0 && 454 fabs(diff) > cfg_connlimit_tolerance) { 455 fprintf(stderr, "Connection bandwidth out of bounds\n"); 456 ok = 0; 457 } 458 } 459 460 variance = total_sq_persec/cfg_n_connections - total_persec*total_persec/(cfg_n_connections*cfg_n_connections); 461 462 printf(" stddev: %f per second\n", sqrt(variance)); 463 if (cfg_stddev_tolerance > 0 && 464 sqrt(variance) > cfg_stddev_tolerance) { 465 fprintf(stderr, "Connection variance out of bounds\n"); 466 ok = 0; 467 } 468 469 event_base_free(base); 470 free(bevs); 471 free(states); 472 473 return ok ? 0 : 1; 474 } 475 476 static struct option { 477 const char *name; int *ptr; int min; int isbool; 478 } options[] = { 479 { "-v", &cfg_verbose, 0, 1 }, 480 { "-h", &cfg_help, 0, 1 }, 481 { "-n", &cfg_n_connections, 1, 0 }, 482 { "-d", &cfg_duration, 1, 0 }, 483 { "-c", &cfg_connlimit, 0, 0 }, 484 { "-g", &cfg_grouplimit, 0, 0 }, 485 { "-G", &cfg_group_drain, -100000, 0 }, 486 { "-t", &cfg_tick_msec, 10, 0 }, 487 { "--min-share", &cfg_min_share, 0, 0 }, 488 { "--check-connlimit", &cfg_connlimit_tolerance, 0, 0 }, 489 { "--check-grouplimit", &cfg_grouplimit_tolerance, 0, 0 }, 490 { "--check-stddev", &cfg_stddev_tolerance, 0, 0 }, 491 #ifdef _WIN32 492 { "--iocp", &cfg_enable_iocp, 0, 1 }, 493 #endif 494 { NULL, NULL, -1, 0 }, 495 }; 496 497 static int 498 handle_option(int argc, char **argv, int *i, const struct option *opt) 499 { 500 long val; 501 char *endptr = NULL; 502 if (opt->isbool) { 503 *opt->ptr = 1; 504 return 0; 505 } 506 if (*i + 1 == argc) { 507 fprintf(stderr, "Too few arguments to '%s'\n",argv[*i]); 508 return -1; 509 } 510 val = strtol(argv[*i+1], &endptr, 10); 511 if (*argv[*i+1] == '\0' || !endptr || *endptr != '\0') { 512 fprintf(stderr, "Couldn't parse numeric value '%s'\n", 513 argv[*i+1]); 514 return -1; 515 } 516 if (val < opt->min || val > 0x7fffffff) { 517 fprintf(stderr, "Value '%s' is out-of-range'\n", 518 argv[*i+1]); 519 return -1; 520 } 521 *opt->ptr = (int)val; 522 ++*i; 523 return 0; 524 } 525 526 static void 527 usage(void) 528 { 529 fprintf(stderr, 530 "test-ratelim [-v] [-n INT] [-d INT] [-c INT] [-g INT] [-t INT]\n\n" 531 "Pushes bytes through a number of possibly rate-limited connections, and\n" 532 "displays average throughput.\n\n" 533 " -n INT: Number of connections to open (default: 30)\n" 534 " -d INT: Duration of the test in seconds (default: 5 sec)\n"); 535 fprintf(stderr, 536 " -c INT: Connection-rate limit applied to each connection in bytes per second\n" 537 " (default: None.)\n" 538 " -g INT: Group-rate limit applied to sum of all usage in bytes per second\n" 539 " (default: None.)\n" 540 " -G INT: drain INT bytes from the group limit every tick. (default: 0)\n" 541 " -t INT: Granularity of timing, in milliseconds (default: 1000 msec)\n"); 542 } 543 544 int 545 main(int argc, char **argv) 546 { 547 int i,j; 548 double ratio; 549 550 #ifdef _WIN32 551 WORD wVersionRequested = MAKEWORD(2,2); 552 WSADATA wsaData; 553 554 (void) WSAStartup(wVersionRequested, &wsaData); 555 #endif 556 557 evutil_weakrand_seed_(&weakrand_state, 0); 558 559 #ifndef _WIN32 560 if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) 561 return 1; 562 #endif 563 for (i = 1; i < argc; ++i) { 564 for (j = 0; options[j].name; ++j) { 565 if (!strcmp(argv[i],options[j].name)) { 566 if (handle_option(argc,argv,&i,&options[j])<0) 567 return 1; 568 goto again; 569 } 570 } 571 fprintf(stderr, "Unknown option '%s'\n", argv[i]); 572 usage(); 573 return 1; 574 again: 575 ; 576 } 577 if (cfg_help) { 578 usage(); 579 return 0; 580 } 581 582 cfg_tick.tv_sec = cfg_tick_msec / 1000; 583 cfg_tick.tv_usec = (cfg_tick_msec % 1000)*1000; 584 585 seconds_per_tick = ratio = cfg_tick_msec / 1000.0; 586 587 cfg_connlimit *= ratio; 588 cfg_grouplimit *= ratio; 589 590 { 591 struct timeval tv; 592 evutil_gettimeofday(&tv, NULL); 593 #ifdef _WIN32 594 srand(tv.tv_usec); 595 #else 596 srandom(tv.tv_usec); 597 #endif 598 } 599 600 #ifndef EVENT__DISABLE_THREAD_SUPPORT 601 evthread_enable_lock_debugging(); 602 #endif 603 604 return test_ratelimiting(); 605 } 606