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