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