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
loud_writecb(struct bufferevent * bev,void * ctx)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
discard_readcb(struct bufferevent * bev,void * ctx)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
write_on_connectedcb(struct bufferevent * bev,short what,void * ctx)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
echo_readcb(struct bufferevent * bev,void * ctx)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
echo_writecb(struct bufferevent * bev,void * ctx)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
echo_eventcb(struct bufferevent * bev,short what,void * ctx)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
echo_listenercb(struct evconnlistener * listener,evutil_socket_t newsock,struct sockaddr * sourceaddr,int socklen,void * ctx)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
check_bucket_levels_cb(evutil_socket_t fd,short events,void * arg)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
check_group_bucket_levels_cb(evutil_socket_t fd,short events,void * arg)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
group_drain_cb(evutil_socket_t fd,short events,void * arg)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
test_ratelimiting(void)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 #ifdef EVTHREAD_USE_WINDOWS_THREADS_IMPLEMENTED
283 evthread_use_windows_threads();
284 #endif
285 event_config_set_flag(base_cfg, EVENT_BASE_FLAG_STARTUP_IOCP);
286 }
287 #endif
288
289 base = event_base_new_with_config(base_cfg);
290 event_config_free(base_cfg);
291 if (! base) {
292 fprintf(stderr, "Couldn't create event_base");
293 return 1;
294 }
295
296 listener = evconnlistener_new_bind(base, echo_listenercb, base,
297 LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1,
298 (struct sockaddr *)&sin, sizeof(sin));
299 if (! listener) {
300 fprintf(stderr, "Couldn't create listener");
301 return 1;
302 }
303
304 slen = sizeof(ss);
305 if (getsockname(evconnlistener_get_fd(listener), (struct sockaddr *)&ss,
306 &slen) < 0) {
307 perror("getsockname");
308 return 1;
309 }
310
311 if (cfg_connlimit > 0) {
312 conn_bucket_cfg = ev_token_bucket_cfg_new(
313 cfg_connlimit, cfg_connlimit * 4,
314 cfg_connlimit, cfg_connlimit * 4,
315 &cfg_tick);
316 assert(conn_bucket_cfg);
317 }
318
319 if (cfg_grouplimit > 0) {
320 group_bucket_cfg = ev_token_bucket_cfg_new(
321 cfg_grouplimit, cfg_grouplimit * 4,
322 cfg_grouplimit, cfg_grouplimit * 4,
323 &cfg_tick);
324 group = ratelim_group = bufferevent_rate_limit_group_new(
325 base, group_bucket_cfg);
326 expected_total_persec = cfg_grouplimit - (cfg_group_drain / seconds_per_tick);
327 expected_avg_persec = cfg_grouplimit / cfg_n_connections;
328 if (cfg_connlimit > 0 && expected_avg_persec > cfg_connlimit)
329 expected_avg_persec = cfg_connlimit;
330 if (cfg_min_share >= 0)
331 bufferevent_rate_limit_group_set_min_share(
332 ratelim_group, cfg_min_share);
333 }
334
335 if (expected_avg_persec < 0 && cfg_connlimit > 0)
336 expected_avg_persec = cfg_connlimit;
337
338 if (expected_avg_persec > 0)
339 expected_avg_persec /= seconds_per_tick;
340 if (expected_total_persec > 0)
341 expected_total_persec /= seconds_per_tick;
342
343 bevs = calloc(cfg_n_connections, sizeof(struct bufferevent *));
344 states = calloc(cfg_n_connections, sizeof(struct client_state));
345
346 for (i = 0; i < cfg_n_connections; ++i) {
347 bevs[i] = bufferevent_socket_new(base, -1,
348 BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE);
349 assert(bevs[i]);
350 bufferevent_setcb(bevs[i], discard_readcb, loud_writecb,
351 write_on_connectedcb, &states[i]);
352 bufferevent_enable(bevs[i], EV_READ|EV_WRITE);
353 bufferevent_socket_connect(bevs[i], (struct sockaddr *)&ss,
354 slen);
355 }
356
357 tv.tv_sec = cfg_duration - 1;
358 tv.tv_usec = 995000;
359
360 event_base_loopexit(base, &tv);
361
362 tv.tv_sec = 0;
363 tv.tv_usec = 100*1000;
364 ms100_common = event_base_init_common_timeout(base, &tv);
365
366 periodic_level_check = event_new(base, -1, EV_PERSIST, check_group_bucket_levels_cb, NULL);
367 event_add(periodic_level_check, ms100_common);
368
369 if (cfg_group_drain && ratelim_group) {
370 group_drain_event = event_new(base, -1, EV_PERSIST, group_drain_cb, NULL);
371 event_add(group_drain_event, &cfg_tick);
372 }
373
374 event_base_dispatch(base);
375
376 ratelim_group = NULL; /* So no more responders get added */
377 event_free(periodic_level_check);
378 if (group_drain_event)
379 event_del(group_drain_event);
380
381 for (i = 0; i < cfg_n_connections; ++i) {
382 bufferevent_free(bevs[i]);
383 }
384 evconnlistener_free(listener);
385
386 /* Make sure no new echo_conns get added to the group. */
387 ratelim_group = NULL;
388
389 /* This should get _everybody_ freed */
390 while (n_echo_conns_open) {
391 printf("waiting for %d conns\n", n_echo_conns_open);
392 tv.tv_sec = 0;
393 tv.tv_usec = 300000;
394 event_base_loopexit(base, &tv);
395 event_base_dispatch(base);
396 }
397
398 if (group)
399 bufferevent_rate_limit_group_free(group);
400
401 if (total_n_bev_checks) {
402 printf("Average read bucket level: %f\n",
403 (double)total_rbucket_level/total_n_bev_checks);
404 printf("Average write bucket level: %f\n",
405 (double)total_wbucket_level/total_n_bev_checks);
406 printf("Highest read bucket level: %f\n",
407 (double)max_bucket_level);
408 printf("Highest write bucket level: %f\n",
409 (double)min_bucket_level);
410 printf("Average max-to-read: %f\n",
411 ((double)total_max_to_read)/total_n_bev_checks);
412 printf("Average max-to-write: %f\n",
413 ((double)total_max_to_write)/total_n_bev_checks);
414 }
415 if (total_n_group_bev_checks) {
416 printf("Average group read bucket level: %f\n",
417 ((double)total_group_rbucket_level)/total_n_group_bev_checks);
418 printf("Average group write bucket level: %f\n",
419 ((double)total_group_wbucket_level)/total_n_group_bev_checks);
420 }
421
422 total_received = 0;
423 total_persec = 0.0;
424 total_sq_persec = 0.0;
425 for (i=0; i < cfg_n_connections; ++i) {
426 double persec = states[i].received;
427 persec /= cfg_duration;
428 total_received += states[i].received;
429 total_persec += persec;
430 total_sq_persec += persec*persec;
431 printf("%d: %f per second\n", i+1, persec);
432 }
433 printf(" total: %f per second\n",
434 ((double)total_received)/cfg_duration);
435 if (expected_total_persec > 0) {
436 double diff = expected_total_persec -
437 ((double)total_received/cfg_duration);
438 printf(" [Off by %lf]\n", diff);
439 if (cfg_grouplimit_tolerance > 0 &&
440 fabs(diff) > cfg_grouplimit_tolerance) {
441 fprintf(stderr, "Group bandwidth out of bounds\n");
442 ok = 0;
443 }
444 }
445
446 printf(" average: %f per second\n",
447 (((double)total_received)/cfg_duration)/cfg_n_connections);
448 if (expected_avg_persec > 0) {
449 double diff = expected_avg_persec - (((double)total_received)/cfg_duration)/cfg_n_connections;
450 printf(" [Off by %lf]\n", diff);
451 if (cfg_connlimit_tolerance > 0 &&
452 fabs(diff) > cfg_connlimit_tolerance) {
453 fprintf(stderr, "Connection bandwidth out of bounds\n");
454 ok = 0;
455 }
456 }
457
458 variance = total_sq_persec/cfg_n_connections - total_persec*total_persec/(cfg_n_connections*cfg_n_connections);
459
460 printf(" stddev: %f per second\n", sqrt(variance));
461 if (cfg_stddev_tolerance > 0 &&
462 sqrt(variance) > cfg_stddev_tolerance) {
463 fprintf(stderr, "Connection variance out of bounds\n");
464 ok = 0;
465 }
466
467 event_base_free(base);
468 free(bevs);
469 free(states);
470
471 return ok ? 0 : 1;
472 }
473
474 static struct option {
475 const char *name; int *ptr; int min; int isbool;
476 } options[] = {
477 { "-v", &cfg_verbose, 0, 1 },
478 { "-h", &cfg_help, 0, 1 },
479 { "-n", &cfg_n_connections, 1, 0 },
480 { "-d", &cfg_duration, 1, 0 },
481 { "-c", &cfg_connlimit, 0, 0 },
482 { "-g", &cfg_grouplimit, 0, 0 },
483 { "-G", &cfg_group_drain, -100000, 0 },
484 { "-t", &cfg_tick_msec, 10, 0 },
485 { "--min-share", &cfg_min_share, 0, 0 },
486 { "--check-connlimit", &cfg_connlimit_tolerance, 0, 0 },
487 { "--check-grouplimit", &cfg_grouplimit_tolerance, 0, 0 },
488 { "--check-stddev", &cfg_stddev_tolerance, 0, 0 },
489 #ifdef _WIN32
490 { "--iocp", &cfg_enable_iocp, 0, 1 },
491 #endif
492 { NULL, NULL, -1, 0 },
493 };
494
495 static int
handle_option(int argc,char ** argv,int * i,const struct option * opt)496 handle_option(int argc, char **argv, int *i, const struct option *opt)
497 {
498 long val;
499 char *endptr = NULL;
500 if (opt->isbool) {
501 *opt->ptr = 1;
502 return 0;
503 }
504 if (*i + 1 == argc) {
505 fprintf(stderr, "Too few arguments to '%s'\n",argv[*i]);
506 return -1;
507 }
508 val = strtol(argv[*i+1], &endptr, 10);
509 if (*argv[*i+1] == '\0' || !endptr || *endptr != '\0') {
510 fprintf(stderr, "Couldn't parse numeric value '%s'\n",
511 argv[*i+1]);
512 return -1;
513 }
514 if (val < opt->min || val > 0x7fffffff) {
515 fprintf(stderr, "Value '%s' is out-of-range'\n",
516 argv[*i+1]);
517 return -1;
518 }
519 *opt->ptr = (int)val;
520 ++*i;
521 return 0;
522 }
523
524 static void
usage(void)525 usage(void)
526 {
527 fprintf(stderr,
528 "test-ratelim [-v] [-n INT] [-d INT] [-c INT] [-g INT] [-t INT]\n\n"
529 "Pushes bytes through a number of possibly rate-limited connections, and\n"
530 "displays average throughput.\n\n"
531 " -n INT: Number of connections to open (default: 30)\n"
532 " -d INT: Duration of the test in seconds (default: 5 sec)\n");
533 fprintf(stderr,
534 " -c INT: Connection-rate limit applied to each connection in bytes per second\n"
535 " (default: None.)\n"
536 " -g INT: Group-rate limit applied to sum of all usage in bytes per second\n"
537 " (default: None.)\n"
538 " -G INT: drain INT bytes from the group limit every tick. (default: 0)\n"
539 " -t INT: Granularity of timing, in milliseconds (default: 1000 msec)\n");
540 }
541
542 int
main(int argc,char ** argv)543 main(int argc, char **argv)
544 {
545 int i,j;
546 double ratio;
547
548 #ifdef _WIN32
549 WORD wVersionRequested = MAKEWORD(2,2);
550 WSADATA wsaData;
551
552 (void) WSAStartup(wVersionRequested, &wsaData);
553 #endif
554
555 evutil_weakrand_seed_(&weakrand_state, 0);
556
557 #ifndef _WIN32
558 if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
559 return 1;
560 #endif
561 for (i = 1; i < argc; ++i) {
562 for (j = 0; options[j].name; ++j) {
563 if (!strcmp(argv[i],options[j].name)) {
564 if (handle_option(argc,argv,&i,&options[j])<0)
565 return 1;
566 goto again;
567 }
568 }
569 fprintf(stderr, "Unknown option '%s'\n", argv[i]);
570 usage();
571 return 1;
572 again:
573 ;
574 }
575 if (cfg_help) {
576 usage();
577 return 0;
578 }
579
580 cfg_tick.tv_sec = cfg_tick_msec / 1000;
581 cfg_tick.tv_usec = (cfg_tick_msec % 1000)*1000;
582
583 seconds_per_tick = ratio = cfg_tick_msec / 1000.0;
584
585 cfg_connlimit *= ratio;
586 cfg_grouplimit *= ratio;
587
588 {
589 struct timeval tv;
590 evutil_gettimeofday(&tv, NULL);
591 #ifdef _WIN32
592 srand(tv.tv_usec);
593 #else
594 srandom(tv.tv_usec);
595 #endif
596 }
597
598 #ifndef EVENT__DISABLE_THREAD_SUPPORT
599 evthread_enable_lock_debugging();
600 #endif
601
602 return test_ratelimiting();
603 }
604