The FreeRADIUS server $Id: f3670dba8951ca10eb4948feb3dc3db9423a334f $
Loading...
Searching...
No Matches
trunk_tests.c
Go to the documentation of this file.
1#include <freeradius-devel/util/test/acutest.h>
2#include <freeradius-devel/util/test/acutest_helpers.h>
3#include <freeradius-devel/util/syserror.h>
4#include <sys/socket.h>
5
6#define TRUNK_TESTS 1
7#include "trunk.c"
8
9typedef struct {
10 trunk_request_t *treq; //!< Trunk request.
11 bool cancelled; //!< Seen by the cancelled callback.
12 bool completed; //!< Seen by the complete callback.
13 bool failed; //!< Seen by the failed callback.
14 bool freed; //!< Seen by the free callback.
15 bool signal_partial; //!< Muxer should signal that this request is partially written.
16 bool signal_cancel_partial; //!< Muxer should signal that this request is partially cancelled.
17 int priority; //!< Priority of request
18} test_proto_request_t;
19
20typedef struct {
21 uint64_t cancelled; //!< Count of tests in this run that were cancelled.
22 uint64_t completed; //!< Count of tests in this run that completed.
23 uint64_t failed; //!< Count of tests in this run that failed.
24 uint64_t freed; //!< Count of tests in this run that were freed.
25} test_proto_stats_t;
26
27#define DEBUG_LVL_SET if (acutest_verbose_level_ >= 3) fr_debug_lvl = L_DBG_LVL_4 + 1
28
29static void test_mux(UNUSED fr_event_list_t *el, trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
30{
31 trunk_request_t *treq;
32 size_t count = 0;
33 int fd = *(talloc_get_type_abort(conn->h, int));
34 ssize_t slen;
35
36 while (trunk_connection_pop_request(&treq, tconn) == 0) {
37 test_proto_request_t *preq = treq->pub.preq;
38 count++;
39
40 /*
41 * Simulate a partial write
42 */
43 if (preq && preq->signal_partial) {
44 trunk_request_signal_partial(treq);
45 preq->signal_partial = false;
46 break;
47 }
48
49 if (acutest_verbose_level_ >= 3) printf("%s - Wrote %p\n", __FUNCTION__, preq);
50
51 slen = write(fd, &preq, sizeof(preq));
52 if (slen < 0) return;
53 if (slen == 0) return;
54 if (slen < (ssize_t)sizeof(preq)) abort();
55
56 trunk_request_signal_sent(treq);
57 }
58 TEST_CHECK(count > 0);
59}
60
61static void test_cancel_mux(UNUSED fr_event_list_t *el, trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
62{
63 trunk_request_t *treq;
64 size_t count = 0;
65 int fd = *(talloc_get_type_abort(conn->h, int));
66 ssize_t slen;
67
68 /*
69 * For cancellation we just do
70 */
71 while ((trunk_connection_pop_cancellation(&treq, tconn) == 0)) {
72 test_proto_request_t *preq = treq->pub.preq;
73 count++;
74
75 /*
76 * Simulate a partial cancel write
77 */
78 if (preq && preq->signal_cancel_partial) {
79 trunk_request_signal_cancel_partial(treq);
80 preq->signal_cancel_partial = false;
81 break;
82 }
83
84 if (acutest_verbose_level_ >= 3) printf("%s - Wrote %p\n", __FUNCTION__, preq);
85 slen = write(fd, &preq, sizeof(preq));
86 if (slen < 0) {
87 fr_perror("%s - %s", __FUNCTION__, fr_syserror(errno));
88 return;
89 }
90 if (slen == 0) return;
91 if (slen < (ssize_t)sizeof(preq)) abort();
92
93 trunk_request_signal_cancel_sent(treq);
94 }
95 TEST_CHECK(count > 0);
96}
97
98static void test_demux(UNUSED fr_event_list_t *el, UNUSED trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
99{
100 int fd = *(talloc_get_type_abort(conn->h, int));
101 test_proto_request_t *preq;
102 ssize_t slen;
103
104 for (;;) {
105 slen = read(fd, &preq, sizeof(preq));
106 if (slen <= 0) break;
107
108 if (acutest_verbose_level_ >= 3) printf("%s - Read %p (%zu)\n", __FUNCTION__, preq, (size_t)slen);
109
110 /*
111 * The pointer came from test_mux() or test_cancel_mux() through
112 * the socket pair, so use the type checked copy.
113 */
114 TEST_CHECK(slen == sizeof(preq));
115 preq = talloc_get_type_abort(preq, test_proto_request_t);
116
117 if (preq->freed) continue;
118
119 /*
120 * Demuxer can handle both normal requests and cancelled ones
121 */
122 switch (preq->treq->pub.state) {
123 case TRUNK_REQUEST_STATE_CANCEL:
124 break; /* Hack - just ignore it */
125
126 case TRUNK_REQUEST_STATE_CANCEL_SENT:
127 trunk_request_signal_cancel_complete(preq->treq);
128 break;
129
130 case TRUNK_REQUEST_STATE_SENT:
131 trunk_request_signal_complete(preq->treq);
132 break;
133
134 default:
135 fr_assert(0);
136 break;
137 }
138 }
139}
140
141static void _conn_io_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags,
142 UNUSED int fd_errno, void *uctx)
143{
144
145 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
146
147 trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
148}
149
150static void _conn_io_read(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
151{
152 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
153 trunk_connection_signal_readable(tconn);
154}
155
156static void _conn_io_write(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
157{
158 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
159 trunk_connection_signal_writable(tconn);
160}
161
162static void _conn_notify(trunk_connection_t *tconn, connection_t *conn,
163 fr_event_list_t *el,
164 trunk_connection_event_t notify_on, UNUSED void *uctx)
165{
166 int fd = *(talloc_get_type_abort(conn->h, int));
167
168 switch (notify_on) {
169 case TRUNK_CONN_EVENT_NONE:
170 fr_event_fd_delete(el, fd, FR_EVENT_FILTER_IO);
171 break;
172
173 case TRUNK_CONN_EVENT_READ:
174 TEST_CHECK(fr_event_fd_insert(conn, NULL, el, fd, _conn_io_read, NULL, _conn_io_error, tconn) == 0);
175 break;
176
177 case TRUNK_CONN_EVENT_WRITE:
178 TEST_CHECK(fr_event_fd_insert(conn, NULL, el, fd, NULL, _conn_io_write, _conn_io_error, tconn) == 0);
179 break;
180
181 case TRUNK_CONN_EVENT_BOTH:
182 TEST_CHECK(fr_event_fd_insert(conn, NULL, el, fd, _conn_io_read, _conn_io_write, _conn_io_error, tconn) == 0);
183 break;
184
185 default:
186 fr_assert(0);
187 }
188}
189
190static void test_request_cancel(UNUSED connection_t *conn, void *preq,
191 UNUSED trunk_cancel_reason_t reason, void *uctx)
192{
193 test_proto_stats_t *stats = uctx;
194 test_proto_request_t *our_preq;
195
196 if (!preq) return;
197
198 our_preq = talloc_get_type_abort(preq, test_proto_request_t);
199 our_preq->cancelled = true;
200 if (stats) stats->cancelled++;
201}
202
203static void test_request_complete(UNUSED request_t *request, void *preq, UNUSED void *rctx, void *uctx)
204{
205 test_proto_stats_t *stats = uctx;
206 test_proto_request_t *our_preq;
207
208 if (!preq) return;
209
210 our_preq = talloc_get_type_abort(preq, test_proto_request_t);
211 our_preq->completed = true;
212 if (stats) stats->completed++;
213}
214
215static void test_request_fail(UNUSED request_t *request, void *preq, UNUSED void *rctx, UNUSED trunk_request_state_t state, void *uctx)
216{
217 test_proto_stats_t *stats = uctx;
218 test_proto_request_t *our_preq;
219
220 if (!preq) return;
221
222 our_preq = talloc_get_type_abort(preq, test_proto_request_t);
223 our_preq->failed = true;
224 if (stats) stats->failed++;
225}
226
227static void test_request_free(UNUSED request_t *request, void *preq, void *uctx)
228{
229 test_proto_stats_t *stats = uctx;
230 test_proto_request_t *our_preq;
231
232 if (!preq) return;
233
234 our_preq = talloc_get_type_abort(preq, test_proto_request_t);
235 our_preq->freed = true;
236 if (stats) stats->freed++;
237}
238
239/** Whenever the second socket in a socket pair is readable, read all pending data, and write it back
240 *
241 */
242static void _conn_io_loopback(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
243{
244 int *our_h = talloc_get_type_abort(uctx, int);
245 static uint8_t buff[1024];
246 static size_t to_write;
247 ssize_t slen;
248
249 fr_assert(fd == our_h[1]);
250
251 while (true) {
252 slen = read(fd, buff, sizeof(buff));
253 if (slen <= 0) return;
254
255 to_write = (size_t)slen;
256
257 if (acutest_verbose_level_ >= 3) printf("%s - Read %zu bytes of data\n", __FUNCTION__, slen);
258 slen = write(our_h[1], buff, (size_t)to_write);
259 if (slen < 0) return;
260
261 if (slen < (ssize_t)to_write) {
262 to_write -= slen;
263 if (acutest_verbose_level_ >= 3) {
264 printf("%s - Partial write %zu bytes left\n", __FUNCTION__, to_write);
265 }
266 return;
267 } else {
268 if (acutest_verbose_level_ >= 3) printf("%s - Wrote %zu bytes of data\n", __FUNCTION__, slen);
269 }
270 }
271}
272
273static void _conn_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
274{
275 int *our_h = talloc_get_type_abort(h, int);
276
277 talloc_free_children(our_h); /* Clear the IO handlers */
278
279 close(our_h[0]);
280 close(our_h[1]);
281
282 talloc_free(our_h);
283}
284
285/** Insert I/O handlers that loop any data back round
286 *
287 */
288static connection_state_t _conn_open(fr_event_list_t *el, void *h, UNUSED void *uctx)
289{
290 int *our_h = talloc_get_type_abort(h, int);
291
292 /*
293 * This always needs to be inserted
294 */
295 TEST_CHECK(fr_event_fd_insert(our_h, NULL, el, our_h[1], _conn_io_loopback, NULL, NULL, our_h) == 0);
296
297 return CONNECTION_STATE_CONNECTED;
298}
299
300/** Allocate a basic socket pair
301 *
302 */
303CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
304static connection_state_t _conn_init(void **h_out, connection_t *conn, UNUSED void *uctx)
305{
306 int *h;
307
308 h = talloc_array(conn, int, 2);
309 socketpair(AF_UNIX, SOCK_STREAM, 0, h);
310
311 (void) fr_nonblock(h[0]);
312 (void) fr_nonblock(h[1]);
313 connection_signal_on_fd(conn, h[0]);
314 *h_out = h;
315
316 return CONNECTION_STATE_CONNECTING;
317}
318
319static connection_t *test_setup_socket_pair_connection_alloc(trunk_connection_t *tconn,
320 fr_event_list_t *el,
321 connection_conf_t const *conn_conf,
322 char const *log_prefix, UNUSED void *uctx)
323{
324 connection_conf_t cstat;
325
326 if (!conn_conf) {
327 memset(&cstat, 0, sizeof(cstat));
328 conn_conf = &cstat;
329 }
330 return connection_alloc(tconn, el,
331 &(connection_funcs_t){
332 .init = _conn_init,
333 .open = _conn_open,
334 .close = _conn_close
335 },
336 conn_conf,
337 log_prefix, tconn);
338}
339
340static fr_cmp_ret_t test_preq_cmp(void const *a, void const *b)
341{
342 test_proto_request_t const *preq_a = a;
343 test_proto_request_t const *preq_b = b;
344
345 return CMP_PREFER_SMALLER(preq_a->priority, preq_b->priority); /* our priorities are NOT request->priority */
346}
347
348static trunk_t *test_setup_trunk(TALLOC_CTX *ctx, fr_event_list_t *el, trunk_conf_t *conf, bool with_cancel_mux, void *uctx)
349{
350 trunk_io_funcs_t io_funcs = {
351 .connection_alloc = test_setup_socket_pair_connection_alloc,
352 .connection_notify = _conn_notify,
353 .request_prioritise = test_preq_cmp,
354 .request_mux = test_mux,
355 .request_demux = test_demux,
356 .request_cancel = test_request_cancel,
357 .request_complete = test_request_complete,
358 .request_fail = test_request_fail,
359 .request_free = test_request_free
360 };
361
362 /*
363 * Function list is copied, so this is OK.
364 */
365 if (with_cancel_mux) io_funcs.request_cancel_mux = test_cancel_mux;
366
367 return trunk_alloc(ctx, el, &io_funcs, conf, "test_socket_pair", uctx, false, NULL);
368}
369
370static void test_socket_pair_alloc_then_free(void)
371{
372 TALLOC_CTX *ctx = talloc_init_const("test");
373 trunk_t *trunk;
374 fr_event_list_t *el;
375 int events;
376
377 trunk_conf_t conf = {
378 .start = 2,
379 .min = 2
380 };
381 trunk_io_funcs_t io_funcs = {
382 .connection_alloc = test_setup_socket_pair_connection_alloc,
383 .request_prioritise = fr_pointer_cmp,
384 };
385
386 DEBUG_LVL_SET;
387
388 el = fr_event_list_alloc(ctx, NULL, NULL);
389
390 fr_timer_list_set_time_func(el->tl, test_time);
391
392 trunk = trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false, NULL);
393 TEST_CHECK(trunk != NULL);
394 if (!trunk) return;
395
396 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 2);
397 events = fr_event_corral(el, test_time_base, true);
398 TEST_CHECK(events == 2); /* Two I/O write events, no timers */
399 fr_event_service(el);
400 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 2);
401
402 events = fr_event_corral(el, test_time_base, false);
403 TEST_CHECK(events == 0); /* I/O events should have been cleared */
404
405 talloc_free(trunk);
406 talloc_free(ctx);
407}
408
409static void test_socket_pair_alloc_then_reconnect_then_free(void)
410{
411 TALLOC_CTX *ctx = talloc_init_const("test");
412 trunk_t *trunk;
413 fr_event_list_t *el;
414 int events;
415 trunk_conf_t conf = {
416 .start = 2,
417 .min = 2,
418 .conn_conf = &(connection_conf_t){
419 .reconnection_delay = fr_time_delta_from_nsec(NSEC / 2)
420 }
421 };
422 trunk_io_funcs_t io_funcs = {
423 .connection_alloc = test_setup_socket_pair_connection_alloc,
424 .request_prioritise = fr_pointer_cmp,
425 };
426 DEBUG_LVL_SET;
427
428 el = fr_event_list_alloc(ctx, NULL, NULL);
429
430 if (!el) return;
431
432 fr_timer_list_set_time_func(el->tl, test_time);
433
434 trunk = trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false, NULL);
435 TEST_CHECK(trunk != NULL);
436 if (!trunk) return;
437
438 events = fr_event_corral(el, test_time_base, true);
439 TEST_CHECK(events == 2); /* Two I/O write events, no timers */
440 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 2);
441 fr_event_service(el);
442 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 2);
443
444 events = fr_event_corral(el, test_time_base, false);
445 TEST_CHECK(events == 0); /* I/O events should have been cleared */
446 TEST_MSG("Got %u events", events);
447
448 trunk_reconnect(trunk, TRUNK_CONN_ACTIVE, CONNECTION_FAILED);
449
450 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(2));
451 events = fr_event_corral(el, test_time_base, true);
452 TEST_CHECK(events == 1); /* Two timer events but event loops only adds one to the total*/
453 TEST_MSG("Got %u events", events);
454 fr_event_service(el);
455
456 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 2);
457
458 events = fr_event_corral(el, test_time_base, true);
459 TEST_CHECK(events == 2); /* Two I/O write events, no timers */
460 fr_event_service(el);
461
462 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 2);
463 events = fr_event_corral(el, test_time_base, false);
464 TEST_CHECK(events == 0); /* I/O events should have been cleared */
465
466 talloc_free(trunk);
467 talloc_free(ctx);
468}
469
470CC_NO_UBSAN(function) /* UBSAN: false positive - public vs private connection_t trips --fsanitize=function*/
471static connection_state_t _conn_init_no_signal(void **h_out, connection_t *conn, UNUSED void *uctx)
472{
473 int *h;
474
475 h = talloc_array(conn, int, 2);
476 socketpair(AF_UNIX, SOCK_STREAM, 0, h);
477 *h_out = h;
478
479 return CONNECTION_STATE_CONNECTING;
480}
481
482static connection_t *test_setup_socket_pair_1s_timeout_connection_alloc(trunk_connection_t *tconn,
483 fr_event_list_t *el,
484 UNUSED connection_conf_t const *conf,
485 char const *log_prefix, void *uctx)
486{
487 return connection_alloc(tconn, el,
488 &(connection_funcs_t){
489 .init = _conn_init_no_signal,
490 .open = _conn_open,
491 .close = _conn_close
492 },
493 &(connection_conf_t){
494 .connection_timeout = fr_time_delta_from_sec(1),
495 .reconnection_delay = fr_time_delta_from_sec(1)
496 },
497 log_prefix, uctx);
498}
499
500static void test_socket_pair_alloc_then_connect_timeout(void)
501{
502 TALLOC_CTX *ctx = talloc_init_const("test");
503 trunk_t *trunk;
504 fr_event_list_t *el;
505 int events;
506 trunk_connection_t *tconn;
507 trunk_conf_t conf = {
508 .start = 1,
509 .min = 1
510 };
511 trunk_io_funcs_t io_funcs = {
512 .connection_alloc = test_setup_socket_pair_1s_timeout_connection_alloc,
513 .request_prioritise = fr_pointer_cmp,
514 };
515
516 DEBUG_LVL_SET;
517
518 el = fr_event_list_alloc(ctx, NULL, NULL);
519
520 fr_timer_list_set_time_func(el->tl, test_time);
521
522
523 trunk = trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false, NULL);
524 TEST_CHECK(trunk != NULL);
525 if (!trunk) return;
526
527 /*
528 * Trigger connection timeout
529 */
530 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 1.5));
531 TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for the connection timeout */
532 events = fr_event_corral(el, test_time_base, true);
533 TEST_CHECK(events == 1); /* We didn't install the I/O events */
534
535 tconn = fr_dlist_head(&trunk->connecting);
536 TEST_CHECK(tconn != NULL);
537 if (tconn == NULL) return;
538
539 TEST_CHECK(connection_get_num_timed_out(tconn->pub.conn) == 0);
540 TEST_CHECK(connection_get_num_reconnected(tconn->pub.conn) == 0);
541
542 /*
543 * Timeout should now fire
544 */
545 fr_event_service(el);
546
547 /*
548 * Connection delay not implemented for timed out connections
549 */
550 TEST_CHECK(connection_get_num_timed_out(tconn->pub.conn) == 1);
551 TEST_CHECK(connection_get_num_reconnected(tconn->pub.conn) == 1);
552
553 events = fr_event_corral(el, test_time_base, false);
554 TEST_CHECK(events == 0); /* I/O events should have been cleared */
555
556 talloc_free(trunk);
557 talloc_free(ctx);
558}
559
560static connection_t *test_setup_socket_pair_1s_reconnection_delay_alloc(trunk_connection_t *tconn,
561 fr_event_list_t *el,
562 UNUSED connection_conf_t const *conn_conf,
563 char const *log_prefix, void *uctx)
564{
565 return connection_alloc(tconn, el,
566 &(connection_funcs_t){
567 .init = _conn_init,
568 .open = _conn_open,
569 .close = _conn_close
570 },
571 &(connection_conf_t){
572 .connection_timeout = fr_time_delta_from_sec(1),
573 .reconnection_delay = fr_time_delta_from_sec(1)
574 },
575 log_prefix, uctx);
576}
577
578static void test_socket_pair_alloc_then_reconnect_check_delay(void)
579{
580 TALLOC_CTX *ctx = talloc_init_const("test");
581 trunk_t *trunk;
582 fr_event_list_t *el;
583 int events;
584 trunk_connection_t *tconn;
585 trunk_conf_t conf = {
586 .start = 1,
587 .min = 1,
588 .conn_conf = &(connection_conf_t){
589 .reconnection_delay = fr_time_delta_from_sec(1),
590 .connection_timeout = fr_time_delta_from_sec(1)
591 }
592 };
593 trunk_io_funcs_t io_funcs = {
594 .connection_alloc = test_setup_socket_pair_1s_reconnection_delay_alloc,
595 .request_prioritise = fr_pointer_cmp,
596 };
597
598 DEBUG_LVL_SET;
599
600 el = fr_event_list_alloc(ctx, NULL, NULL);
601 fr_timer_list_set_time_func(el->tl, test_time);
602
603 trunk = trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false, NULL);
604 TEST_CHECK(trunk != NULL);
605 if (!trunk) return;
606
607 /*
608 * Trigger connection timeout
609 */
610 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 1.5));
611 TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for the connection timeout */
612 events = fr_event_corral(el, test_time_base, true);
613 TEST_CHECK(events == 2); /* We didn't install the I/O events */
614 fr_event_service(el);
615
616 tconn = fr_minmax_heap_min_peek(trunk->active);
617 TEST_CHECK(tconn != NULL);
618 if (tconn == NULL) return;
619
620 TEST_CHECK(connection_get_num_timed_out(tconn->pub.conn) == 0);
621 TEST_CHECK(connection_get_num_reconnected(tconn->pub.conn) == 0);
622
623 /*
624 * Trigger reconnection
625 */
626 connection_signal_reconnect(tconn->pub.conn, CONNECTION_FAILED);
627 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
628
629 events = fr_event_corral(el, test_time_base, false);
630 TEST_CHECK(events == 0); /* Reconnect delay not ready to fire yet, no I/O handlers installed */
631 TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for reconnect delay */
632
633 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
634 events = fr_event_corral(el, test_time_base, false);
635 TEST_CHECK(events == 1); /* Reconnect delay should now be ready to fire */
636
637 fr_event_service(el); /* Services the timer, which then triggers init */
638
639 TEST_CHECK(connection_get_num_timed_out(tconn->pub.conn) == 0);
640 TEST_CHECK(connection_get_num_reconnected(tconn->pub.conn) == 1);
641
642 events = fr_event_corral(el, test_time_base, true);
643 TEST_CHECK(events == 1); /* Should have a pending I/O event and a timer */
644
645 talloc_free(trunk);
646 talloc_free(ctx);
647}
648
649/*
650 * Test basic enqueue and dequeue
651 */
652static void test_enqueue_basic(void)
653{
654 TALLOC_CTX *ctx = talloc_init_const("test");
655 trunk_t *trunk;
656 fr_event_list_t *el;
657 trunk_conf_t conf = {
658 .start = 1,
659 .min = 1,
660 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
661 };
662 test_proto_request_t *preq;
663 trunk_request_t *treq = NULL;
664 trunk_enqueue_t rcode;
665
666 DEBUG_LVL_SET;
667
668 el = fr_event_list_alloc(ctx, NULL, NULL);
669 fr_timer_list_set_time_func(el->tl, test_time);
670
671 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
672
673 /*
674 * Our preq is a pointer to the trunk
675 * request so we don't have to manage
676 * a tree of requests and responses.
677 */
678 preq = talloc_zero(NULL, test_proto_request_t);
679
680 /*
681 * The trunk is active, but there's no
682 * connections.
683 *
684 * We're under the current request limit
685 * so the request should enter the
686 * backlog.
687 */
688 rcode = trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
689 preq->treq = treq;
690 TEST_CHECK(rcode == TRUNK_ENQUEUE_IN_BACKLOG);
691
692 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_BACKLOG) == 1);
693
694 /*
695 * Allow the connection to establish
696 */
697 fr_event_corral(el, test_time_base, false);
698 fr_event_service(el);
699
700 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_BACKLOG) == 0);
701 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 1);
702
703 /*
704 * Should now be active and have a write event
705 * inserted into the event loop.
706 */
707 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
708
709 /*
710 * Trunk should be signalled the connection is
711 * writable.
712 *
713 * We should then:
714 * - Pop a request from the pending queue.
715 * - Write the request to the socket pair
716 */
717 fr_event_corral(el, test_time_base, false);
718 fr_event_service(el);
719
720 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
721
722 /*
723 * Gives the loopback function a chance
724 * to read the data, and write it back.
725 */
726 fr_event_corral(el, test_time_base, false);
727 fr_event_service(el);
728
729 /*
730 * Trunk should be signalled the connection is
731 * readable.
732 *
733 * We should then:
734 * - Read the (looped back) response.
735 * - Signal the trunk that the connection is readable.
736 */
737 fr_event_corral(el, test_time_base, false);
738 fr_event_service(el);
739
740 TEST_CHECK(preq->completed == true);
741 TEST_CHECK(preq->failed == false);
742 TEST_CHECK(preq->cancelled == false);
743 TEST_CHECK(preq->freed == true);
744 talloc_free(preq);
745
746 talloc_free(trunk);
747 talloc_free(ctx);
748}
749
750/*
751 * Test request cancellations when the connection is in various states
752 */
753static void test_enqueue_cancellation_points(void)
754{
755 TALLOC_CTX *ctx = talloc_init_const("test");
756 trunk_t *trunk;
757 fr_event_list_t *el;
758 trunk_conf_t conf = {
759 .start = 1,
760 .min = 1,
761 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
762 };
763 test_proto_request_t *preq;
764 trunk_request_t *treq = NULL;
765
766 DEBUG_LVL_SET;
767
768 el = fr_event_list_alloc(ctx, NULL, NULL);
769 fr_timer_list_set_time_func(el->tl, test_time);
770
771 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
772 preq = talloc_zero(NULL, test_proto_request_t);
773 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
774
775 TEST_CASE("cancellation via trunk free - TRUNK_REQUEST_STATE_BACKLOG");
776 talloc_free(trunk);
777 TEST_CHECK(preq->completed == false);
778 TEST_CHECK(preq->failed == true);
779 TEST_CHECK(preq->cancelled == false);
780 TEST_CHECK(preq->freed == true);
781 talloc_free(preq);
782
783 TEST_CASE("cancellation via signal - TRUNK_REQUEST_STATE_BACKLOG");
784 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
785 preq = talloc_zero(NULL, test_proto_request_t);
786 treq = NULL;
787 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
788 preq->treq = treq;
789 trunk_request_signal_cancel(treq);
790 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
791
792 TEST_CHECK(preq->completed == false);
793 TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
794 TEST_CHECK(preq->cancelled == false);
795 TEST_CHECK(preq->freed == true);
796 talloc_free(preq);
797 talloc_free(trunk);
798
799 TEST_CASE("cancellation via trunk free - TRUNK_REQUEST_STATE_PARTIAL");
800 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
801 preq = talloc_zero(NULL, test_proto_request_t);
802 preq->signal_partial = true;
803 treq = NULL;
804 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
805 preq->treq = treq;
806
807 fr_event_corral(el, test_time_base, false); /* Connect the connection */
808 fr_event_service(el);
809
810 fr_event_corral(el, test_time_base, false); /* Send the request */
811 fr_event_service(el);
812
813 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PARTIAL));
814
815 talloc_free(trunk);
816
817 TEST_CHECK(preq->completed == false);
818 TEST_CHECK(preq->failed == true);
819 TEST_CHECK(preq->cancelled == true);
820 TEST_CHECK(preq->freed == true);
821 talloc_free(preq);
822
823 TEST_CASE("cancellation via signal - TRUNK_REQUEST_STATE_PARTIAL");
824 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
825 preq = talloc_zero(NULL, test_proto_request_t);
826 preq->signal_partial = true;
827 treq = NULL;
828 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
829 preq->treq = treq;
830
831 fr_event_corral(el, test_time_base, false); /* Connect the connection */
832 fr_event_service(el);
833
834 fr_event_corral(el, test_time_base, false); /* Send the request */
835 fr_event_service(el);
836
837 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PARTIAL) == 1);
838 trunk_request_signal_cancel(treq);
839 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
840
841 TEST_CHECK(preq->completed == false);
842 TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
843 TEST_CHECK(preq->cancelled == true);
844 TEST_CHECK(preq->freed == true);
845 talloc_free(preq);
846 talloc_free(trunk);
847
848 TEST_CASE("cancellation via trunk free - TRUNK_REQUEST_STATE_SENT");
849 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
850 preq = talloc_zero(NULL, test_proto_request_t);
851 treq = NULL;
852 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
853 preq->treq = treq;
854
855 fr_event_corral(el, test_time_base, false); /* Connect the connection */
856 fr_event_service(el);
857
858 fr_event_corral(el, test_time_base, false); /* Send the request */
859 fr_event_service(el);
860
861 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
862 talloc_free(trunk);
863
864 TEST_CHECK(preq->completed == false);
865 TEST_CHECK(preq->failed == true);
866 TEST_CHECK(preq->cancelled == true);
867 TEST_CHECK(preq->freed == true);
868 talloc_free(preq);
869
870 TEST_CASE("cancellation via signal - TRUNK_REQUEST_STATE_SENT");
871 trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
872 preq = talloc_zero(NULL, test_proto_request_t);
873 treq = NULL;
874 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
875 preq->treq = treq;
876
877 fr_event_corral(el, test_time_base, false); /* Connect the connection */
878 fr_event_service(el);
879
880 fr_event_corral(el, test_time_base, false); /* Send the request */
881 fr_event_service(el);
882
883 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
884 trunk_request_signal_cancel(treq);
885 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
886
887 TEST_CHECK(preq->completed == false);
888 TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
889 TEST_CHECK(preq->cancelled == true);
890 TEST_CHECK(preq->freed == true);
891 talloc_free(preq);
892 talloc_free(trunk);
893
894 TEST_CASE("cancellation via trunk free - TRUNK_REQUEST_STATE_CANCEL_PARTIAL");
895 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
896 preq = talloc_zero(NULL, test_proto_request_t);
897 preq->signal_cancel_partial = true;
898 treq = NULL;
899 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
900 preq->treq = treq;
901
902 fr_event_corral(el, test_time_base, false); /* Connect the connection */
903 fr_event_service(el);
904
905 fr_event_corral(el, test_time_base, false); /* Send the request */
906 fr_event_service(el);
907
908 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
909 trunk_request_signal_cancel(treq);
910 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL) == 1);
911
912 fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
913 fr_event_service(el);
914
915 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_PARTIAL) == 1);
916
917 talloc_free(trunk);
918
919 TEST_CHECK(preq->completed == false);
920 TEST_CHECK(preq->failed == false);
921 TEST_CHECK(preq->cancelled == true);
922 TEST_CHECK(preq->freed == true);
923 talloc_free(preq);
924
925 TEST_CASE("cancellation via trunk free - TRUNK_REQUEST_STATE_CANCEL_SENT");
926 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
927 preq = talloc_zero(NULL, test_proto_request_t);
928 treq = NULL;
929 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
930 preq->treq = treq;
931
932 fr_event_corral(el, test_time_base, false); /* Connect the connection */
933 fr_event_service(el);
934
935 fr_event_corral(el, test_time_base, false); /* Send the request */
936 fr_event_service(el);
937
938 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
939 trunk_request_signal_cancel(treq);
940 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL) == 1);
941
942 fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
943 fr_event_service(el);
944
945 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
946
947 talloc_free(trunk);
948
949 TEST_CHECK(preq->completed == false);
950 TEST_CHECK(preq->failed == false);
951 TEST_CHECK(preq->cancelled == true);
952 TEST_CHECK(preq->freed == true);
953 talloc_free(preq);
954
955 TEST_CASE("trunk free after TRUNK_REQUEST_STATE_CANCEL_COMPLETE");
956 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
957 preq = talloc_zero(NULL, test_proto_request_t);
958 treq = NULL;
959 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
960 preq->treq = treq;
961
962 fr_event_corral(el, test_time_base, false); /* Connect the connection */
963 fr_event_service(el);
964
965 fr_event_corral(el, test_time_base, false); /* Send the request */
966 fr_event_service(el);
967
968 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
969 trunk_request_signal_cancel(treq);
970 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL) == 1);
971
972 fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
973 fr_event_service(el);
974
975 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
976
977 fr_event_corral(el, test_time_base, false); /* Loop the cancel request back round */
978 fr_event_service(el);
979
980 fr_event_corral(el, test_time_base, false); /* Read the cancel ACK (such that it is) */
981 fr_event_service(el);
982
983 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
984
985 talloc_free(trunk);
986
987 TEST_CHECK(preq->completed == false);
988 TEST_CHECK(preq->failed == false);
989 TEST_CHECK(preq->cancelled == true);
990 TEST_CHECK(preq->freed == true);
991 talloc_free(preq);
992
993
994 talloc_free(ctx);
995}
996
997/*
998 * Test PARTIAL -> SENT and CANCEL-PARTIAL -> CANCEL-SENT
999 */
1000static void test_partial_to_complete_states(void)
1001{
1002 TALLOC_CTX *ctx = talloc_init_const("test");
1003 trunk_t *trunk;
1004 fr_event_list_t *el;
1005 trunk_conf_t conf = {
1006 .start = 1,
1007 .min = 1,
1008 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1009 };
1010 test_proto_request_t *preq;
1011 trunk_request_t *treq = NULL;
1012
1013 DEBUG_LVL_SET;
1014
1015 el = fr_event_list_alloc(ctx, NULL, NULL);
1016 fr_timer_list_set_time_func(el->tl, test_time);
1017
1018 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1019 preq = talloc_zero(NULL, test_proto_request_t);
1020 preq->signal_partial = true;
1021 preq->signal_cancel_partial = true;
1022
1023 TEST_CASE("TRUNK_REQUEST_STATE_PARTIAL -> TRUNK_REQUEST_STATE_SENT");
1024
1025 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1026 preq->treq = treq;
1027
1028 fr_event_corral(el, test_time_base, false); /* Connect the connection */
1029 fr_event_service(el);
1030
1031 fr_event_corral(el, test_time_base, false); /* Send the request */
1032 fr_event_service(el);
1033
1034 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PARTIAL) == 1);
1035
1036 fr_event_corral(el, test_time_base, false); /* Complete the partial request */
1037 fr_event_service(el);
1038
1039 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 1);
1040
1041 trunk_request_signal_cancel(treq);
1042 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL) == 1);
1043
1044 TEST_CASE("TRUNK_REQUEST_STATE_CANCEL_PARTIAL -> TRUNK_REQUEST_STATE_CANCEL_SENT");
1045
1046 fr_event_corral(el, test_time_base, false); /* Send partial cancel request */
1047 fr_event_service(el);
1048
1049 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_PARTIAL) == 1);
1050
1051 fr_event_corral(el, test_time_base, false); /* Complete the partial cancellation */
1052 fr_event_service(el);
1053
1054 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
1055
1056 fr_event_corral(el, test_time_base, false); /* Loop the cancellation request back */
1057 fr_event_service(el);
1058
1059 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
1060
1061 talloc_free(trunk);
1062
1063 TEST_CHECK(preq->completed == false);
1064 TEST_CHECK(preq->failed == false);
1065 TEST_CHECK(preq->cancelled == true);
1066 TEST_CHECK(preq->freed == true);
1067 talloc_free(preq);
1068
1069 talloc_free(ctx);
1070}
1071
1072/*
1073 * Test calling reconnect with requests in each different state
1074 */
1075static void test_requeue_on_reconnect(void)
1076{
1077 TALLOC_CTX *ctx = talloc_init_const("test");
1078 trunk_t *trunk;
1079 fr_event_list_t *el;
1080 trunk_conf_t conf = {
1081 .start = 2,
1082 .min = 2,
1083 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5),
1084 .conn_conf = &(connection_conf_t){
1085 .reconnection_delay = fr_time_delta_from_nsec(NSEC / 10)
1086 },
1087 .backlog_on_failed_conn = true
1088 };
1089 test_proto_request_t *preq;
1090 trunk_request_t *treq = NULL;
1091 trunk_connection_t *tconn;
1092
1093 DEBUG_LVL_SET;
1094 fr_talloc_fault_setup();
1095
1096 el = fr_event_list_alloc(ctx, NULL, NULL);
1097 fr_timer_list_set_time_func(el->tl, test_time);
1098
1099 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1100 preq = talloc_zero(ctx, test_proto_request_t);
1101 preq->signal_partial = true;
1102 preq->signal_cancel_partial = true;
1103
1104 fr_event_corral(el, test_time_base, false); /* Connect the connection(s) */
1105 fr_event_service(el);
1106
1107 TEST_CASE("dequeue on reconnect - TRUNK_REQUEST_STATE_PENDING");
1108
1109 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE), 2);
1110
1111 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1112 preq->treq = treq;
1113
1114 tconn = treq->pub.tconn; /* Store the conn the request was assigned to */
1115 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1116
1117 trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED);
1118
1119 /*
1120 * Should be reassigned to the other connection
1121 */
1122 TEST_CHECK(tconn != treq->pub.tconn);
1123 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1124
1125 /*
1126 * Should be reassigned to the backlog
1127 */
1128 trunk_connection_signal_reconnect(treq->pub.tconn, CONNECTION_FAILED);
1129 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_BACKLOG), 1);
1130 TEST_CHECK(!treq->pub.tconn);
1131
1132 TEST_CASE("cancel on reconnect - TRUNK_REQUEST_STATE_PARTIAL");
1133
1134 /*
1135 * Allow the connections to reconnect
1136 */
1137 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1138 fr_event_corral(el, test_time_base, false);
1139 fr_event_service(el); /* run management function */
1140 fr_event_corral(el, test_time_base, false);
1141 fr_event_service(el); /* service any I/O callbacks */
1142
1143 /*
1144 * Request should now be assigned back to one of the reconnected
1145 * connections.
1146 */
1147 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1148 TEST_CHECK(treq->pub.tconn != NULL);
1149
1150 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1151 fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1152 fr_event_service(el);
1153
1154 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PARTIAL), 1);
1155
1156 /*
1157 * Reconnect the connection.
1158 *
1159 * preq should pass through the cancel function,
1160 * then be re-assigned.
1161 */
1162 tconn = treq->pub.tconn;
1163 trunk_connection_signal_reconnect(treq->pub.tconn, CONNECTION_FAILED);
1164
1165 TEST_CHECK(preq->completed == false);
1166 TEST_CHECK(preq->failed == false);
1167 TEST_CHECK(preq->cancelled == true);
1168 TEST_CHECK(preq->freed == false);
1169
1170 preq->cancelled = false; /* Reset */
1171
1172 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1173 TEST_CHECK(tconn != treq->pub.tconn); /* Ensure it moved */
1174
1175 TEST_CASE("cancel on reconnect - TRUNK_REQUEST_STATE_SENT");
1176
1177 /*
1178 * Sent the request (fully)
1179 */
1180 fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1181 fr_event_service(el);
1182
1183 /*
1184 * The above indeed appears to send the request partially;
1185 * this appears to be required to send it fully, judging by
1186 * the following check, which fails without it.
1187 */
1188 fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1189 fr_event_service(el);
1190 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT), 1);
1191
1192 tconn = treq->pub.tconn;
1193 trunk_connection_signal_reconnect(treq->pub.tconn, CONNECTION_FAILED);
1194
1195 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1196
1197 /*
1198 * Allow the connections to reconnect
1199 * and send the request.
1200 */
1201 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1202 fr_event_corral(el, test_time_base, false);
1203 fr_event_service(el);
1204 TEST_CHECK(tconn != treq->pub.tconn); /* Ensure it moved */
1205
1206 TEST_CHECK(preq->completed == false);
1207 TEST_CHECK(preq->failed == false);
1208 TEST_CHECK(preq->cancelled == true);
1209 TEST_CHECK(preq->freed == false);
1210
1211 preq->cancelled = false; /* Reset */
1212
1213 TEST_CASE("free on reconnect - TRUNK_REQUEST_STATE_CANCEL");
1214
1215 /*
1216 * Signal the request should be cancelled
1217 */
1218 trunk_request_signal_cancel(treq);
1219 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL), 1);
1220
1221 /*
1222 * Requests in the cancel state, are
1223 * freed instead of being moved between
1224 * connections.
1225 */
1226 trunk_connection_signal_reconnect(tconn, CONNECTION_FAILED); /* treq->pub.tconn, now invalid due to cancel */
1227
1228 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1229 fr_event_corral(el, test_time_base, false);
1230 fr_event_service(el);
1231
1232 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1233 fr_event_corral(el, test_time_base, false);
1234 fr_event_service(el);
1235
1236 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1237 fr_event_corral(el, test_time_base, false);
1238 fr_event_service(el);
1239
1240 TEST_CHECK(preq->completed == false);
1241 TEST_CHECK(preq->failed == false);
1242 TEST_CHECK(preq->cancelled == true);
1243 TEST_CHECK(preq->freed == true);
1244
1245 /*
1246 * Allow the connection we just reconnected
1247 * to open so it doesn't interfere with
1248 * the next test.
1249 */
1250 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1251 fr_event_corral(el, test_time_base, false);
1252 fr_event_service(el);
1253
1254 TEST_CASE("free on reconnect - TRUNK_REQUEST_STATE_CANCEL_PARTIAL");
1255
1256 /*
1257 * Queue up a new request, and get it to the cancel-partial state.
1258 */
1259 preq = talloc_zero(ctx, test_proto_request_t);
1260 preq->signal_cancel_partial = true;
1261 treq = NULL;
1262 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1263 preq->treq = treq;
1264
1265 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1266
1267 /*
1268 * Sent the request (fully)
1269 */
1270 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1271 fr_event_corral(el, test_time_base, false); /* Send the request (fully) */
1272 fr_event_service(el);
1273
1274 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT), 1);
1275 trunk_request_signal_cancel(treq); /* Cancel the request */
1276
1277 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL), 1);
1278
1279 /*
1280 * Transition to cancel partial
1281 */
1282 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1283 fr_event_corral(el, test_time_base, false);
1284 fr_event_service(el);
1285
1286 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_PARTIAL), 1);
1287
1288 /*
1289 * Trigger a reconnection
1290 */
1291 trunk_connection_signal_reconnect(treq->pub.tconn, CONNECTION_FAILED);
1292
1293 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1294 fr_event_corral(el, test_time_base, false);
1295 fr_event_service(el);
1296
1297 TEST_CHECK(preq->completed == false);
1298 TEST_CHECK(preq->failed == false);
1299 TEST_CHECK(preq->cancelled == true);
1300 TEST_CHECK(preq->freed == true);
1301
1302 /*
1303 * Allow the connection we just reconnected
1304 * top open so it doesn't interfere with
1305 * the next test.
1306 */
1307 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1308 fr_event_corral(el, test_time_base, false);
1309 fr_event_service(el);
1310
1311 TEST_CASE("free on reconnect - TRUNK_REQUEST_STATE_CANCEL_SENT");
1312
1313 /*
1314 * Queue up a new request, and get it to the cancel-sent state.
1315 */
1316 preq = talloc_zero(NULL, test_proto_request_t);
1317 treq = NULL;
1318 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1319 preq->treq = treq;
1320
1321 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 1);
1322
1323 /*
1324 * Sent the request (fully)
1325 */
1326 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1327 fr_event_corral(el, test_time_base, false); /* Send the request (fully) */
1328 fr_event_service(el);
1329
1330 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT), 1);
1331 trunk_request_signal_cancel(treq); /* Cancel the request */
1332
1333 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL), 1);
1334
1335 /*
1336 * Transition to cancel
1337 */
1338 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1339 fr_event_corral(el, test_time_base, false);
1340 fr_event_service(el);
1341
1342 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_CANCEL_SENT), 1);
1343
1344 /*
1345 * Trigger a reconnection
1346 */
1347 trunk_connection_signal_reconnect(treq->pub.tconn, CONNECTION_FAILED);
1348
1349 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1350 fr_event_corral(el, test_time_base, false);
1351 fr_event_service(el);
1352
1353 TEST_CHECK(preq->completed == false);
1354 TEST_CHECK(preq->failed == false);
1355 TEST_CHECK(preq->cancelled == true);
1356 TEST_CHECK(preq->freed == true);
1357
1358 talloc_free(preq);
1359
1360 talloc_free(ctx);
1361}
1362
1363static void test_connection_start_on_enqueue(void)
1364{
1365 TALLOC_CTX *ctx = talloc_init_const("test");
1366 trunk_t *trunk;
1367 fr_event_list_t *el;
1368 trunk_conf_t conf = {
1369 .start = 0,
1370 .min = 0, /* No connections on start */
1371 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1372 };
1373 test_proto_request_t *preq;
1374 trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1375
1376 DEBUG_LVL_SET;
1377
1378 el = fr_event_list_alloc(ctx, NULL, NULL);
1379 fr_timer_list_set_time_func(el->tl, test_time);
1380
1381 /* Need to provide a timer starting value above zero */
1382 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1383
1384 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1385 preq = talloc_zero(NULL, test_proto_request_t);
1386
1387 TEST_CASE("C0 - Enqueue should spawn");
1388 trunk_request_enqueue(&treq_a, trunk, NULL, preq, NULL);
1389
1390 /*
1391 * This causes the event associated with the request left on
1392 * the backlog queue to be handled, which (along with the other
1393 * corral; service sequence, makes the checks all pass.
1394 */
1395 fr_event_corral(el, test_time_base, false);
1396 fr_event_service(el);
1397
1398 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1399
1400 TEST_CASE("C1 connecting, !max_req_per_conn - Enqueue MUST NOT spawn");
1401 trunk_request_enqueue(&treq_b, trunk, NULL, preq, NULL);
1402
1403 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1404
1405 /*
1406 * Allow the connections to open
1407 */
1408 fr_event_corral(el, test_time_base, false);
1409 fr_event_service(el);
1410
1411 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
1412
1413 TEST_CASE("C1 active, !max_req_per_conn - Enqueue MUST NOT spawn");
1414 trunk_request_enqueue(&treq_c, trunk, NULL, preq, NULL);
1415
1416 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
1417 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 3);
1418
1419 talloc_free(ctx);
1420 talloc_free(preq);
1421}
1422
1423static void test_connection_rebalance_requests(void)
1424{
1425 TALLOC_CTX *ctx = talloc_init_const("test");
1426 trunk_t *trunk;
1427 fr_event_list_t *el;
1428 trunk_conf_t conf = {
1429 .start = 2,
1430 .min = 2, /* No connections on start */
1431 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1432 };
1433 test_proto_request_t *preq;
1434 trunk_connection_t *tconn;
1435 trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1436
1437 DEBUG_LVL_SET;
1438
1439 el = fr_event_list_alloc(ctx, NULL, NULL);
1440 fr_timer_list_set_time_func(el->tl, test_time);
1441
1442 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1443 preq = talloc_zero(NULL, test_proto_request_t);
1444 printf("Rebalance %p\n", preq);
1445
1446 /*
1447 * Allow the connections to open
1448 */
1449 fr_event_corral(el, test_time_base, false);
1450 fr_event_service(el);
1451
1452 /*
1453 * Mark one of the connections as full, and
1454 * enqueue three requests on the other.
1455 */
1456 tconn = fr_minmax_heap_min_peek(trunk->active);
1457
1458 TEST_CASE("C2 connected, R0 - Signal inactive");
1459 trunk_connection_signal_inactive(tconn);
1460
1461
1462 trunk_request_enqueue(&treq_a, trunk, NULL, preq, NULL);
1463 trunk_request_enqueue(&treq_b, trunk, NULL, preq, NULL);
1464 trunk_request_enqueue(&treq_c, trunk, NULL, preq, NULL);
1465
1466 TEST_CASE("C1 connected, C2 inactive, R3 - Enqueued");
1467 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 3);
1468 TEST_CHECK(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
1469
1470 /*
1471 * Now mark the previous connection as
1472 * active. It should receive at least
1473 * one of the requests.
1474 */
1475 TEST_CASE("C2 active, R3 - Signal active, should balance");
1476 trunk_connection_signal_active(tconn);
1477
1478 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 3);
1479 TEST_CHECK(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) >= 1);
1480
1481 talloc_free(ctx);
1482 talloc_free(preq);
1483}
1484
1485#define ALLOC_REQ(_id) \
1486do { \
1487 treq_##_id = trunk_request_alloc(trunk, NULL); \
1488 preq_##_id = talloc_zero(ctx, test_proto_request_t); \
1489 preq_##_id->treq = treq_##_id; \
1490 preq_##_id->priority = next_prio++; \
1491} while (0)
1492
1493static void test_connection_levels_max(void)
1494{
1495 TALLOC_CTX *ctx = talloc_init_const("test");
1496 trunk_t *trunk;
1497 fr_event_list_t *el;
1498 trunk_conf_t conf = {
1499 .start = 0, /* No connections on start */
1500 .min = 0,
1501 .max = 2,
1502 .max_req_per_conn = 2,
1503 .target_req_per_conn = 2, /* One request per connection */
1504 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1505 };
1506 test_proto_request_t *preq_a, *preq_b, *preq_c, *preq_d, *preq_e;
1507 trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL, *treq_d = NULL, *treq_e = NULL;
1508 int next_prio = 0;
1509
1510 DEBUG_LVL_SET;
1511
1512 el = fr_event_list_alloc(ctx, NULL, NULL);
1513 fr_timer_list_set_time_func(el->tl, test_time);
1514
1515 /* Need to provide a timer starting value above zero */
1516 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1517
1518 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1519 TRUNK_VERIFY(trunk);
1520
1521 /*
1522 * Queuing a request should start a connection.
1523 */
1524 TEST_CASE("C0, R1 - Enqueue should spawn");
1525 ALLOC_REQ(a);
1526 TEST_CHECK(trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1527 TRUNK_VERIFY(trunk);
1528
1529 /*
1530 * Like test_connection_start_on_enqueue(), you have to process the backlog
1531 * to start the chain of events.
1532 */
1533 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1534 fr_event_corral(el, test_time_base, false);
1535 fr_event_service(el);
1536
1537 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING), 1);
1538 TRUNK_VERIFY(trunk);
1539
1540 /*
1541 * Queuing another request should *NOT* start another connection
1542 */
1543 TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1544 ALLOC_REQ(b);
1545 TEST_CHECK(trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1546 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING), 1);
1547 TRUNK_VERIFY(trunk);
1548
1549 TEST_CASE("C1 connecting, R3 - MUST NOT spawn");
1550 ALLOC_REQ(c);
1551 TEST_CHECK(trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1552 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING), 1);
1553 TRUNK_VERIFY(trunk);
1554
1555 TEST_CASE("C1 connecting, R4 - MUST NOT spawn");
1556 ALLOC_REQ(d);
1557 TEST_CHECK(trunk_request_enqueue(&treq_d, trunk, NULL, preq_d, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1558 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING), 1);
1559 TRUNK_VERIFY(trunk);
1560
1561 TEST_CASE("C1 connecting, R5 - MUST NOT spawn, NO CAPACITY");
1562 ALLOC_REQ(e);
1563 TEST_CHECK(trunk_request_enqueue(&treq_e, trunk, NULL, preq_e, NULL) == TRUNK_ENQUEUE_NO_CAPACITY);
1564 TEST_CHECK_LEN(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING), 1);
1565 TRUNK_VERIFY(trunk);
1566
1567 /*
1568 * Allowing connection to open
1569 */
1570 fr_event_corral(el, test_time_base, false);
1571 fr_event_service(el);
1572
1573 TEST_CASE("C1 active, R4 - Check pending 2");
1574 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 2);
1575 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_BACKLOG), 2);
1576 TRUNK_VERIFY(trunk);
1577
1578 /*
1579 * Sending requests
1580 */
1581 fr_event_corral(el, test_time_base, false);
1582 fr_event_service(el);
1583
1584 TEST_CASE("C1 active, R4 - Check sent 2");
1585 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT), 2);
1586 TRUNK_VERIFY(trunk);
1587
1588 /*
1589 * Looping I/O
1590 */
1591 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1592 fr_event_corral(el, test_time_base, false);
1593 fr_event_service(el);
1594
1595 /*
1596 * Receiving responses
1597 */
1598 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(2));
1599 fr_event_corral(el, test_time_base, false);
1600 fr_event_service(el);
1601
1602 TEST_CHECK(preq_a->completed == true);
1603 TEST_CHECK(preq_a->failed == false);
1604 TEST_CHECK(preq_a->cancelled == false);
1605 TEST_CHECK(preq_a->freed == true);
1606
1607 TEST_CHECK(preq_b->completed == true);
1608 TEST_CHECK(preq_b->failed == false);
1609 TEST_CHECK(preq_b->cancelled == false);
1610 TEST_CHECK(preq_b->freed == true);
1611
1612 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING), 2);
1613 TEST_CHECK_LEN(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_BACKLOG), 0);
1614 TRUNK_VERIFY(trunk);
1615
1616 TEST_CASE("C1 active, R0 - Check complete 2, pending 0");
1617
1618 /*
1619 * Sending requests
1620 */
1621 fr_event_corral(el, test_time_base, false);
1622 fr_event_service(el);
1623
1624 /*
1625 * Looping I/O
1626 */
1627 fr_event_corral(el, test_time_base, false);
1628 fr_event_service(el);
1629
1630 /*
1631 * Receiving responses
1632 */
1633 fr_event_corral(el, test_time_base, false);
1634 fr_event_service(el);
1635
1636 TEST_CHECK(preq_c->completed == true);
1637 TEST_CHECK(preq_c->failed == false);
1638 TEST_CHECK(preq_c->cancelled == false);
1639 TEST_CHECK(preq_c->freed == true);
1640
1641 TEST_CHECK(preq_d->completed == true);
1642 TEST_CHECK(preq_d->failed == false);
1643 TEST_CHECK(preq_d->cancelled == false);
1644 TEST_CHECK(preq_d->freed == true);
1645
1646 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
1647 TRUNK_VERIFY(trunk);
1648
1649 talloc_free(trunk);
1650 talloc_free(ctx);
1651}
1652
1653static void test_connection_levels_alternating_edges(void)
1654{
1655 TALLOC_CTX *ctx = talloc_init_const("test");
1656 trunk_t *trunk;
1657 fr_event_list_t *el;
1658 trunk_conf_t conf = {
1659 .start = 0, /* No connections on start */
1660 .min = 0,
1661 .max = 0,
1662 .max_req_per_conn = 0,
1663 .target_req_per_conn = 2, /* One request per connection */
1664 .manage_interval = fr_time_delta_from_nsec(NSEC / 10)
1665 };
1666
1667 test_proto_request_t *preq_a, *preq_b, *preq_c;
1668 trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1669 test_proto_stats_t stats;
1670 int next_prio = 0;
1671
1672 DEBUG_LVL_SET;
1673
1674 el = fr_event_list_alloc(ctx, NULL, NULL);
1675 fr_timer_list_set_time_func(el->tl, test_time);
1676
1677 /* Need to provide a timer starting value above zero */
1678 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1679
1680 memset(&stats, 0, sizeof(stats));
1681 trunk = test_setup_trunk(ctx, el, &conf, true, &stats);
1682
1683 /*
1684 * Queuing a request should start a connection.
1685 */
1686 TEST_CASE("C0, R1 - Enqueue should spawn");
1687 ALLOC_REQ(a);
1688 TEST_CHECK(trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1689
1690 /*
1691 * Processing the event associated with the backlog creates
1692 * the connection in connecting state..
1693 */
1694 fr_event_corral(el, test_time_base, false);
1695 fr_event_service(el);
1696
1697 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1698
1699 TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1700 ALLOC_REQ(b);
1701 TEST_CHECK(trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1702 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1703 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1704
1705 /*
1706 * Open connection
1707 */
1708 fr_event_corral(el, test_time_base, false);
1709 fr_event_service(el);
1710
1711 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 2);
1712
1713 TEST_CASE("C1 connected, R3 - should spawn");
1714 ALLOC_REQ(c);
1715 TEST_CHECK(trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == TRUNK_ENQUEUE_OK);
1716 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1717
1718 fr_event_corral(el, test_time_base, false);
1719 fr_event_service(el);
1720
1721 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 3);
1722 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
1723 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1724
1725 /*
1726 * Complete requests
1727 */
1728 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1729
1730 fr_event_corral(el, test_time_base, false);
1731 fr_event_service(el);
1732
1733 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1734
1735 TEST_CASE("C1 connected, C2 connecting, R2 - MUST NOT spawn");
1736 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 3);
1737 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 2);
1738
1739 /*
1740 * Finish the last request, should close one connection
1741 */
1742 fr_event_corral(el, test_time_base, false);
1743 fr_event_service(el);
1744
1745 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1746
1747 TEST_CASE("C1 connected, R0");
1748 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_ALL) == 0);
1749 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
1750
1751 /*
1752 * Requests now done, should close another connection
1753 */
1754 fr_event_corral(el, test_time_base, false);
1755 fr_event_service(el);
1756
1757 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1758
1759 TEST_CASE("C0, R0");
1760 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 0);
1761
1762 TEST_CHECK(stats.completed == 3);
1763 TEST_CHECK(stats.failed == 0);
1764 TEST_CHECK(stats.cancelled == 0);
1765 TEST_CHECK(stats.freed == 3);
1766
1767 /*
1768 * Queuing a request should start a connection.
1769 */
1770 TEST_CASE("C0, R1 - Enqueue should spawn");
1771 ALLOC_REQ(a);
1772 TEST_CHECK(trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1773
1774 /*
1775 * ...once the event associated with the backlogged request is handled.
1776 */
1777 fr_event_corral(el, test_time_base, false);
1778 fr_event_service(el);
1779
1780 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1781
1782 TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1783 ALLOC_REQ(b);
1784 TEST_CHECK(trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1785 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1786 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1787
1788 /*
1789 * Open connection
1790 */
1791 fr_event_corral(el, test_time_base, false);
1792 fr_event_service(el);
1793
1794 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_PENDING) == 2);
1795
1796 TEST_CASE("C1 connected, R3 - should spawn");
1797 ALLOC_REQ(c);
1798 TEST_CHECK(trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == TRUNK_ENQUEUE_OK);
1799 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1800
1801 fr_event_corral(el, test_time_base, false);
1802 fr_event_service(el);
1803
1804 TEST_CHECK(trunk_request_count_by_state(trunk, TRUNK_CONN_ALL, TRUNK_REQUEST_STATE_SENT) == 3);
1805 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 1);
1806 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1807
1808 talloc_free(trunk);
1809 talloc_free(ctx);
1810}
1811
1812/*
1813 * Test the global trunk state transitions: PENDING -> ACTIVE -> FAILED.
1814 */
1815static void test_trunk_state_pending_active_failed(void)
1816{
1817 TALLOC_CTX *ctx = talloc_init_const("test");
1818 trunk_t *trunk;
1819 fr_event_list_t *el;
1820 int events;
1821 trunk_conf_t conf = {
1822 .start = 2,
1823 .min = 2,
1824 .conn_conf = &(connection_conf_t){
1825 /*
1826 * Long, so failed connections stay CLOSED
1827 * while we check the FAILED state.
1828 */
1829 .reconnection_delay = fr_time_delta_from_sec(10)
1830 }
1831 };
1832 trunk_io_funcs_t io_funcs = {
1833 .connection_alloc = test_setup_socket_pair_connection_alloc,
1834 .request_prioritise = fr_pointer_cmp,
1835 };
1836
1837 DEBUG_LVL_SET;
1838
1839 el = fr_event_list_alloc(ctx, NULL, NULL);
1840 fr_timer_list_set_time_func(el->tl, test_time);
1841
1842 trunk = trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false, NULL);
1843 TEST_CHECK(trunk != NULL);
1844 if (!trunk) return;
1845
1846 /*
1847 * start = 2 spawns two connections which are connecting. No
1848 * active connections yet, so the trunk is PENDING.
1849 */
1850 TEST_CASE("PENDING - connections are connecting");
1851 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 2);
1852 TEST_CHECK(trunk->pub.state == TRUNK_STATE_PENDING);
1853
1854 /*
1855 * Let the connections open. The trunk becomes ACTIVE.
1856 */
1857 events = fr_event_corral(el, test_time_base, true);
1858 TEST_CHECK(events == 2); /* Two I/O write events, no timers */
1859 fr_event_service(el);
1860
1861 TEST_CASE("ACTIVE - connections are active");
1862 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 2);
1863 TEST_CHECK(trunk->pub.state == TRUNK_STATE_ACTIVE);
1864
1865 /*
1866 * Fail all connections. With a long reconnection_delay they stay
1867 * CLOSED, so the trunk becomes FAILED.
1868 */
1869 TEST_CASE("FAILED - all connections closed / in reconnect backoff");
1870 trunk_reconnect(trunk, TRUNK_CONN_ACTIVE, CONNECTION_FAILED);
1871
1872 (void) fr_event_corral(el, test_time_base, false);
1873 fr_event_service(el);
1874
1875 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CLOSED) == 2);
1876 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 0);
1877 TEST_CHECK(trunk->pub.state == TRUNK_STATE_FAILED);
1878
1879 talloc_free(trunk);
1880 talloc_free(ctx);
1881}
1882
1883/*
1884 * Test the global trunk FULL state: no active connections, but the
1885 * maximum permitted connections are all connected and full.
1886 */
1887static void test_trunk_state_full(void)
1888{
1889 TALLOC_CTX *ctx = talloc_init_const("test");
1890 trunk_t *trunk;
1891 fr_event_list_t *el;
1892 trunk_conf_t conf = {
1893 .start = 0, /* No connections on start */
1894 .min = 0,
1895 .max = 1,
1896 .max_req_per_conn = 1,
1897 .target_req_per_conn = 1,
1898 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1899 };
1900 test_proto_request_t *preq_a;
1901 trunk_request_t *treq_a = NULL;
1902 int next_prio = 0;
1903
1904 DEBUG_LVL_SET;
1905
1906 el = fr_event_list_alloc(ctx, NULL, NULL);
1907 fr_timer_list_set_time_func(el->tl, test_time);
1908
1909 /* Need to provide a timer starting value above zero */
1910 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1911
1912 trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1913
1914 /*
1915 * No connections at all -> IDLE.
1916 */
1917 TEST_CASE("IDLE - no connections");
1918 TEST_CHECK(trunk->pub.state == TRUNK_STATE_IDLE);
1919
1920 /*
1921 * Enqueue a request. It enters the backlog and spawns a connection.
1922 */
1923 ALLOC_REQ(a);
1924 TEST_CHECK(trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == TRUNK_ENQUEUE_IN_BACKLOG);
1925
1926 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1927 fr_event_corral(el, test_time_base, false);
1928 fr_event_service(el);
1929
1930 /*
1931 * The connection spawned by the backlog is connecting -> PENDING.
1932 */
1933 TEST_CASE("PENDING - connection connecting");
1934 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) == 1);
1935 TEST_CHECK(trunk->pub.state == TRUNK_STATE_PENDING);
1936
1937 /*
1938 * Open the connection. The backlogged request is assigned to it,
1939 * which fills it (max_req_per_conn == 1). With max == 1 and that
1940 * one connection full (and none active or connecting), the trunk
1941 * is FULL.
1942 */
1943 fr_event_corral(el, test_time_base, false);
1944 fr_event_service(el);
1945
1946 TEST_CASE("FULL - the maximum number of connections are all full");
1947 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_FULL) == 1);
1948 TEST_CHECK(trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE) == 0);
1949 TEST_CHECK(trunk->pub.state == TRUNK_STATE_FULL);
1950
1951 talloc_free(trunk);
1952 talloc_free(ctx);
1953}
1954
1955#undef fr_time /* Need to the real time */
1956static void test_enqueue_and_io_speed(void)
1957{
1958 TALLOC_CTX *ctx = talloc_init_const("test");
1959 trunk_t *trunk;
1960 fr_event_list_t *el;
1961 int events;
1962 trunk_conf_t conf = {
1963 .start = 1,
1964 .min = 1,
1965 .max = 0,
1966 .max_req_per_conn = 0,
1967 .target_req_per_conn = 0, /* One request per connection */
1968 .req_pool_headers = 1,
1969 .req_pool_size = sizeof(test_proto_request_t),
1970 .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1971 };
1972 size_t i = 0, requests = 100000;
1973 fr_time_t enqueue_start, enqueue_stop, io_start, io_stop;
1974 fr_time_delta_t enqueue_time, io_time, total_time;
1975 trunk_request_t **treq_array;
1976 test_proto_request_t **preq_array;
1977 test_proto_stats_t stats;
1978
1979 DEBUG_LVL_SET;
1980
1981 el = fr_event_list_alloc(ctx, NULL, NULL);
1982 fr_timer_list_set_time_func(el->tl, test_time);
1983
1984 /* Need to provide a timer starting value above zero */
1985 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1986
1987 memset(&stats, 0, sizeof(stats));
1988 trunk = test_setup_trunk(ctx, el, &conf, true, &stats);
1989
1990 /*
1991 * Open the connections
1992 */
1993 fr_event_corral(el, test_time_base, false);
1994 fr_event_service(el);
1995
1996 /*
1997 * Build up a cache of requests
1998 * This prevents all mallocs on request enqueue.
1999 *
2000 * When the server's running, this does represent
2001 * close to what we'd have as a steady state.
2002 */
2003 MEM(treq_array = talloc_array(ctx, trunk_request_t *, requests));
2004 for (i = 0; i < requests; i++) treq_array[i] = trunk_request_alloc(trunk, NULL);
2005 for (i = 0; i < requests; i++) trunk_request_free(&treq_array[i]);
2006
2007 MEM(preq_array = talloc_array(ctx, test_proto_request_t *, requests));
2008
2009 DEBUG_LVL_SET;
2010
2011 TEST_CASE("Enqueue requests");
2012 enqueue_start = fr_time();
2013// ProfilerStart(getenv("FR_PROFILE"));
2014 for (i = 0; i < requests; i++) {
2015 trunk_request_t *treq;
2016 test_proto_request_t *preq = NULL;
2017
2018 treq = trunk_request_alloc(trunk, NULL);
2019 preq = talloc_zero(treq, test_proto_request_t);
2020 preq->treq = treq;
2021 trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
2022 }
2023 enqueue_stop = fr_time();
2024 enqueue_time = fr_time_sub(enqueue_stop, enqueue_start);
2025 if (acutest_verbose_level_ >= 1) {
2026 INFO("Enqueue time %pV (%u rps) (%"PRIu64"/%"PRIu64")",
2027 fr_box_time_delta(enqueue_time),
2028 (uint32_t)(requests / ((float)(fr_time_delta_unwrap(enqueue_time)) / NSEC)),
2029 trunk->pub.req_alloc_new, trunk->pub.req_alloc_reused);
2030 }
2031
2032 TEST_CASE("Perform I/O operations");
2033 io_start = fr_time();
2034 while (true) {
2035 events = fr_event_corral(el, test_time_base, false);
2036 if (!events) break;
2037 fr_event_service(el);
2038 test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.25));
2039 }
2040 io_stop = fr_time();
2041 io_time = fr_time_sub(io_stop, io_start);
2042
2043 if (acutest_verbose_level_ >= 1) {
2044 INFO("I/O time %pV (%u rps)",
2045 fr_box_time_delta(io_time),
2046 (uint32_t)(requests / ((float)(fr_time_delta_unwrap(io_time)) / NSEC)));
2047 }
2048
2049 if (acutest_verbose_level_ >= 1) {
2050 total_time = fr_time_sub(io_stop, enqueue_start);
2051 INFO("Total time %pV (%u rps)",
2052 fr_box_time_delta(total_time),
2053 (uint32_t)(requests / ((float)(fr_time_delta_unwrap(total_time)) / NSEC)));
2054 }
2055
2056 TEST_CHECK_LEN(stats.completed, requests);
2057 TEST_CHECK_LEN(stats.failed, 0);
2058 TEST_CHECK_LEN(stats.cancelled, 0);
2059 TEST_CHECK_LEN(stats.freed, requests);
2060
2061// ProfilerStop();
2062
2063 talloc_free(ctx);
2064}
2065
2066/*
2067 * Connection spawning
2068 */
2069TEST_LIST = {
2070 /*
2071 * Basic tests
2072 */
2073 { "Basic - Alloc then free", test_socket_pair_alloc_then_free },
2074 { "Basic - Alloc then reconnect then free", test_socket_pair_alloc_then_reconnect_then_free },
2075
2076 /*
2077 * Connection timeout
2078 */
2079 { "Timeouts - Connection", test_socket_pair_alloc_then_connect_timeout },
2080 { "Timeouts - Reconnect delay", test_socket_pair_alloc_then_reconnect_check_delay },
2081
2082 /*
2083 * Basic enqueue/dequeue
2084 */
2085 { "Enqueue - Basic", test_enqueue_basic },
2086 { "Enqueue - Cancellation points", test_enqueue_cancellation_points },
2087 { "Enqueue - Partial state transitions", test_partial_to_complete_states },
2088 { "Requeue - On reconnect", test_requeue_on_reconnect },
2089
2090 /*
2091 * Rebalance
2092 */
2093 { "Rebalance - Connection rebalance", test_connection_rebalance_requests },
2094
2095 /*
2096 * Connection spawning tests
2097 */
2098 { "Spawn - Test connection start on enqueue", test_connection_start_on_enqueue },
2099 { "Spawn - Connection levels max", test_connection_levels_max },
2100 { "Spawn - Connection levels alternating edges",test_connection_levels_alternating_edges },
2101
2102 /*
2103 * Trunk state transitions
2104 */
2105 { "State - Pending, active, failed", test_trunk_state_pending_active_failed },
2106 { "State - Full", test_trunk_state_full },
2107
2108 /*
2109 * Performance tests
2110 */
2111 { "Speed Test - Enqueue, and I/O", test_enqueue_and_io_speed },
2112 TEST_TERMINATOR
2113};
TEST_CHECK
#define TEST_CHECK(cond)
Definition acutest.h:101
acutest_verbose_level_
static int acutest_verbose_level_
Definition acutest.h:472
TEST_CASE
#define TEST_CASE(name)
Definition acutest.h:203
TEST_TERMINATOR
#define TEST_TERMINATOR
Definition acutest.h:77
TEST_MSG
#define TEST_MSG(...)
Definition acutest.h:234
TEST_CHECK_LEN
#define TEST_CHECK_LEN(_got, _exp)
Definition acutest_helpers.h:32
request
static uint8_t request[4]
Definition bio_dedup_tests.c:35
CMP_PREFER_SMALLER
#define CMP_PREFER_SMALLER(_a, _b)
Evaluates to +1 for a > b, and -1 for a < b.
Definition build.h:105
CC_NO_UBSAN
#define CC_NO_UBSAN(_sanitize)
Definition build.h:503
UNUSED
#define UNUSED
Definition build.h:384
fr_talloc_fault_setup
void fr_talloc_fault_setup(void)
Register talloc fault handlers.
Definition debug.c:1050
MEM
#define MEM(x)
Definition debug.h:38
fr_dlist_head
static void * fr_dlist_head(fr_dlist_head_t const *list_head)
Return the HEAD item of a list or NULL if the list is empty.
Definition dlist.h:468
fr_event_fd_insert
#define fr_event_fd_insert(...)
Definition event.h:247
FR_EVENT_FILTER_IO
@ FR_EVENT_FILTER_IO
Combined filter for read/write functions/.
Definition event.h:83
talloc_free
talloc_free(hp)
fr_event_service
void fr_event_service(fr_event_list_t *el)
Service any outstanding timer or file descriptor events.
Definition event.c:2222
fr_event_corral
int fr_event_corral(fr_event_list_t *el, fr_time_t now, bool wait)
Gather outstanding timer and file descriptor events.
Definition event.c:2090
fr_event_list_num_timers
uint64_t fr_event_list_num_timers(fr_event_list_t *el)
Return the number of timer events currently scheduled.
Definition event.c:553
fr_time
#define fr_time()
Definition event.c:60
fr_event_list_alloc
fr_event_list_t * fr_event_list_alloc(TALLOC_CTX *ctx, fr_event_status_cb_t status, void *status_uctx)
Initialise a new event list.
Definition event.c:2553
fr_event_fd_delete
int fr_event_fd_delete(fr_event_list_t *el, int fd, fr_event_filter_t filter)
Remove a file descriptor from the event loop.
Definition event.c:1206
fr_event_list_s
Stores all information relating to an event list.
Definition event.c:377
uint32_t
unsigned int uint32_t
Definition merged_model.c:33
ssize_t
long int ssize_t
Definition merged_model.c:24
uint8_t
unsigned char uint8_t
Definition merged_model.c:30
size_t
unsigned long int size_t
Definition merged_model.c:25
request_t
Definition merged_model.c:232
fr_minmax_heap_min_peek
void * fr_minmax_heap_min_peek(fr_minmax_heap_t *hp)
Definition minmax_heap.c:544
fr_pointer_cmp
fr_cmp_ret_t fr_pointer_cmp(void const *a, void const *b)
Compares two pointers.
Definition misc.c:449
fr_nonblock
int fr_nonblock(UNUSED int fd)
Definition misc.c:293
fr_cmp_ret_t
fr_cmp_ret_t
Result of an ordering comparison.
Definition misc.h:50
io_funcs
static const trunk_io_funcs_t io_funcs
Definition bio.c:2719
fr_assert
#define fr_assert(_expr)
Definition rad_assert.h:37
INFO
#define INFO(fmt,...)
Definition radict.c:67
events
static fr_event_list_t * events
Definition radsniff.c:58
conf
static rs_t * conf
Definition radsniff.c:52
connection_get_num_timed_out
uint64_t connection_get_num_timed_out(connection_t const *conn)
Return the number of times this connection has timed out whilst connecting.
Definition connection.c:598
connection_get_num_reconnected
uint64_t connection_get_num_reconnected(connection_t const *conn)
Return the number of times we've attempted to establish or re-establish this connection.
Definition connection.c:586
connection_signal_reconnect
void connection_signal_reconnect(connection_t *conn, connection_reason_t reason)
Asynchronously signal the connection should be reconnected.
Definition connection.c:1197
connection_signal_on_fd
int connection_signal_on_fd(connection_t *conn, int fd)
Setup the connection to change states to connected or failed based on I/O events.
Definition connection.c:1438
connection_alloc
connection_t * connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, connection_funcs_t const *funcs, connection_conf_t const *conf, char const *log_prefix, void const *uctx)
Allocate a new connection.
Definition connection.c:1546
connection_s
Definition connection.c:87
connection_state_t
connection_state_t
Definition connection.h:47
CONNECTION_STATE_CONNECTED
@ CONNECTION_STATE_CONNECTED
File descriptor is open (ready for writing).
Definition connection.h:54
CONNECTION_STATE_CONNECTING
@ CONNECTION_STATE_CONNECTING
Waiting for connection to establish.
Definition connection.h:52
CONNECTION_FAILED
@ CONNECTION_FAILED
Connection is being reconnected because it failed.
Definition connection.h:89
connection_conf_t
Definition connection.h:96
connection_funcs_t
Holds a complete set of functions for a connection.
Definition connection.h:199
buff
static char buff[sizeof("18446744073709551615")+3]
Definition size_tests.c:37
test_time
static fr_time_t test_time(void)
Definition slab_tests.c:43
test_time_base
static fr_time_t test_time_base
Definition slab_tests.c:42
count
return count
Definition module.c:155
fr_syserror
char const * fr_syserror(int num)
Guaranteed to be thread-safe version of strerror.
Definition syserror.c:243
talloc_init_const
static TALLOC_CTX * talloc_init_const(char const *name)
Allocate a top level chunk with a constant name.
Definition talloc.h:127
fr_time_add_time_delta
static fr_time_t fr_time_add_time_delta(fr_time_t a, fr_time_delta_t b)
Definition time.h:173
fr_time_delta_unwrap
static int64_t fr_time_delta_unwrap(fr_time_delta_t time)
Definition time.h:154
fr_time_delta_from_sec
static fr_time_delta_t fr_time_delta_from_sec(int64_t sec)
Definition time.h:590
NSEC
#define NSEC
Definition time.h:379
fr_time_delta_from_nsec
static fr_time_delta_t fr_time_delta_from_nsec(int64_t nsec)
Definition time.h:563
fr_time_sub
#define fr_time_sub(_a, _b)
Subtract one time from another.
Definition time.h:229
fr_time_delta_s
A time delta, a difference in time measured in nanoseconds.
Definition time.h:80
fr_time_s
"server local" time.
Definition time.h:69
fr_timer_list_set_time_func
void fr_timer_list_set_time_func(fr_timer_list_t *tl, fr_event_time_source_t func)
Override event list time source.
Definition timer.c:1199
trunk.c
A management API for bonding multiple connections together.
trunk_connection_pop_cancellation
int trunk_connection_pop_cancellation(trunk_request_t **treq_out, trunk_connection_t *tconn)
Pop a cancellation request off a connection's cancellation queue.
Definition trunk.c:3931
trunk_reconnect
void trunk_reconnect(trunk_t *trunk, int states, connection_reason_t reason)
Force the trunk to re-establish its connections.
Definition trunk.c:4902
trunk_request_signal_partial
void trunk_request_signal_partial(trunk_request_t *treq)
Signal a partial write.
Definition trunk.c:2093
trunk_request_signal_cancel_sent
void trunk_request_signal_cancel_sent(trunk_request_t *treq)
Signal that a remote server has been notified of the cancellation.
Definition trunk.c:2324
trunk_connection_signal_readable
void trunk_connection_signal_readable(trunk_connection_t *tconn)
Signal that a trunk connection is readable.
Definition trunk.c:4017
trunk_request_alloc
trunk_request_t * trunk_request_alloc(trunk_t *trunk, request_t *request)
(Pre-)Allocate a new trunk request
Definition trunk.c:2542
trunk_request_count_by_connection
uint32_t trunk_request_count_by_connection(trunk_connection_t const *tconn, int req_state)
Return the count number of requests associated with a trunk connection.
Definition trunk.c:2963
trunk_request_s::pub
struct trunk_request_pub_s pub
Public fields in the trunk request.
Definition trunk.c:100
trunk_s::pub
struct trunk_pub_s pub
Public fields in the trunk connection.
Definition trunk.c:220
trunk_request_enqueue
trunk_enqueue_t trunk_request_enqueue(trunk_request_t **treq_out, trunk_t *trunk, request_t *request, void *preq, void *rctx)
Enqueue a request that needs data written to the trunk.
Definition trunk.c:2657
trunk_request_signal_cancel_complete
void trunk_request_signal_cancel_complete(trunk_request_t *treq)
Signal that a remote server acked our cancellation.
Definition trunk.c:2348
trunk_connection_pop_request
int trunk_connection_pop_request(trunk_request_t **treq_out, trunk_connection_t *tconn)
Pop a request off a connection's pending queue.
Definition trunk.c:3979
trunk_s::connecting
fr_dlist_head_t connecting
Connections which are not yet in the open state.
Definition trunk.c:246
trunk_request_signal_cancel
void trunk_request_signal_cancel(trunk_request_t *treq)
Cancel a trunk request.
Definition trunk.c:2216
trunk_connection_s::pub
struct trunk_connection_pub_s pub
Public fields in the trunk connection.
Definition trunk.c:138
trunk_connection_count_by_state
uint16_t trunk_connection_count_by_state(trunk_t *trunk, int conn_state)
Return the count number of connections in the specified states.
Definition trunk.c:2939
trunk_alloc
trunk_t * trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, trunk_io_funcs_t const *funcs, trunk_conf_t const *conf, char const *log_prefix, void const *uctx, bool delay_start, fr_pair_list_t *trigger_args)
Allocate a new collection of connections.
Definition trunk.c:5124
trunk_s::active
fr_minmax_heap_t * active
Connections which can service requests.
Definition trunk.c:248
trunk_request_free
void trunk_request_free(trunk_request_t **treq_to_free)
If the trunk request is freed then update the target requests.
Definition trunk.c:2386
trunk_connection_signal_active
void trunk_connection_signal_active(trunk_connection_t *tconn)
Signal a trunk connection is no longer full.
Definition trunk.c:4056
trunk_connection_signal_inactive
void trunk_connection_signal_inactive(trunk_connection_t *tconn)
Signal a trunk connection cannot accept more requests.
Definition trunk.c:4033
trunk_request_count_by_state
uint64_t trunk_request_count_by_state(trunk_t *trunk, int conn_state, int req_state)
Return a count of requests on a connection in a specific state.
Definition trunk.c:4690
trunk_request_signal_cancel_partial
void trunk_request_signal_cancel_partial(trunk_request_t *treq)
Signal a partial cancel write.
Definition trunk.c:2300
trunk_request_signal_sent
void trunk_request_signal_sent(trunk_request_t *treq)
Signal that the request was written to a connection successfully.
Definition trunk.c:2114
trunk_request_signal_complete
void trunk_request_signal_complete(trunk_request_t *treq)
Signal that a trunk request is complete.
Definition trunk.c:2158
trunk_connection_signal_reconnect
void trunk_connection_signal_reconnect(trunk_connection_t *tconn, connection_reason_t reason)
Signal a trunk connection is no longer viable.
Definition trunk.c:4095
trunk_connection_signal_writable
void trunk_connection_signal_writable(trunk_connection_t *tconn)
Signal that a trunk connection is writable.
Definition trunk.c:3999
trunk_connection_s
Associates request queues with a connection.
Definition trunk.c:137
trunk_request_s
Wraps a normal request.
Definition trunk.c:99
trunk_s
Main trunk management handle.
Definition trunk.c:219
TRUNK_VERIFY
#define TRUNK_VERIFY(_trunk)
Definition trunk.h:946
TRUNK_REQUEST_STATE_ALL
#define TRUNK_REQUEST_STATE_ALL
All request states.
Definition trunk.h:205
TRUNK_CONN_FULL
@ TRUNK_CONN_FULL
Connection is full and can't accept any more requests.
Definition trunk.h:104
TRUNK_CONN_CONNECTING
@ TRUNK_CONN_CONNECTING
Connection is connecting.
Definition trunk.h:99
TRUNK_CONN_CLOSED
@ TRUNK_CONN_CLOSED
Connection was closed, either explicitly or due to failure.
Definition trunk.h:103
TRUNK_CONN_ACTIVE
@ TRUNK_CONN_ACTIVE
Connection is connected and ready to service requests.
Definition trunk.h:100
trunk_pub_s::req_alloc_reused
uint64_t _CONST req_alloc_reused
How many requests were reused.
Definition trunk.h:344
trunk_request_pub_s::state
trunk_request_state_t _CONST state
Which list the request is now located in.
Definition trunk.h:359
trunk_request_pub_s::tconn
trunk_connection_t *_CONST tconn
Connection this request belongs to.
Definition trunk.h:363
trunk_io_funcs_t::connection_alloc
trunk_connection_alloc_t connection_alloc
Allocate a new connection_t.
Definition trunk.h:747
trunk_connection_event_t
trunk_connection_event_t
What type of I/O events the trunk connection is currently interested in receiving.
Definition trunk.h:81
TRUNK_CONN_EVENT_BOTH
@ TRUNK_CONN_EVENT_BOTH
Trunk should be notified if a connection is readable or writable.
Definition trunk.h:88
TRUNK_CONN_EVENT_WRITE
@ TRUNK_CONN_EVENT_WRITE
Trunk should be notified if a connection is writable.
Definition trunk.h:86
TRUNK_CONN_EVENT_NONE
@ TRUNK_CONN_EVENT_NONE
Don't notify the trunk on connection state changes.
Definition trunk.h:82
TRUNK_CONN_EVENT_READ
@ TRUNK_CONN_EVENT_READ
Trunk should be notified if a connection is readable.
Definition trunk.h:84
TRUNK_CONN_ALL
#define TRUNK_CONN_ALL
All connection states.
Definition trunk.h:120
trunk_cancel_reason_t
trunk_cancel_reason_t
Reasons for a request being cancelled.
Definition trunk.h:55
trunk_pub_s::req_alloc_new
uint64_t _CONST req_alloc_new
How many requests we've allocated.
Definition trunk.h:342
trunk_connection_pub_s::conn
connection_t *_CONST conn
The underlying connection.
Definition trunk.h:383
TRUNK_STATE_PENDING
@ TRUNK_STATE_PENDING
Trunk has connections, but none are usable yet; connections are being opened (INIT / CONNECTING).
Definition trunk.h:66
TRUNK_STATE_FAILED
@ TRUNK_STATE_FAILED
Trunk has connections, but they have all failed and are closed / in reconnect backoff.
Definition trunk.h:72
TRUNK_STATE_ACTIVE
@ TRUNK_STATE_ACTIVE
Trunk has at least one active connection which can service requests.
Definition trunk.h:64
TRUNK_STATE_FULL
@ TRUNK_STATE_FULL
Trunk has no active connections, but has one or more connected connections which are all full (at cap...
Definition trunk.h:68
TRUNK_STATE_IDLE
@ TRUNK_STATE_IDLE
Trunk has no connections.
Definition trunk.h:63
trunk_enqueue_t
trunk_enqueue_t
Definition trunk.h:158
TRUNK_ENQUEUE_OK
@ TRUNK_ENQUEUE_OK
Operation was successful.
Definition trunk.h:160
TRUNK_ENQUEUE_NO_CAPACITY
@ TRUNK_ENQUEUE_NO_CAPACITY
At maximum number of connections, and no connection has capacity.
Definition trunk.h:161
TRUNK_ENQUEUE_IN_BACKLOG
@ TRUNK_ENQUEUE_IN_BACKLOG
Request should be enqueued in backlog.
Definition trunk.h:159
trunk_request_pub_s::preq
void *_CONST preq
Data for the muxer to write to the connection.
Definition trunk.h:365
trunk_io_funcs_t::request_cancel_mux
trunk_request_cancel_mux_t request_cancel_mux
!< Read one or more requests from a connection.
Definition trunk.h:760
trunk_request_state_t
trunk_request_state_t
Used for sanity checks and to simplify freeing.
Definition trunk.h:171
TRUNK_REQUEST_STATE_PARTIAL
@ TRUNK_REQUEST_STATE_PARTIAL
Some of the request was written to the socket, more of it should be written later.
Definition trunk.h:180
TRUNK_REQUEST_STATE_CANCEL_SENT
@ TRUNK_REQUEST_STATE_CANCEL_SENT
We've informed the remote server that the request has been cancelled.
Definition trunk.h:195
TRUNK_REQUEST_STATE_CANCEL
@ TRUNK_REQUEST_STATE_CANCEL
A request on a particular socket was cancel.
Definition trunk.h:194
TRUNK_REQUEST_STATE_CANCEL_PARTIAL
@ TRUNK_REQUEST_STATE_CANCEL_PARTIAL
We partially wrote a cancellation request.
Definition trunk.h:197
TRUNK_REQUEST_STATE_BACKLOG
@ TRUNK_REQUEST_STATE_BACKLOG
In the backlog.
Definition trunk.h:177
TRUNK_REQUEST_STATE_PENDING
@ TRUNK_REQUEST_STATE_PENDING
In the queue of a connection and is pending writing.
Definition trunk.h:178
TRUNK_REQUEST_STATE_SENT
@ TRUNK_REQUEST_STATE_SENT
Was written to a socket. Waiting for a response.
Definition trunk.h:182
trunk_pub_s::state
trunk_state_t _CONST state
Current state of the trunk.
Definition trunk.h:347
trunk_conf_t
Common configuration parameters for a trunk.
Definition trunk.h:234
trunk_io_funcs_t
I/O functions to pass to trunk_alloc.
Definition trunk.h:746
test_proto_request_t::treq
trunk_request_t * treq
Trunk request.
Definition trunk_tests.c:10
DEBUG_LVL_SET
#define DEBUG_LVL_SET
Definition trunk_tests.c:27
TEST_LIST
TEST_LIST
Definition trunk_tests.c:2069
test_demux
static void test_demux(UNUSED fr_event_list_t *el, UNUSED trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
Definition trunk_tests.c:98
test_proto_stats_t::cancelled
uint64_t cancelled
Count of tests in this run that were cancelled.
Definition trunk_tests.c:21
test_trunk_state_full
static void test_trunk_state_full(void)
Definition trunk_tests.c:1887
test_proto_stats_t::freed
uint64_t freed
Count of tests in this run that were freed.
Definition trunk_tests.c:24
test_setup_socket_pair_connection_alloc
static connection_t * test_setup_socket_pair_connection_alloc(trunk_connection_t *tconn, fr_event_list_t *el, connection_conf_t const *conn_conf, char const *log_prefix, UNUSED void *uctx)
Definition trunk_tests.c:319
test_enqueue_and_io_speed
static void test_enqueue_and_io_speed(void)
Definition trunk_tests.c:1956
test_proto_request_t::cancelled
bool cancelled
Seen by the cancelled callback.
Definition trunk_tests.c:11
_conn_io_error
static void _conn_io_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, UNUSED int fd_errno, void *uctx)
Definition trunk_tests.c:141
_conn_io_loopback
static void _conn_io_loopback(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
Whenever the second socket in a socket pair is readable, read all pending data, and write it back.
Definition trunk_tests.c:242
test_mux
static void test_mux(UNUSED fr_event_list_t *el, trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
Definition trunk_tests.c:29
test_trunk_state_pending_active_failed
static void test_trunk_state_pending_active_failed(void)
Definition trunk_tests.c:1815
test_socket_pair_alloc_then_connect_timeout
static void test_socket_pair_alloc_then_connect_timeout(void)
Definition trunk_tests.c:500
test_cancel_mux
static void test_cancel_mux(UNUSED fr_event_list_t *el, trunk_connection_t *tconn, connection_t *conn, UNUSED void *uctx)
Definition trunk_tests.c:61
_conn_io_write
static void _conn_io_write(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
Definition trunk_tests.c:156
test_partial_to_complete_states
static void test_partial_to_complete_states(void)
Definition trunk_tests.c:1000
test_setup_socket_pair_1s_timeout_connection_alloc
static connection_t * test_setup_socket_pair_1s_timeout_connection_alloc(trunk_connection_t *tconn, fr_event_list_t *el, UNUSED connection_conf_t const *conf, char const *log_prefix, void *uctx)
Definition trunk_tests.c:482
test_request_free
static void test_request_free(UNUSED request_t *request, void *preq, void *uctx)
Definition trunk_tests.c:227
test_socket_pair_alloc_then_free
static void test_socket_pair_alloc_then_free(void)
Definition trunk_tests.c:370
test_proto_request_t::failed
bool failed
Seen by the failed callback.
Definition trunk_tests.c:13
test_request_fail
static void test_request_fail(UNUSED request_t *request, void *preq, UNUSED void *rctx, UNUSED trunk_request_state_t state, void *uctx)
Definition trunk_tests.c:215
test_connection_rebalance_requests
static void test_connection_rebalance_requests(void)
Definition trunk_tests.c:1423
_conn_init_no_signal
static connection_state_t _conn_init_no_signal(void **h_out, connection_t *conn, UNUSED void *uctx)
Definition trunk_tests.c:471
test_setup_socket_pair_1s_reconnection_delay_alloc
static connection_t * test_setup_socket_pair_1s_reconnection_delay_alloc(trunk_connection_t *tconn, fr_event_list_t *el, UNUSED connection_conf_t const *conn_conf, char const *log_prefix, void *uctx)
Definition trunk_tests.c:560
test_connection_levels_max
static void test_connection_levels_max(void)
Definition trunk_tests.c:1493
test_socket_pair_alloc_then_reconnect_check_delay
static void test_socket_pair_alloc_then_reconnect_check_delay(void)
Definition trunk_tests.c:578
test_proto_request_t::freed
bool freed
Seen by the free callback.
Definition trunk_tests.c:14
test_proto_stats_t::failed
uint64_t failed
Count of tests in this run that failed.
Definition trunk_tests.c:23
_conn_io_read
static void _conn_io_read(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
Definition trunk_tests.c:150
test_proto_request_t::completed
bool completed
Seen by the complete callback.
Definition trunk_tests.c:12
_conn_notify
static void _conn_notify(trunk_connection_t *tconn, connection_t *conn, fr_event_list_t *el, trunk_connection_event_t notify_on, UNUSED void *uctx)
Definition trunk_tests.c:162
ALLOC_REQ
#define ALLOC_REQ(_id)
Definition trunk_tests.c:1485
test_proto_request_t::priority
int priority
Priority of request.
Definition trunk_tests.c:17
test_socket_pair_alloc_then_reconnect_then_free
static void test_socket_pair_alloc_then_reconnect_then_free(void)
Definition trunk_tests.c:409
test_setup_trunk
static trunk_t * test_setup_trunk(TALLOC_CTX *ctx, fr_event_list_t *el, trunk_conf_t *conf, bool with_cancel_mux, void *uctx)
Definition trunk_tests.c:348
_conn_open
static connection_state_t _conn_open(fr_event_list_t *el, void *h, UNUSED void *uctx)
Insert I/O handlers that loop any data back round.
Definition trunk_tests.c:288
test_enqueue_cancellation_points
static void test_enqueue_cancellation_points(void)
Definition trunk_tests.c:753
test_preq_cmp
static fr_cmp_ret_t test_preq_cmp(void const *a, void const *b)
Definition trunk_tests.c:340
test_requeue_on_reconnect
static void test_requeue_on_reconnect(void)
Definition trunk_tests.c:1075
test_connection_start_on_enqueue
static void test_connection_start_on_enqueue(void)
Definition trunk_tests.c:1363
test_proto_request_t::signal_partial
bool signal_partial
Muxer should signal that this request is partially written.
Definition trunk_tests.c:15
test_request_complete
static void test_request_complete(UNUSED request_t *request, void *preq, UNUSED void *rctx, void *uctx)
Definition trunk_tests.c:203
test_request_cancel
static void test_request_cancel(UNUSED connection_t *conn, void *preq, UNUSED trunk_cancel_reason_t reason, void *uctx)
Definition trunk_tests.c:190
test_enqueue_basic
static void test_enqueue_basic(void)
Definition trunk_tests.c:652
test_proto_stats_t::completed
uint64_t completed
Count of tests in this run that completed.
Definition trunk_tests.c:22
test_connection_levels_alternating_edges
static void test_connection_levels_alternating_edges(void)
Definition trunk_tests.c:1653
_conn_close
static void _conn_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
Definition trunk_tests.c:273
_conn_init
static connection_state_t _conn_init(void **h_out, connection_t *conn, UNUSED void *uctx)
Allocate a basic socket pair.
Definition trunk_tests.c:304
test_proto_request_t::signal_cancel_partial
bool signal_cancel_partial
Muxer should signal that this request is partially cancelled.
Definition trunk_tests.c:16
test_proto_request_t
Definition trunk_tests.c:9
test_proto_stats_t
Definition trunk_tests.c:20
el
static fr_event_list_t * el
Definition unit_test_attribute.c:270
fr_perror
void fr_perror(char const *fmt,...)
Print the current error to stderr with a prefix.
Definition strerror.c:737
fr_box_time_delta
#define fr_box_time_delta(_val)
Definition value.h:397
Generated by   doxygen 1.9.8