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