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