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  fr_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, fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
32 {
33  fr_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 (fr_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  fr_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  fr_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, fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
64 {
65  fr_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 ((fr_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  fr_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  fr_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 fr_trunk_connection_t *tconn, fr_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 FR_TRUNK_REQUEST_STATE_CANCEL:
132  break; /* Hack - just ignore it */
133 
134  case FR_TRUNK_REQUEST_STATE_CANCEL_SENT:
135  /* coverity[tainted_data] */
136  fr_trunk_request_signal_cancel_complete(preq->treq);
137  break;
138 
139  case FR_TRUNK_REQUEST_STATE_SENT:
140  /* coverity[tainted_data] */
141  fr_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  fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t);
156 
157  fr_trunk_connection_signal_reconnect(tconn, FR_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  fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t);
163  fr_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  fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t);
169  fr_trunk_connection_signal_writable(tconn);
170 }
171 
172 static void _conn_notify(fr_trunk_connection_t *tconn, fr_connection_t *conn,
173  fr_event_list_t *el,
174  fr_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 FR_TRUNK_CONN_EVENT_NONE:
180  fr_event_fd_delete(el, fd, FR_EVENT_FILTER_IO);
181  break;
182 
183  case FR_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 FR_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 FR_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 fr_connection_t *conn, void *preq,
201  UNUSED fr_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 fr_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 fr_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 FR_CONNECTION_STATE_CONNECTED;
308 }
309 
310 /** Allocate a basic socket pair
311  *
312  */
313 static fr_connection_state_t _conn_init(void **h_out, fr_connection_t *conn, UNUSED void *uctx)
314 {
315  int *h;
316 
317  h = talloc_array(conn, int, 2);
318  socketpair(AF_UNIX, SOCK_STREAM, 0, h);
319 
320  fr_nonblock(h[0]);
321  fr_nonblock(h[1]);
322  fr_connection_signal_on_fd(conn, h[0]);
323  *h_out = h;
324 
325  return FR_CONNECTION_STATE_CONNECTING;
326 }
327 
328 static fr_connection_t *test_setup_socket_pair_connection_alloc(fr_trunk_connection_t *tconn,
329  fr_event_list_t *el,
330  fr_connection_conf_t const *conn_conf,
331  char const *log_prefix, UNUSED void *uctx)
332 {
333  fr_connection_conf_t cstat;
334 
335  if (!conn_conf) {
336  memset(&cstat, 0, sizeof(cstat));
337  conn_conf = &cstat;
338  }
339  return fr_connection_alloc(tconn, el,
340  &(fr_connection_funcs_t){
341  .init = _conn_init,
342  .open = _conn_open,
343  .close = _conn_close
344  },
345  conn_conf,
346  log_prefix, tconn);
347 }
348 
349 static int8_t test_preq_cmp(void const *a, void const *b)
350 {
351  test_proto_request_t const *preq_a = a;
352  test_proto_request_t const *preq_b = b;
353  return CMP(preq_a->priority, preq_b->priority);
354 }
355 
356 static fr_trunk_t *test_setup_trunk(TALLOC_CTX *ctx, fr_event_list_t *el, fr_trunk_conf_t *conf, bool with_cancel_mux, void *uctx)
357 {
358  fr_trunk_io_funcs_t io_funcs = {
359  .connection_alloc = test_setup_socket_pair_connection_alloc,
360  .connection_notify = _conn_notify,
361  .request_prioritise = test_preq_cmp,
362  .request_mux = test_mux,
363  .request_demux = test_demux,
364  .request_cancel = test_request_cancel,
365  .request_complete = test_request_complete,
366  .request_fail = test_request_fail,
367  .request_free = test_request_free
368  };
369 
370  /*
371  * Function list is copied, so this is OK.
372  */
373  if (with_cancel_mux) io_funcs.request_cancel_mux = test_cancel_mux;
374 
375  return fr_trunk_alloc(ctx, el, &io_funcs, conf, "test_socket_pair", uctx, false);
376 }
377 
378 static void test_socket_pair_alloc_then_free(void)
379 {
380  TALLOC_CTX *ctx = talloc_init_const("test");
381  fr_trunk_t *trunk;
382  fr_event_list_t *el;
383  int events;
384 
385  fr_trunk_conf_t conf = {
386  .start = 2,
387  .min = 2
388  };
389  fr_trunk_io_funcs_t io_funcs = {
390  .connection_alloc = test_setup_socket_pair_connection_alloc,
391  .request_prioritise = fr_pointer_cmp,
392  };
393 
394  DEBUG_LVL_SET;
395 
396  el = fr_event_list_alloc(ctx, NULL, NULL);
397 
398  fr_event_list_set_time_func(el, test_time);
399 
400  trunk = fr_trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false);
401  TEST_CHECK(trunk != NULL);
402  if (!trunk) return;
403 
404  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 2);
405  events = fr_event_corral(el, test_time_base, true);
406  TEST_CHECK(events == 2); /* Two I/O write events, no timers */
407  fr_event_service(el);
408  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 2);
409 
410  events = fr_event_corral(el, test_time_base, false);
411  TEST_CHECK(events == 0); /* I/O events should have been cleared */
412 
413  talloc_free(trunk);
414  talloc_free(ctx);
415 }
416 
417 static void test_socket_pair_alloc_then_reconnect_then_free(void)
418 {
419  TALLOC_CTX *ctx = talloc_init_const("test");
420  fr_trunk_t *trunk;
421  fr_event_list_t *el;
422  int events;
423  fr_trunk_conf_t conf = {
424  .start = 2,
425  .min = 2,
426  .conn_conf = &(fr_connection_conf_t){
427  .reconnection_delay = fr_time_delta_from_nsec(NSEC / 2)
428  }
429  };
430  fr_trunk_io_funcs_t io_funcs = {
431  .connection_alloc = test_setup_socket_pair_connection_alloc,
432  .request_prioritise = fr_pointer_cmp,
433  };
434  DEBUG_LVL_SET;
435 
436  el = fr_event_list_alloc(ctx, NULL, NULL);
437 
438  if (!el) return;
439 
440  fr_event_list_set_time_func(el, test_time);
441 
442  trunk = fr_trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false);
443  TEST_CHECK(trunk != NULL);
444  if (!trunk) return;
445 
446  events = fr_event_corral(el, test_time_base, true);
447  TEST_CHECK(events == 2); /* Two I/O write events, no timers */
448  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 2);
449  fr_event_service(el);
450  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 2);
451 
452  events = fr_event_corral(el, test_time_base, false);
453  TEST_CHECK(events == 0); /* I/O events should have been cleared */
454  TEST_MSG("Got %u events", events);
455 
456  fr_trunk_reconnect(trunk, FR_TRUNK_CONN_ACTIVE, FR_CONNECTION_FAILED);
457 
458  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(2));
459  events = fr_event_corral(el, test_time_base, true);
460  TEST_CHECK(events == 1); /* Two timer events but event loops only adds one to the total*/
461  TEST_MSG("Got %u events", events);
462  fr_event_service(el);
463 
464  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 2);
465 
466  events = fr_event_corral(el, test_time_base, true);
467  TEST_CHECK(events == 2); /* Two I/O write events, no timers */
468  fr_event_service(el);
469 
470  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 2);
471  events = fr_event_corral(el, test_time_base, false);
472  TEST_CHECK(events == 0); /* I/O events should have been cleared */
473 
474  talloc_free(trunk);
475  talloc_free(ctx);
476 }
477 
478 static fr_connection_state_t _conn_init_no_signal(void **h_out, fr_connection_t *conn, UNUSED void *uctx)
479 {
480  int *h;
481 
482  h = talloc_array(conn, int, 2);
483  socketpair(AF_UNIX, SOCK_STREAM, 0, h);
484  *h_out = h;
485 
486  return FR_CONNECTION_STATE_CONNECTING;
487 }
488 
489 static fr_connection_t *test_setup_socket_pair_1s_timeout_connection_alloc(fr_trunk_connection_t *tconn,
490  fr_event_list_t *el,
491  UNUSED fr_connection_conf_t const *conf,
492  char const *log_prefix, void *uctx)
493 {
494  return fr_connection_alloc(tconn, el,
495  &(fr_connection_funcs_t){
496  .init = _conn_init_no_signal,
497  .open = _conn_open,
498  .close = _conn_close
499  },
500  &(fr_connection_conf_t){
501  .connection_timeout = fr_time_delta_from_sec(1),
502  .reconnection_delay = fr_time_delta_from_sec(1)
503  },
504  log_prefix, uctx);
505 }
506 
507 static void test_socket_pair_alloc_then_connect_timeout(void)
508 {
509  TALLOC_CTX *ctx = talloc_init_const("test");
510  fr_trunk_t *trunk;
511  fr_event_list_t *el;
512  int events;
513  fr_trunk_connection_t *tconn;
514  fr_trunk_conf_t conf = {
515  .start = 1,
516  .min = 1
517  };
518  fr_trunk_io_funcs_t io_funcs = {
519  .connection_alloc = test_setup_socket_pair_1s_timeout_connection_alloc,
520  .request_prioritise = fr_pointer_cmp,
521  };
522 
523  DEBUG_LVL_SET;
524 
525  el = fr_event_list_alloc(ctx, NULL, NULL);
526 
527  fr_event_list_set_time_func(el, test_time);
528 
529 
530  trunk = fr_trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false);
531  TEST_CHECK(trunk != NULL);
532  if (!trunk) return;
533 
534  /*
535  * Trigger connection timeout
536  */
537  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 1.5));
538  TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for the connection timeout */
539  events = fr_event_corral(el, test_time_base, true);
540  TEST_CHECK(events == 1); /* We didn't install the I/O events */
541 
542  tconn = fr_dlist_head(&trunk->connecting);
543  TEST_CHECK(tconn != NULL);
544  if (tconn == NULL) return;
545 
546  TEST_CHECK(fr_connection_get_num_timed_out(tconn->pub.conn) == 0);
547  TEST_CHECK(fr_connection_get_num_reconnected(tconn->pub.conn) == 0);
548 
549  /*
550  * Timeout should now fire
551  */
552  fr_event_service(el);
553 
554  /*
555  * Connection delay not implemented for timed out connections
556  */
557  TEST_CHECK(fr_connection_get_num_timed_out(tconn->pub.conn) == 1);
558  TEST_CHECK(fr_connection_get_num_reconnected(tconn->pub.conn) == 1);
559 
560  events = fr_event_corral(el, test_time_base, false);
561  TEST_CHECK(events == 0); /* I/O events should have been cleared */
562 
563  talloc_free(trunk);
564  talloc_free(ctx);
565 }
566 
567 static fr_connection_t *test_setup_socket_pair_1s_reconnection_delay_alloc(fr_trunk_connection_t *tconn,
568  fr_event_list_t *el,
569  UNUSED fr_connection_conf_t const *conn_conf,
570  char const *log_prefix, void *uctx)
571 {
572  return fr_connection_alloc(tconn, el,
573  &(fr_connection_funcs_t){
574  .init = _conn_init,
575  .open = _conn_open,
576  .close = _conn_close
577  },
578  &(fr_connection_conf_t){
579  .connection_timeout = fr_time_delta_from_sec(1),
580  .reconnection_delay = fr_time_delta_from_sec(1)
581  },
582  log_prefix, uctx);
583 }
584 
585 static void test_socket_pair_alloc_then_reconnect_check_delay(void)
586 {
587  TALLOC_CTX *ctx = talloc_init_const("test");
588  fr_trunk_t *trunk;
589  fr_event_list_t *el;
590  int events;
591  fr_trunk_connection_t *tconn;
592  fr_trunk_conf_t conf = {
593  .start = 1,
594  .min = 1,
595  .conn_conf = &(fr_connection_conf_t){
596  .reconnection_delay = fr_time_delta_from_sec(1),
597  .connection_timeout = fr_time_delta_from_sec(1)
598  }
599  };
600  fr_trunk_io_funcs_t io_funcs = {
601  .connection_alloc = test_setup_socket_pair_1s_reconnection_delay_alloc,
602  .request_prioritise = fr_pointer_cmp,
603  };
604 
605  DEBUG_LVL_SET;
606 
607  el = fr_event_list_alloc(ctx, NULL, NULL);
608  fr_event_list_set_time_func(el, test_time);
609 
610  trunk = fr_trunk_alloc(ctx, el, &io_funcs, &conf, "test_socket_pair", NULL, false);
611  TEST_CHECK(trunk != NULL);
612  if (!trunk) return;
613 
614  /*
615  * Trigger connection timeout
616  */
617  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 1.5));
618  TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for the connection timeout */
619  events = fr_event_corral(el, test_time_base, true);
620  TEST_CHECK(events == 2); /* We didn't install the I/O events */
621  fr_event_service(el);
622 
623  tconn = fr_minmax_heap_min_peek(trunk->active);
624  TEST_CHECK(tconn != NULL);
625  if (tconn == NULL) return;
626 
627  TEST_CHECK(fr_connection_get_num_timed_out(tconn->pub.conn) == 0);
628  TEST_CHECK(fr_connection_get_num_reconnected(tconn->pub.conn) == 0);
629 
630  /*
631  * Trigger reconnection
632  */
633  fr_connection_signal_reconnect(tconn->pub.conn, FR_CONNECTION_FAILED);
634  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
635 
636  events = fr_event_corral(el, test_time_base, false);
637  TEST_CHECK(events == 0); /* Reconnect delay not ready to fire yet, no I/O handlers installed */
638  TEST_CHECK(fr_event_list_num_timers(el) == 1); /* One timer event for reconnect delay */
639 
640  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
641  events = fr_event_corral(el, test_time_base, false);
642  TEST_CHECK(events == 1); /* Reconnect delay should now be ready to fire */
643 
644  fr_event_service(el); /* Services the timer, which then triggers init */
645 
646  TEST_CHECK(fr_connection_get_num_timed_out(tconn->pub.conn) == 0);
647  TEST_CHECK(fr_connection_get_num_reconnected(tconn->pub.conn) == 1);
648 
649  events = fr_event_corral(el, test_time_base, true);
650  TEST_CHECK(events == 1); /* Should have a pending I/O event and a timer */
651 
652  talloc_free(trunk);
653  talloc_free(ctx);
654 }
655 
656 /*
657  * Test basic enqueue and dequeue
658  */
659 static void test_enqueue_basic(void)
660 {
661  TALLOC_CTX *ctx = talloc_init_const("test");
662  fr_trunk_t *trunk;
663  fr_event_list_t *el;
664  fr_trunk_conf_t conf = {
665  .start = 1,
666  .min = 1,
667  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
668  };
669  test_proto_request_t *preq;
670  fr_trunk_request_t *treq = NULL;
671  fr_trunk_enqueue_t rcode;
672 
673  DEBUG_LVL_SET;
674 
675  el = fr_event_list_alloc(ctx, NULL, NULL);
676  fr_event_list_set_time_func(el, test_time);
677 
678  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
679 
680  /*
681  * Our preq is a pointer to the trunk
682  * request so we don't have to manage
683  * a tree of requests and responses.
684  */
685  preq = talloc_zero(NULL, test_proto_request_t);
686 
687  /*
688  * The trunk is active, but there's no
689  * connections.
690  *
691  * We're under the current request limit
692  * so the request should enter the
693  * backlog.
694  */
695  rcode = fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
696  preq->treq = treq;
697  TEST_CHECK(rcode == FR_TRUNK_ENQUEUE_IN_BACKLOG);
698 
699  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_BACKLOG) == 1);
700 
701  /*
702  * Allow the connection to establish
703  */
704  fr_event_corral(el, test_time_base, false);
705  fr_event_service(el);
706 
707  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_BACKLOG) == 0);
708  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 1);
709 
710  /*
711  * Should now be active and have a write event
712  * inserted into the event loop.
713  */
714  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
715 
716  /*
717  * Trunk should be signalled the connection is
718  * writable.
719  *
720  * We should then:
721  * - Pop a request from the pending queue.
722  * - Write the request to the socket pair
723  */
724  fr_event_corral(el, test_time_base, false);
725  fr_event_service(el);
726 
727  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
728 
729  /*
730  * Gives the loopback function a chance
731  * to read the data, and write it back.
732  */
733  fr_event_corral(el, test_time_base, false);
734  fr_event_service(el);
735 
736  /*
737  * Trunk should be signalled the connection is
738  * readable.
739  *
740  * We should then:
741  * - Read the (looped back) response.
742  * - Signal the trunk that the connection is readable.
743  */
744  fr_event_corral(el, test_time_base, false);
745  fr_event_service(el);
746 
747  TEST_CHECK(preq->completed == true);
748  TEST_CHECK(preq->failed == false);
749  TEST_CHECK(preq->cancelled == false);
750  TEST_CHECK(preq->freed == true);
751  talloc_free(preq);
752 
753  talloc_free(trunk);
754  talloc_free(ctx);
755 }
756 
757 /*
758  * Test request cancellations when the connection is in various states
759  */
760 static void test_enqueue_cancellation_points(void)
761 {
762  TALLOC_CTX *ctx = talloc_init_const("test");
763  fr_trunk_t *trunk;
764  fr_event_list_t *el;
765  fr_trunk_conf_t conf = {
766  .start = 1,
767  .min = 1,
768  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
769  };
770  test_proto_request_t *preq;
771  fr_trunk_request_t *treq = NULL;
772 
773  DEBUG_LVL_SET;
774 
775  el = fr_event_list_alloc(ctx, NULL, NULL);
776  fr_event_list_set_time_func(el, test_time);
777 
778  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
779  preq = talloc_zero(NULL, test_proto_request_t);
780  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
781 
782  TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_STATE_BACKLOG");
783  talloc_free(trunk);
784  TEST_CHECK(preq->completed == false);
785  TEST_CHECK(preq->failed == true);
786  TEST_CHECK(preq->cancelled == false);
787  TEST_CHECK(preq->freed == true);
788  talloc_free(preq);
789 
790  TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_STATE_BACKLOG");
791  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
792  preq = talloc_zero(NULL, test_proto_request_t);
793  treq = NULL;
794  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
795  preq->treq = treq;
796  fr_trunk_request_signal_cancel(treq);
797  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
798 
799  TEST_CHECK(preq->completed == false);
800  TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
801  TEST_CHECK(preq->cancelled == false);
802  TEST_CHECK(preq->freed == true);
803  talloc_free(preq);
804  talloc_free(trunk);
805 
806  TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_STATE_PARTIAL");
807  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
808  preq = talloc_zero(NULL, test_proto_request_t);
809  preq->signal_partial = true;
810  treq = NULL;
811  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
812  preq->treq = treq;
813 
814  fr_event_corral(el, test_time_base, false); /* Connect the connection */
815  fr_event_service(el);
816 
817  fr_event_corral(el, test_time_base, false); /* Send the request */
818  fr_event_service(el);
819 
820  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PARTIAL));
821 
822  talloc_free(trunk);
823 
824  TEST_CHECK(preq->completed == false);
825  TEST_CHECK(preq->failed == true);
826  TEST_CHECK(preq->cancelled == true);
827  TEST_CHECK(preq->freed == true);
828  talloc_free(preq);
829 
830  TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_STATE_PARTIAL");
831  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
832  preq = talloc_zero(NULL, test_proto_request_t);
833  preq->signal_partial = true;
834  treq = NULL;
835  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
836  preq->treq = treq;
837 
838  fr_event_corral(el, test_time_base, false); /* Connect the connection */
839  fr_event_service(el);
840 
841  fr_event_corral(el, test_time_base, false); /* Send the request */
842  fr_event_service(el);
843 
844  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PARTIAL) == 1);
845  fr_trunk_request_signal_cancel(treq);
846  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
847 
848  TEST_CHECK(preq->completed == false);
849  TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
850  TEST_CHECK(preq->cancelled == true);
851  TEST_CHECK(preq->freed == true);
852  talloc_free(preq);
853  talloc_free(trunk);
854 
855  TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_STATE_SENT");
856  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
857  preq = talloc_zero(NULL, test_proto_request_t);
858  treq = NULL;
859  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
860  preq->treq = treq;
861 
862  fr_event_corral(el, test_time_base, false); /* Connect the connection */
863  fr_event_service(el);
864 
865  fr_event_corral(el, test_time_base, false); /* Send the request */
866  fr_event_service(el);
867 
868  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
869  talloc_free(trunk);
870 
871  TEST_CHECK(preq->completed == false);
872  TEST_CHECK(preq->failed == true);
873  TEST_CHECK(preq->cancelled == true);
874  TEST_CHECK(preq->freed == true);
875  talloc_free(preq);
876 
877  TEST_CASE("cancellation via signal - FR_TRUNK_REQUEST_STATE_SENT");
878  trunk = test_setup_trunk(ctx, el, &conf, false, NULL);
879  preq = talloc_zero(NULL, test_proto_request_t);
880  treq = NULL;
881  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
882  preq->treq = treq;
883 
884  fr_event_corral(el, test_time_base, false); /* Connect the connection */
885  fr_event_service(el);
886 
887  fr_event_corral(el, test_time_base, false); /* Send the request */
888  fr_event_service(el);
889 
890  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
891  fr_trunk_request_signal_cancel(treq);
892  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
893 
894  TEST_CHECK(preq->completed == false);
895  TEST_CHECK(preq->failed == false); /* Request/rctx not guaranteed after signal, so can't call fail */
896  TEST_CHECK(preq->cancelled == true);
897  TEST_CHECK(preq->freed == true);
898  talloc_free(preq);
899  talloc_free(trunk);
900 
901  TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL");
902  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
903  preq = talloc_zero(NULL, test_proto_request_t);
904  preq->signal_cancel_partial = true;
905  treq = NULL;
906  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
907  preq->treq = treq;
908 
909  fr_event_corral(el, test_time_base, false); /* Connect the connection */
910  fr_event_service(el);
911 
912  fr_event_corral(el, test_time_base, false); /* Send the request */
913  fr_event_service(el);
914 
915  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
916  fr_trunk_request_signal_cancel(treq);
917  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL) == 1);
918 
919  fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
920  fr_event_service(el);
921 
922  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL) == 1);
923 
924  talloc_free(trunk);
925 
926  TEST_CHECK(preq->completed == false);
927  TEST_CHECK(preq->failed == false);
928  TEST_CHECK(preq->cancelled == true);
929  TEST_CHECK(preq->freed == true);
930  talloc_free(preq);
931 
932  TEST_CASE("cancellation via trunk free - FR_TRUNK_REQUEST_STATE_CANCEL_SENT");
933  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
934  preq = talloc_zero(NULL, test_proto_request_t);
935  treq = NULL;
936  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
937  preq->treq = treq;
938 
939  fr_event_corral(el, test_time_base, false); /* Connect the connection */
940  fr_event_service(el);
941 
942  fr_event_corral(el, test_time_base, false); /* Send the request */
943  fr_event_service(el);
944 
945  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
946  fr_trunk_request_signal_cancel(treq);
947  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL) == 1);
948 
949  fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
950  fr_event_service(el);
951 
952  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
953 
954  talloc_free(trunk);
955 
956  TEST_CHECK(preq->completed == false);
957  TEST_CHECK(preq->failed == false);
958  TEST_CHECK(preq->cancelled == true);
959  TEST_CHECK(preq->freed == true);
960  talloc_free(preq);
961 
962  TEST_CASE("trunk free after FR_TRUNK_REQUEST_STATE_CANCEL_COMPLETE");
963  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
964  preq = talloc_zero(NULL, test_proto_request_t);
965  treq = NULL;
966  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
967  preq->treq = treq;
968 
969  fr_event_corral(el, test_time_base, false); /* Connect the connection */
970  fr_event_service(el);
971 
972  fr_event_corral(el, test_time_base, false); /* Send the request */
973  fr_event_service(el);
974 
975  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
976  fr_trunk_request_signal_cancel(treq);
977  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL) == 1);
978 
979  fr_event_corral(el, test_time_base, false); /* Send the cancellation request */
980  fr_event_service(el);
981 
982  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
983 
984  fr_event_corral(el, test_time_base, false); /* Loop the cancel request back round */
985  fr_event_service(el);
986 
987  fr_event_corral(el, test_time_base, false); /* Read the cancel ACK (such that it is) */
988  fr_event_service(el);
989 
990  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
991 
992  talloc_free(trunk);
993 
994  TEST_CHECK(preq->completed == false);
995  TEST_CHECK(preq->failed == false);
996  TEST_CHECK(preq->cancelled == true);
997  TEST_CHECK(preq->freed == true);
998  talloc_free(preq);
999 
1000 
1001  talloc_free(ctx);
1002 }
1003 
1004 /*
1005  * Test PARTIAL -> SENT and CANCEL-PARTIAL -> CANCEL-SENT
1006  */
1007 static void test_partial_to_complete_states(void)
1008 {
1009  TALLOC_CTX *ctx = talloc_init_const("test");
1010  fr_trunk_t *trunk;
1011  fr_event_list_t *el;
1012  fr_trunk_conf_t conf = {
1013  .start = 1,
1014  .min = 1,
1015  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1016  };
1017  test_proto_request_t *preq;
1018  fr_trunk_request_t *treq = NULL;
1019 
1020  DEBUG_LVL_SET;
1021 
1022  el = fr_event_list_alloc(ctx, NULL, NULL);
1023  fr_event_list_set_time_func(el, test_time);
1024 
1025  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1026  preq = talloc_zero(NULL, test_proto_request_t);
1027  preq->signal_partial = true;
1028  preq->signal_cancel_partial = true;
1029 
1030  TEST_CASE("FR_TRUNK_REQUEST_STATE_PARTIAL -> FR_TRUNK_REQUEST_STATE_SENT");
1031 
1032  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1033  preq->treq = treq;
1034 
1035  fr_event_corral(el, test_time_base, false); /* Connect the connection */
1036  fr_event_service(el);
1037 
1038  fr_event_corral(el, test_time_base, false); /* Send the request */
1039  fr_event_service(el);
1040 
1041  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PARTIAL) == 1);
1042 
1043  fr_event_corral(el, test_time_base, false); /* Complete the partial request */
1044  fr_event_service(el);
1045 
1046  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 1);
1047 
1048  fr_trunk_request_signal_cancel(treq);
1049  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL) == 1);
1050 
1051  TEST_CASE("FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL -> FR_TRUNK_REQUEST_STATE_CANCEL_SENT");
1052 
1053  fr_event_corral(el, test_time_base, false); /* Send partial cancel request */
1054  fr_event_service(el);
1055 
1056  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL) == 1);
1057 
1058  fr_event_corral(el, test_time_base, false); /* Complete the partial cancellation */
1059  fr_event_service(el);
1060 
1061  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_SENT) == 1);
1062 
1063  fr_event_corral(el, test_time_base, false); /* Loop the cancellation request back */
1064  fr_event_service(el);
1065 
1066  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
1067 
1068  talloc_free(trunk);
1069 
1070  TEST_CHECK(preq->completed == false);
1071  TEST_CHECK(preq->failed == false);
1072  TEST_CHECK(preq->cancelled == true);
1073  TEST_CHECK(preq->freed == true);
1074  talloc_free(preq);
1075 
1076  talloc_free(ctx);
1077 }
1078 
1079 /*
1080  * Test calling reconnect with requests in each different state
1081  */
1082 static void test_requeue_on_reconnect(void)
1083 {
1084  TALLOC_CTX *ctx = talloc_init_const("test");
1085  fr_trunk_t *trunk;
1086  fr_event_list_t *el;
1087  fr_trunk_conf_t conf = {
1088  .start = 2,
1089  .min = 2,
1090  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5),
1091  .conn_conf = &(fr_connection_conf_t){
1092  .reconnection_delay = fr_time_delta_from_nsec(NSEC / 10)
1093  },
1094  .backlog_on_failed_conn = true
1095  };
1096  test_proto_request_t *preq;
1097  fr_trunk_request_t *treq = NULL;
1098  fr_trunk_connection_t *tconn;
1099 
1100  DEBUG_LVL_SET;
1101  fr_talloc_fault_setup();
1102 
1103  el = fr_event_list_alloc(ctx, NULL, NULL);
1104  fr_event_list_set_time_func(el, test_time);
1105 
1106  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1107  preq = talloc_zero(ctx, test_proto_request_t);
1108  preq->signal_partial = true;
1109  preq->signal_cancel_partial = true;
1110 
1111  fr_event_corral(el, test_time_base, false); /* Connect the connection(s) */
1112  fr_event_service(el);
1113 
1114  TEST_CASE("dequeue on reconnect - FR_TRUNK_REQUEST_STATE_PENDING");
1115 
1116  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE), 2);
1117 
1118  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1119  preq->treq = treq;
1120 
1121  tconn = treq->pub.tconn; /* Store the conn the request was assigned to */
1122  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1123 
1124  fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED);
1125 
1126  /*
1127  * Should be reassigned to the other connection
1128  */
1129  TEST_CHECK(tconn != treq->pub.tconn);
1130  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1131 
1132  /*
1133  * Should be reassigned to the backlog
1134  */
1135  fr_trunk_connection_signal_reconnect(treq->pub.tconn, FR_CONNECTION_FAILED);
1136  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_BACKLOG), 1);
1137  TEST_CHECK(!treq->pub.tconn);
1138 
1139  TEST_CASE("cancel on reconnect - FR_TRUNK_REQUEST_STATE_PARTIAL");
1140 
1141  /*
1142  * Allow the connections to reconnect
1143  */
1144  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1145  fr_event_corral(el, test_time_base, false);
1146  fr_event_service(el); /* run management function */
1147  fr_event_corral(el, test_time_base, false);
1148  fr_event_service(el); /* service any I/O callbacks */
1149 
1150  /*
1151  * Request should now be assigned back to one of the reconnected
1152  * connections.
1153  */
1154  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1155  TEST_CHECK(treq->pub.tconn != NULL);
1156 
1157  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1158  fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1159  fr_event_service(el);
1160 
1161  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PARTIAL), 1);
1162 
1163  /*
1164  * Reconnect the connection.
1165  *
1166  * preq should pass through the cancel function,
1167  * then be re-assigned.
1168  */
1169  tconn = treq->pub.tconn;
1170  fr_trunk_connection_signal_reconnect(treq->pub.tconn, FR_CONNECTION_FAILED);
1171 
1172  TEST_CHECK(preq->completed == false);
1173  TEST_CHECK(preq->failed == false);
1174  TEST_CHECK(preq->cancelled == true);
1175  TEST_CHECK(preq->freed == false);
1176 
1177  preq->cancelled = false; /* Reset */
1178 
1179  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1180  TEST_CHECK(tconn != treq->pub.tconn); /* Ensure it moved */
1181 
1182  TEST_CASE("cancel on reconnect - FR_TRUNK_REQUEST_STATE_SENT");
1183 
1184  /*
1185  * Sent the request (fully)
1186  */
1187  fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1188  fr_event_service(el);
1189 
1190  /*
1191  * The above indeed appears to send the request partially;
1192  * this appears to be required to send it fully, judging by
1193  * the following check, which fails without it.
1194  */
1195  fr_event_corral(el, test_time_base, false); /* Send the request (partially) */
1196  fr_event_service(el);
1197  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT), 1);
1198 
1199  tconn = treq->pub.tconn;
1200  fr_trunk_connection_signal_reconnect(treq->pub.tconn, FR_CONNECTION_FAILED);
1201 
1202  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1203 
1204  /*
1205  * Allow the connections to reconnect
1206  * and send the request.
1207  */
1208  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1209  fr_event_corral(el, test_time_base, false);
1210  fr_event_service(el);
1211  TEST_CHECK(tconn != treq->pub.tconn); /* Ensure it moved */
1212 
1213  TEST_CHECK(preq->completed == false);
1214  TEST_CHECK(preq->failed == false);
1215  TEST_CHECK(preq->cancelled == true);
1216  TEST_CHECK(preq->freed == false);
1217 
1218  preq->cancelled = false; /* Reset */
1219 
1220  TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_STATE_CANCEL");
1221 
1222  /*
1223  * Signal the request should be cancelled
1224  */
1225  fr_trunk_request_signal_cancel(treq);
1226  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL), 1);
1227 
1228  /*
1229  * Requests in the cancel state, are
1230  * freed instead of being moved between
1231  * connections.
1232  */
1233  fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED); /* treq->pub.tconn, now invalid due to cancel */
1234 
1235  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1236  fr_event_corral(el, test_time_base, false);
1237  fr_event_service(el);
1238 
1239  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1240  fr_event_corral(el, test_time_base, false);
1241  fr_event_service(el);
1242 
1243  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1244  fr_event_corral(el, test_time_base, false);
1245  fr_event_service(el);
1246 
1247  TEST_CHECK(preq->completed == false);
1248  TEST_CHECK(preq->failed == false);
1249  TEST_CHECK(preq->cancelled == true);
1250  TEST_CHECK(preq->freed == true);
1251 
1252  /*
1253  * Allow the connection we just reconnected
1254  * to open so it doesn't interfere with
1255  * the next test.
1256  */
1257  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1258  fr_event_corral(el, test_time_base, false);
1259  fr_event_service(el);
1260 
1261  TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL");
1262 
1263  /*
1264  * Queue up a new request, and get it to the cancel-partial state.
1265  */
1266  preq = talloc_zero(ctx, test_proto_request_t);
1267  preq->signal_cancel_partial = true;
1268  treq = NULL;
1269  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1270  preq->treq = treq;
1271 
1272  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1273 
1274  /*
1275  * Sent the request (fully)
1276  */
1277  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1278  fr_event_corral(el, test_time_base, false); /* Send the request (fully) */
1279  fr_event_service(el);
1280 
1281  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT), 1);
1282  fr_trunk_request_signal_cancel(treq); /* Cancel the request */
1283 
1284  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL), 1);
1285 
1286  /*
1287  * Transition to cancel partial
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_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL), 1);
1294 
1295  /*
1296  * Trigger a reconnection
1297  */
1298  fr_trunk_connection_signal_reconnect(treq->pub.tconn, FR_CONNECTION_FAILED);
1299 
1300  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1301  fr_event_corral(el, test_time_base, false);
1302  fr_event_service(el);
1303 
1304  TEST_CHECK(preq->completed == false);
1305  TEST_CHECK(preq->failed == false);
1306  TEST_CHECK(preq->cancelled == true);
1307  TEST_CHECK(preq->freed == true);
1308 
1309  /*
1310  * Allow the connection we just reconnected
1311  * top open so it doesn't interfere with
1312  * the next test.
1313  */
1314  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1315  fr_event_corral(el, test_time_base, false);
1316  fr_event_service(el);
1317 
1318  TEST_CASE("free on reconnect - FR_TRUNK_REQUEST_STATE_CANCEL_SENT");
1319 
1320  /*
1321  * Queue up a new request, and get it to the cancel-sent state.
1322  */
1323  preq = talloc_zero(NULL, test_proto_request_t);
1324  treq = NULL;
1325  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1326  preq->treq = treq;
1327 
1328  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 1);
1329 
1330  /*
1331  * Sent the request (fully)
1332  */
1333  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1334  fr_event_corral(el, test_time_base, false); /* Send the request (fully) */
1335  fr_event_service(el);
1336 
1337  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT), 1);
1338  fr_trunk_request_signal_cancel(treq); /* Cancel the request */
1339 
1340  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL), 1);
1341 
1342  /*
1343  * Transition to cancel
1344  */
1345  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1346  fr_event_corral(el, test_time_base, false);
1347  fr_event_service(el);
1348 
1349  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_CANCEL_SENT), 1);
1350 
1351  /*
1352  * Trigger a reconnection
1353  */
1354  fr_trunk_connection_signal_reconnect(treq->pub.tconn, FR_CONNECTION_FAILED);
1355 
1356  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1357  fr_event_corral(el, test_time_base, false);
1358  fr_event_service(el);
1359 
1360  TEST_CHECK(preq->completed == false);
1361  TEST_CHECK(preq->failed == false);
1362  TEST_CHECK(preq->cancelled == true);
1363  TEST_CHECK(preq->freed == true);
1364 
1365  talloc_free(preq);
1366 
1367  talloc_free(ctx);
1368 }
1369 
1370 static void test_connection_start_on_enqueue(void)
1371 {
1372  TALLOC_CTX *ctx = talloc_init_const("test");
1373  fr_trunk_t *trunk;
1374  fr_event_list_t *el;
1375  fr_trunk_conf_t conf = {
1376  .start = 0,
1377  .min = 0, /* No connections on start */
1378  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1379  };
1380  test_proto_request_t *preq;
1381  fr_trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1382 
1383  DEBUG_LVL_SET;
1384 
1385  el = fr_event_list_alloc(ctx, NULL, NULL);
1386  fr_event_list_set_time_func(el, test_time);
1387 
1388  /* Need to provide a timer starting value above zero */
1389  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1390 
1391  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1392  preq = talloc_zero(NULL, test_proto_request_t);
1393 
1394  TEST_CASE("C0 - Enqueue should spawn");
1395  fr_trunk_request_enqueue(&treq_a, trunk, NULL, preq, NULL);
1396 
1397  /*
1398  * This causes the event associated with the request left on
1399  * the backlog queue to be handled, which (along with the other
1400  * corral; service sequence, makes the checks all pass.
1401  */
1402  fr_event_corral(el, test_time_base, false);
1403  fr_event_service(el);
1404 
1405  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1406 
1407  TEST_CASE("C1 connecting, !max_req_per_conn - Enqueue MUST NOT spawn");
1408  fr_trunk_request_enqueue(&treq_b, trunk, NULL, preq, NULL);
1409 
1410  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1411 
1412  /*
1413  * Allow the connections to open
1414  */
1415  fr_event_corral(el, test_time_base, false);
1416  fr_event_service(el);
1417 
1418  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
1419 
1420  TEST_CASE("C1 active, !max_req_per_conn - Enqueue MUST NOT spawn");
1421  fr_trunk_request_enqueue(&treq_c, trunk, NULL, preq, NULL);
1422 
1423  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
1424  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 3);
1425 
1426  talloc_free(ctx);
1427  talloc_free(preq);
1428 }
1429 
1430 static void test_connection_rebalance_requests(void)
1431 {
1432  TALLOC_CTX *ctx = talloc_init_const("test");
1433  fr_trunk_t *trunk;
1434  fr_event_list_t *el;
1435  fr_trunk_conf_t conf = {
1436  .start = 2,
1437  .min = 2, /* No connections on start */
1438  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1439  };
1440  test_proto_request_t *preq;
1441  fr_trunk_connection_t *tconn;
1442  fr_trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1443 
1444  DEBUG_LVL_SET;
1445 
1446  el = fr_event_list_alloc(ctx, NULL, NULL);
1447  fr_event_list_set_time_func(el, test_time);
1448 
1449  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1450  preq = talloc_zero(NULL, test_proto_request_t);
1451  printf("Rebalance %p\n", preq);
1452 
1453  /*
1454  * Allow the connections to open
1455  */
1456  fr_event_corral(el, test_time_base, false);
1457  fr_event_service(el);
1458 
1459  /*
1460  * Mark one of the connections as full, and
1461  * enqueue three requests on the other.
1462  */
1463  tconn = fr_minmax_heap_min_peek(trunk->active);
1464 
1465  TEST_CASE("C2 connected, R0 - Signal inactive");
1466  fr_trunk_connection_signal_inactive(tconn);
1467 
1468 
1469  fr_trunk_request_enqueue(&treq_a, trunk, NULL, preq, NULL);
1470  fr_trunk_request_enqueue(&treq_b, trunk, NULL, preq, NULL);
1471  fr_trunk_request_enqueue(&treq_c, trunk, NULL, preq, NULL);
1472 
1473  TEST_CASE("C1 connected, C2 inactive, R3 - Enqueued");
1474  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 3);
1475  TEST_CHECK(fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_STATE_ALL) == 0);
1476 
1477  /*
1478  * Now mark the previous connection as
1479  * active. It should receive at least
1480  * one of the requests.
1481  */
1482  TEST_CASE("C2 active, R3 - Signal active, should balance");
1483  fr_trunk_connection_signal_active(tconn);
1484 
1485  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 3);
1486  TEST_CHECK(fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_STATE_ALL) >= 1);
1487 
1488  talloc_free(ctx);
1489  talloc_free(preq);
1490 }
1491 
1492 #define ALLOC_REQ(_id) \
1493 do { \
1494  treq_##_id = fr_trunk_request_alloc(trunk, NULL); \
1495  preq_##_id = talloc_zero(ctx, test_proto_request_t); \
1496  preq_##_id->treq = treq_##_id; \
1497  preq_##_id->priority = next_prio++; \
1498 } while (0)
1499 
1500 static void test_connection_levels_max(void)
1501 {
1502  TALLOC_CTX *ctx = talloc_init_const("test");
1503  fr_trunk_t *trunk;
1504  fr_event_list_t *el;
1505  fr_trunk_conf_t conf = {
1506  .start = 0, /* No connections on start */
1507  .min = 0,
1508  .max = 2,
1509  .max_req_per_conn = 2,
1510  .target_req_per_conn = 2, /* One request per connection */
1511  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1512  };
1513  test_proto_request_t *preq_a, *preq_b, *preq_c, *preq_d, *preq_e;
1514  fr_trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL, *treq_d = NULL, *treq_e = NULL;
1515  int next_prio = 0;
1516 
1517  DEBUG_LVL_SET;
1518 
1519  el = fr_event_list_alloc(ctx, NULL, NULL);
1520  fr_event_list_set_time_func(el, test_time);
1521 
1522  /* Need to provide a timer starting value above zero */
1523  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1524 
1525  trunk = test_setup_trunk(ctx, el, &conf, true, NULL);
1526  FR_TRUNK_VERIFY(trunk);
1527 
1528  /*
1529  * Queuing a request should start a connection.
1530  */
1531  TEST_CASE("C0, R1 - Enqueue should spawn");
1532  ALLOC_REQ(a);
1533  TEST_CHECK(fr_trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1534  FR_TRUNK_VERIFY(trunk);
1535 
1536  /*
1537  * Like test_connection_start_on_enqueue(), you have to process the backlog
1538  * to start the chain of events.
1539  */
1540  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1541  fr_event_corral(el, test_time_base, false);
1542  fr_event_service(el);
1543 
1544  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING), 1);
1545  FR_TRUNK_VERIFY(trunk);
1546 
1547  /*
1548  * Queuing another request should *NOT* start another connection
1549  */
1550  TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1551  ALLOC_REQ(b);
1552  TEST_CHECK(fr_trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1553  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING), 1);
1554  FR_TRUNK_VERIFY(trunk);
1555 
1556  TEST_CASE("C1 connecting, R3 - MUST NOT spawn");
1557  ALLOC_REQ(c);
1558  TEST_CHECK(fr_trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1559  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING), 1);
1560  FR_TRUNK_VERIFY(trunk);
1561 
1562  TEST_CASE("C1 connecting, R4 - MUST NOT spawn");
1563  ALLOC_REQ(d);
1564  TEST_CHECK(fr_trunk_request_enqueue(&treq_d, trunk, NULL, preq_d, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1565  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING), 1);
1566  FR_TRUNK_VERIFY(trunk);
1567 
1568  TEST_CASE("C1 connecting, R5 - MUST NOT spawn, NO CAPACITY");
1569  ALLOC_REQ(e);
1570  TEST_CHECK(fr_trunk_request_enqueue(&treq_e, trunk, NULL, preq_e, NULL) == FR_TRUNK_ENQUEUE_NO_CAPACITY);
1571  TEST_CHECK_LEN(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING), 1);
1572  FR_TRUNK_VERIFY(trunk);
1573 
1574  /*
1575  * Allowing connection to open
1576  */
1577  fr_event_corral(el, test_time_base, false);
1578  fr_event_service(el);
1579 
1580  TEST_CASE("C1 active, R4 - Check pending 2");
1581  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 2);
1582  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_BACKLOG), 2);
1583  FR_TRUNK_VERIFY(trunk);
1584 
1585  /*
1586  * Sending requests
1587  */
1588  fr_event_corral(el, test_time_base, false);
1589  fr_event_service(el);
1590 
1591  TEST_CASE("C1 active, R4 - Check sent 2");
1592  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT), 2);
1593  FR_TRUNK_VERIFY(trunk);
1594 
1595  /*
1596  * Looping I/O
1597  */
1598  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1599  fr_event_corral(el, test_time_base, false);
1600  fr_event_service(el);
1601 
1602  /*
1603  * Receiving responses
1604  */
1605  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(2));
1606  fr_event_corral(el, test_time_base, false);
1607  fr_event_service(el);
1608 
1609  TEST_CHECK(preq_a->completed == true);
1610  TEST_CHECK(preq_a->failed == false);
1611  TEST_CHECK(preq_a->cancelled == false);
1612  TEST_CHECK(preq_a->freed == true);
1613 
1614  TEST_CHECK(preq_b->completed == true);
1615  TEST_CHECK(preq_b->failed == false);
1616  TEST_CHECK(preq_b->cancelled == false);
1617  TEST_CHECK(preq_b->freed == true);
1618 
1619  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING), 2);
1620  TEST_CHECK_LEN(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_BACKLOG), 0);
1621  FR_TRUNK_VERIFY(trunk);
1622 
1623  TEST_CASE("C1 active, R0 - Check complete 2, pending 0");
1624 
1625  /*
1626  * Sending requests
1627  */
1628  fr_event_corral(el, test_time_base, false);
1629  fr_event_service(el);
1630 
1631  /*
1632  * Looping I/O
1633  */
1634  fr_event_corral(el, test_time_base, false);
1635  fr_event_service(el);
1636 
1637  /*
1638  * Receiving responses
1639  */
1640  fr_event_corral(el, test_time_base, false);
1641  fr_event_service(el);
1642 
1643  TEST_CHECK(preq_c->completed == true);
1644  TEST_CHECK(preq_c->failed == false);
1645  TEST_CHECK(preq_c->cancelled == false);
1646  TEST_CHECK(preq_c->freed == true);
1647 
1648  TEST_CHECK(preq_d->completed == true);
1649  TEST_CHECK(preq_d->failed == false);
1650  TEST_CHECK(preq_d->cancelled == false);
1651  TEST_CHECK(preq_d->freed == true);
1652 
1653  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
1654  FR_TRUNK_VERIFY(trunk);
1655 
1656  talloc_free(trunk);
1657  talloc_free(ctx);
1658 }
1659 
1660 static void test_connection_levels_alternating_edges(void)
1661 {
1662  TALLOC_CTX *ctx = talloc_init_const("test");
1663  fr_trunk_t *trunk;
1664  fr_event_list_t *el;
1665  fr_trunk_conf_t conf = {
1666  .start = 0, /* No connections on start */
1667  .min = 0,
1668  .max = 0,
1669  .max_req_per_conn = 0,
1670  .target_req_per_conn = 2, /* One request per connection */
1671  .manage_interval = fr_time_delta_from_nsec(NSEC / 10)
1672  };
1673 
1674  test_proto_request_t *preq_a, *preq_b, *preq_c;
1675  fr_trunk_request_t *treq_a = NULL, *treq_b = NULL, *treq_c = NULL;
1676  test_proto_stats_t stats;
1677  int next_prio = 0;
1678 
1679  DEBUG_LVL_SET;
1680 
1681  el = fr_event_list_alloc(ctx, NULL, NULL);
1682  fr_event_list_set_time_func(el, test_time);
1683 
1684  /* Need to provide a timer starting value above zero */
1685  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1686 
1687  memset(&stats, 0, sizeof(stats));
1688  trunk = test_setup_trunk(ctx, el, &conf, true, &stats);
1689 
1690  /*
1691  * Queuing a request should start a connection.
1692  */
1693  TEST_CASE("C0, R1 - Enqueue should spawn");
1694  ALLOC_REQ(a);
1695  TEST_CHECK(fr_trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1696 
1697  /*
1698  * Processing the event associated with the backlog creates
1699  * the connection in connecting state..
1700  */
1701  fr_event_corral(el, test_time_base, false);
1702  fr_event_service(el);
1703 
1704  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1705 
1706  TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1707  ALLOC_REQ(b);
1708  TEST_CHECK(fr_trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1709  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1710  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1711 
1712  /*
1713  * Open connection
1714  */
1715  fr_event_corral(el, test_time_base, false);
1716  fr_event_service(el);
1717 
1718  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 2);
1719 
1720  TEST_CASE("C1 connected, R3 - should spawn");
1721  ALLOC_REQ(c);
1722  TEST_CHECK(fr_trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == FR_TRUNK_ENQUEUE_OK);
1723  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1724 
1725  fr_event_corral(el, test_time_base, false);
1726  fr_event_service(el);
1727 
1728  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 3);
1729  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
1730  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1731 
1732  /*
1733  * Complete requests
1734  */
1735  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1736 
1737  fr_event_corral(el, test_time_base, false);
1738  fr_event_service(el);
1739 
1740  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1741 
1742  TEST_CASE("C1 connected, C2 connecting, R2 - MUST NOT spawn");
1743  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 3);
1744  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 2);
1745 
1746  /*
1747  * Finish the last request, should close one connection
1748  */
1749  fr_event_corral(el, test_time_base, false);
1750  fr_event_service(el);
1751 
1752  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1753 
1754  TEST_CASE("C1 connected, R0");
1755  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_ALL) == 0);
1756  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
1757 
1758  /*
1759  * Requests now done, should close another connection
1760  */
1761  fr_event_corral(el, test_time_base, false);
1762  fr_event_service(el);
1763 
1764  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1765 
1766  TEST_CASE("C0, R0");
1767  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 0);
1768 
1769  TEST_CHECK(stats.completed == 3);
1770  TEST_CHECK(stats.failed == 0);
1771  TEST_CHECK(stats.cancelled == 0);
1772  TEST_CHECK(stats.freed == 3);
1773 
1774  /*
1775  * Queuing a request should start a connection.
1776  */
1777  TEST_CASE("C0, R1 - Enqueue should spawn");
1778  ALLOC_REQ(a);
1779  TEST_CHECK(fr_trunk_request_enqueue(&treq_a, trunk, NULL, preq_a, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1780 
1781  /*
1782  * ...once the event associated with the backlogged request is handled.
1783  */
1784  fr_event_corral(el, test_time_base, false);
1785  fr_event_service(el);
1786 
1787  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1788 
1789  TEST_CASE("C1 connecting, R2 - MUST NOT spawn");
1790  ALLOC_REQ(b);
1791  TEST_CHECK(fr_trunk_request_enqueue(&treq_b, trunk, NULL, preq_b, NULL) == FR_TRUNK_ENQUEUE_IN_BACKLOG);
1792  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1793  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1794 
1795  /*
1796  * Open connection
1797  */
1798  fr_event_corral(el, test_time_base, false);
1799  fr_event_service(el);
1800 
1801  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_PENDING) == 2);
1802 
1803  TEST_CASE("C1 connected, R3 - should spawn");
1804  ALLOC_REQ(c);
1805  TEST_CHECK(fr_trunk_request_enqueue(&treq_c, trunk, NULL, preq_c, NULL) == FR_TRUNK_ENQUEUE_OK);
1806  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_sec(1));
1807 
1808  fr_event_corral(el, test_time_base, false);
1809  fr_event_service(el);
1810 
1811  TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_STATE_SENT) == 3);
1812  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_ACTIVE) == 1);
1813  TEST_CHECK(fr_trunk_connection_count_by_state(trunk, FR_TRUNK_CONN_CONNECTING) == 1);
1814 
1815  talloc_free(trunk);
1816  talloc_free(ctx);
1817 }
1818 
1819 #undef fr_time /* Need to the real time */
1820 static void test_enqueue_and_io_speed(void)
1821 {
1822  TALLOC_CTX *ctx = talloc_init_const("test");
1823  fr_trunk_t *trunk;
1824  fr_event_list_t *el;
1825  int events;
1826  fr_trunk_conf_t conf = {
1827  .start = 1,
1828  .min = 1,
1829  .max = 0,
1830  .max_req_per_conn = 0,
1831  .target_req_per_conn = 0, /* One request per connection */
1832  .req_pool_headers = 1,
1833  .req_pool_size = sizeof(test_proto_request_t),
1834  .manage_interval = fr_time_delta_from_nsec(NSEC * 0.5)
1835  };
1836  size_t i = 0, requests = 100000;
1837  fr_time_t enqueue_start, enqueue_stop, io_start, io_stop;
1838  fr_time_delta_t enqueue_time, io_time, total_time;
1839  fr_trunk_request_t **treq_array;
1840  test_proto_request_t **preq_array;
1841  test_proto_stats_t stats;
1842 
1843  DEBUG_LVL_SET;
1844 
1845  el = fr_event_list_alloc(ctx, NULL, NULL);
1846  fr_event_list_set_time_func(el, test_time);
1847 
1848  /* Need to provide a timer starting value above zero */
1849  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.5));
1850 
1851  memset(&stats, 0, sizeof(stats));
1852  trunk = test_setup_trunk(ctx, el, &conf, true, &stats);
1853 
1854  /*
1855  * Open the connections
1856  */
1857  fr_event_corral(el, test_time_base, false);
1858  fr_event_service(el);
1859 
1860  /*
1861  * Build up a cache of requests
1862  * This prevents all mallocs on request enqueue.
1863  *
1864  * When the server's running, this does represent
1865  * close to what we'd have as a steady state.
1866  */
1867  MEM(treq_array = talloc_array(ctx, fr_trunk_request_t *, requests));
1868  for (i = 0; i < requests; i++) treq_array[i] = fr_trunk_request_alloc(trunk, NULL);
1869  for (i = 0; i < requests; i++) fr_trunk_request_free(&treq_array[i]);
1870 
1871  MEM(preq_array = talloc_array(ctx, test_proto_request_t *, requests));
1872 
1873  DEBUG_LVL_SET;
1874 
1875  TEST_CASE("Enqueue requests");
1876  enqueue_start = fr_time();
1877 // ProfilerStart(getenv("FR_PROFILE"));
1878  for (i = 0; i < requests; i++) {
1879  fr_trunk_request_t *treq;
1880  test_proto_request_t *preq = NULL;
1881 
1882  treq = fr_trunk_request_alloc(trunk, NULL);
1883  preq = talloc_zero(treq, test_proto_request_t);
1884  preq->treq = treq;
1885  fr_trunk_request_enqueue(&treq, trunk, NULL, preq, NULL);
1886  }
1887  enqueue_stop = fr_time();
1888  enqueue_time = fr_time_sub(enqueue_stop, enqueue_start);
1889  if (acutest_verbose_level_ >= 1) {
1890  INFO("Enqueue time %pV (%u rps) (%"PRIu64"/%"PRIu64")",
1891  fr_box_time_delta(enqueue_time),
1892  (uint32_t)(requests / ((float)(fr_time_delta_unwrap(enqueue_time)) / NSEC)),
1893  trunk->pub.req_alloc_new, trunk->pub.req_alloc_reused);
1894  }
1895 
1896  TEST_CASE("Perform I/O operations");
1897  io_start = fr_time();
1898  while (true) {
1899  events = fr_event_corral(el, test_time_base, false);
1900  if (!events) break;
1901  fr_event_service(el);
1902  test_time_base = fr_time_add_time_delta(test_time_base, fr_time_delta_from_nsec(NSEC * 0.25));
1903  }
1904  io_stop = fr_time();
1905  io_time = fr_time_sub(io_stop, io_start);
1906 
1907  if (acutest_verbose_level_ >= 1) {
1908  INFO("I/O time %pV (%u rps)",
1909  fr_box_time_delta(io_time),
1910  (uint32_t)(requests / ((float)(fr_time_delta_unwrap(io_time)) / NSEC)));
1911  }
1912 
1913  if (acutest_verbose_level_ >= 1) {
1914  total_time = fr_time_sub(io_stop, enqueue_start);
1915  INFO("Total time %pV (%u rps)",
1916  fr_box_time_delta(total_time),
1917  (uint32_t)(requests / ((float)(fr_time_delta_unwrap(total_time)) / NSEC)));
1918  }
1919 
1920  TEST_CHECK_LEN(stats.completed, requests);
1921  TEST_CHECK_LEN(stats.failed, 0);
1922  TEST_CHECK_LEN(stats.cancelled, 0);
1923  TEST_CHECK_LEN(stats.freed, requests);
1924 
1925 // ProfilerStop();
1926 
1927  talloc_free(ctx);
1928 }
1929 
1930 /*
1931  * Connection spawning
1932  */
1933 TEST_LIST = {
1934  /*
1935  * Basic tests
1936  */
1937  { "Basic - Alloc then free", test_socket_pair_alloc_then_free },
1938  { "Basic - Alloc then reconnect then free", test_socket_pair_alloc_then_reconnect_then_free },
1939 
1940  /*
1941  * Connection timeout
1942  */
1943  { "Timeouts - Connection", test_socket_pair_alloc_then_connect_timeout },
1944  { "Timeouts - Reconnect delay", test_socket_pair_alloc_then_reconnect_check_delay },
1945 
1946  /*
1947  * Basic enqueue/dequeue
1948  */
1949  { "Enqueue - Basic", test_enqueue_basic },
1950  { "Enqueue - Cancellation points", test_enqueue_cancellation_points },
1951  { "Enqueue - Partial state transitions", test_partial_to_complete_states },
1952  { "Requeue - On reconnect", test_requeue_on_reconnect },
1953 
1954  /*
1955  * Rebalance
1956  */
1957  { "Rebalance - Connection rebalance", test_connection_rebalance_requests },
1958 
1959  /*
1960  * Connection spawning tests
1961  */
1962  { "Spawn - Test connection start on enqueue", test_connection_start_on_enqueue },
1963  { "Spawn - Connection levels max", test_connection_levels_max },
1964  { "Spawn - Connection levels alternating edges",test_connection_levels_alternating_edges },
1965 
1966  /*
1967  * Performance tests
1968  */
1969  { "Speed Test - Enqueue, and I/O", test_enqueue_and_io_speed },
1970  { NULL }
1971 };
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
fr_connection_state_t
fr_connection_state_t
Definition: connection.h:45
FR_CONNECTION_STATE_CONNECTING
@ FR_CONNECTION_STATE_CONNECTING
Waiting for connection to establish.
Definition: connection.h:50
FR_CONNECTION_STATE_CONNECTED
@ FR_CONNECTION_STATE_CONNECTED
File descriptor is open (ready for writing).
Definition: connection.h:52
FR_CONNECTION_FAILED
@ FR_CONNECTION_FAILED
Connection is being reconnected because it failed.
Definition: connection.h:84
fr_connection_conf_t
Definition: connection.h:91
fr_connection_funcs_t
Holds a complete set of functions for a connection.
Definition: connection.h:186
fr_talloc_fault_setup
void fr_talloc_fault_setup(void)
Register talloc fault handlers.
Definition: debug.c:1196
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:2542
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:2892
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:2964
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:2407
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:606
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:1253
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:284
fr_pointer_cmp
int8_t fr_pointer_cmp(void const *a, void const *b)
Compares two pointers.
Definition: misc.c:408
stats
static rc_stats_t stats
Definition: radclient-ng.c:72
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
fr_connection_get_num_timed_out
uint64_t fr_connection_get_num_timed_out(fr_connection_t const *conn)
Return the number of times this connection has timed out whilst connecting.
Definition: connection.c:612
fr_connection_get_num_reconnected
uint64_t fr_connection_get_num_reconnected(fr_connection_t const *conn)
Return the number of times we've attempted to establish or re-establish this connection.
Definition: connection.c:600
fr_connection_signal_on_fd
int fr_connection_signal_on_fd(fr_connection_t *conn, int fd)
Setup the connection to change states to connected or failed based on I/O events.
Definition: connection.c:1400
fr_connection_alloc
fr_connection_t * fr_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, fr_connection_funcs_t const *funcs, fr_connection_conf_t const *conf, char const *log_prefix, void const *uctx)
Allocate a new connection.
Definition: connection.c:1507
fr_connection_signal_reconnect
void fr_connection_signal_reconnect(fr_connection_t *conn, fr_connection_reason_t reason)
Asynchronously signal the connection should be reconnected.
Definition: connection.c:1166
fr_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:175
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:588
NSEC
#define NSEC
Definition: time.h:377
fr_time_delta_from_nsec
static fr_time_delta_t fr_time_delta_from_nsec(int64_t nsec)
Definition: time.h:561
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.
fr_trunk_request_free
void fr_trunk_request_free(fr_trunk_request_t **treq_to_free)
If the trunk request is freed then update the target requests.
Definition: trunk.c:2217
fr_trunk_request_signal_cancel_sent
void fr_trunk_request_signal_cancel_sent(fr_trunk_request_t *treq)
Signal that a remote server has been notified of the cancellation.
Definition: trunk.c:2155
fr_trunk_s::connecting
fr_dlist_head_t connecting
Connections which are not yet in the open state.
Definition: trunk.c:216
fr_trunk_s::pub
struct fr_trunk_pub_s pub
Public fields in the trunk connection.
Definition: trunk.c:190
fr_trunk_reconnect
void fr_trunk_reconnect(fr_trunk_t *trunk, int states, fr_connection_reason_t reason)
Force the trunk to re-establish its connections.
Definition: trunk.c:4557
fr_trunk_connection_signal_active
void fr_trunk_connection_signal_active(fr_trunk_connection_t *tconn)
Signal a trunk connection is no longer full.
Definition: trunk.c:3833
fr_trunk_request_signal_cancel
void fr_trunk_request_signal_cancel(fr_trunk_request_t *treq)
Cancel a trunk request.
Definition: trunk.c:2047
fr_trunk_connection_s::pub
struct fr_trunk_connection_pub_s pub
Public fields in the trunk connection.
Definition: trunk.c:128
fr_trunk_request_alloc
fr_trunk_request_t * fr_trunk_request_alloc(fr_trunk_t *trunk, request_t *request)
(Pre-)Allocate a new trunk request
Definition: trunk.c:2369
fr_trunk_s::active
fr_minmax_heap_t * active
Connections which can service requests.
Definition: trunk.c:218
fr_trunk_connection_pop_cancellation
int fr_trunk_connection_pop_cancellation(fr_trunk_request_t **treq_out, fr_trunk_connection_t *tconn)
Pop a cancellation request off a connection's cancellation queue.
Definition: trunk.c:3708
fr_trunk_request_signal_sent
void fr_trunk_request_signal_sent(fr_trunk_request_t *treq)
Signal that the request was written to a connection successfully.
Definition: trunk.c:1973
fr_trunk_request_enqueue
fr_trunk_enqueue_t fr_trunk_request_enqueue(fr_trunk_request_t **treq_out, fr_trunk_t *trunk, request_t *request, void *preq, void *rctx)
Enqueue a request that needs data written to the trunk.
Definition: trunk.c:2481
fr_trunk_request_count_by_connection
uint32_t fr_trunk_request_count_by_connection(fr_trunk_connection_t const *tconn, int req_state)
Return the count number of requests associated with a trunk connection.
Definition: trunk.c:2767
fr_trunk_connection_count_by_state
uint16_t fr_trunk_connection_count_by_state(fr_trunk_t *trunk, int conn_state)
Return the count number of connections in the specified states.
Definition: trunk.c:2743
fr_trunk_request_signal_complete
void fr_trunk_request_signal_complete(fr_trunk_request_t *treq)
Signal that a trunk request is complete.
Definition: trunk.c:1995
fr_trunk_request_count_by_state
uint64_t fr_trunk_request_count_by_state(fr_trunk_t *trunk, int conn_state, int req_state)
Return a count of requests on a connection in a specific state.
Definition: trunk.c:4354
fr_trunk_request_signal_partial
void fr_trunk_request_signal_partial(fr_trunk_request_t *treq)
Signal a partial write.
Definition: trunk.c:1952
fr_trunk_alloc
fr_trunk_t * fr_trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, fr_trunk_io_funcs_t const *funcs, fr_trunk_conf_t const *conf, char const *log_prefix, void const *uctx, bool delay_start)
Allocate a new collection of connections.
Definition: trunk.c:4767
fr_trunk_connection_pop_request
int fr_trunk_connection_pop_request(fr_trunk_request_t **treq_out, fr_trunk_connection_t *tconn)
Pop a request off a connection's pending queue.
Definition: trunk.c:3756
fr_trunk_connection_signal_inactive
void fr_trunk_connection_signal_inactive(fr_trunk_connection_t *tconn)
Signal a trunk connection cannot accept more requests.
Definition: trunk.c:3810
fr_trunk_connection_signal_readable
void fr_trunk_connection_signal_readable(fr_trunk_connection_t *tconn)
Signal that a trunk connection is readable.
Definition: trunk.c:3794
fr_trunk_request_s::pub
struct fr_trunk_request_pub_s pub
Public fields in the trunk request.
Definition: trunk.c:98
fr_trunk_connection_signal_reconnect
void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn, fr_connection_reason_t reason)
Signal a trunk connection is no longer viable.
Definition: trunk.c:3872
fr_trunk_connection_signal_writable
void fr_trunk_connection_signal_writable(fr_trunk_connection_t *tconn)
Signal that a trunk connection is writable.
Definition: trunk.c:3776
fr_trunk_request_signal_cancel_complete
void fr_trunk_request_signal_cancel_complete(fr_trunk_request_t *treq)
Signal that a remote server acked our cancellation.
Definition: trunk.c:2179
fr_trunk_request_signal_cancel_partial
void fr_trunk_request_signal_cancel_partial(fr_trunk_request_t *treq)
Signal a partial cancel write.
Definition: trunk.c:2131
fr_trunk_connection_s
Associates request queues with a connection.
Definition: trunk.c:127
fr_trunk_request_s
Wraps a normal request.
Definition: trunk.c:97
fr_trunk_s
Main trunk management handle.
Definition: trunk.c:189
fr_trunk_pub_s::req_alloc_reused
uint64_t _CONST req_alloc_reused
How many requests were reused.
Definition: trunk.h:311
fr_trunk_cancel_reason_t
fr_trunk_cancel_reason_t
Reasons for a request being cancelled.
Definition: trunk.h:55
fr_trunk_request_pub_s::tconn
fr_trunk_connection_t *_CONST tconn
Connection this request belongs to.
Definition: trunk.h:332
FR_TRUNK_CONN_CONNECTING
@ FR_TRUNK_CONN_CONNECTING
Connection is connecting.
Definition: trunk.h:90
FR_TRUNK_CONN_ACTIVE
@ FR_TRUNK_CONN_ACTIVE
Connection is connected and ready to service requests.
Definition: trunk.h:91
fr_trunk_request_state_t
fr_trunk_request_state_t
Used for sanity checks and to simplify freeing.
Definition: trunk.h:161
FR_TRUNK_REQUEST_STATE_BACKLOG
@ FR_TRUNK_REQUEST_STATE_BACKLOG
In the backlog.
Definition: trunk.h:167
FR_TRUNK_REQUEST_STATE_CANCEL
@ FR_TRUNK_REQUEST_STATE_CANCEL
A request on a particular socket was cancel.
Definition: trunk.h:175
FR_TRUNK_REQUEST_STATE_SENT
@ FR_TRUNK_REQUEST_STATE_SENT
Was written to a socket. Waiting for a response.
Definition: trunk.h:172
FR_TRUNK_REQUEST_STATE_PENDING
@ FR_TRUNK_REQUEST_STATE_PENDING
In the queue of a connection and is pending writing.
Definition: trunk.h:168
FR_TRUNK_REQUEST_STATE_CANCEL_SENT
@ FR_TRUNK_REQUEST_STATE_CANCEL_SENT
We've informed the remote server that the request has been cancelled.
Definition: trunk.h:176
FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL
@ FR_TRUNK_REQUEST_STATE_CANCEL_PARTIAL
We partially wrote a cancellation request.
Definition: trunk.h:178
FR_TRUNK_REQUEST_STATE_PARTIAL
@ FR_TRUNK_REQUEST_STATE_PARTIAL
Some of the request was written to the socket, more of it should be written later.
Definition: trunk.h:170
FR_TRUNK_CONN_ALL
#define FR_TRUNK_CONN_ALL
All connection states.
Definition: trunk.h:111
fr_trunk_request_pub_s::state
fr_trunk_request_state_t _CONST state
Which list the request is now located in.
Definition: trunk.h:328
fr_trunk_pub_s::req_alloc_new
uint64_t _CONST req_alloc_new
How many requests we've allocated.
Definition: trunk.h:309
fr_trunk_io_funcs_t::connection_alloc
fr_trunk_connection_alloc_t connection_alloc
Allocate a new fr_connection_t.
Definition: trunk.h:712
FR_TRUNK_VERIFY
#define FR_TRUNK_VERIFY(_trunk)
Definition: trunk.h:906
FR_TRUNK_REQUEST_STATE_ALL
#define FR_TRUNK_REQUEST_STATE_ALL
All request states.
Definition: trunk.h:185
fr_trunk_connection_event_t
fr_trunk_connection_event_t
What type of I/O events the trunk connection is currently interested in receiving.
Definition: trunk.h:72
FR_TRUNK_CONN_EVENT_NONE
@ FR_TRUNK_CONN_EVENT_NONE
Don't notify the trunk on connection state changes.
Definition: trunk.h:73
FR_TRUNK_CONN_EVENT_WRITE
@ FR_TRUNK_CONN_EVENT_WRITE
Trunk should be notified if a connection is writable.
Definition: trunk.h:77
FR_TRUNK_CONN_EVENT_READ
@ FR_TRUNK_CONN_EVENT_READ
Trunk should be notified if a connection is readable.
Definition: trunk.h:75
FR_TRUNK_CONN_EVENT_BOTH
@ FR_TRUNK_CONN_EVENT_BOTH
Trunk should be notified if a connection is readable or writable.
Definition: trunk.h:79
fr_trunk_connection_pub_s::conn
fr_connection_t *_CONST conn
The underlying connection.
Definition: trunk.h:352
fr_trunk_request_pub_s::preq
void *_CONST preq
Data for the muxer to write to the connection.
Definition: trunk.h:334
fr_trunk_io_funcs_t::request_cancel_mux
fr_trunk_request_cancel_mux_t request_cancel_mux
!< Read one or more requests from a connection.
Definition: trunk.h:725
fr_trunk_enqueue_t
fr_trunk_enqueue_t
Definition: trunk.h:148
FR_TRUNK_ENQUEUE_IN_BACKLOG
@ FR_TRUNK_ENQUEUE_IN_BACKLOG
Request should be enqueued in backlog.
Definition: trunk.h:149
FR_TRUNK_ENQUEUE_NO_CAPACITY
@ FR_TRUNK_ENQUEUE_NO_CAPACITY
At maximum number of connections, and no connection has capacity.
Definition: trunk.h:151
FR_TRUNK_ENQUEUE_OK
@ FR_TRUNK_ENQUEUE_OK
Operation was successful.
Definition: trunk.h:150
fr_trunk_conf_t
Common configuration parameters for a trunk.
Definition: trunk.h:213
fr_trunk_io_funcs_t
I/O functions to pass to fr_trunk_alloc.
Definition: trunk.h:711
DEBUG_LVL_SET
#define DEBUG_LVL_SET
Definition: trunk_tests.c:29
test_cancel_mux
static void test_cancel_mux(UNUSED fr_event_list_t *el, fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
Definition: trunk_tests.c:63
TEST_LIST
TEST_LIST
Definition: trunk_tests.c:1933
test_proto_request_t::treq
fr_trunk_request_t * treq
Trunk request.
Definition: trunk_tests.c:12
_conn_notify
static void _conn_notify(fr_trunk_connection_t *tconn, fr_connection_t *conn, fr_event_list_t *el, fr_trunk_connection_event_t notify_on, UNUSED void *uctx)
Definition: trunk_tests.c:172
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_setup_socket_pair_connection_alloc
static fr_connection_t * test_setup_socket_pair_connection_alloc(fr_trunk_connection_t *tconn, fr_event_list_t *el, fr_connection_conf_t const *conn_conf, char const *log_prefix, UNUSED void *uctx)
Definition: trunk_tests.c:328
test_enqueue_and_io_speed
static void test_enqueue_and_io_speed(void)
Definition: trunk_tests.c:1820
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
test_mux
static void test_mux(UNUSED fr_event_list_t *el, fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
Definition: trunk_tests.c:31
_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_socket_pair_alloc_then_connect_timeout
static void test_socket_pair_alloc_then_connect_timeout(void)
Definition: trunk_tests.c:507
test_setup_socket_pair_1s_reconnection_delay_alloc
static fr_connection_t * test_setup_socket_pair_1s_reconnection_delay_alloc(fr_trunk_connection_t *tconn, fr_event_list_t *el, UNUSED fr_connection_conf_t const *conn_conf, char const *log_prefix, void *uctx)
Definition: trunk_tests.c:567
_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:1007
test_demux
static void test_demux(UNUSED fr_event_list_t *el, UNUSED fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
Definition: trunk_tests.c:100
test_setup_trunk
static fr_trunk_t * test_setup_trunk(TALLOC_CTX *ctx, fr_event_list_t *el, fr_trunk_conf_t *conf, bool with_cancel_mux, void *uctx)
Definition: trunk_tests.c:356
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:378
test_proto_request_t::failed
bool failed
Seen by the failed callback.
Definition: trunk_tests.c:15
test_connection_rebalance_requests
static void test_connection_rebalance_requests(void)
Definition: trunk_tests.c:1430
test_connection_levels_max
static void test_connection_levels_max(void)
Definition: trunk_tests.c:1500
test_socket_pair_alloc_then_reconnect_check_delay
static void test_socket_pair_alloc_then_reconnect_check_delay(void)
Definition: trunk_tests.c:585
test_proto_request_t::freed
bool freed
Seen by the free callback.
Definition: trunk_tests.c:16
_conn_open
static fr_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_proto_stats_t::failed
uint64_t failed
Count of tests in this run that failed.
Definition: trunk_tests.c:25
_conn_init
static fr_connection_state_t _conn_init(void **h_out, fr_connection_t *conn, UNUSED void *uctx)
Allocate a basic socket pair.
Definition: trunk_tests.c:313
_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_setup_socket_pair_1s_timeout_connection_alloc
static fr_connection_t * test_setup_socket_pair_1s_timeout_connection_alloc(fr_trunk_connection_t *tconn, fr_event_list_t *el, UNUSED fr_connection_conf_t const *conf, char const *log_prefix, void *uctx)
Definition: trunk_tests.c:489
test_proto_request_t::completed
bool completed
Seen by the complete callback.
Definition: trunk_tests.c:14
ALLOC_REQ
#define ALLOC_REQ(_id)
Definition: trunk_tests.c:1492
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:417
test_enqueue_cancellation_points
static void test_enqueue_cancellation_points(void)
Definition: trunk_tests.c:760
test_requeue_on_reconnect
static void test_requeue_on_reconnect(void)
Definition: trunk_tests.c:1082
test_connection_start_on_enqueue
static void test_connection_start_on_enqueue(void)
Definition: trunk_tests.c:1370
test_request_cancel
static void test_request_cancel(UNUSED fr_connection_t *conn, void *preq, UNUSED fr_trunk_cancel_reason_t reason, void *uctx)
Definition: trunk_tests.c:200
test_preq_cmp
static int8_t test_preq_cmp(void const *a, void const *b)
Definition: trunk_tests.c:349
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_enqueue_basic
static void test_enqueue_basic(void)
Definition: trunk_tests.c:659
test_proto_stats_t::completed
uint64_t completed
Count of tests in this run that completed.
Definition: trunk_tests.c:24
_conn_init_no_signal
static fr_connection_state_t _conn_init_no_signal(void **h_out, fr_connection_t *conn, UNUSED void *uctx)
Definition: trunk_tests.c:478
test_connection_levels_alternating_edges
static void test_connection_levels_alternating_edges(void)
Definition: trunk_tests.c:1660
_conn_close
static void _conn_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx)
Definition: trunk_tests.c:283
test_request_fail
static void test_request_fail(UNUSED request_t *request, void *preq, UNUSED void *rctx, UNUSED fr_trunk_request_state_t state, void *uctx)
Definition: trunk_tests.c:225
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:268
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:336
Generated by   doxygen 1.9.1