The FreeRADIUS server $Id: f3670dba8951ca10eb4948feb3dc3db9423a334f $
Loading...
Searching...
No Matches
trunk.c
Go to the documentation of this file.
1/*
2 * This program is free software; you can redistribute it and/or modify
3 * it under the terms of the GNU General Public License as published by
4 * the Free Software Foundation; either version 2 of the License, or (at
5 * your option) any later version.
6 *
7 * This program is distributed in the hope that it will be useful,
8 * but WITHOUT ANY WARRANTY; without even the implied warranty of
9 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10 * GNU General Public License for more details.
11 *
12 * You should have received a copy of the GNU General Public License
13 * along with this program; if not, write to the Free Software
14 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
15 */
16
17/**
18 * $Id: 1acce256acdacc2d09c56ab3c2d47affb27635b9 $
19 *
20 * @file src/lib/server/trunk.c
21 * @brief A management API for bonding multiple connections together.
22 *
23 * @copyright 2019-2020 Arran Cudbard-Bell (a.cudbardb@freeradius.org)
24 * @copyright 2019-2020 The FreeRADIUS server project
25 */
26
27#define LOG_PREFIX trunk->log_prefix
28
29#ifdef NDEBUG
30# define TALLOC_GET_TYPE_ABORT_NOOP 1
31#endif
32
33typedef struct trunk_request_s trunk_request_t;
34typedef struct trunk_connection_s trunk_connection_t;
35typedef struct trunk_s trunk_t;
36#define _TRUNK_PRIVATE 1
37#include <freeradius-devel/server/trunk.h>
38
39#include <freeradius-devel/server/trigger.h>
40#include <freeradius-devel/util/debug.h>
41#include <freeradius-devel/util/misc.h>
42#include <freeradius-devel/util/syserror.h>
43#include <freeradius-devel/util/minmax_heap.h>
44
45#ifdef HAVE_STDATOMIC_H
46# include <stdatomic.h>
47# ifndef ATOMIC_VAR_INIT
48# define ATOMIC_VAR_INIT(_x) (_x)
49# endif
50#else
51# include <freeradius-devel/util/stdatomic.h>
52#endif
53
54static atomic_uint_fast64_t request_counter = ATOMIC_VAR_INIT(1);
55
56#ifdef TESTING_TRUNK
57static fr_time_t test_time_base = fr_time_wrap(1);
58
59static fr_time_t test_time(void)
60{
61 return test_time_base;
62}
63
64#define fr_time test_time
65#endif
66
67#ifndef NDEBUG
68/** The maximum number of state logs to record per request
69 *
70 */
71#define TRUNK_REQUEST_STATE_LOG_MAX 20
72
73/** Trace state machine changes for a particular request
74 *
75 */
76typedef struct {
77 fr_dlist_head_t *log_head; //!< To allow the log entry to remove itself on free.
78 fr_dlist_t entry; //!< Entry in the linked list.
79 trunk_request_state_t from; //!< What state we transitioned from.
80 trunk_request_state_t to; //!< What state we transitioned to.
81
82 trunk_connection_t *tconn; //!< The request was associated with.
83 ///< Pointer may now be invalid, do no de-reference.
84
85 uint64_t tconn_id; //!< If the treq was associated with a connection
86 ///< the connection ID.
87 trunk_connection_state_t tconn_state; //!< If the treq was associated with a connection
88 ///< the connection state at the time of the
89 ///< state transition.
90
91 char const *function; //!< State change occurred in.
92 int line; //!< Line change occurred on.
93} trunk_request_state_log_t;
94#endif
95
96/** Wraps a normal request
97 *
98 */
99struct trunk_request_s {
100 struct trunk_request_pub_s pub; //!< Public fields in the trunk request.
101 ///< This *MUST* be the first field in this
102 ///< structure.
103
104 uint64_t id; //!< Trunk request ID.
105
106 fr_heap_index_t heap_id; //!< Used to track the request conn->pending heap.
107
108 fr_dlist_t entry; //!< Used to track the trunk request in the conn->sent
109 ///< or trunk->backlog request.
110
111 trunk_cancel_reason_t cancel_reason; //!< Why this request was cancelled.
112
113 fr_time_t last_freed; //!< Last time this request was freed.
114
115 bool bound_to_conn; //!< Fail the request if there's an attempt to
116 ///< re-enqueue it.
117
118 bool sent; //!< Trunk request has been sent at least once.
119 ///< Used so that re-queueing doesn't increase trunk
120 ///< `sent` count.
121
122#ifndef NDEBUG
123 fr_dlist_head_t log; //!< State change log.
124#endif
125};
126
127
128/** Associates request queues with a connection
129 *
130 * @dotfile src/lib/server/trunk_conn.gv "Trunk connection state machine"
131 * @dotfile src/lib/server/trunk_req.gv "Trunk request state machine"
132 */
133struct trunk_connection_s {
134 struct trunk_connection_pub_s pub; //!< Public fields in the trunk connection.
135 ///< This *MUST* be the first field in this
136 ///< structure.
137
138 fr_heap_index_t heap_id; //!< Used to track the connection in the connected
139 ///< heap.
140
141 fr_dlist_t entry; //!< Used to track the connection in the connecting,
142 ///< full and failed lists.
143
144 /** @name State
145 * @{
146 */
147 trunk_connection_event_t events; //!< The current events we expect to be notified on.
148 /** @} */
149
150 /** @name Request lists
151 * @{
152 */
153 fr_heap_t *pending; //!< Requests waiting to be sent.
154
155 trunk_request_t *partial; //!< Partially written request.
156
157 fr_dlist_head_t sent; //!< Sent request.
158
159 fr_dlist_head_t reapable; //!< Idle request.
160
161 fr_dlist_head_t cancel; //!< Requests in the cancel state.
162
163 trunk_request_t *cancel_partial; //!< Partially written cancellation request.
164
165 fr_dlist_head_t cancel_sent; //!< Sent cancellation request.
166 /** @} */
167
168 /** @name Statistics
169 * @{
170 */
171 uint64_t sent_count; //!< The number of requests that have been sent using
172 ///< this connection.
173 /** @} */
174
175 /** @name Timers
176 * @{
177 */
178 fr_timer_t *lifetime_ev; //!< Maximum time this connection can be open.
179 /** @} */
180};
181
182/** An entry in a trunk watch function list
183 *
184 */
185typedef struct trunk_watch_entry_s {
186 fr_dlist_t entry; //!< List entry.
187 trunk_watch_t func; //!< Function to call when a trunk enters
188 ///< the state this list belongs to
189 bool oneshot; //!< Remove the function after it's called once.
190 bool enabled; //!< Whether the watch entry is enabled.
191 void *uctx; //!< User data to pass to the function.
192} trunk_watch_entry_t;
193
194/** Map connection states to trigger names
195 *
196 * Must stay in the same order as #trunk_connection_state_t
197 */
198static fr_table_num_indexed_bit_pos_t const trunk_conn_trigger_names[] = {
199 { L("pool.connection_halted"), TRUNK_CONN_HALTED }, /* 0x0000 - bit 0 */
200 { L("pool.connection_init"), TRUNK_CONN_INIT }, /* 0x0001 - bit 1 */
201 { L("pool.connection_connecting"), TRUNK_CONN_CONNECTING }, /* 0x0002 - bit 2 */
202 { L("pool.connection_active"), TRUNK_CONN_ACTIVE }, /* 0x0004 - bit 3 */
203 { L("pool.connection_closed"), TRUNK_CONN_CLOSED }, /* 0x0008 - bit 4 */
204 { L("pool.connection_full"), TRUNK_CONN_FULL }, /* 0x0010 - bit 5 */
205 { L("pool.connection_inactive"), TRUNK_CONN_INACTIVE }, /* 0x0020 - bit 6 */
206 { L("pool.connection_inactive_draining"), TRUNK_CONN_INACTIVE_DRAINING }, /* 0x0040 - bit 7 */
207 { L("pool.connection_draining"), TRUNK_CONN_DRAINING }, /* 0x0080 - bit 8 */
208 { L("pool.connection_draining_to_free"), TRUNK_CONN_DRAINING_TO_FREE } /* 0x0100 - bit 9 */
209};
210static size_t trunk_conn_trigger_names_len = NUM_ELEMENTS(trunk_conn_trigger_names);
211
212/** Main trunk management handle
213 *
214 */
215struct trunk_s {
216 struct trunk_pub_s pub; //!< Public fields in the trunk connection.
217 ///< This *MUST* be the first field in this
218 ///< structure.
219
220 char const *log_prefix; //!< What to prepend to messages.
221
222 fr_event_list_t *el; //!< Event list used by this trunk and the connection.
223
224 trunk_conf_t conf; //!< Trunk common configuration.
225
226 fr_dlist_head_t free_requests; //!< Requests in the unassigned state. Waiting to be
227 ///< enqueued.
228
229 fr_heap_t *backlog; //!< The request backlog. Requests we couldn't
230 ///< immediately assign to a connection.
231
232 /** @name Connection lists
233 *
234 * A connection must always be in exactly one of these lists
235 * or trees.
236 *
237 * @{
238 */
239 fr_dlist_head_t init; //!< Connections which have not yet started
240 ///< connecting.
241
242 fr_dlist_head_t connecting; //!< Connections which are not yet in the open state.
243
244 fr_minmax_heap_t *active; //!< Connections which can service requests.
245
246 fr_dlist_head_t full; //!< Connections which have too many outstanding
247 ///< requests.
248
249 fr_dlist_head_t inactive; //!< Connections which have been signalled to be
250 ///< inactive by the API client.
251
252 fr_dlist_head_t inactive_draining; //!< Connections which have been signalled to be
253 ///< inactive by the API client, which the trunk
254 ///< manager is draining to close.
255
256 fr_dlist_head_t failed; //!< Connections that'll be reconnected shortly.
257
258 fr_dlist_head_t closed; //!< Connections that have closed. Either due to
259 ///< shutdown, reconnection or failure.
260
261 fr_dlist_head_t draining; //!< Connections that will be freed once all their
262 ///< requests are complete, but can be reactivated.
263
264 fr_dlist_head_t draining_to_free; //!< Connections that will be freed once all their
265 ///< requests are complete.
266
267 fr_dlist_head_t to_free; //!< Connections we're done with and will free on
268 //!< the next call to trunk_manage.
269 //!< This prevents connections from being freed
270 //!< whilst we're inside callbacks.
271 /** @} */
272
273 /** @name Callbacks
274 * @{
275 */
276 trunk_io_funcs_t funcs; //!< I/O functions.
277
278 void *in_handler; //!< Which handler we're inside.
279
280 void *uctx; //!< Uctx data to pass to alloc.
281
282 fr_dlist_head_t watch[TRUNK_STATE_MAX]; //!< To be called when trunk changes state.
283
284 trunk_watch_entry_t *next_watcher; //!< Watcher about to be run. Used to prevent nested watchers.
285 /** @} */
286
287 /** @name Timers
288 * @{
289 */
290 fr_timer_t *manage_ev; //!< Periodic connection management event.
291 /** @} */
292
293 /** @name Log rate limiting entries
294 * @{
295 */
296 fr_rate_limit_t limit_max_requests_alloc_log; //!< Rate limit on "Refusing to alloc requests - Limit of * requests reached"
297
298 fr_rate_limit_t limit_last_failure_log; //!< Rate limit on "Refusing to enqueue requests - No active conns"
299 /** @} */
300
301 /** @name State
302 * @{
303 */
304 bool freeing; //!< Trunk is being freed, don't spawn new
305 ///< connections or re-enqueue.
306
307 bool started; //!< Has the trunk been started.
308
309 bool managing_connections; //!< Whether the trunk is allowed to manage
310 ///< (open/close) connections.
311
312 uint64_t last_req_per_conn; //!< The last request to connection ratio we calculated.
313 /** @} */
314
315 fr_pair_list_t *trigger_args; //!< Passed to trigger
316
317 bool trigger_undef[NUM_ELEMENTS(trunk_conn_trigger_names)]; //!< Record that a specific trigger is undefined.
318
319 CONF_PAIR *trigger_cp[NUM_ELEMENTS(trunk_conn_trigger_names)]; //!< Cached trigger CONF_PAIRs
320};
321
322int trunk_trigger_cf_parse(TALLOC_CTX *ctx, void *out, void *parent, CONF_ITEM *ci, conf_parser_t const *rule);
323
324static conf_parser_t const trunk_config_request[] = {
325 { FR_CONF_OFFSET("per_connection_max", trunk_conf_t, max_req_per_conn), .dflt = "2000" },
326 { FR_CONF_OFFSET("per_connection_target", trunk_conf_t, target_req_per_conn), .dflt = "1000" },
327 { FR_CONF_OFFSET("free_delay", trunk_conf_t, req_cleanup_delay), .dflt = "10.0" },
328 { FR_CONF_OFFSET("triggers", trunk_conf_t, req_triggers), .func = trunk_trigger_cf_parse },
329
330 CONF_PARSER_TERMINATOR
331};
332
333static conf_parser_t const trunk_config_connection[] = {
334 { FR_CONF_OFFSET("connect_timeout", connection_conf_t, connection_timeout), .dflt = "3.0" },
335 { FR_CONF_OFFSET("reconnect_delay", connection_conf_t, reconnection_delay), .dflt = "1" },
336
337 CONF_PARSER_TERMINATOR
338};
339
340#ifndef TRUNK_TESTS
341conf_parser_t const trunk_config[] = {
342 { FR_CONF_OFFSET("start", trunk_conf_t, start), .dflt = "1" },
343 { FR_CONF_OFFSET("min", trunk_conf_t, min), .dflt = "1" },
344 { FR_CONF_OFFSET("max", trunk_conf_t, max), .dflt = "5" },
345 { FR_CONF_OFFSET("connecting", trunk_conf_t, connecting), .dflt = "2" },
346 { FR_CONF_OFFSET("uses", trunk_conf_t, max_uses), .dflt = "0" },
347 { FR_CONF_OFFSET("lifetime", trunk_conf_t, lifetime), .dflt = "0" },
348 { FR_CONF_OFFSET("idle_timeout", trunk_conf_t, idle_timeout), .dflt = "0" },
349
350 { FR_CONF_OFFSET("open_delay", trunk_conf_t, open_delay), .dflt = "0.2" },
351 { FR_CONF_OFFSET("close_delay", trunk_conf_t, close_delay), .dflt = "10.0" },
352
353 { FR_CONF_OFFSET("manage_interval", trunk_conf_t, manage_interval), .dflt = "0.2" },
354
355 { FR_CONF_OFFSET("max_backlog", trunk_conf_t, max_backlog), .dflt = "1000" },
356
357 { FR_CONF_OFFSET("backlog_on_failed_conn", trunk_conf_t, backlog_on_failed_conn), },
358
359 { FR_CONF_OFFSET("triggers", trunk_conf_t, conn_triggers), .func = trunk_trigger_cf_parse },
360
361 { FR_CONF_OFFSET_SUBSECTION("connection", 0, trunk_conf_t, conn_conf, trunk_config_connection), .subcs_size = sizeof(trunk_config_connection) },
362 { FR_CONF_POINTER("request", 0, CONF_FLAG_SUBSECTION, NULL), .subcs = (void const *) trunk_config_request },
363
364 CONF_PARSER_TERMINATOR
365};
366#endif
367
368#ifndef NDEBUG
369/** Map request states to trigger names
370 *
371 * Must stay in the same order as #trunk_connection_state_t
372 */
373static fr_table_num_indexed_bit_pos_t const trunk_req_trigger_names[] = {
374 { L("pool.request_init"), TRUNK_REQUEST_STATE_INIT }, /* 0x0000 - bit 0 */
375 { L("pool.request_unassigned"), TRUNK_REQUEST_STATE_UNASSIGNED }, /* 0x0001 - bit 1 */
376 { L("pool.request_backlog"), TRUNK_REQUEST_STATE_BACKLOG }, /* 0x0002 - bit 2 */
377 { L("pool.request_pending"), TRUNK_REQUEST_STATE_PENDING }, /* 0x0004 - bit 3 */
378 { L("pool.request_partial"), TRUNK_REQUEST_STATE_PARTIAL }, /* 0x0008 - bit 4 */
379 { L("pool.request_sent"), TRUNK_REQUEST_STATE_SENT }, /* 0x0010 - bit 5 */
380 { L("pool.request_state_reapable"), TRUNK_REQUEST_STATE_REAPABLE }, /* 0x0020 - bit 6 */
381 { L("pool.request_complete"), TRUNK_REQUEST_STATE_COMPLETE }, /* 0x0040 - bit 7 */
382 { L("pool.request_state_failed"), TRUNK_REQUEST_STATE_FAILED }, /* 0x0080 - bit 8 */
383 { L("pool.request_state_cancel"), TRUNK_REQUEST_STATE_CANCEL }, /* 0x0100 - bit 9 */
384 { L("pool.request_state_cancel_sent"), TRUNK_REQUEST_STATE_CANCEL_SENT }, /* 0x0200 - bit 10 */
385 { L("pool.request_state_cancel_partial"), TRUNK_REQUEST_STATE_CANCEL_PARTIAL }, /* 0x0400 - bit 11 */
386 { L("pool.request_state_cancel_complete"), TRUNK_REQUEST_STATE_CANCEL_COMPLETE }, /* 0x0800 - bit 12 */
387};
388static size_t trunk_req_trigger_names_len = NUM_ELEMENTS(trunk_req_trigger_names);
389#endif
390
391static fr_table_num_ordered_t const trunk_request_states[] = {
392 { L("INIT"), TRUNK_REQUEST_STATE_INIT },
393 { L("UNASSIGNED"), TRUNK_REQUEST_STATE_UNASSIGNED },
394 { L("BACKLOG"), TRUNK_REQUEST_STATE_BACKLOG },
395 { L("PENDING"), TRUNK_REQUEST_STATE_PENDING },
396 { L("PARTIAL"), TRUNK_REQUEST_STATE_PARTIAL },
397 { L("SENT"), TRUNK_REQUEST_STATE_SENT },
398 { L("REAPABLE"), TRUNK_REQUEST_STATE_REAPABLE },
399 { L("COMPLETE"), TRUNK_REQUEST_STATE_COMPLETE },
400 { L("FAILED"), TRUNK_REQUEST_STATE_FAILED },
401 { L("CANCEL"), TRUNK_REQUEST_STATE_CANCEL },
402 { L("CANCEL-SENT"), TRUNK_REQUEST_STATE_CANCEL_SENT },
403 { L("CANCEL-PARTIAL"), TRUNK_REQUEST_STATE_CANCEL_PARTIAL },
404 { L("CANCEL-COMPLETE"), TRUNK_REQUEST_STATE_CANCEL_COMPLETE }
405};
406static size_t trunk_request_states_len = NUM_ELEMENTS(trunk_request_states);
407
408static fr_table_num_ordered_t const trunk_states[] = {
409 { L("IDLE"), TRUNK_STATE_IDLE },
410 { L("ACTIVE"), TRUNK_STATE_ACTIVE },
411 { L("PENDING"), TRUNK_STATE_PENDING }
412};
413static size_t trunk_states_len = NUM_ELEMENTS(trunk_states);
414
415static fr_table_num_ordered_t const trunk_connection_states[] = {
416 { L("INIT"), TRUNK_CONN_INIT },
417 { L("HALTED"), TRUNK_CONN_HALTED },
418 { L("CONNECTING"), TRUNK_CONN_CONNECTING },
419 { L("ACTIVE"), TRUNK_CONN_ACTIVE },
420 { L("CLOSED"), TRUNK_CONN_CLOSED },
421 { L("FULL"), TRUNK_CONN_FULL },
422 { L("INACTIVE"), TRUNK_CONN_INACTIVE },
423 { L("INACTIVE-DRAINING"), TRUNK_CONN_INACTIVE_DRAINING },
424 { L("DRAINING"), TRUNK_CONN_DRAINING },
425 { L("DRAINING-TO-FREE"), TRUNK_CONN_DRAINING_TO_FREE }
426};
427static size_t trunk_connection_states_len = NUM_ELEMENTS(trunk_connection_states);
428
429static fr_table_num_ordered_t const trunk_cancellation_reasons[] = {
430 { L("TRUNK_CANCEL_REASON_NONE"), TRUNK_CANCEL_REASON_NONE },
431 { L("TRUNK_CANCEL_REASON_SIGNAL"), TRUNK_CANCEL_REASON_SIGNAL },
432 { L("TRUNK_CANCEL_REASON_MOVE"), TRUNK_CANCEL_REASON_MOVE },
433 { L("TRUNK_CANCEL_REASON_REQUEUE"), TRUNK_CANCEL_REASON_REQUEUE }
434};
435static size_t trunk_cancellation_reasons_len = NUM_ELEMENTS(trunk_cancellation_reasons);
436
437static fr_table_num_ordered_t const trunk_connection_events[] = {
438 { L("TRUNK_CONN_EVENT_NONE"), TRUNK_CONN_EVENT_NONE },
439 { L("TRUNK_CONN_EVENT_READ"), TRUNK_CONN_EVENT_READ },
440 { L("TRUNK_CONN_EVENT_WRITE"), TRUNK_CONN_EVENT_WRITE },
441 { L("TRUNK_CONN_EVENT_BOTH"), TRUNK_CONN_EVENT_BOTH },
442};
443static size_t trunk_connection_events_len = NUM_ELEMENTS(trunk_connection_events);
444
445#define CONN_TRIGGER(_state) do { \
446 uint8_t idx = fr_high_bit_pos(_state); \
447 if (trunk->conf.conn_triggers && !trunk->trigger_undef[idx]) { \
448 if (trigger(unlang_interpret_get_thread_default(), trunk->conf.conn_trigger_cs, \
449 &trunk->trigger_cp[idx], \
450 fr_table_str_by_value(trunk_conn_trigger_names, _state, \
451 "<INVALID>"), true, trunk->trigger_args) == -1) { \
452 trunk->trigger_undef[idx] = true; \
453 } \
454 } \
455} while (0)
456
457#define CONN_STATE_TRANSITION(_new, _log) \
458do { \
459 _log("[%" PRIu64 "] Trunk connection changed state %s -> %s", \
460 tconn->pub.conn->id, \
461 fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"), \
462 fr_table_str_by_value(trunk_connection_states, _new, "<INVALID>")); \
463 tconn->pub.state = _new; \
464 CONN_TRIGGER(_new); \
465 trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false); \
466} while (0)
467
468#define CONN_BAD_STATE_TRANSITION(_new) \
469do { \
470 if (!fr_cond_assert_msg(0, "[%" PRIu64 "] Trunk connection invalid transition %s -> %s", \
471 tconn->pub.conn->id, \
472 fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"), \
473 fr_table_str_by_value(trunk_connection_states, _new, "<INVALID>"))) return; \
474} while (0)
475
476#ifndef NDEBUG
477void trunk_request_state_log_entry_add(char const *function, int line,
478 trunk_request_t *treq, trunk_request_state_t new) CC_HINT(nonnull);
479
480#define REQUEST_TRIGGER(_state) do { \
481 if (trunk->conf.req_triggers) { \
482 trigger(unlang_interpret_get_thread_default(), \
483 trunk->conf.req_trigger_cs, NULL, fr_table_str_by_value(trunk_req_trigger_names, _state, \
484 "<INVALID>"), true, trunk->trigger_args); \
485 } \
486} while (0)
487
488/** Record a request state transition and log appropriate output
489 *
490 */
491#define REQUEST_STATE_TRANSITION(_new) \
492do { \
493 request_t *request = treq->pub.request; \
494 ROPTIONAL(RDEBUG3, DEBUG3, "Trunk request %" PRIu64 " changed state %s -> %s", \
495 treq->id, \
496 fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
497 fr_table_str_by_value(trunk_request_states, _new, "<INVALID>")); \
498 trunk_request_state_log_entry_add(__FUNCTION__, __LINE__, treq, _new); \
499 treq->pub.state = _new; \
500 REQUEST_TRIGGER(_new); \
501} while (0)
502#define REQUEST_BAD_STATE_TRANSITION(_new) \
503do { \
504 trunk_request_state_log(&default_log, L_ERR, __FILE__, __LINE__, treq); \
505 if (!fr_cond_assert_msg(0, "Trunk request %" PRIu64 " invalid transition %s -> %s", \
506 treq->id, \
507 fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
508 fr_table_str_by_value(trunk_request_states, _new, "<INVALID>"))) return; \
509} while (0)
510#else
511/** Record a request state transition
512 *
513 */
514#define REQUEST_STATE_TRANSITION(_new) \
515do { \
516 request_t *request = treq->pub.request; \
517 ROPTIONAL(RDEBUG3, DEBUG3, "Trunk request %" PRIu64 " changed state %s -> %s", \
518 treq->id, \
519 fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
520 fr_table_str_by_value(trunk_request_states, _new, "<INVALID>")); \
521 treq->pub.state = _new; \
522} while (0)
523#define REQUEST_BAD_STATE_TRANSITION(_new) \
524do { \
525 if (!fr_cond_assert_msg(0, "Trunk request %" PRIu64 " invalid transition %s -> %s", \
526 treq->id, \
527 fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"), \
528 fr_table_str_by_value(trunk_request_states, _new, "<INVALID>"))) return; \
529} while (0)
530#endif
531
532
533/** Call the cancel callback if set
534 *
535 */
536#define DO_REQUEST_CANCEL(_treq, _reason) \
537do { \
538 if ((_treq)->pub.trunk->funcs.request_cancel) { \
539 request_t *request = (_treq)->pub.request; \
540 void *_prev = (_treq)->pub.trunk->in_handler; \
541 (_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_cancel; \
542 ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_cancel(conn=%p, preq=%p, reason=%s, uctx=%p)", \
543 (_treq)->pub.tconn->pub.conn, \
544 (_treq)->pub.preq, \
545 fr_table_str_by_value(trunk_cancellation_reasons, \
546 (_reason), \
547 "<INVALID>"), \
548 (_treq)->pub.trunk->uctx); \
549 (_treq)->pub.trunk->funcs.request_cancel((_treq)->pub.tconn->pub.conn, (_treq)->pub.preq, (_reason), (_treq)->pub.trunk->uctx); \
550 (_treq)->pub.trunk->in_handler = _prev; \
551 } \
552} while(0)
553
554/** Call the "conn_release" callback (if set)
555 *
556 */
557#define DO_REQUEST_CONN_RELEASE(_treq) \
558do { \
559 if ((_treq)->pub.trunk->funcs.request_conn_release) { \
560 request_t *request = (_treq)->pub.request; \
561 void *_prev = (_treq)->pub.trunk->in_handler; \
562 (_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_conn_release; \
563 ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_conn_release(conn=%p, preq=%p, uctx=%p)", \
564 (_treq)->pub.tconn->pub.conn, \
565 (_treq)->pub.preq, \
566 (_treq)->pub.trunk->uctx); \
567 (_treq)->pub.trunk->funcs.request_conn_release((_treq)->pub.tconn->pub.conn, (_treq)->pub.preq, (_treq)->pub.trunk->uctx); \
568 (_treq)->pub.trunk->in_handler = _prev; \
569 } \
570} while(0)
571
572/** Call the complete callback (if set)
573 *
574 */
575#define DO_REQUEST_COMPLETE(_treq) \
576do { \
577 if ((_treq)->pub.trunk->funcs.request_complete) { \
578 request_t *request = (_treq)->pub.request; \
579 void *_prev = (_treq)->pub.trunk->in_handler; \
580 ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_complete(request=%p, preq=%p, rctx=%p, uctx=%p)", \
581 (_treq)->pub.request, \
582 (_treq)->pub.preq, \
583 (_treq)->pub.rctx, \
584 (_treq)->pub.trunk->uctx); \
585 (_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_complete; \
586 (_treq)->pub.trunk->funcs.request_complete((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.rctx, (_treq)->pub.trunk->uctx); \
587 (_treq)->pub.trunk->in_handler = _prev; \
588 } \
589} while(0)
590
591/** Call the fail callback (if set)
592 *
593 */
594#define DO_REQUEST_FAIL(_treq, _prev_state) \
595do { \
596 if ((_treq)->pub.trunk->funcs.request_fail) { \
597 request_t *request = (_treq)->pub.request; \
598 void *_prev = (_treq)->pub.trunk->in_handler; \
599 ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_fail(request=%p, preq=%p, rctx=%p, state=%s uctx=%p)", \
600 (_treq)->pub.request, \
601 (_treq)->pub.preq, \
602 (_treq)->pub.rctx, \
603 fr_table_str_by_value(trunk_request_states, (_prev_state), "<INVALID>"), \
604 (_treq)->pub.trunk->uctx); \
605 (_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_fail; \
606 (_treq)->pub.trunk->funcs.request_fail((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.rctx, _prev_state, (_treq)->pub.trunk->uctx); \
607 (_treq)->pub.trunk->in_handler = _prev; \
608 } \
609} while(0)
610
611/** Call the free callback (if set)
612 *
613 */
614#define DO_REQUEST_FREE(_treq) \
615do { \
616 if ((_treq)->pub.trunk->funcs.request_free) { \
617 request_t *request = (_treq)->pub.request; \
618 void *_prev = (_treq)->pub.trunk->in_handler; \
619 ROPTIONAL(RDEBUG3, DEBUG3, "Calling request_free(request=%p, preq=%p, uctx=%p)", \
620 (_treq)->pub.request, \
621 (_treq)->pub.preq, \
622 (_treq)->pub.trunk->uctx); \
623 (_treq)->pub.trunk->in_handler = (void *)(_treq)->pub.trunk->funcs.request_free; \
624 (_treq)->pub.trunk->funcs.request_free((_treq)->pub.request, (_treq)->pub.preq, (_treq)->pub.trunk->uctx); \
625 (_treq)->pub.trunk->in_handler = _prev; \
626 } \
627} while(0)
628
629/** Write one or more requests to a connection
630 *
631 */
632#define DO_REQUEST_MUX(_tconn) \
633do { \
634 void *_prev = (_tconn)->pub.trunk->in_handler; \
635 DEBUG3("[%" PRIu64 "] Calling request_mux(el=%p, tconn=%p, conn=%p, uctx=%p)", \
636 (_tconn)->pub.conn->id, \
637 (_tconn)->pub.trunk->el, \
638 (_tconn), \
639 (_tconn)->pub.conn, \
640 (_tconn)->pub.trunk->uctx); \
641 (_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_mux; \
642 (_tconn)->pub.trunk->funcs.request_mux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
643 (_tconn)->pub.trunk->in_handler = _prev; \
644} while(0)
645
646/** Read one or more requests from a connection
647 *
648 */
649#define DO_REQUEST_DEMUX(_tconn) \
650do { \
651 void *_prev = (_tconn)->pub.trunk->in_handler; \
652 DEBUG3("[%" PRIu64 "] Calling request_demux(tconn=%p, conn=%p, uctx=%p)", \
653 (_tconn)->pub.conn->id, \
654 (_tconn), \
655 (_tconn)->pub.conn, \
656 (_tconn)->pub.trunk->uctx); \
657 (_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_demux; \
658 (_tconn)->pub.trunk->funcs.request_demux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
659 (_tconn)->pub.trunk->in_handler = _prev; \
660} while(0)
661
662/** Write one or more cancellation requests to a connection
663 *
664 */
665#define DO_REQUEST_CANCEL_MUX(_tconn) \
666do { \
667 if ((_tconn)->pub.trunk->funcs.request_cancel_mux) { \
668 void *_prev = (_tconn)->pub.trunk->in_handler; \
669 DEBUG3("[%" PRIu64 "] Calling request_cancel_mux(tconn=%p, conn=%p, uctx=%p)", \
670 (_tconn)->pub.conn->id, \
671 (_tconn), \
672 (_tconn)->pub.conn, \
673 (_tconn)->pub.trunk->uctx); \
674 (_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.request_cancel_mux; \
675 (_tconn)->pub.trunk->funcs.request_cancel_mux((_tconn)->pub.trunk->el, (_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->uctx); \
676 (_tconn)->pub.trunk->in_handler = _prev; \
677 } \
678} while(0)
679
680/** Allocate a new connection
681 *
682 */
683#define DO_CONNECTION_ALLOC(_tconn) \
684do { \
685 void *_prev = trunk->in_handler; \
686 DEBUG3("Calling connection_alloc(tconn=%p, el=%p, conf=%p, log_prefix=\"%s\", uctx=%p)", \
687 (_tconn), \
688 (_tconn)->pub.trunk->el, \
689 (_tconn)->pub.trunk->conf.conn_conf, \
690 trunk->log_prefix, \
691 (_tconn)->pub.trunk->uctx); \
692 (_tconn)->pub.trunk->in_handler = (void *) (_tconn)->pub.trunk->funcs.connection_alloc; \
693 (_tconn)->pub.conn = trunk->funcs.connection_alloc((_tconn), (_tconn)->pub.trunk->el, (_tconn)->pub.trunk->conf.conn_conf, (_tconn)->pub.trunk->log_prefix, trunk->uctx); \
694 (_tconn)->pub.trunk->in_handler = _prev; \
695 if (!(_tconn)->pub.conn) { \
696 ERROR("Failed creating new connection"); \
697 talloc_free(tconn); \
698 return -1; \
699 } \
700} while(0)
701
702/** Change what events the connection should be notified about
703 *
704 */
705#define DO_CONNECTION_NOTIFY(_tconn, _events) \
706do { \
707 if ((_tconn)->pub.trunk->funcs.connection_notify) { \
708 void *_prev = (_tconn)->pub.trunk->in_handler; \
709 DEBUG3("[%" PRIu64 "] Calling connection_notify(tconn=%p, conn=%p, el=%p, events=%s, uctx=%p)", \
710 (_tconn)->pub.conn->id, \
711 (_tconn), \
712 (_tconn)->pub.conn, \
713 (_tconn)->pub.trunk->el, \
714 fr_table_str_by_value(trunk_connection_events, (_events), "<INVALID>"), \
715 (_tconn)->pub.trunk->uctx); \
716 (_tconn)->pub.trunk->in_handler = (void *)(_tconn)->pub.trunk->funcs.connection_notify; \
717 (_tconn)->pub.trunk->funcs.connection_notify((_tconn), (_tconn)->pub.conn, (_tconn)->pub.trunk->el, (_events), (_tconn)->pub.trunk->uctx); \
718 (_tconn)->pub.trunk->in_handler = _prev; \
719 } \
720} while(0)
721
722#define IN_HANDLER(_trunk) (((_trunk)->in_handler) != NULL)
723#define IN_REQUEST_MUX(_trunk) (((_trunk)->funcs.request_mux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_mux))
724#define IN_REQUEST_DEMUX(_trunk) (((_trunk)->funcs.request_demux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_demux))
725#define IN_REQUEST_CANCEL_MUX(_trunk) (((_trunk)->funcs.request_cancel_mux) && ((_trunk)->in_handler == (void *)(_trunk)->funcs.request_cancel_mux))
726
727#define IS_SERVICEABLE(_tconn) ((_tconn)->pub.state & TRUNK_CONN_SERVICEABLE)
728#define IS_PROCESSING(_tconn) ((_tconn)->pub.state & TRUNK_CONN_PROCESSING)
729
730/** Remove the current request from the backlog
731 *
732 */
733#define REQUEST_EXTRACT_BACKLOG(_treq) \
734do { \
735 int _ret; \
736 _ret = fr_heap_extract(&(_treq)->pub.trunk->backlog, _treq); \
737 if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from backlog heap: %s", fr_strerror())) break; \
738} while (0)
739
740/** Remove the current request from the pending list
741 *
742 */
743#define REQUEST_EXTRACT_PENDING(_treq) \
744do { \
745 int _ret; \
746 _ret = fr_heap_extract(&(_treq)->pub.tconn->pending, _treq); \
747 if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from pending heap: %s", fr_strerror())) break; \
748} while (0)
749
750/** Remove the current request from the partial slot
751 *
752 */
753#define REQUEST_EXTRACT_PARTIAL(_treq) \
754do { \
755 fr_assert((_treq)->pub.tconn->partial == treq); \
756 tconn->partial = NULL; \
757} while (0)
758
759/** Remove the current request from the sent list
760 *
761 */
762#define REQUEST_EXTRACT_SENT(_treq) fr_dlist_remove(&tconn->sent, treq)
763
764/** Remove the current request from the reapable list
765 *
766 */
767#define REQUEST_EXTRACT_REAPABLE(_treq) fr_dlist_remove(&tconn->reapable, treq)
768
769/** Remove the current request from the cancel list
770 *
771 */
772#define REQUEST_EXTRACT_CANCEL(_treq) fr_dlist_remove(&tconn->cancel, treq)
773
774/** Remove the current request from the cancel_partial slot
775 *
776 */
777#define REQUEST_EXTRACT_CANCEL_PARTIAL(_treq) \
778do { \
779 fr_assert((_treq)->pub.tconn->cancel_partial == treq); \
780 tconn->cancel_partial = NULL; \
781} while (0)
782
783/** Remove the current request from the cancel sent list
784 *
785 */
786#define REQUEST_EXTRACT_CANCEL_SENT(_treq) fr_dlist_remove(&tconn->cancel_sent, treq)
787
788/** Reorder the connections in the active heap
789 *
790 * fr_heap_extract will also error out if heap_id is bad - no need for assert
791 */
792#define CONN_REORDER(_tconn) \
793do { \
794 int _ret; \
795 if ((fr_minmax_heap_num_elements((_tconn)->pub.trunk->active) == 1)) break; \
796 if (!fr_cond_assert((_tconn)->pub.state == TRUNK_CONN_ACTIVE)) break; \
797 _ret = fr_minmax_heap_extract((_tconn)->pub.trunk->active, (_tconn)); \
798 if (!fr_cond_assert_msg(_ret == 0, "Failed extracting conn from active heap: %s", fr_strerror())) break; \
799 fr_minmax_heap_insert((_tconn)->pub.trunk->active, (_tconn)); \
800} while (0)
801
802DIAG_OFF(unused-function)
803
804#define FR_TRUNK_LIST_FUNC(_list,_type) \
805static inline CC_HINT(nonnull, always_inline) void trunk_list_ ## _list ## _add(trunk_t *trunk, _type *arg) \
806{ \
807 fr_dlist_insert_head(&trunk->_list, arg); \
808} \
809static inline CC_HINT(nonnull, always_inline) _type *trunk_list_ ## _list ##_peek(trunk_t *trunk) \
810{ \
811 return fr_dlist_tail(&trunk->_list); \
812} \
813static inline CC_HINT(nonnull, always_inline) _type *trunk_list_ ## _list ##_pop(trunk_t *trunk) \
814{ \
815 return fr_dlist_pop_head(&trunk->_list); \
816} \
817static inline CC_HINT(nonnull, always_inline) void trunk_list_ ## _list ##_remove(trunk_t *trunk, _type *arg) \
818{ \
819 fr_dlist_remove(&trunk->_list, arg); \
820}
821
822FR_TRUNK_LIST_FUNC(free_requests, trunk_request_t)
823FR_TRUNK_LIST_FUNC(full, trunk_connection_t)
824FR_TRUNK_LIST_FUNC(inactive, trunk_connection_t)
825FR_TRUNK_LIST_FUNC(inactive_draining, trunk_connection_t)
826FR_TRUNK_LIST_FUNC(draining, trunk_connection_t)
827
828DIAG_ON(unused-function)
829
830/** Call a list of watch functions associated with a state
831 *
832 */
833static inline void trunk_watch_call(trunk_t *trunk, fr_dlist_head_t *list, trunk_state_t state)
834{
835 /*
836 * Nested watcher calls are not allowed
837 * and shouldn't be possible because of
838 * deferred signal processing.
839 */
840 fr_assert(trunk->next_watcher == NULL);
841
842 while ((trunk->next_watcher = fr_dlist_next(list, trunk->next_watcher))) {
843 trunk_watch_entry_t *entry = trunk->next_watcher;
844 bool oneshot = entry->oneshot; /* Watcher could be freed, so store now */
845
846 if (!entry->enabled) continue;
847 if (oneshot) trunk->next_watcher = fr_dlist_remove(list, entry);
848
849 entry->func(trunk, trunk->pub.state, state, entry->uctx);
850
851 if (oneshot) talloc_free(entry);
852 }
853 trunk->next_watcher = NULL;
854}
855
856/** Call the state change watch functions
857 *
858 */
859#define CALL_WATCHERS(_trunk, _state) \
860do { \
861 if (fr_dlist_empty(&(_trunk)->watch[_state])) break; \
862 trunk_watch_call((_trunk), &(_trunk)->watch[_state], _state); \
863} while(0)
864
865/** Remove a watch function from a trunk state list
866 *
867 * @param[in] trunk The trunk to remove the watcher from.
868 * @param[in] state to remove the watch from.
869 * @param[in] watch Function to remove.
870 * @return
871 * - 0 if the function was removed successfully.
872 * - -1 if the function wasn't present in the watch list.
873 * - -2 if an invalid state was passed.
874 */
875int trunk_del_watch(trunk_t *trunk, trunk_state_t state, trunk_watch_t watch)
876{
877 trunk_watch_entry_t *entry = NULL;
878 fr_dlist_head_t *list;
879
880 if (state >= TRUNK_STATE_MAX) return -2;
881
882 list = &trunk->watch[state];
883 while ((entry = fr_dlist_next(list, entry))) {
884 if (entry->func == watch) {
885 if (trunk->next_watcher == entry) {
886 trunk->next_watcher = fr_dlist_remove(list, entry);
887 } else {
888 fr_dlist_remove(list, entry);
889 }
890 talloc_free(entry);
891 return 0;
892 }
893 }
894
895 return -1;
896}
897
898/** Add a watch entry to the trunk state list
899 *
900 * @param[in] trunk The trunk to add the watcher to.
901 * @param[in] state to watch for.
902 * @param[in] watch Function to add.
903 * @param[in] oneshot Should this watcher only be run once.
904 * @param[in] uctx Context to pass to function.
905 * @return
906 * - NULL if an invalid state is passed.
907 * - A new watch entry handle on success.
908 */
909trunk_watch_entry_t *trunk_add_watch(trunk_t *trunk, trunk_state_t state,
910 trunk_watch_t watch, bool oneshot, void const *uctx)
911{
912 trunk_watch_entry_t *entry;
913 fr_dlist_head_t *list;
914
915 if (state >= TRUNK_STATE_MAX) return NULL;
916
917 list = &trunk->watch[state];
918 MEM(entry = talloc_zero(trunk, trunk_watch_entry_t));
919
920 entry->func = watch;
921 entry->oneshot = oneshot;
922 entry->enabled = true;
923 memcpy(&entry->uctx, &uctx, sizeof(entry->uctx));
924 fr_dlist_insert_tail(list, entry);
925
926 return entry;
927}
928
929#define TRUNK_STATE_TRANSITION(_new) \
930do { \
931 DEBUG3("Trunk changed state %s -> %s", \
932 fr_table_str_by_value(trunk_states, trunk->pub.state, "<INVALID>"), \
933 fr_table_str_by_value(trunk_states, _new, "<INVALID>")); \
934 CALL_WATCHERS(trunk, _new); \
935 trunk->pub.state = _new; \
936} while (0)
937
938static void trunk_request_enter_backlog(trunk_request_t *treq, bool new);
939static void trunk_request_enter_pending(trunk_request_t *treq, trunk_connection_t *tconn, bool new);
940static void trunk_request_enter_partial(trunk_request_t *treq);
941static void trunk_request_enter_sent(trunk_request_t *treq);
942static void trunk_request_enter_reapable(trunk_request_t *treq);
943static void trunk_request_enter_failed(trunk_request_t *treq);
944static void trunk_request_enter_complete(trunk_request_t *treq);
945static void trunk_request_enter_cancel(trunk_request_t *treq, trunk_cancel_reason_t reason);
946static void trunk_request_enter_cancel_sent(trunk_request_t *treq);
947static void trunk_request_enter_cancel_complete(trunk_request_t *treq);
948
949static uint64_t trunk_requests_per_connection(uint16_t *conn_count_out, uint32_t *req_conn_out,
950 trunk_t *trunk, fr_time_t now, NDEBUG_UNUSED bool verify);
951
952static int trunk_connection_spawn(trunk_t *trunk, fr_time_t now);
953static inline void trunk_connection_auto_full(trunk_connection_t *tconn);
954static inline void trunk_connection_auto_unfull(trunk_connection_t *tconn);
955static inline void trunk_connection_readable(trunk_connection_t *tconn);
956static inline void trunk_connection_writable(trunk_connection_t *tconn);
957static void trunk_connection_event_update(trunk_connection_t *tconn);
958static void trunk_connection_enter_full(trunk_connection_t *tconn);
959static void trunk_connection_enter_inactive(trunk_connection_t *tconn);
960static void trunk_connection_enter_inactive_draining(trunk_connection_t *tconn);
961static void trunk_connection_enter_draining(trunk_connection_t *tconn);
962static void trunk_connection_enter_draining_to_free(trunk_connection_t *tconn);
963static void trunk_connection_enter_active(trunk_connection_t *tconn);
964
965static void trunk_rebalance(trunk_t *trunk);
966static void trunk_manage(trunk_t *trunk, fr_time_t now);
967static void _trunk_timer(fr_timer_list_t *tl, fr_time_t now, void *uctx);
968static void trunk_backlog_drain(trunk_t *trunk);
969
970/** Compare two protocol requests
971 *
972 * Allows protocol requests to be prioritised with a function
973 * specified by the API client. Defaults to by pointer address
974 * if no function is specified.
975 *
976 * @param[in] a treq to compare to b.
977 * @param[in] b treq to compare to a.
978 * @return
979 * - +1 if a > b.
980 * - 0 if a == b.
981 * - -1 if a < b.
982 */
983static int8_t _trunk_request_prioritise(void const *a, void const *b)
984{
985 trunk_request_t const *treq_a = talloc_get_type_abort_const(a, trunk_request_t);
986 trunk_request_t const *treq_b = talloc_get_type_abort_const(b, trunk_request_t);
987
988 fr_assert(treq_a->pub.trunk == treq_b->pub.trunk);
989
990 return treq_a->pub.trunk->funcs.request_prioritise(treq_a->pub.preq, treq_b->pub.preq);
991}
992
993/** Remove a request from all connection lists
994 *
995 * A common function used by init, fail, complete state functions to disassociate
996 * a request from a connection in preparation for freeing or reassignment.
997 *
998 * Despite its unassuming name, this function is *the* place to put calls to
999 * functions which need to be called when the number of requests associated with
1000 * a connection changes.
1001 *
1002 * Trunk requests will always be passed to this function before they're removed
1003 * from a connection, even if the requests are being freed.
1004 *
1005 * @param[in] treq to trigger a state change for.
1006 */
1007static void trunk_request_remove_from_conn(trunk_request_t *treq)
1008{
1009 trunk_connection_t *tconn = treq->pub.tconn;
1010 trunk_t *trunk = treq->pub.trunk;
1011
1012 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1013
1014 switch (treq->pub.state) {
1015 case TRUNK_REQUEST_STATE_UNASSIGNED:
1016 return; /* Not associated with connection */
1017
1018 case TRUNK_REQUEST_STATE_PENDING:
1019 REQUEST_EXTRACT_PENDING(treq);
1020 break;
1021
1022 case TRUNK_REQUEST_STATE_PARTIAL:
1023 REQUEST_EXTRACT_PARTIAL(treq);
1024 break;
1025
1026 case TRUNK_REQUEST_STATE_SENT:
1027 REQUEST_EXTRACT_SENT(treq);
1028 break;
1029
1030 case TRUNK_REQUEST_STATE_REAPABLE:
1031 REQUEST_EXTRACT_REAPABLE(treq);
1032 break;
1033
1034 case TRUNK_REQUEST_STATE_CANCEL:
1035 REQUEST_EXTRACT_CANCEL(treq);
1036 break;
1037
1038 case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
1039 REQUEST_EXTRACT_CANCEL_PARTIAL(treq);
1040 break;
1041
1042 case TRUNK_REQUEST_STATE_CANCEL_SENT:
1043 REQUEST_EXTRACT_CANCEL_SENT(treq);
1044 break;
1045
1046 default:
1047 fr_assert(0);
1048 break;
1049 }
1050
1051 /*
1052 * If the request wasn't associated with a
1053 * connection, then there's nothing more
1054 * to do.
1055 */
1056 if (!tconn) return;
1057
1058 {
1059 request_t *request = treq->pub.request;
1060
1061 ROPTIONAL(RDEBUG3, DEBUG3, "%s Trunk connection released request %" PRIu64,
1062 tconn->pub.conn->name, treq->id);
1063 }
1064 /*
1065 * Release any connection specific resources the
1066 * treq holds.
1067 */
1068 DO_REQUEST_CONN_RELEASE(treq);
1069
1070 switch (tconn->pub.state){
1071 case TRUNK_CONN_FULL:
1072 trunk_connection_auto_unfull(tconn); /* Check if we can switch back to active */
1073 if (tconn->pub.state == TRUNK_CONN_FULL) break; /* Only fallthrough if conn is now active */
1074 FALL_THROUGH;
1075
1076 case TRUNK_CONN_ACTIVE:
1077 CONN_REORDER(tconn);
1078 break;
1079
1080 default:
1081 break;
1082 }
1083
1084 treq->pub.tconn = NULL;
1085
1086 /*
1087 * Request removed from the connection
1088 * see if we need up deregister I/O events.
1089 */
1090 trunk_connection_event_update(tconn);
1091}
1092
1093/** Transition a request to the unassigned state, in preparation for re-assignment
1094 *
1095 * @note treq->tconn may be inviable after calling
1096 * if treq->conn and connection_signals_pause are not used.
1097 * This is due to call to trunk_request_remove_from_conn.
1098 *
1099 * @param[in] treq to trigger a state change for.
1100 */
1101static void trunk_request_enter_unassigned(trunk_request_t *treq)
1102{
1103 trunk_t *trunk = treq->pub.trunk;
1104
1105 switch (treq->pub.state) {
1106 case TRUNK_REQUEST_STATE_UNASSIGNED:
1107 return;
1108
1109 case TRUNK_REQUEST_STATE_BACKLOG:
1110 REQUEST_EXTRACT_BACKLOG(treq);
1111 break;
1112
1113 case TRUNK_REQUEST_STATE_PENDING:
1114 case TRUNK_REQUEST_STATE_CANCEL:
1115 case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
1116 case TRUNK_REQUEST_STATE_CANCEL_SENT:
1117 trunk_request_remove_from_conn(treq);
1118 break;
1119
1120 default:
1121 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_UNASSIGNED);
1122 }
1123
1124 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_UNASSIGNED);
1125}
1126
1127/** Transition a request to the backlog state, adding it to the backlog of the trunk
1128 *
1129 * @note treq->tconn and treq may be inviable after calling
1130 * if treq->conn and connection_signals_pause are not used.
1131 * This is due to call to trunk_manage.
1132 *
1133 * @param[in] treq to trigger a state change for.
1134 * @param[in] new Whether this is a new request.
1135 */
1136static void trunk_request_enter_backlog(trunk_request_t *treq, bool new)
1137{
1138 trunk_connection_t *tconn = treq->pub.tconn;
1139 trunk_t *trunk = treq->pub.trunk;
1140
1141 switch (treq->pub.state) {
1142 case TRUNK_REQUEST_STATE_INIT:
1143 case TRUNK_REQUEST_STATE_UNASSIGNED:
1144 break;
1145
1146 case TRUNK_REQUEST_STATE_PENDING:
1147 REQUEST_EXTRACT_PENDING(treq);
1148 break;
1149
1150 case TRUNK_REQUEST_STATE_CANCEL:
1151 REQUEST_EXTRACT_CANCEL(treq);
1152 break;
1153
1154 default:
1155 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_BACKLOG);
1156 }
1157
1158 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_BACKLOG);
1159 fr_heap_insert(&trunk->backlog, treq); /* Insert into the backlog heap */
1160
1161 /*
1162 * A new request has entered the trunk.
1163 * Re-calculate request/connection ratios.
1164 */
1165 if (new) trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);
1166
1167 /*
1168 * To reduce latency, if there's no connections
1169 * in the connecting state, call the trunk manage
1170 * function immediately.
1171 *
1172 * Likewise, if there's draining connections
1173 * which could be moved back to active call
1174 * the trunk manage function.
1175 *
1176 * Remember requests only enter the backlog if
1177 * there's no connections which can service them.
1178 */
1179 if ((trunk_connection_count_by_state(treq->pub.trunk, TRUNK_CONN_CONNECTING) == 0) ||
1180 (trunk_connection_count_by_state(treq->pub.trunk, TRUNK_CONN_DRAINING) > 0)) {
1181 trunk_connection_manage_schedule(treq->pub.trunk);
1182 }
1183}
1184
1185/** Transition a request to the pending state, adding it to the backlog of an active connection
1186 *
1187 * All trunk requests being added to a connection get passed to this function.
1188 * All trunk requests being removed from a connection get passed to #trunk_request_remove_from_conn.
1189 *
1190 * @note treq->tconn and treq may be inviable after calling
1191 * if treq->conn and connection_signals_pause is not used.
1192 * This is due to call to trunk_connection_event_update.
1193 *
1194 * @param[in] treq to trigger a state change for.
1195 * @param[in] tconn to enqueue the request on.
1196 * @param[in] new Whether this is a new request.
1197 */
1198static void trunk_request_enter_pending(trunk_request_t *treq, trunk_connection_t *tconn, bool new)
1199{
1200 trunk_t *trunk = treq->pub.trunk;
1201
1202 fr_assert(tconn->pub.trunk == trunk);
1203 fr_assert(IS_PROCESSING(tconn));
1204
1205 switch (treq->pub.state) {
1206 case TRUNK_REQUEST_STATE_INIT:
1207 case TRUNK_REQUEST_STATE_UNASSIGNED:
1208 fr_assert(!treq->pub.tconn);
1209 break;
1210
1211 case TRUNK_REQUEST_STATE_BACKLOG:
1212 fr_assert(!treq->pub.tconn);
1213 REQUEST_EXTRACT_BACKLOG(treq);
1214 break;
1215
1216 case TRUNK_REQUEST_STATE_CANCEL: /* Moved from another connection */
1217 REQUEST_EXTRACT_CANCEL(treq);
1218 break;
1219
1220 default:
1221 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_PENDING);
1222 }
1223
1224 /*
1225 * Assign the new connection first this first so
1226 * it appears in the state log.
1227 */
1228 treq->pub.tconn = tconn;
1229
1230 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_PENDING);
1231
1232 {
1233 request_t *request = treq->pub.request;
1234
1235 ROPTIONAL(RDEBUG, DEBUG3, "%s Trunk connection assigned request %"PRIu64,
1236 tconn->pub.conn->name, treq->id);
1237 }
1238 fr_heap_insert(&tconn->pending, treq);
1239
1240 /*
1241 * A new request has entered the trunk.
1242 * Re-calculate request/connection ratios.
1243 */
1244 if (new) trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);
1245
1246 /*
1247 * Check if we need to automatically transition the
1248 * connection to full.
1249 */
1250 trunk_connection_auto_full(tconn);
1251
1252 /*
1253 * Reorder the connection in the heap now it has an
1254 * additional request.
1255 */
1256 if (tconn->pub.state == TRUNK_CONN_ACTIVE) CONN_REORDER(tconn);
1257
1258 /*
1259 * We have a new request, see if we need to register
1260 * for I/O events.
1261 */
1262 trunk_connection_event_update(tconn);
1263}
1264
1265/** Transition a request to the partial state, indicating that is has been partially sent
1266 *
1267 * @param[in] treq to trigger a state change for.
1268 */
1269static void trunk_request_enter_partial(trunk_request_t *treq)
1270{
1271 trunk_connection_t *tconn = treq->pub.tconn;
1272 trunk_t *trunk = treq->pub.trunk;
1273
1274 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1275
1276 switch (treq->pub.state) {
1277 case TRUNK_REQUEST_STATE_PENDING: /* All requests go through pending, even requeued ones */
1278 REQUEST_EXTRACT_PENDING(treq);
1279 break;
1280
1281 default:
1282 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_PARTIAL);
1283 }
1284
1285 fr_assert(!tconn->partial);
1286 tconn->partial = treq;
1287
1288 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_PARTIAL);
1289}
1290
1291/** Transition a request to the sent state, indicating that it's been sent in its entirety
1292 *
1293 * @note treq->tconn and treq may be inviable after calling
1294 * if treq->conn and connection_signals_pause is not used.
1295 * This is due to call to trunk_connection_event_update.
1296 *
1297 * @param[in] treq to trigger a state change for.
1298 */
1299static void trunk_request_enter_sent(trunk_request_t *treq)
1300{
1301 trunk_connection_t *tconn = treq->pub.tconn;
1302 trunk_t *trunk = treq->pub.trunk;
1303
1304 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1305
1306 switch (treq->pub.state) {
1307 case TRUNK_REQUEST_STATE_PENDING:
1308 REQUEST_EXTRACT_PENDING(treq);
1309 break;
1310
1311 case TRUNK_REQUEST_STATE_PARTIAL:
1312 REQUEST_EXTRACT_PARTIAL(treq);
1313 break;
1314
1315 default:
1316 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_SENT);
1317 }
1318
1319 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_SENT);
1320 fr_dlist_insert_tail(&tconn->sent, treq);
1321
1322 /*
1323 * Update the connection's sent stats if this is the
1324 * first time this request is being sent.
1325 */
1326 if (!treq->sent) {
1327 trunk->pub.last_write_success = fr_time();
1328
1329 tconn->pub.last_write_success = trunk->pub.last_write_success;
1330 tconn->sent_count++;
1331 treq->sent = true;
1332
1333 /*
1334 * Enforces max_uses
1335 */
1336 if ((trunk->conf.max_uses > 0) && (tconn->sent_count >= trunk->conf.max_uses)) {
1337 DEBUG3("Trunk hit max uses %" PRIu64 " at %d", trunk->conf.max_uses, __LINE__);
1338 trunk_connection_enter_draining_to_free(tconn);
1339 }
1340 }
1341
1342 /*
1343 * We just sent a request, we probably need
1344 * to tell the event loop we want to be
1345 * notified if there's data available.
1346 */
1347 trunk_connection_event_update(tconn);
1348}
1349
1350/** Transition a request to the reapable state, indicating that it's been sent in its entirety, but no response is expected
1351 *
1352 * @note Largely a replica of trunk_request_enter_sent.
1353 *
1354 * @param[in] treq to trigger a state change for.
1355 */
1356static void trunk_request_enter_reapable(trunk_request_t *treq)
1357{
1358 trunk_connection_t *tconn = treq->pub.tconn;
1359 trunk_t *trunk = treq->pub.trunk;
1360
1361 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1362
1363 switch (treq->pub.state) {
1364 case TRUNK_REQUEST_STATE_PENDING:
1365 REQUEST_EXTRACT_PENDING(treq);
1366 break;
1367
1368 case TRUNK_REQUEST_STATE_PARTIAL:
1369 REQUEST_EXTRACT_PARTIAL(treq);
1370 break;
1371
1372 default:
1373 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_REAPABLE);
1374 }
1375
1376 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_REAPABLE);
1377 fr_dlist_insert_tail(&tconn->reapable, treq);
1378
1379 if (!treq->sent) {
1380 tconn->sent_count++;
1381 treq->sent = true;
1382
1383 if ((trunk->conf.max_uses > 0) && (tconn->sent_count >= trunk->conf.max_uses)) {
1384 DEBUG3("Trunk hit max uses %" PRIu64 " at %d", trunk->conf.max_uses, __LINE__);
1385 trunk_connection_enter_draining_to_free(tconn);
1386 }
1387 }
1388
1389 trunk_connection_event_update(tconn);
1390}
1391
1392/** Transition a request to the cancel state, placing it in a connection's cancellation list
1393 *
1394 * If a request_cancel_send callback is provided, that callback will
1395 * be called periodically for requests which were cancelled due to
1396 * a signal.
1397 *
1398 * The request_cancel_send callback will dequeue cancelled requests
1399 * and inform a remote server that the result is no longer required.
1400 *
1401 * A request must enter this state before being added to the backlog
1402 * of another connection if it's been sent or partially sent.
1403 *
1404 * @note treq->tconn and treq may be inviable after calling
1405 * if treq->conn and connection_signals_pause is not used.
1406 * This is due to call to trunk_connection_event_update.
1407 *
1408 * @param[in] treq to trigger a state change for.
1409 * @param[in] reason Why the request was cancelled.
1410 * Should be one of:
1411 * - TRUNK_CANCEL_REASON_SIGNAL request cancelled
1412 * because of a signal from the interpreter.
1413 * - TRUNK_CANCEL_REASON_MOVE request cancelled
1414 * because the connection failed and it needs
1415 * to be assigned to a new connection.
1416 * - TRUNK_CANCEL_REASON_REQUEUE request cancelled
1417 * as it needs to be resent on the same connection.
1418 */
1419static void trunk_request_enter_cancel(trunk_request_t *treq, trunk_cancel_reason_t reason)
1420{
1421 trunk_connection_t *tconn = treq->pub.tconn;
1422 trunk_t *trunk = treq->pub.trunk;
1423
1424 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1425
1426 switch (treq->pub.state) {
1427 case TRUNK_REQUEST_STATE_PARTIAL:
1428 REQUEST_EXTRACT_PARTIAL(treq);
1429 break;
1430
1431 case TRUNK_REQUEST_STATE_SENT:
1432 REQUEST_EXTRACT_SENT(treq);
1433 break;
1434
1435 case TRUNK_REQUEST_STATE_REAPABLE:
1436 REQUEST_EXTRACT_REAPABLE(treq);
1437 break;
1438
1439 default:
1440 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL);
1441 }
1442
1443 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL);
1444 fr_dlist_insert_tail(&tconn->cancel, treq);
1445 treq->cancel_reason = reason;
1446
1447 DO_REQUEST_CANCEL(treq, reason);
1448
1449 /*
1450 * Our treq is no longer bound to an actual
1451 * request_t *, as we can't guarantee the
1452 * lifetime of the original request_t *.
1453 */
1454 if (treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL) treq->pub.request = NULL;
1455
1456 /*
1457 * Register for I/O write events if we need to.
1458 */
1459 trunk_connection_event_update(treq->pub.tconn);
1460}
1461
1462/** Transition a request to the cancel_partial state, placing it in a connection's cancel_partial slot
1463 *
1464 * The request_demux function is then responsible for signalling
1465 * that the cancel request is complete when the remote server
1466 * acknowledges the cancellation request.
1467 *
1468 * @param[in] treq to trigger a state change for.
1469 */
1470static void trunk_request_enter_cancel_partial(trunk_request_t *treq)
1471{
1472 trunk_connection_t *tconn = treq->pub.tconn;
1473 trunk_t *trunk = treq->pub.trunk;
1474
1475 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1476 fr_assert(trunk->funcs.request_cancel_mux);
1477 fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL);
1478
1479 switch (treq->pub.state) {
1480 case TRUNK_REQUEST_STATE_CANCEL: /* The only valid state cancel_sent can be reached from */
1481 REQUEST_EXTRACT_CANCEL(treq);
1482 break;
1483
1484 default:
1485 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
1486 }
1487
1488 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
1489 fr_assert(!tconn->cancel_partial);
1490 tconn->cancel_partial = treq;
1491}
1492
1493/** Transition a request to the cancel_sent state, placing it in a connection's cancel_sent list
1494 *
1495 * The request_demux function is then responsible for signalling
1496 * that the cancel request is complete when the remote server
1497 * acknowledges the cancellation request.
1498 *
1499 * @note treq->tconn and treq may be inviable after calling
1500 * if treq->conn and connection_signals_pause is not used.
1501 * This is due to call to trunk_connection_event_update.
1502 *
1503 * @param[in] treq to trigger a state change for.
1504 */
1505static void trunk_request_enter_cancel_sent(trunk_request_t *treq)
1506{
1507 trunk_connection_t *tconn = treq->pub.tconn;
1508 trunk_t *trunk = treq->pub.trunk;
1509
1510 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1511 fr_assert(trunk->funcs.request_cancel_mux);
1512 fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_SIGNAL);
1513
1514 switch (treq->pub.state) {
1515 case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
1516 REQUEST_EXTRACT_CANCEL_PARTIAL(treq);
1517 break;
1518
1519 case TRUNK_REQUEST_STATE_CANCEL:
1520 REQUEST_EXTRACT_CANCEL(treq);
1521 break;
1522
1523 default:
1524 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_SENT);
1525 }
1526
1527 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_SENT);
1528 fr_dlist_insert_tail(&tconn->cancel_sent, treq);
1529
1530 /*
1531 * De-register for I/O write events
1532 * and register the read events
1533 * to drain the cancel ACKs.
1534 */
1535 trunk_connection_event_update(treq->pub.tconn);
1536}
1537
1538/** Cancellation was acked, the request is complete, free it
1539 *
1540 * The API client will not be informed, as the original request_t *
1541 * will likely have been freed by this point.
1542 *
1543 * @note treq will be inviable after a call to this function.
1544 * treq->tconn may be inviable after calling
1545 * if treq->conn and connection_signals_pause is not used.
1546 * This is due to call to trunk_request_remove_from_conn.
1547 *
1548 * @param[in] treq to mark as complete.
1549 */
1550static void trunk_request_enter_cancel_complete(trunk_request_t *treq)
1551{
1552 trunk_connection_t *tconn = treq->pub.tconn;
1553 trunk_t *trunk = treq->pub.trunk;
1554
1555 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1556 if (!fr_cond_assert(!treq->pub.request)) return; /* Only a valid state for request_t * which have been cancelled */
1557
1558 switch (treq->pub.state) {
1559 case TRUNK_REQUEST_STATE_CANCEL_SENT:
1560 case TRUNK_REQUEST_STATE_CANCEL:
1561 break;
1562
1563 default:
1564 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_COMPLETE);
1565 }
1566
1567 trunk_request_remove_from_conn(treq);
1568
1569 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_CANCEL_COMPLETE);
1570 trunk_request_free(&treq); /* Free the request */
1571}
1572
1573/** Request completed successfully, inform the API client and free the request
1574 *
1575 * @note treq will be inviable after a call to this function.
1576 * treq->tconn may also be inviable due to call to
1577 * trunk_request_remove_from_conn.
1578 *
1579 * @param[in] treq to mark as complete.
1580 */
1581static void trunk_request_enter_complete(trunk_request_t *treq)
1582{
1583 trunk_connection_t *tconn = treq->pub.tconn;
1584 trunk_t *trunk = treq->pub.trunk;
1585
1586 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1587
1588 switch (treq->pub.state) {
1589 case TRUNK_REQUEST_STATE_SENT:
1590 case TRUNK_REQUEST_STATE_PENDING:
1591 case TRUNK_REQUEST_STATE_REAPABLE:
1592 trunk_request_remove_from_conn(treq);
1593 break;
1594
1595 default:
1596 REQUEST_BAD_STATE_TRANSITION(TRUNK_REQUEST_STATE_COMPLETE);
1597 }
1598
1599 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_COMPLETE);
1600 DO_REQUEST_COMPLETE(treq);
1601 trunk_request_free(&treq); /* Free the request */
1602}
1603
1604/** Request failed, inform the API client and free the request
1605 *
1606 * @note treq will be inviable after a call to this function.
1607 * treq->tconn may also be inviable due to call to
1608 * trunk_request_remove_from_conn.
1609 *
1610 * @param[in] treq to mark as failed.
1611 */
1612static void trunk_request_enter_failed(trunk_request_t *treq)
1613{
1614 trunk_connection_t *tconn = treq->pub.tconn;
1615 trunk_t *trunk = treq->pub.trunk;
1616 trunk_request_state_t prev = treq->pub.state;
1617
1618 if (!fr_cond_assert(!tconn || (tconn->pub.trunk == trunk))) return;
1619
1620 switch (treq->pub.state) {
1621 case TRUNK_REQUEST_STATE_BACKLOG:
1622 REQUEST_EXTRACT_BACKLOG(treq);
1623 break;
1624
1625 default:
1626 trunk_request_remove_from_conn(treq);
1627 break;
1628 }
1629
1630 REQUEST_STATE_TRANSITION(TRUNK_REQUEST_STATE_FAILED);
1631 DO_REQUEST_FAIL(treq, prev);
1632 trunk_request_free(&treq); /* Free the request */
1633}
1634
1635/** Check to see if a trunk request can be enqueued
1636 *
1637 * @param[out] tconn_out Connection the request may be enqueued on.
1638 * @param[in] trunk To enqueue requests on.
1639 * @param[in] request associated with the treq (if any).
1640 * @return
1641 * - TRUNK_ENQUEUE_OK caller should enqueue request on provided tconn.
1642 * - TRUNK_ENQUEUE_IN_BACKLOG Request should be queued in the backlog.
1643 * - TRUNK_ENQUEUE_NO_CAPACITY Unable to enqueue request as we have no spare
1644 * connections or backlog space.
1645 * - TRUNK_ENQUEUE_DST_UNAVAILABLE Can't enqueue because the destination is
1646 * unreachable.
1647 */
1648static trunk_enqueue_t trunk_request_check_enqueue(trunk_connection_t **tconn_out, trunk_t *trunk,
1649 request_t *request)
1650{
1651 trunk_connection_t *tconn;
1652 /*
1653 * If we have an active connection then
1654 * return that.
1655 */
1656 tconn = fr_minmax_heap_min_peek(trunk->active);
1657 if (tconn) {
1658 *tconn_out = tconn;
1659 return TRUNK_ENQUEUE_OK;
1660 }
1661
1662 /*
1663 * Unlike the connection pool, we don't need
1664 * to drive any internal processes by feeding
1665 * it requests.
1666 *
1667 * If the last event to occur was a failure
1668 * we refuse to enqueue new requests until
1669 * one or more connections comes online.
1670 */
1671 if (!trunk->conf.backlog_on_failed_conn &&
1672 fr_time_gt(trunk->pub.last_failed, fr_time_wrap(0)) &&
1673 fr_time_lt(trunk->pub.last_connected, trunk->pub.last_failed)) {
1674 RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_last_failure_log,
1675 RWARN, WARN, "Refusing to enqueue requests - "
1676 "No active connections and last event was a connection failure");
1677
1678 return TRUNK_ENQUEUE_DST_UNAVAILABLE;
1679 }
1680
1681
1682 /*
1683 * Only enforce if we're limiting maximum
1684 * number of connections, and maximum
1685 * number of requests per connection.
1686 *
1687 * The alloc function also checks this
1688 * which is why this is only done for
1689 * debug builds.
1690 */
1691 if (trunk->conf.max_req_per_conn && trunk->conf.max) {
1692 uint64_t limit;
1693
1694 limit = trunk->conf.max * (uint64_t)trunk->conf.max_req_per_conn;
1695 if (limit > 0) {
1696 uint64_t total_reqs;
1697
1698 total_reqs = trunk_request_count_by_state(trunk, TRUNK_CONN_ALL,
1699 TRUNK_REQUEST_STATE_ALL);
1700 if (total_reqs >= (limit + trunk->conf.max_backlog)) {
1701 RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_max_requests_alloc_log,
1702 RWARN, WARN, "Refusing to alloc requests - "
1703 "Limit of %"PRIu64" (max = %u * per_connection_max = %u) "
1704 "plus %u backlog requests reached",
1705 limit, trunk->conf.max, trunk->conf.max_req_per_conn,
1706 trunk->conf.max_backlog);
1707 return TRUNK_ENQUEUE_NO_CAPACITY;
1708 }
1709 }
1710 }
1711
1712 return TRUNK_ENQUEUE_IN_BACKLOG;
1713}
1714
1715/** Enqueue a request which has never been assigned to a connection or was previously cancelled
1716 *
1717 * @param[in] treq to re enqueue. Must have been removed
1718 * from its existing connection with
1719 * #trunk_connection_requests_dequeue.
1720 * @return
1721 * - TRUNK_ENQUEUE_OK Request was re-enqueued.
1722 * - TRUNK_ENQUEUE_NO_CAPACITY Request enqueueing failed because we're at capacity.
1723 * - TRUNK_ENQUEUE_DST_UNAVAILABLE Enqueuing failed for some reason.
1724 * Usually because the connection to the resource is down.
1725 */
1726static trunk_enqueue_t trunk_request_enqueue_existing(trunk_request_t *treq)
1727{
1728 trunk_t *trunk = treq->pub.trunk;
1729 trunk_connection_t *tconn = NULL;
1730 trunk_enqueue_t ret;
1731
1732 /*
1733 * Must *NOT* still be assigned to another connection
1734 */
1735 fr_assert(!treq->pub.tconn);
1736
1737 ret = trunk_request_check_enqueue(&tconn, trunk, treq->pub.request);
1738 switch (ret) {
1739 case TRUNK_ENQUEUE_OK:
1740 if (trunk->conf.always_writable) {
1741 connection_signals_pause(tconn->pub.conn);
1742 trunk_request_enter_pending(treq, tconn, false);
1743 trunk_connection_writable(tconn);
1744 connection_signals_resume(tconn->pub.conn);
1745 } else {
1746 trunk_request_enter_pending(treq, tconn, false);
1747 }
1748 break;
1749
1750 case TRUNK_ENQUEUE_IN_BACKLOG:
1751 /*
1752 * No more connections and request
1753 * is already in the backlog.
1754 *
1755 * Signal our caller it should stop
1756 * trying to drain the backlog.
1757 */
1758 if (treq->pub.state == TRUNK_REQUEST_STATE_BACKLOG) return TRUNK_ENQUEUE_NO_CAPACITY;
1759 trunk_request_enter_backlog(treq, false);
1760 break;
1761
1762 default:
1763 break;
1764 }
1765
1766 return ret;
1767}
1768
1769/** Shift requests in the specified states onto new connections
1770 *
1771 * This function will blindly dequeue any requests in the specified state and get
1772 * them back to the unassigned state, cancelling any sent or partially sent requests.
1773 *
1774 * This function does not check that dequeuing a request in a particular state is a
1775 * sane or sensible thing to do, that's up to the caller!
1776 *
1777 * @param[out] out A list to insert the newly dequeued and unassigned
1778 * requests into.
1779 * @param[in] tconn to dequeue requests from.
1780 * @param[in] states Dequeue request in these states.
1781 * @param[in] max The maximum number of requests to dequeue. 0 for unlimited.
1782 */
1783static uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, trunk_connection_t *tconn,
1784 int states, uint64_t max)
1785{
1786 trunk_request_t *treq;
1787 uint64_t count = 0;
1788
1789 if (max == 0) max = UINT64_MAX;
1790
1791#define OVER_MAX_CHECK if (++count > max) return (count - 1)
1792
1793#define DEQUEUE_ALL(_src_list, _state) do { \
1794 while ((treq = fr_dlist_head(_src_list))) { \
1795 OVER_MAX_CHECK; \
1796 fr_assert(treq->pub.state == (_state)); \
1797 trunk_request_enter_unassigned(treq); \
1798 fr_dlist_insert_tail(out, treq); \
1799 } } while (0)
1800
1801 /*
1802 * Don't need to do anything with
1803 * cancellation requests.
1804 */
1805 if (states & TRUNK_REQUEST_STATE_CANCEL) DEQUEUE_ALL(&tconn->cancel,
1806 TRUNK_REQUEST_STATE_CANCEL);
1807
1808 /*
1809 * ...same with cancel inform
1810 */
1811 if (states & TRUNK_REQUEST_STATE_CANCEL_SENT) DEQUEUE_ALL(&tconn->cancel_sent,
1812 TRUNK_REQUEST_STATE_CANCEL_SENT);
1813
1814 /*
1815 * ....same with cancel partial
1816 */
1817 if (states & TRUNK_REQUEST_STATE_CANCEL_PARTIAL) {
1818 OVER_MAX_CHECK;
1819 treq = tconn->cancel_partial;
1820 if (treq) {
1821 fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
1822 trunk_request_enter_unassigned(treq);
1823 fr_dlist_insert_tail(out, treq);
1824 }
1825 }
1826
1827 /*
1828 * ...and pending.
1829 */
1830 if (states & TRUNK_REQUEST_STATE_PENDING) {
1831 while ((treq = fr_heap_peek(tconn->pending))) {
1832 OVER_MAX_CHECK;
1833 fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_PENDING);
1834 trunk_request_enter_unassigned(treq);
1835 fr_dlist_insert_tail(out, treq);
1836 }
1837 }
1838
1839 /*
1840 * Cancel partially sent requests
1841 */
1842 if (states & TRUNK_REQUEST_STATE_PARTIAL) {
1843 OVER_MAX_CHECK;
1844 treq = tconn->partial;
1845 if (treq) {
1846 fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_PARTIAL);
1847
1848 /*
1849 * Don't allow the connection to change state whilst
1850 * we're draining requests from it.
1851 */
1852 connection_signals_pause(tconn->pub.conn);
1853 trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_MOVE);
1854 trunk_request_enter_unassigned(treq);
1855 fr_dlist_insert_tail(out, treq);
1856 connection_signals_resume(tconn->pub.conn);
1857 }
1858 }
1859
1860 /*
1861 * Cancel sent requests
1862 */
1863 if (states & TRUNK_REQUEST_STATE_SENT) {
1864 /*
1865 * Don't allow the connection to change state whilst
1866 * we're draining requests from it.
1867 */
1868 connection_signals_pause(tconn->pub.conn);
1869 while ((treq = fr_dlist_head(&tconn->sent))) {
1870 OVER_MAX_CHECK;
1871 fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_SENT);
1872
1873 trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_MOVE);
1874 trunk_request_enter_unassigned(treq);
1875 fr_dlist_insert_tail(out, treq);
1876 }
1877 connection_signals_resume(tconn->pub.conn);
1878 }
1879
1880 return count;
1881}
1882
1883/** Remove requests in specified states from a connection, attempting to distribute them to new connections
1884 *
1885 * @param[in] tconn To remove requests from.
1886 * @param[in] states One or more states or'd together.
1887 * @param[in] max The maximum number of requests to dequeue.
1888 * 0 for unlimited.
1889 * @param[in] fail_bound If true causes any requests bound to the connection to fail.
1890 * If false bound requests will not be moved.
1891 *
1892 * @return the number of requests re-queued.
1893 */
1894static uint64_t trunk_connection_requests_requeue_priv(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
1895{
1896 trunk_t *trunk = tconn->pub.trunk;
1897 fr_dlist_head_t to_process;
1898 trunk_request_t *treq = NULL;
1899 uint64_t moved = 0;
1900
1901 if (max == 0) max = UINT64_MAX;
1902
1903 fr_dlist_talloc_init(&to_process, trunk_request_t, entry);
1904
1905 /*
1906 * Prevent the connection changing state whilst we're
1907 * working with it.
1908 *
1909 * There's a user callback that can be called by
1910 * trunk_request_enqueue_existing which can reconnect
1911 * the connection.
1912 */
1913 connection_signals_pause(tconn->pub.conn);
1914
1915 /*
1916 * Remove non-cancelled requests from the connection
1917 */
1918 moved += trunk_connection_requests_dequeue(&to_process, tconn, states & ~TRUNK_REQUEST_STATE_CANCEL_ALL, max);
1919
1920 /*
1921 * Prevent requests being requeued on the same trunk
1922 * connection, which would break rebalancing.
1923 *
1924 * This is a bit of a hack, but nothing should test
1925 * for connection/list consistency in this code,
1926 * and if something is added later, it'll be flagged
1927 * by the tests.
1928 */
1929 if (tconn->pub.state == TRUNK_CONN_ACTIVE) {
1930 int ret;
1931
1932 ret = fr_minmax_heap_extract(trunk->active, tconn);
1933 if (!fr_cond_assert_msg(ret == 0,
1934 "Failed extracting conn from active heap: %s", fr_strerror())) goto done;
1935
1936 }
1937
1938 /*
1939 * Loop over all the requests we gathered and
1940 * redistribute them to new connections.
1941 */
1942 while ((treq = fr_dlist_next(&to_process, treq))) {
1943 trunk_request_t *prev;
1944
1945 prev = fr_dlist_remove(&to_process, treq);
1946
1947 /*
1948 * Attempts to re-queue a request
1949 * that's bound to a connection
1950 * results in a failure.
1951 */
1952 if (treq->bound_to_conn) {
1953 if (fail_bound || !IS_SERVICEABLE(tconn)) {
1954 trunk_request_enter_failed(treq);
1955 } else {
1956 trunk_request_enter_pending(treq, tconn, false);
1957 }
1958 goto next;
1959 }
1960
1961 switch (trunk_request_enqueue_existing(treq)) {
1962 case TRUNK_ENQUEUE_OK:
1963 break;
1964
1965 /*
1966 * A connection failed, and
1967 * there's no other connections
1968 * available to deal with the
1969 * load, it's been placed back
1970 * in the backlog.
1971 */
1972 case TRUNK_ENQUEUE_IN_BACKLOG:
1973 break;
1974
1975 /*
1976 * If we fail to re-enqueue then
1977 * there's nothing to do except
1978 * fail the request.
1979 */
1980 case TRUNK_ENQUEUE_DST_UNAVAILABLE:
1981 case TRUNK_ENQUEUE_NO_CAPACITY:
1982 case TRUNK_ENQUEUE_FAIL:
1983 trunk_request_enter_failed(treq);
1984 break;
1985 }
1986 next:
1987 treq = prev;
1988 }
1989
1990 /*
1991 * Add the connection back into the active list
1992 */
1993 if (tconn->pub.state == TRUNK_CONN_ACTIVE) {
1994 int ret;
1995
1996 ret = fr_minmax_heap_insert(trunk->active, tconn);
1997 if (!fr_cond_assert_msg(ret == 0,
1998 "Failed re-inserting conn into active heap: %s", fr_strerror())) goto done;
1999 }
2000 if (moved >= max) goto done;
2001
2002 /*
2003 * Deal with the cancelled requests specially we can't
2004 * queue them up again as they were only valid on that
2005 * specific connection.
2006 *
2007 * We just need to run them to completion which, as
2008 * they should already be in the unassigned state,
2009 * just means freeing them.
2010 */
2011 moved += trunk_connection_requests_dequeue(&to_process, tconn,
2012 states & TRUNK_REQUEST_STATE_CANCEL_ALL, max - moved);
2013 while ((treq = fr_dlist_next(&to_process, treq))) {
2014 trunk_request_t *prev;
2015
2016 prev = fr_dlist_remove(&to_process, treq);
2017 trunk_request_free(&treq);
2018 treq = prev;
2019 }
2020
2021done:
2022
2023 /*
2024 * Always re-calculate the request/connection
2025 * ratio at the end.
2026 *
2027 * This avoids having the state transition
2028 * functions do it.
2029 *
2030 * The ratio would be wrong when they calculated
2031 * it anyway, because a bunch of requests are
2032 * dequeued from the connection and temporarily
2033 * cease to exist from the perspective of the
2034 * trunk_requests_per_connection code.
2035 */
2036 trunk_requests_per_connection(NULL, NULL, trunk, fr_time(), false);
2037
2038 connection_signals_resume(tconn->pub.conn);
2039 return moved;
2040}
2041
2042/** Move requests off of a connection and requeue elsewhere
2043 *
2044 * @note We don't re-queue on draining or draining to free, as requests should have already been
2045 * moved off of the connection. It's also dangerous as the trunk management code main
2046 * clean up a connection in this state when it's run on re-queue, and then the caller
2047 * may try and access a now freed connection.
2048 *
2049 * @param[in] tconn to move requests off of.
2050 * @param[in] states Only move requests in this state.
2051 * @param[in] max The maximum number of requests to dequeue. 0 for unlimited.
2052 * @param[in] fail_bound If true causes any requests bound to the connection to fail.
2053 * If false bound requests will not be moved.
2054 * @return The number of requests requeued.
2055 */
2056uint64_t trunk_connection_requests_requeue(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
2057{
2058 switch (tconn->pub.state) {
2059 case TRUNK_CONN_ACTIVE:
2060 case TRUNK_CONN_FULL:
2061 case TRUNK_CONN_INACTIVE:
2062 return trunk_connection_requests_requeue_priv(tconn, states, max, fail_bound);
2063
2064 default:
2065 return 0;
2066 }
2067}
2068
2069/** Signal a partial write
2070 *
2071 * Where there's high load, and the outbound write buffer is full
2072 *
2073 * @param[in] treq to signal state change for.
2074 */
2075void trunk_request_signal_partial(trunk_request_t *treq)
2076{
2077 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2078
2079 if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
2080 "%s can only be called from within request_mux handler", __FUNCTION__)) return;
2081
2082 switch (treq->pub.state) {
2083 case TRUNK_REQUEST_STATE_PENDING:
2084 trunk_request_enter_partial(treq);
2085 break;
2086
2087 default:
2088 return;
2089 }
2090}
2091
2092/** Signal that the request was written to a connection successfully
2093 *
2094 * @param[in] treq to signal state change for.
2095 */
2096void trunk_request_signal_sent(trunk_request_t *treq)
2097{
2098 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2099
2100 if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
2101 "%s can only be called from within request_mux handler", __FUNCTION__)) return;
2102
2103 switch (treq->pub.state) {
2104 case TRUNK_REQUEST_STATE_PENDING:
2105 case TRUNK_REQUEST_STATE_PARTIAL:
2106 trunk_request_enter_sent(treq);
2107 break;
2108
2109 default:
2110 return;
2111 }
2112}
2113
2114/** Signal that the request was written to a connection successfully, but no response is expected
2115 *
2116 * @param[in] treq to signal state change for.
2117 */
2118void trunk_request_signal_reapable(trunk_request_t *treq)
2119{
2120 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2121
2122 if (!fr_cond_assert_msg(IN_REQUEST_MUX(treq->pub.trunk),
2123 "%s can only be called from within request_mux handler", __FUNCTION__)) return;
2124
2125 switch (treq->pub.state) {
2126 case TRUNK_REQUEST_STATE_PENDING:
2127 case TRUNK_REQUEST_STATE_PARTIAL:
2128 trunk_request_enter_reapable(treq);
2129 break;
2130
2131 default:
2132 return;
2133 }
2134}
2135
2136/** Signal that a trunk request is complete
2137 *
2138 * The API client will be informed that the request is now complete.
2139 */
2140void trunk_request_signal_complete(trunk_request_t *treq)
2141{
2142 trunk_t *trunk = treq->pub.trunk;
2143
2144 if (!fr_cond_assert_msg(trunk, "treq not associated with trunk")) return;
2145
2146 /*
2147 * We assume that if the request is being signalled
2148 * as complete from the demux function, that it was
2149 * a successful read.
2150 *
2151 * If this assumption turns out to be incorrect
2152 * then we need to add an argument to signal_complete
2153 * to indicate if this is a successful read.
2154 */
2155 if (IN_REQUEST_DEMUX(trunk)) {
2156 trunk_connection_t *tconn = treq->pub.tconn;
2157
2158 trunk->pub.last_read_success = fr_time();
2159 tconn->pub.last_read_success = trunk->pub.last_read_success;
2160 }
2161
2162 switch (treq->pub.state) {
2163 case TRUNK_REQUEST_STATE_SENT:
2164 case TRUNK_REQUEST_STATE_PENDING: /* Got immediate response, i.e. cached */
2165 case TRUNK_REQUEST_STATE_REAPABLE:
2166 trunk_request_enter_complete(treq);
2167 break;
2168
2169 default:
2170 return;
2171 }
2172}
2173
2174/** Signal that a trunk request failed
2175 *
2176 * The API client will be informed that the request has failed.
2177 */
2178void trunk_request_signal_fail(trunk_request_t *treq)
2179{
2180 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2181
2182 trunk_request_enter_failed(treq);
2183}
2184
2185/** Cancel a trunk request
2186 *
2187 * treq can be in any state, but requests to cancel if the treq is not in
2188 * the TRUNK_REQUEST_STATE_PARTIAL or TRUNK_REQUEST_STATE_SENT state will be ignored.
2189 *
2190 * The complete or failed callbacks will not be called here, as it's assumed the request_t *
2191 * is now inviable as it's being cancelled.
2192 *
2193 * The free function however, is called, and that should be used to perform necessary
2194 * cleanup.
2195 *
2196 * @param[in] treq to signal state change for.
2197 */
2198void trunk_request_signal_cancel(trunk_request_t *treq)
2199{
2200 trunk_t *trunk;
2201
2202 /*
2203 * Ensure treq hasn't been freed
2204 */
2205 (void)talloc_get_type_abort(treq, trunk_request_t);
2206
2207 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2208
2209 if (!fr_cond_assert_msg(!IN_HANDLER(treq->pub.trunk),
2210 "%s cannot be called within a handler", __FUNCTION__)) return;
2211
2212 trunk = treq->pub.trunk;
2213
2214 switch (treq->pub.state) {
2215 /*
2216 * We don't call the complete or failed callbacks
2217 * as the request and rctx are no longer viable.
2218 */
2219 case TRUNK_REQUEST_STATE_PARTIAL:
2220 case TRUNK_REQUEST_STATE_SENT:
2221 {
2222 trunk_connection_t *tconn = treq->pub.tconn;
2223
2224 /*
2225 * Don't allow connection state changes
2226 */
2227 connection_signals_pause(tconn->pub.conn);
2228 trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_SIGNAL);
2229 if (!fr_cond_assert_msg(treq->pub.state == TRUNK_REQUEST_STATE_CANCEL,
2230 "Bad state %s after cancellation",
2231 fr_table_str_by_value(trunk_request_states, treq->pub.state, "<INVALID>"))) {
2232 connection_signals_resume(tconn->pub.conn);
2233 return;
2234 }
2235 /*
2236 * No cancel muxer. We're done.
2237 *
2238 * If we do have a cancel mux function,
2239 * the next time this connection becomes
2240 * writable, we'll call the cancel mux
2241 * function.
2242 *
2243 * We don't run the complete or failed
2244 * callbacks here as the request is
2245 * being cancelled.
2246 */
2247 if (!trunk->funcs.request_cancel_mux) {
2248 trunk_request_enter_unassigned(treq);
2249 trunk_request_free(&treq);
2250 }
2251 connection_signals_resume(tconn->pub.conn);
2252 }
2253 break;
2254
2255 /*
2256 * We're already in the process of cancelling a
2257 * request, so ignore duplicate signals.
2258 */
2259 case TRUNK_REQUEST_STATE_CANCEL:
2260 case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
2261 case TRUNK_REQUEST_STATE_CANCEL_SENT:
2262 case TRUNK_REQUEST_STATE_CANCEL_COMPLETE:
2263 break;
2264
2265 /*
2266 * For any other state, we just release the request
2267 * from its current connection and free it.
2268 */
2269 default:
2270 trunk_request_enter_unassigned(treq);
2271 trunk_request_free(&treq);
2272 break;
2273 }
2274}
2275
2276/** Signal a partial cancel write
2277 *
2278 * Where there's high load, and the outbound write buffer is full
2279 *
2280 * @param[in] treq to signal state change for.
2281 */
2282void trunk_request_signal_cancel_partial(trunk_request_t *treq)
2283{
2284 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2285
2286 if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
2287 "%s can only be called from within request_cancel_mux handler", __FUNCTION__)) return;
2288
2289 switch (treq->pub.state) {
2290 case TRUNK_REQUEST_STATE_CANCEL:
2291 trunk_request_enter_cancel_partial(treq);
2292 break;
2293
2294 default:
2295 return;
2296 }
2297}
2298
2299/** Signal that a remote server has been notified of the cancellation
2300 *
2301 * Called from request_cancel_mux to indicate that the datastore has been informed
2302 * that the response is no longer needed.
2303 *
2304 * @param[in] treq to signal state change for.
2305 */
2306void trunk_request_signal_cancel_sent(trunk_request_t *treq)
2307{
2308 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2309
2310 if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
2311 "%s can only be called from within request_cancel_mux handler", __FUNCTION__)) return;
2312
2313 switch (treq->pub.state) {
2314 case TRUNK_REQUEST_STATE_CANCEL:
2315 case TRUNK_REQUEST_STATE_CANCEL_PARTIAL:
2316 trunk_request_enter_cancel_sent(treq);
2317 break;
2318
2319 default:
2320 break;
2321 }
2322}
2323
2324/** Signal that a remote server acked our cancellation
2325 *
2326 * Called from request_demux to indicate that it got an ack for the cancellation.
2327 *
2328 * @param[in] treq to signal state change for.
2329 */
2330void trunk_request_signal_cancel_complete(trunk_request_t *treq)
2331{
2332 if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return;
2333
2334 if (!fr_cond_assert_msg(IN_REQUEST_DEMUX(treq->pub.trunk) || IN_REQUEST_CANCEL_MUX(treq->pub.trunk),
2335 "%s can only be called from within request_demux or request_cancel_mux handlers",
2336 __FUNCTION__)) return;
2337
2338 switch (treq->pub.state) {
2339 case TRUNK_REQUEST_STATE_CANCEL_SENT:
2340 /*
2341 * This is allowed, as we may not need to wait
2342 * for the database to ACK our cancellation
2343 * request.
2344 *
2345 * Note: TRUNK_REQUEST_STATE_CANCEL_PARTIAL
2346 * is not allowed here, as that'd mean we'd half
2347 * written the cancellation request out to the
2348 * socket, and then decided to abandon it.
2349 *
2350 * That'd leave the socket in an unusable state.
2351 */
2352 case TRUNK_REQUEST_STATE_CANCEL:
2353 trunk_request_enter_cancel_complete(treq);
2354 break;
2355
2356 default:
2357 break;
2358 }
2359}
2360
2361/** If the trunk request is freed then update the target requests
2362 *
2363 * gperftools showed calling the request free function directly was slightly faster
2364 * than using talloc_free.
2365 *
2366 * @param[in] treq_to_free request.
2367 */
2368void trunk_request_free(trunk_request_t **treq_to_free)
2369{
2370 trunk_request_t *treq = *treq_to_free;
2371 trunk_t *trunk;
2372
2373 if (unlikely(!treq)) return;
2374
2375 trunk = treq->pub.trunk;
2376
2377 /*
2378 * The only valid states a trunk request can be
2379 * freed from.
2380 */
2381 switch (treq->pub.state) {
2382 case TRUNK_REQUEST_STATE_INIT:
2383 case TRUNK_REQUEST_STATE_UNASSIGNED:
2384 case TRUNK_REQUEST_STATE_COMPLETE:
2385 case TRUNK_REQUEST_STATE_FAILED:
2386 case TRUNK_REQUEST_STATE_CANCEL_COMPLETE:
2387 break;
2388
2389 default:
2390 if (!fr_cond_assert(0)) return;
2391 }
2392
2393 /*
2394 * Zero out the pointer to prevent double frees
2395 */
2396 *treq_to_free = NULL;
2397
2398 /*
2399 * Call the API client callback to free
2400 * any associated memory.
2401 */
2402 DO_REQUEST_FREE(treq);
2403
2404 /*
2405 * Update the last above/below target stats
2406 * We only do this when we alloc or free
2407 * connections, or on connection
2408 * state changes.
2409 */
2410 trunk_requests_per_connection(NULL, NULL, treq->pub.trunk, fr_time(), false);
2411
2412 /*
2413 * This tracks the total number of requests
2414 * allocated and not freed or returned to
2415 * the free list.
2416 */
2417 if (fr_cond_assert(trunk->pub.req_alloc > 0)) trunk->pub.req_alloc--;
2418
2419 /*
2420 * No cleanup delay, means cleanup immediately
2421 */
2422 if (!fr_time_delta_ispos(trunk->conf.req_cleanup_delay)) {
2423 treq->pub.state = TRUNK_REQUEST_STATE_INIT;
2424
2425#ifndef NDEBUG
2426 /*
2427 * Ensure anything parented off the treq
2428 * is freed. We do this to trigger
2429 * the destructors for the log entries.
2430 */
2431 talloc_free_children(treq);
2432
2433 /*
2434 * State log should now be empty as entries
2435 * remove themselves from the dlist
2436 * on free.
2437 */
2438 fr_assert_msg(fr_dlist_num_elements(&treq->log) == 0,
2439 "Should have 0 remaining log entries, have %u", fr_dlist_num_elements(&treq->log));
2440#endif
2441
2442 talloc_free(treq);
2443 return;
2444 }
2445
2446 /*
2447 * Ensure anything parented off the treq
2448 * is freed.
2449 */
2450 talloc_free_children(treq);
2451
2452#ifndef NDEBUG
2453 /*
2454 * State log should now be empty as entries
2455 * remove themselves from the dlist
2456 * on free.
2457 */
2458 fr_assert_msg(fr_dlist_num_elements(&treq->log) == 0,
2459 "Should have 0 remaining log entries, have %u", fr_dlist_num_elements(&treq->log));
2460#endif
2461
2462 /*
2463 *
2464 * Return the trunk request back to the init state.
2465 */
2466 *treq = (trunk_request_t){
2467 .pub = {
2468 .state = TRUNK_REQUEST_STATE_INIT,
2469 .trunk = treq->pub.trunk,
2470 },
2471 .cancel_reason = TRUNK_CANCEL_REASON_NONE,
2472 .last_freed = fr_time(),
2473#ifndef NDEBUG
2474 .log = treq->log /* Keep the list head, to save reinitialisation */
2475#endif
2476 };
2477
2478
2479 /*
2480 * Insert at the head, so that we can free
2481 * requests that have been unused for N
2482 * seconds from the tail.
2483 */
2484 trunk_list_free_requests_add(trunk, treq);
2485
2486}
2487
2488/** Actually free the trunk request
2489 *
2490 */
2491static int _trunk_request_free(trunk_request_t *treq)
2492{
2493 trunk_t *trunk = treq->pub.trunk;
2494
2495 switch (treq->pub.state) {
2496 case TRUNK_REQUEST_STATE_INIT:
2497 case TRUNK_REQUEST_STATE_UNASSIGNED:
2498 break;
2499
2500 default:
2501 fr_assert(0);
2502 break;
2503 }
2504
2505 trunk_list_free_requests_remove(trunk, treq);
2506
2507 return 0;
2508}
2509
2510/** (Pre-)Allocate a new trunk request
2511 *
2512 * If trunk->conf.req_pool_headers or trunk->conf.req_pool_size are not zero then the
2513 * request will be a talloc pool, which can be used to hold the preq.
2514 *
2515 * @note Do not use MEM to check the result of this allocated as it may fail for
2516 * non-fatal reasons.
2517 *
2518 * @param[in] trunk to add request to.
2519 * @param[in] request to wrap in a trunk request (treq).
2520 * @return
2521 * - A newly allocated request.
2522 * - NULL if too many requests are allocated.
2523 */
2524trunk_request_t *trunk_request_alloc(trunk_t *trunk, request_t *request)
2525{
2526 trunk_request_t *treq;
2527
2528 /*
2529 * The number of treqs currently allocated
2530 * exceeds the maximum number allowed.
2531 */
2532 if (trunk->conf.max_req_per_conn && trunk->conf.max) {
2533 uint64_t limit;
2534
2535 limit = (uint64_t) trunk->conf.max_req_per_conn * trunk->conf.max;
2536 if (trunk->pub.req_alloc >= (limit + trunk->conf.max_backlog)) {
2537 RATE_LIMIT_LOCAL_ROPTIONAL(&trunk->limit_max_requests_alloc_log,
2538 RWARN, WARN, "Refusing to alloc requests - "
2539 "Limit of %"PRIu64" (max = %u * per_connection_max = %u) "
2540 "plus %u backlog requests reached",
2541 limit, trunk->conf.max, trunk->conf.max_req_per_conn,
2542 trunk->conf.max_backlog);
2543 return NULL;
2544 }
2545 }
2546
2547 /*
2548 * Re-use a recently freed request, which might have some
2549 * better cache locality than getting a request from the tail.
2550 *
2551 * If we can't do that, just allocate a new one.
2552 */
2553 treq = trunk_list_free_requests_pop(trunk);
2554 if (treq) {
2555 fr_assert(treq->pub.state == TRUNK_REQUEST_STATE_INIT);
2556 fr_assert(treq->pub.trunk == trunk);
2557 fr_assert(treq->pub.tconn == NULL);
2558 fr_assert(treq->cancel_reason == TRUNK_CANCEL_REASON_NONE);
2559 fr_assert(fr_time_gt(treq->last_freed, fr_time_wrap(0)));
2560 trunk->pub.req_alloc_reused++;
2561 } else {
2562 MEM(treq = talloc_pooled_object(trunk, trunk_request_t,
2563 trunk->conf.req_pool_headers, trunk->conf.req_pool_size));
2564 talloc_set_destructor(treq, _trunk_request_free);
2565
2566 *treq = (trunk_request_t){
2567 .pub = {
2568 .state = TRUNK_REQUEST_STATE_INIT,
2569 .trunk = trunk
2570 },
2571 .cancel_reason = TRUNK_CANCEL_REASON_NONE
2572 };
2573 trunk->pub.req_alloc_new++;
2574#ifndef NDEBUG
2575 fr_dlist_init(&treq->log, trunk_request_state_log_t, entry);
2576#endif
2577 }
2578
2579 trunk->pub.req_alloc++;
2580 treq->id = atomic_fetch_add_explicit(&request_counter, 1, memory_order_relaxed);
2581 /* heap_id - initialised when treq inserted into pending */
2582 /* list - empty */
2583 /* preq - populated later */
2584 /* rctx - populated later */
2585 treq->pub.request = request;
2586
2587 return treq;
2588}
2589
2590/** Enqueue a request that needs data written to the trunk
2591 *
2592 * When a request_t * needs to make an asynchronous request to an external datastore
2593 * it should call this function, specifying a preq (protocol request) containing
2594 * the data necessary to request information from the external datastore, and an
2595 * rctx (resume ctx) used to hold the decoded response and/or any error codes.
2596 *
2597 * After a treq is successfully enqueued it will either be assigned immediately
2598 * to the pending queue of a connection, or if no connections are available,
2599 * (depending on the trunk configuration) the treq will be placed in the trunk's
2600 * global backlog.
2601 *
2602 * After receiving a positive return code from this function the caller should
2603 * immediately yield, to allow the various timers and I/O handlers that drive tconn
2604 * (trunk connection) and treq state changes to be called.
2605 *
2606 * When a tconn becomes writable (or the trunk is configured to be always writable)
2607 * the #trunk_request_mux_t callback will be called to dequeue, encode and
2608 * send any pending requests for that tconn. The #trunk_request_mux_t callback
2609 * is also responsible for tracking the outbound requests to allow the
2610 * #trunk_request_demux_t callback to match inbound responses with the original
2611 * treq. Once the #trunk_request_mux_t callback is done processing the treq
2612 * it signals what state the treq should enter next using one of the
2613 * trunk_request_signal_* functions.
2614 *
2615 * When a tconn becomes readable the user specified #trunk_request_demux_t
2616 * callback is called to process any responses, match them with the original treq.
2617 * and signal what state they should enter next using one of the
2618 * trunk_request_signal_* functions.
2619 *
2620 * @param[in,out] treq_out A trunk request handle. If the memory pointed to
2621 * is NULL, a new treq will be allocated.
2622 * Otherwise treq should point to memory allocated
2623 * with trunk_request_alloc.
2624 * @param[in] trunk to enqueue request on.
2625 * @param[in] request to enqueue.
2626 * @param[in] preq Protocol request to write out. Will be freed when
2627 * treq is freed. Should ideally be parented by the
2628 * treq if possible.
2629 * Use #trunk_request_alloc for pre-allocation of
2630 * the treq.
2631 * @param[in] rctx The resume context to write any result to.
2632 * @return
2633 * - TRUNK_ENQUEUE_OK.
2634 * - TRUNK_ENQUEUE_IN_BACKLOG.
2635 * - TRUNK_ENQUEUE_NO_CAPACITY.
2636 * - TRUNK_ENQUEUE_DST_UNAVAILABLE
2637 * - TRUNK_ENQUEUE_FAIL
2638 */
2639trunk_enqueue_t trunk_request_enqueue(trunk_request_t **treq_out, trunk_t *trunk,
2640 request_t *request, void *preq, void *rctx)
2641{
2642 trunk_connection_t *tconn = NULL;
2643 trunk_request_t *treq;
2644 trunk_enqueue_t ret;
2645
2646 if (!fr_cond_assert_msg(!IN_HANDLER(trunk),
2647 "%s cannot be called within a handler", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;
2648
2649 if (!fr_cond_assert_msg(!*treq_out || ((*treq_out)->pub.state == TRUNK_REQUEST_STATE_INIT),
2650 "%s requests must be in \"init\" state", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;
2651
2652 /*
2653 * If delay_start was set, we may need
2654 * to insert the timer for the connection manager.
2655 */
2656 if (unlikely(!trunk->started)) {
2657 if (trunk_start(trunk) < 0) return TRUNK_ENQUEUE_FAIL;
2658 }
2659
2660 ret = trunk_request_check_enqueue(&tconn, trunk, request);
2661 switch (ret) {
2662 case TRUNK_ENQUEUE_OK:
2663 if (*treq_out) {
2664 treq = *treq_out;
2665 } else {
2666 *treq_out = treq = trunk_request_alloc(trunk, request);
2667 if (!treq) return TRUNK_ENQUEUE_NO_CAPACITY;
2668 }
2669 treq->pub.preq = preq;
2670 treq->pub.rctx = rctx;
2671 if (trunk->conf.always_writable) {
2672 connection_signals_pause(tconn->pub.conn);
2673 trunk_request_enter_pending(treq, tconn, true);
2674 trunk_connection_writable(tconn);
2675 connection_signals_resume(tconn->pub.conn);
2676 } else {
2677 trunk_request_enter_pending(treq, tconn, true);
2678 }
2679 break;
2680
2681 case TRUNK_ENQUEUE_IN_BACKLOG:
2682 if (*treq_out) {
2683 treq = *treq_out;
2684 } else {
2685 *treq_out = treq = trunk_request_alloc(trunk, request);
2686 if (!treq) return TRUNK_ENQUEUE_NO_CAPACITY;
2687 }
2688 treq->pub.preq = preq;
2689 treq->pub.rctx = rctx;
2690 trunk_request_enter_backlog(treq, true);
2691 break;
2692
2693 default:
2694 /*
2695 * If a trunk request was provided
2696 * populate the preq and rctx fields
2697 * so that if it's freed with
2698 * trunk_request_free, the free
2699 * function works as intended.
2700 */
2701 if (*treq_out) {
2702 treq = *treq_out;
2703 treq->pub.preq = preq;
2704 treq->pub.rctx = rctx;
2705 }
2706 return ret;
2707 }
2708
2709 return ret;
2710}
2711
2712/** Re-enqueue a request on the same connection
2713 *
2714 * If the treq has been sent, we assume that we're being signalled to requeue
2715 * because something outside of the trunk API has determined that a retransmission
2716 * is required. The easiest way to perform that retransmission is to clean up
2717 * any tracking information for the request, and the requeue it for transmission.
2718 *
2719 * IF re-queueing fails, the request will enter the fail state. It should not be
2720 * accessed if this occurs.
2721 *
2722 * @param[in] treq to requeue (retransmit).
2723 * @return
2724 * - TRUNK_ENQUEUE_OK.
2725 * - TRUNK_ENQUEUE_DST_UNAVAILABLE - Connection cannot service requests.
2726 * - TRUNK_ENQUEUE_FAIL - Request isn't in a valid state to be reassigned.
2727 */
2728trunk_enqueue_t trunk_request_requeue(trunk_request_t *treq)
2729{
2730 trunk_connection_t *tconn = treq->pub.tconn; /* Existing conn */
2731
2732 if (!tconn) return TRUNK_ENQUEUE_FAIL;
2733
2734 if (!IS_PROCESSING(tconn)) {
2735 trunk_request_enter_failed(treq);
2736 return TRUNK_ENQUEUE_DST_UNAVAILABLE;
2737 }
2738
2739 switch (treq->pub.state) {
2740 case TRUNK_REQUEST_STATE_PARTIAL:
2741 case TRUNK_REQUEST_STATE_SENT:
2742 case TRUNK_REQUEST_STATE_REAPABLE:
2743 connection_signals_pause(tconn->pub.conn);
2744 trunk_request_enter_cancel(treq, TRUNK_CANCEL_REASON_REQUEUE);
2745 trunk_request_enter_pending(treq, tconn, false);
2746 if (treq->pub.trunk->conf.always_writable) {
2747 trunk_connection_writable(tconn);
2748 }
2749 connection_signals_resume(tconn->pub.conn);
2750 break;
2751
2752 case TRUNK_REQUEST_STATE_BACKLOG: /* Do nothing.... */
2753 case TRUNK_REQUEST_STATE_PENDING: /* Do nothing.... */
2754 break;
2755
2756 default:
2757 trunk_request_enter_failed(treq);
2758 return TRUNK_ENQUEUE_FAIL;
2759 }
2760
2761 return TRUNK_ENQUEUE_OK;
2762}
2763
2764/** Enqueue additional requests on a specific connection
2765 *
2766 * This may be used to create a series of requests on a single connection, or to generate
2767 * in-band status checks.
2768 *
2769 * @note If conf->always_writable, then the muxer will be called immediately. The caller
2770 * must be able to handle multiple calls to its muxer gracefully.
2771 *
2772 * @param[in,out] treq_out A trunk request handle. If the memory pointed to
2773 * is NULL, a new treq will be allocated.
2774 * Otherwise treq should point to memory allocated
2775 * with trunk_request_alloc.
2776 * @param[in] tconn to enqueue request on.
2777 * @param[in] request to enqueue.
2778 * @param[in] preq Protocol request to write out. Will be freed when
2779 * treq is freed. Should ideally be parented by the
2780 * treq if possible.
2781 * Use #trunk_request_alloc for pre-allocation of
2782 * the treq.
2783 * @param[in] rctx The resume context to write any result to.
2784 * @param[in] ignore_limits Ignore max_req_per_conn. Useful to force status
2785 * checks through even if the connection is at capacity.
2786 * Will also allow enqueuing on "inactive", "draining",
2787 * "draining-to-free" connections.
2788 * @return
2789 * - TRUNK_ENQUEUE_OK.
2790 * - TRUNK_ENQUEUE_NO_CAPACITY - At max_req_per_conn_limit
2791 * - TRUNK_ENQUEUE_DST_UNAVAILABLE - Connection cannot service requests.
2792 */
2793trunk_enqueue_t trunk_request_enqueue_on_conn(trunk_request_t **treq_out, trunk_connection_t *tconn,
2794 request_t *request, void *preq, void *rctx,
2795 bool ignore_limits)
2796{
2797 trunk_request_t *treq;
2798 trunk_t *trunk = tconn->pub.trunk;
2799
2800 if (!fr_cond_assert_msg(!*treq_out || ((*treq_out)->pub.state == TRUNK_REQUEST_STATE_INIT),
2801 "%s requests must be in \"init\" state", __FUNCTION__)) return TRUNK_ENQUEUE_FAIL;
2802
2803 if (!IS_SERVICEABLE(tconn)) return TRUNK_ENQUEUE_DST_UNAVAILABLE;
2804
2805 /*
2806 * Limits check
2807 */
2808 if (!ignore_limits) {
2809 if (trunk->conf.max_req_per_conn &&
2810 (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) >=
2811 trunk->conf.max_req_per_conn)) return TRUNK_ENQUEUE_NO_CAPACITY;
2812
2813 if (tconn->pub.state != TRUNK_CONN_ACTIVE) return TRUNK_ENQUEUE_NO_CAPACITY;
2814 }
2815
2816 if (*treq_out) {
2817 treq = *treq_out;
2818 } else {
2819 *treq_out = treq = trunk_request_alloc(trunk, request);
2820 if (!treq) return TRUNK_ENQUEUE_NO_CAPACITY;
2821 }
2822
2823 treq->pub.preq = preq;
2824 treq->pub.rctx = rctx;
2825 treq->bound_to_conn = true; /* Don't let the request be transferred */
2826
2827 if (trunk->conf.always_writable) {
2828 connection_signals_pause(tconn->pub.conn);
2829 trunk_request_enter_pending(treq, tconn, true);
2830 trunk_connection_writable(tconn);
2831 connection_signals_resume(tconn->pub.conn);
2832 } else {
2833 trunk_request_enter_pending(treq, tconn, true);
2834 }
2835
2836 return TRUNK_ENQUEUE_OK;
2837}
2838
2839#ifndef NDEBUG
2840/** Used for sanity checks to ensure all log entries have been freed
2841 *
2842 */
2843static int _state_log_entry_free(trunk_request_state_log_t *slog)
2844{
2845 fr_dlist_remove(slog->log_head, slog);
2846
2847 return 0;
2848}
2849
2850void trunk_request_state_log_entry_add(char const *function, int line,
2851 trunk_request_t *treq, trunk_request_state_t new)
2852{
2853 trunk_request_state_log_t *slog = NULL;
2854
2855 if (fr_dlist_num_elements(&treq->log) >= TRUNK_REQUEST_STATE_LOG_MAX) {
2856 slog = fr_dlist_head(&treq->log);
2857 fr_assert_msg(slog, "slog list head NULL but element counter was %u",
2858 fr_dlist_num_elements(&treq->log));
2859 (void)fr_dlist_remove(&treq->log, slog); /* Returns NULL when removing the list head */
2860 memset(slog, 0, sizeof(*slog));
2861 } else {
2862 MEM(slog = talloc_zero(treq, trunk_request_state_log_t));
2863 talloc_set_destructor(slog, _state_log_entry_free);
2864 }
2865
2866 slog->log_head = &treq->log;
2867 slog->from = treq->pub.state;
2868 slog->to = new;
2869 slog->function = function;
2870 slog->line = line;
2871 if (treq->pub.tconn) {
2872 slog->tconn = treq->pub.tconn;
2873 slog->tconn_id = treq->pub.tconn->pub.conn->id;
2874 slog->tconn_state = treq->pub.tconn->pub.state;
2875 }
2876
2877 fr_dlist_insert_tail(&treq->log, slog);
2878
2879}
2880
2881void trunk_request_state_log(fr_log_t const *log, fr_log_type_t log_type, char const *file, int line,
2882 trunk_request_t const *treq)
2883{
2884 trunk_request_state_log_t *slog = NULL;
2885
2886 int i;
2887
2888 for (slog = fr_dlist_head(&treq->log), i = 0;
2889 slog;
2890 slog = fr_dlist_next(&treq->log, slog), i++) {
2891 fr_log(log, log_type, file, line, "[%u] %s:%i - in conn %"PRIu64" in state %s - %s -> %s",
2892 i, slog->function, slog->line,
2893 slog->tconn_id,
2894 slog->tconn ? fr_table_str_by_value(trunk_connection_states,
2895 slog->tconn_state, "<INVALID>") : "none",
2896 fr_table_str_by_value(trunk_request_states, slog->from, "<INVALID>"),
2897 fr_table_str_by_value(trunk_request_states, slog->to, "<INVALID>"));
2898 }
2899}
2900#endif
2901
2902/** Return the count number of connections in the specified states
2903 *
2904 * @param[in] trunk to retrieve counts for.
2905 * @param[in] conn_state One or more #trunk_connection_state_t states or'd together.
2906 * @return The number of connections in the specified states.
2907 */
2908uint16_t trunk_connection_count_by_state(trunk_t *trunk, int conn_state)
2909{
2910 uint16_t count = 0;
2911
2912 if (conn_state & TRUNK_CONN_INIT) count += fr_dlist_num_elements(&trunk->init);
2913 if (conn_state & TRUNK_CONN_CONNECTING) count += fr_dlist_num_elements(&trunk->connecting);
2914 if (conn_state & TRUNK_CONN_ACTIVE) count += fr_minmax_heap_num_elements(trunk->active);
2915 if (conn_state & TRUNK_CONN_FULL) count += fr_dlist_num_elements(&trunk->full);
2916 if (conn_state & TRUNK_CONN_INACTIVE) count += fr_dlist_num_elements(&trunk->inactive);
2917 if (conn_state & TRUNK_CONN_INACTIVE_DRAINING) count += fr_dlist_num_elements(&trunk->inactive_draining);
2918 if (conn_state & TRUNK_CONN_CLOSED) count += fr_dlist_num_elements(&trunk->closed);
2919 if (conn_state & TRUNK_CONN_DRAINING) count += fr_dlist_num_elements(&trunk->draining);
2920 if (conn_state & TRUNK_CONN_DRAINING_TO_FREE) count += fr_dlist_num_elements(&trunk->draining_to_free);
2921
2922 return count;
2923}
2924
2925/** Return the count number of requests associated with a trunk connection
2926 *
2927 * @param[in] tconn to return request count for.
2928 * @param[in] req_state One or more request states or'd together.
2929 *
2930 * @return The number of requests in the specified states, associated with a tconn.
2931 */
2932uint32_t trunk_request_count_by_connection(trunk_connection_t const *tconn, int req_state)
2933{
2934 uint32_t count = 0;
2935
2936 if (req_state & TRUNK_REQUEST_STATE_PENDING) count += fr_heap_num_elements(tconn->pending);
2937 if (req_state & TRUNK_REQUEST_STATE_PARTIAL) count += tconn->partial ? 1 : 0;
2938 if (req_state & TRUNK_REQUEST_STATE_SENT) count += fr_dlist_num_elements(&tconn->sent);
2939 if (req_state & TRUNK_REQUEST_STATE_REAPABLE) count += fr_dlist_num_elements(&tconn->reapable);
2940 if (req_state & TRUNK_REQUEST_STATE_CANCEL) count += fr_dlist_num_elements(&tconn->cancel);
2941 if (req_state & TRUNK_REQUEST_STATE_CANCEL_PARTIAL) count += tconn->cancel_partial ? 1 : 0;
2942 if (req_state & TRUNK_REQUEST_STATE_CANCEL_SENT) count += fr_dlist_num_elements(&tconn->cancel_sent);
2943
2944 return count;
2945}
2946
2947/** Automatically mark a connection as full
2948 *
2949 * @param[in] tconn to potentially mark as full.
2950 */
2951static inline void trunk_connection_auto_full(trunk_connection_t *tconn)
2952{
2953 trunk_t *trunk = tconn->pub.trunk;
2954 uint32_t count;
2955
2956 if (tconn->pub.state != TRUNK_CONN_ACTIVE) return;
2957
2958 /*
2959 * Enforces max_req_per_conn
2960 */
2961 if (trunk->conf.max_req_per_conn > 0) {
2962 count = trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL);
2963 if (count >= trunk->conf.max_req_per_conn) trunk_connection_enter_full(tconn);
2964 }
2965}
2966
2967/** Return whether a trunk connection should currently be considered full
2968 *
2969 * @param[in] tconn to check.
2970 * @return
2971 * - true if the connection is full.
2972 * - false if the connection is not full.
2973 */
2974static inline bool trunk_connection_is_full(trunk_connection_t *tconn)
2975{
2976 trunk_t *trunk = tconn->pub.trunk;
2977 uint32_t count;
2978
2979 /*
2980 * Enforces max_req_per_conn
2981 */
2982 count = trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL);
2983 if ((trunk->conf.max_req_per_conn == 0) || (count < trunk->conf.max_req_per_conn)) return false;
2984
2985 return true;
2986}
2987
2988/** Automatically mark a connection as active or reconnect it
2989 *
2990 * @param[in] tconn to potentially mark as active or reconnect.
2991 */
2992static inline void trunk_connection_auto_unfull(trunk_connection_t *tconn)
2993{
2994 if (tconn->pub.state != TRUNK_CONN_FULL) return;
2995
2996 /*
2997 * Enforces max_req_per_conn
2998 */
2999 if (!trunk_connection_is_full(tconn)) trunk_connection_enter_active(tconn);
3000}
3001
3002/** A connection is readable. Call the request_demux function to read pending requests
3003 *
3004 */
3005static inline void trunk_connection_readable(trunk_connection_t *tconn)
3006{
3007 trunk_t *trunk = tconn->pub.trunk;
3008
3009 DO_REQUEST_DEMUX(tconn);
3010}
3011
3012/** A connection is writable. Call the request_mux function to write pending requests
3013 *
3014 */
3015static inline void trunk_connection_writable(trunk_connection_t *tconn)
3016{
3017 trunk_t *trunk = tconn->pub.trunk;
3018
3019 /*
3020 * Call the cancel_sent function (if we have one)
3021 * to inform a backend datastore we no longer
3022 * care about the result
3023 */
3024 if (trunk->funcs.request_cancel_mux && trunk_request_count_by_connection(tconn,
3025 TRUNK_REQUEST_STATE_CANCEL |
3026 TRUNK_REQUEST_STATE_CANCEL_PARTIAL)) {
3027 DO_REQUEST_CANCEL_MUX(tconn);
3028 }
3029 if (!trunk_request_count_by_connection(tconn,
3030 TRUNK_REQUEST_STATE_PENDING |
3031 TRUNK_REQUEST_STATE_PARTIAL)) return;
3032 DO_REQUEST_MUX(tconn);
3033}
3034
3035/** Update the registrations for I/O events we're interested in
3036 *
3037 */
3038static void trunk_connection_event_update(trunk_connection_t *tconn)
3039{
3040 trunk_t *trunk = tconn->pub.trunk;
3041 trunk_connection_event_t events = TRUNK_CONN_EVENT_NONE;
3042
3043 switch (tconn->pub.state) {
3044 /*
3045 * We only register I/O events if the trunk connection is
3046 * in one of these states.
3047 *
3048 * For the other states the trunk shouldn't be processing
3049 * requests.
3050 */
3051 case TRUNK_CONN_ACTIVE:
3052 case TRUNK_CONN_FULL:
3053 case TRUNK_CONN_INACTIVE:
3054 case TRUNK_CONN_INACTIVE_DRAINING:
3055 case TRUNK_CONN_DRAINING:
3056 case TRUNK_CONN_DRAINING_TO_FREE:
3057 /*
3058 * If the connection is always writable,
3059 * then we don't care about write events.
3060 */
3061 if (!trunk->conf.always_writable &&
3062 trunk_request_count_by_connection(tconn,
3063 TRUNK_REQUEST_STATE_PARTIAL |
3064 TRUNK_REQUEST_STATE_PENDING |
3065 (trunk->funcs.request_cancel_mux ?
3066 TRUNK_REQUEST_STATE_CANCEL |
3067 TRUNK_REQUEST_STATE_CANCEL_PARTIAL : 0)) > 0) {
3068 events |= TRUNK_CONN_EVENT_WRITE;
3069 }
3070
3071 if (trunk_request_count_by_connection(tconn,
3072 TRUNK_REQUEST_STATE_SENT |
3073 (trunk->funcs.request_cancel_mux ?
3074 TRUNK_REQUEST_STATE_CANCEL_SENT : 0)) > 0) {
3075 events |= TRUNK_CONN_EVENT_READ;
3076 }
3077 break;
3078
3079 default:
3080 break;
3081 }
3082
3083 if (tconn->events != events) {
3084 /*
3085 * There may be a fatal error which results
3086 * in the connection being freed.
3087 *
3088 * Stop that from happening until after
3089 * we're done using it.
3090 */
3091 connection_signals_pause(tconn->pub.conn);
3092 DO_CONNECTION_NOTIFY(tconn, events);
3093 tconn->events = events;
3094 connection_signals_resume(tconn->pub.conn);
3095 }
3096}
3097
3098/** Remove a trunk connection from whichever list it's currently in
3099 *
3100 * @param[in] tconn to remove.
3101 */
3102static void trunk_connection_remove(trunk_connection_t *tconn)
3103{
3104 trunk_t *trunk = tconn->pub.trunk;
3105
3106 switch (tconn->pub.state) {
3107 case TRUNK_CONN_ACTIVE:
3108 {
3109 int ret;
3110
3111 ret = fr_minmax_heap_extract(trunk->active, tconn);
3112 if (!fr_cond_assert_msg(ret == 0, "Failed extracting conn from active heap: %s", fr_strerror())) return;
3113 }
3114 return;
3115
3116 case TRUNK_CONN_INIT:
3117 fr_dlist_remove(&trunk->init, tconn);
3118 break;
3119
3120 case TRUNK_CONN_CONNECTING:
3121 fr_dlist_remove(&trunk->connecting, tconn);
3122 return;
3123
3124 case TRUNK_CONN_CLOSED:
3125 fr_dlist_remove(&trunk->closed, tconn);
3126 return;
3127
3128 case TRUNK_CONN_FULL:
3129 trunk_list_full_remove(trunk, tconn);
3130 return;
3131
3132 case TRUNK_CONN_INACTIVE:
3133 trunk_list_inactive_remove(trunk, tconn);
3134 return;
3135
3136 case TRUNK_CONN_INACTIVE_DRAINING:
3137 trunk_list_inactive_draining_remove(trunk, tconn);
3138 return;
3139
3140 case TRUNK_CONN_DRAINING:
3141 trunk_list_draining_remove(trunk, tconn);
3142 return;
3143
3144 case TRUNK_CONN_DRAINING_TO_FREE:
3145 fr_dlist_remove(&trunk->draining_to_free, tconn);
3146 return;
3147
3148 case TRUNK_CONN_HALTED:
3149 return;
3150 }
3151}
3152
3153/** Transition a connection to the full state
3154 *
3155 * Called whenever a trunk connection is at the maximum number of requests.
3156 * Removes the connection from the connected heap, and places it in the full list.
3157 */
3158static void trunk_connection_enter_full(trunk_connection_t *tconn)
3159{
3160 trunk_t *trunk = tconn->pub.trunk;
3161
3162 switch (tconn->pub.state) {
3163 case TRUNK_CONN_ACTIVE:
3164 trunk_connection_remove(tconn);
3165 break;
3166
3167 default:
3168 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_FULL);
3169 }
3170
3171 trunk_list_full_add(trunk, tconn);
3172 CONN_STATE_TRANSITION(TRUNK_CONN_FULL, DEBUG2);
3173}
3174
3175/** Transition a connection to the inactive state
3176 *
3177 * Called whenever the API client wants to stop new requests being enqueued
3178 * on a trunk connection.
3179 */
3180static void trunk_connection_enter_inactive(trunk_connection_t *tconn)
3181{
3182 trunk_t *trunk = tconn->pub.trunk;
3183
3184 switch (tconn->pub.state) {
3185 case TRUNK_CONN_ACTIVE:
3186 case TRUNK_CONN_FULL:
3187 trunk_connection_remove(tconn);
3188 break;
3189
3190 default:
3191 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INACTIVE);
3192 }
3193
3194 trunk_list_inactive_add(trunk, tconn);
3195 CONN_STATE_TRANSITION(TRUNK_CONN_INACTIVE, DEBUG2);
3196}
3197
3198/** Transition a connection to the inactive-draining state
3199 *
3200 * Called whenever the trunk manager wants to drain an inactive connection
3201 * of its requests.
3202 */
3203static void trunk_connection_enter_inactive_draining(trunk_connection_t *tconn)
3204{
3205 trunk_t *trunk = tconn->pub.trunk;
3206
3207 switch (tconn->pub.state) {
3208 case TRUNK_CONN_INACTIVE:
3209 case TRUNK_CONN_DRAINING:
3210 trunk_connection_remove(tconn);
3211 break;
3212
3213 default:
3214 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INACTIVE_DRAINING);
3215 }
3216
3217 trunk_list_inactive_draining_add(trunk, tconn);
3218 CONN_STATE_TRANSITION(TRUNK_CONN_INACTIVE_DRAINING, INFO);
3219
3220 /*
3221 * Immediately re-enqueue all pending
3222 * requests, so the connection is drained
3223 * quicker.
3224 */
3225 trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
3226}
3227
3228/** Transition a connection to the draining state
3229 *
3230 * Removes the connection from the active heap so it won't be assigned any new
3231 * connections.
3232 */
3233static void trunk_connection_enter_draining(trunk_connection_t *tconn)
3234{
3235 trunk_t *trunk = tconn->pub.trunk;
3236
3237 switch (tconn->pub.state) {
3238 case TRUNK_CONN_ACTIVE:
3239 case TRUNK_CONN_FULL:
3240 case TRUNK_CONN_INACTIVE:
3241 case TRUNK_CONN_INACTIVE_DRAINING:
3242 trunk_connection_remove(tconn);
3243 break;
3244
3245 default:
3246 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING);
3247 }
3248
3249 trunk_list_draining_add(trunk, tconn);
3250 CONN_STATE_TRANSITION(TRUNK_CONN_DRAINING, INFO);
3251
3252 /*
3253 * Immediately re-enqueue all pending
3254 * requests, so the connection is drained
3255 * quicker.
3256 */
3257 trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
3258}
3259
3260/** Transition a connection to the draining-to-reconnect state
3261 *
3262 * Removes the connection from the active heap so it won't be assigned any new
3263 * connections.
3264 */
3265static void trunk_connection_enter_draining_to_free(trunk_connection_t *tconn)
3266{
3267 trunk_t *trunk = tconn->pub.trunk;
3268
3269 FR_TIMER_DISARM(tconn->lifetime_ev);
3270
3271 switch (tconn->pub.state) {
3272 case TRUNK_CONN_ACTIVE:
3273 case TRUNK_CONN_FULL:
3274 case TRUNK_CONN_INACTIVE:
3275 case TRUNK_CONN_INACTIVE_DRAINING:
3276 case TRUNK_CONN_DRAINING:
3277 trunk_connection_remove(tconn);
3278 break;
3279
3280 default:
3281 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE);
3282 }
3283
3284 fr_dlist_insert_head(&trunk->draining_to_free, tconn);
3285 CONN_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE, INFO);
3286
3287 /*
3288 * Immediately re-enqueue all pending
3289 * requests, so the connection is drained
3290 * quicker.
3291 */
3292 trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_PENDING, 0, false);
3293}
3294
3295
3296/** Transition a connection back to the active state
3297 *
3298 * This should only be called on a connection which is in the full state,
3299 * inactive state, draining state or connecting state.
3300 */
3301static void trunk_connection_enter_active(trunk_connection_t *tconn)
3302{
3303 trunk_t *trunk = tconn->pub.trunk;
3304 int ret;
3305
3306 switch (tconn->pub.state) {
3307 case TRUNK_CONN_FULL:
3308 case TRUNK_CONN_INACTIVE:
3309 case TRUNK_CONN_INACTIVE_DRAINING:
3310 case TRUNK_CONN_DRAINING:
3311 trunk_connection_remove(tconn);
3312 break;
3313
3314 case TRUNK_CONN_INIT:
3315 case TRUNK_CONN_CONNECTING:
3316 trunk_connection_remove(tconn);
3317 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_PENDING));
3318 break;
3319
3320 default:
3321 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_ACTIVE);
3322 }
3323
3324 ret = fr_minmax_heap_insert(trunk->active, tconn); /* re-insert into the active heap*/
3325 if (!fr_cond_assert_msg(ret == 0, "Failed inserting connection into active heap: %s", fr_strerror())) {
3326 trunk_connection_enter_inactive_draining(tconn);
3327 return;
3328 }
3329
3330 CONN_STATE_TRANSITION(TRUNK_CONN_ACTIVE, DEBUG2);
3331
3332 /*
3333 * Reorder the connections
3334 */
3335 CONN_REORDER(tconn);
3336
3337 /*
3338 * Rebalance requests
3339 */
3340 trunk_rebalance(trunk);
3341
3342 /*
3343 * We place requests into the backlog
3344 * because there were no connections
3345 * available to handle them.
3346 *
3347 * If a connection has become active
3348 * chances are those backlogged requests
3349 * can now be enqueued, so try and do
3350 * that now.
3351 *
3352 * If there's requests sitting in the
3353 * backlog indefinitely, it's because
3354 * they were inserted there erroneously
3355 * when there were active connections
3356 * which could have handled them.
3357 */
3358 trunk_backlog_drain(trunk);
3359}
3360
3361/** Connection transitioned to the init state
3362 *
3363 * Reflect the connection state change in the lists we use to track connections.
3364 *
3365 * @note This function is only called from the connection API as a watcher.
3366 *
3367 * @param[in] conn The connection which changes state.
3368 * @param[in] prev The connection is was in.
3369 * @param[in] state The connection is now in.
3370 * @param[in] uctx The trunk_connection_t wrapping the connection.
3371 */
3372static void _trunk_connection_on_init(UNUSED connection_t *conn,
3373 UNUSED connection_state_t prev,
3374 UNUSED connection_state_t state,
3375 void *uctx)
3376{
3377 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3378 trunk_t *trunk = tconn->pub.trunk;
3379
3380 switch (tconn->pub.state) {
3381 case TRUNK_CONN_HALTED:
3382 break;
3383
3384 case TRUNK_CONN_CLOSED:
3385 trunk_connection_remove(tconn);
3386 break;
3387
3388 default:
3389 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_INIT);
3390 }
3391
3392 fr_dlist_insert_head(&trunk->init, tconn);
3393 CONN_STATE_TRANSITION(TRUNK_CONN_INIT, DEBUG2);
3394}
3395
3396/** Connection transitioned to the connecting state
3397 *
3398 * Reflect the connection state change in the lists we use to track connections.
3399 *
3400 * @note This function is only called from the connection API as a watcher.
3401 *
3402 * @param[in] conn The connection which changes state.
3403 * @param[in] prev The connection is was in.
3404 * @param[in] state The connection is now in.
3405 * @param[in] uctx The trunk_connection_t wrapping the connection.
3406 */
3407static void _trunk_connection_on_connecting(UNUSED connection_t *conn,
3408 UNUSED connection_state_t prev,
3409 UNUSED connection_state_t state,
3410 void *uctx)
3411{
3412 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3413 trunk_t *trunk = tconn->pub.trunk;
3414
3415 switch (tconn->pub.state) {
3416 case TRUNK_CONN_INIT:
3417 case TRUNK_CONN_CLOSED:
3418 trunk_connection_remove(tconn);
3419 break;
3420
3421 default:
3422 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_CONNECTING);
3423 }
3424
3425 /*
3426 * If a connection just entered the
3427 * connecting state, it should have
3428 * no requests associated with it.
3429 */
3430 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
3431
3432 fr_dlist_insert_head(&trunk->connecting, tconn); /* MUST remain a head insertion for reconnect logic */
3433 CONN_STATE_TRANSITION(TRUNK_CONN_CONNECTING, INFO);
3434}
3435
3436/** Connection transitioned to the shutdown state
3437 *
3438 * If we're not already in the draining-to-free state, transition there now.
3439 *
3440 * The idea is that if something signalled the connection to shutdown, we need
3441 * to reflect that by dequeuing any pending requests, not accepting new ones,
3442 * and waiting for the existing requests to complete.
3443 *
3444 * @note This function is only called from the connection API as a watcher.
3445 *
3446 * @param[in] conn The connection which changes state.
3447 * @param[in] prev The connection is was in.
3448 * @param[in] state The connection is now in.
3449 * @param[in] uctx The trunk_connection_t wrapping the connection.
3450 */
3451static void _trunk_connection_on_shutdown(UNUSED connection_t *conn,
3452 UNUSED connection_state_t prev,
3453 UNUSED connection_state_t state,
3454 void *uctx)
3455{
3456 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3457
3458 switch (tconn->pub.state) {
3459 case TRUNK_CONN_DRAINING_TO_FREE:
3460 /*
3461 * Shutdown from draining-to-free means no outstanding requests.
3462 * Now signal to halt.
3463 */
3464 connection_signal_halt(conn);
3465 return;
3466
3467 case TRUNK_CONN_ACTIVE: /* Transition to draining-to-free */
3468 case TRUNK_CONN_FULL:
3469 case TRUNK_CONN_INACTIVE:
3470 case TRUNK_CONN_INACTIVE_DRAINING:
3471 case TRUNK_CONN_DRAINING:
3472 break;
3473
3474 case TRUNK_CONN_INIT:
3475 case TRUNK_CONN_CONNECTING:
3476 case TRUNK_CONN_CLOSED:
3477 case TRUNK_CONN_HALTED:
3478 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_DRAINING_TO_FREE);
3479 }
3480
3481 trunk_connection_enter_draining_to_free(tconn);
3482}
3483
3484/** Trigger a reconnection of the trunk connection
3485 *
3486 * @param[in] tl timer list the timer was inserted into.
3487 * @param[in] now Current time.
3488 * @param[in] uctx The tconn.
3489 */
3490static void _trunk_connection_lifetime_expire(UNUSED fr_timer_list_t *tl, UNUSED fr_time_t now, void *uctx)
3491{
3492 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3493
3494 trunk_connection_enter_draining_to_free(tconn);
3495}
3496
3497/** Connection transitioned to the connected state
3498 *
3499 * Reflect the connection state change in the lists we use to track connections.
3500 *
3501 * @note This function is only called from the connection API as a watcher.
3502 *
3503 * @param[in] conn The connection which changes state.
3504 * @param[in] prev The connection is was in.
3505 * @param[in] state The connection is now in.
3506 * @param[in] uctx The trunk_connection_t wrapping the connection.
3507 */
3508static void _trunk_connection_on_connected(UNUSED connection_t *conn,
3509 UNUSED connection_state_t prev,
3510 UNUSED connection_state_t state,
3511 void *uctx)
3512{
3513 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3514 trunk_t *trunk = tconn->pub.trunk;
3515
3516 /*
3517 * If a connection was just connected, it should only
3518 * have a pending list of requests. This state is found
3519 * in the rlm_radius module, which starts a new trunk,
3520 * and then immediately enqueues a request onto it. The
3521 * alternative for rlm_radius is to keep it's own queue
3522 * of pending requests before the trunk is fully
3523 * initialized. And then enqueue them onto the trunk
3524 * when the trunk is connected.
3525 *
3526 * It's instead easier (and makes more sense) to allow
3527 * the trunk to accept packets into its queue. If there
3528 * are no connections within a period of time, then the
3529 * requests will retry, or will time out.
3530 */
3531 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_PENDING));
3532
3533 /*
3534 * Set here, as the active state can
3535 * be transitioned to from full and
3536 * draining too.
3537 */
3538 trunk->pub.last_connected = fr_time();
3539
3540 /*
3541 * Set last_write_success so that idle timeout checks will run
3542 * from when the connection has connected if they fire before
3543 * any requests are written, rather than from server start time.
3544 */
3545 tconn->pub.last_write_success = fr_time();
3546
3547 /*
3548 * Insert a timer to reconnect the
3549 * connection periodically.
3550 */
3551 if (fr_time_delta_ispos(trunk->conf.lifetime)) {
3552 if (fr_timer_in(tconn, trunk->el->tl, &tconn->lifetime_ev,
3553 trunk->conf.lifetime, false, _trunk_connection_lifetime_expire, tconn) < 0) {
3554 PERROR("Failed inserting connection reconnection timer event, halting connection");
3555 connection_signal_shutdown(tconn->pub.conn);
3556 return;
3557 }
3558 }
3559
3560 trunk_connection_enter_active(tconn);
3561}
3562
3563/** Connection failed after it was connected
3564 *
3565 * Reflect the connection state change in the lists we use to track connections.
3566 *
3567 * @note This function is only called from the connection API as a watcher.
3568 *
3569 * @param[in] conn The connection which changes state.
3570 * @param[in] prev The connection is was in.
3571 * @param[in] state The connection is now in.
3572 * @param[in] uctx The trunk_connection_t wrapping the connection.
3573 */
3574static void _trunk_connection_on_closed(UNUSED connection_t *conn,
3575 UNUSED connection_state_t prev,
3576 UNUSED connection_state_t state,
3577 void *uctx)
3578{
3579 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3580 trunk_t *trunk = tconn->pub.trunk;
3581 bool need_requeue = false;
3582
3583 switch (tconn->pub.state) {
3584 case TRUNK_CONN_ACTIVE:
3585 case TRUNK_CONN_FULL:
3586 case TRUNK_CONN_INACTIVE:
3587 case TRUNK_CONN_INACTIVE_DRAINING:
3588 case TRUNK_CONN_DRAINING:
3589 case TRUNK_CONN_DRAINING_TO_FREE:
3590 need_requeue = true;
3591 trunk_connection_remove(tconn);
3592 break;
3593
3594 case TRUNK_CONN_INIT: /* Initialisation failed */
3595 case TRUNK_CONN_CONNECTING:
3596 trunk_connection_remove(tconn);
3597 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
3598 break;
3599
3600 case TRUNK_CONN_CLOSED:
3601 case TRUNK_CONN_HALTED: /* Can't move backwards? */
3602 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_CLOSED);
3603 }
3604
3605 fr_dlist_insert_head(&trunk->closed, tconn); /* MUST remain a head insertion for reconnect logic */
3606 CONN_STATE_TRANSITION(TRUNK_CONN_CLOSED, INFO);
3607
3608 /*
3609 * Now *AFTER* the connection has been
3610 * removed from the active, pool
3611 * re-enqueue the requests.
3612 */
3613 if (need_requeue) trunk_connection_requests_requeue_priv(tconn, TRUNK_REQUEST_STATE_ALL, 0, true);
3614
3615 /*
3616 * There should be no requests left on this
3617 * connection. They should have all been
3618 * moved off or failed.
3619 */
3620 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
3621
3622 /*
3623 * Clear statistics and flags
3624 */
3625 tconn->sent_count = 0;
3626
3627 /*
3628 * Remove the reconnect event
3629 */
3630 if (fr_time_delta_ispos(trunk->conf.lifetime)) FR_TIMER_DELETE(&tconn->lifetime_ev);
3631
3632 /*
3633 * Remove the I/O events
3634 */
3635 trunk_connection_event_update(tconn);
3636}
3637
3638/** Connection failed
3639 *
3640 * @param[in] conn The connection which changes state.
3641 * @param[in] prev The connection is was in.
3642 * @param[in] state The connection is now in.
3643 * @param[in] uctx The trunk_connection_t wrapping the connection.
3644 */
3645static void _trunk_connection_on_failed(connection_t *conn,
3646 connection_state_t prev,
3647 connection_state_t state,
3648 void *uctx)
3649{
3650 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3651 trunk_t *trunk = tconn->pub.trunk;
3652
3653 /*
3654 * Need to set this first as it
3655 * determines whether requests are
3656 * re-queued or fail outright.
3657 */
3658 trunk->pub.last_failed = fr_time();
3659
3660 /*
3661 * Failed in the init state, transition the
3662 * connection to closed, else we get an
3663 * INIT -> INIT transition which triggers
3664 * an assert.
3665 */
3666 if (prev == CONNECTION_STATE_INIT) _trunk_connection_on_closed(conn, prev, state, uctx);
3667
3668 /*
3669 * See what the state of the trunk is
3670 * if there are no connections that could
3671 * potentially accept requests in the near
3672 * future, then fail all the requests in the
3673 * trunk backlog.
3674 */
3675 if ((prev == CONNECTION_STATE_CONNECTED) &&
3676 (trunk_connection_count_by_state(trunk,
3677 (TRUNK_CONN_ACTIVE |
3678 TRUNK_CONN_FULL |
3679 TRUNK_CONN_DRAINING)) == 0)) trunk_backlog_drain(trunk);
3680}
3681
3682/** Connection transitioned to the halted state
3683 *
3684 * Remove the connection remove all lists, as it's likely about to be freed.
3685 *
3686 * Setting the trunk back to the init state ensures that if the code is ever
3687 * refactored and #connection_signal_reconnect is used after a connection
3688 * is halted, then everything is maintained in a valid state.
3689 *
3690 * @note This function is only called from the connection API as a watcher.
3691 *
3692 * @param[in] conn The connection which changes state.
3693 * @param[in] prev The connection is was in.
3694 * @param[in] state The connection is now in.
3695 * @param[in] uctx The trunk_connection_t wrapping the connection.
3696 */
3697static void _trunk_connection_on_halted(UNUSED connection_t *conn,
3698 UNUSED connection_state_t prev,
3699 UNUSED connection_state_t state,
3700 void *uctx)
3701{
3702 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
3703 trunk_t *trunk = tconn->pub.trunk;
3704
3705 switch (tconn->pub.state) {
3706 case TRUNK_CONN_INIT:
3707 case TRUNK_CONN_CLOSED:
3708 trunk_connection_remove(tconn);
3709 break;
3710
3711 default:
3712 CONN_BAD_STATE_TRANSITION(TRUNK_CONN_HALTED);
3713 }
3714
3715 /*
3716 * It began life in the halted state,
3717 * and will end life in the halted state.
3718 */
3719 CONN_STATE_TRANSITION(TRUNK_CONN_HALTED, DEBUG2);
3720
3721 /*
3722 * There should be no requests left on this
3723 * connection. They should have all been
3724 * moved off or failed.
3725 */
3726 fr_assert(trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0);
3727
3728 /*
3729 * And free the connection...
3730 */
3731 if (trunk->in_handler) {
3732 /*
3733 * ...later.
3734 */
3735 fr_dlist_insert_tail(&trunk->to_free, tconn);
3736 return;
3737 }
3738 talloc_free(tconn);
3739}
3740
3741/** Free a connection
3742 *
3743 * Enforces orderly free order of children of the tconn
3744 */
3745static int _trunk_connection_free(trunk_connection_t *tconn)
3746{
3747 fr_assert(tconn->pub.state == TRUNK_CONN_HALTED);
3748 fr_assert(!fr_dlist_entry_in_list(&tconn->entry)); /* Should not be in a list */
3749
3750 /*
3751 * Loop over all the requests we gathered
3752 * and transition them to the failed state,
3753 * freeing them.
3754 *
3755 * Usually, requests will be re-queued when
3756 * a connection enters the closed state,
3757 * but in this case because the whole trunk
3758 * is being freed, we don't bother, and
3759 * just signal to the API client that the
3760 * requests failed.
3761 */
3762 if (tconn->pub.trunk->freeing) {
3763 fr_dlist_head_t to_fail;
3764 trunk_request_t *treq = NULL;
3765
3766 fr_dlist_talloc_init(&to_fail, trunk_request_t, entry);
3767
3768 /*
3769 * Remove requests from this connection
3770 */
3771 trunk_connection_requests_dequeue(&to_fail, tconn, TRUNK_REQUEST_STATE_ALL, 0);
3772 while ((treq = fr_dlist_next(&to_fail, treq))) {
3773 trunk_request_t *prev;
3774
3775 prev = fr_dlist_remove(&to_fail, treq);
3776 trunk_request_enter_failed(treq);
3777 treq = prev;
3778 }
3779 }
3780
3781 /*
3782 * Ensure we're not signalled by the connection
3783 * as it processes its backlog of state changes,
3784 * as we are about to be freed.
3785 */
3786 connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_INIT, _trunk_connection_on_init);
3787 connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTING, _trunk_connection_on_connecting);
3788 connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTED, _trunk_connection_on_connected);
3789 connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_CLOSED, _trunk_connection_on_closed);
3790 connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_SHUTDOWN, _trunk_connection_on_shutdown);
3791 connection_del_watch_pre(tconn->pub.conn, CONNECTION_STATE_FAILED, _trunk_connection_on_failed);
3792 connection_del_watch_post(tconn->pub.conn, CONNECTION_STATE_HALTED, _trunk_connection_on_halted);
3793
3794 /*
3795 * This may return -1, indicating the free was deferred
3796 * this is fine. It just means the conn will be freed
3797 * after all the handlers have exited.
3798 */
3799 (void)talloc_free(tconn->pub.conn);
3800 tconn->pub.conn = NULL;
3801
3802 return 0;
3803}
3804
3805/** Attempt to spawn a new connection
3806 *
3807 * Calls the API client's alloc() callback to create a new connection_t,
3808 * then inserts the connection into the 'connecting' list.
3809 *
3810 * @param[in] trunk to spawn connection in.
3811 * @param[in] now The current time.
3812 */
3813static int trunk_connection_spawn(trunk_t *trunk, fr_time_t now)
3814{
3815 trunk_connection_t *tconn;
3816
3817
3818 /*
3819 * Call the API client's callback to create
3820 * a new connection_t.
3821 */
3822 MEM(tconn = talloc_zero(trunk, trunk_connection_t));
3823 tconn->pub.trunk = trunk;
3824 tconn->pub.state = TRUNK_CONN_HALTED; /* All connections start in the halted state */
3825
3826 /*
3827 * Allocate a new connection_t or fail.
3828 */
3829 DO_CONNECTION_ALLOC(tconn);
3830
3831 MEM(tconn->pending = fr_heap_talloc_alloc(tconn, _trunk_request_prioritise, trunk_request_t, heap_id, 0));
3832 fr_dlist_talloc_init(&tconn->sent, trunk_request_t, entry);
3833 fr_dlist_talloc_init(&tconn->reapable, trunk_request_t, entry);
3834 fr_dlist_talloc_init(&tconn->cancel, trunk_request_t, entry);
3835 fr_dlist_talloc_init(&tconn->cancel_sent, trunk_request_t, entry);
3836
3837 /*
3838 * OK, we have the connection, now setup watch
3839 * points so we know when it changes state.
3840 *
3841 * This lets us automatically move the tconn
3842 * between the different lists in the trunk
3843 * with minimum extra code.
3844 */
3845 connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_INIT,
3846 _trunk_connection_on_init, false, tconn); /* Before init() has been called */
3847
3848 connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTING,
3849 _trunk_connection_on_connecting, false, tconn); /* After init() has been called */
3850
3851 connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_CONNECTED,
3852 _trunk_connection_on_connected, false, tconn); /* After open() has been called */
3853
3854 connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_CLOSED,
3855 _trunk_connection_on_closed, false, tconn); /* Before close() has been called */
3856
3857 connection_add_watch_pre(tconn->pub.conn, CONNECTION_STATE_FAILED,
3858 _trunk_connection_on_failed, false, tconn); /* Before failed() has been called */
3859
3860 connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_SHUTDOWN,
3861 _trunk_connection_on_shutdown, false, tconn); /* After shutdown() has been called */
3862
3863 connection_add_watch_post(tconn->pub.conn, CONNECTION_STATE_HALTED,
3864 _trunk_connection_on_halted, false, tconn); /* About to be freed */
3865
3866 talloc_set_destructor(tconn, _trunk_connection_free);
3867
3868 connection_signal_init(tconn->pub.conn); /* annnnd GO! */
3869
3870 trunk->pub.last_open = now;
3871
3872 return 0;
3873}
3874
3875/** Pop a cancellation request off a connection's cancellation queue
3876 *
3877 * The request we return is advanced by the request moving out of the
3878 * cancel state and into the cancel_sent or cancel_complete state.
3879 *
3880 * One of these signalling functions must be called after the request
3881 * has been popped:
3882 *
3883 * - #trunk_request_signal_cancel_sent
3884 * The remote datastore has been informed, but we need to wait for acknowledgement.
3885 * The #trunk_request_demux_t callback must handle the acks calling
3886 * #trunk_request_signal_cancel_complete when an ack is received.
3887 *
3888 * - #trunk_request_signal_cancel_complete
3889 * The request was cancelled and we don't need to wait, clean it up immediately.
3890 *
3891 * @param[out] treq_out to process
3892 * @param[in] tconn Connection to drain cancellation request from.
3893 * @return
3894 * - 1 if no more requests.
3895 * - 0 if a new request was written to treq_out.
3896 * - -1 if the connection was previously freed. Caller *MUST NOT* touch any
3897 * memory or requests associated with the connection.
3898 * - -2 if called outside of the cancel muxer.
3899 */
3900int trunk_connection_pop_cancellation(trunk_request_t **treq_out, trunk_connection_t *tconn)
3901{
3902 if (unlikely(tconn->pub.state == TRUNK_CONN_HALTED)) return -1;
3903
3904 if (!fr_cond_assert_msg(IN_REQUEST_CANCEL_MUX(tconn->pub.trunk),
3905 "%s can only be called from within request_cancel_mux handler",
3906 __FUNCTION__)) return -2;
3907
3908 *treq_out = tconn->cancel_partial ? tconn->cancel_partial : fr_dlist_head(&tconn->cancel);
3909 if (!*treq_out) return 1;
3910
3911 return 0;
3912}
3913
3914/** Pop a request off a connection's pending queue
3915 *
3916 * The request we return is advanced by the request moving out of the partial or
3917 * pending states, when the mux function signals us.
3918 *
3919 * If the same request is returned again and again, it means the muxer isn't actually
3920 * doing anything with the request we returned, and it's and error in the muxer code.
3921 *
3922 * One of these signalling functions must be used after the request has been popped:
3923 *
3924 * - #trunk_request_signal_complete
3925 * The request was completed. Either we got a synchronous response, or we knew the
3926 * response without contacting an external server (cache).
3927 *
3928 * - #trunk_request_signal_fail
3929 * Failed muxing the request due to a permanent issue, i.e. an invalid request.
3930 *
3931 * - #trunk_request_signal_partial
3932 * Wrote part of a request. This request will be returned on the next call to this
3933 * function so that the request_mux function can finish writing it. Only useful
3934 * for stream type connections. Datagram type connections cannot have partial
3935 * writes.
3936 *
3937 * - #trunk_request_signal_sent Successfully sent a request.
3938 *
3939 * @param[out] treq_out to process
3940 * @param[in] tconn to pop a request from.
3941 * @return
3942 * - 1 if no more requests.
3943 * - 0 if a new request was written to treq_out.
3944 * - -1 if the connection was previously freed. Caller *MUST NOT* touch any
3945 * memory or requests associated with the connection.
3946 * - -2 if called outside of the muxer.
3947 */
3948int trunk_connection_pop_request(trunk_request_t **treq_out, trunk_connection_t *tconn)
3949{
3950 if (unlikely(tconn->pub.state == TRUNK_CONN_HALTED)) return -1;
3951
3952 if (!fr_cond_assert_msg(IN_REQUEST_MUX(tconn->pub.trunk),
3953 "%s can only be called from within request_mux handler",
3954 __FUNCTION__)) return -2;
3955
3956 *treq_out = tconn->partial ? tconn->partial : fr_heap_peek(tconn->pending);
3957 if (!*treq_out) return 1;
3958
3959 return 0;
3960}
3961
3962/** Signal that a trunk connection is writable
3963 *
3964 * Should be called from the 'write' I/O handler to signal that requests can be enqueued.
3965 *
3966 * @param[in] tconn to signal.
3967 */
3968void trunk_connection_signal_writable(trunk_connection_t *tconn)
3969{
3970 trunk_t *trunk = tconn->pub.trunk;
3971
3972 if (!fr_cond_assert_msg(!IN_HANDLER(tconn->pub.trunk),
3973 "%s cannot be called within a handler", __FUNCTION__)) return;
3974
3975 DEBUG3("[%" PRIu64 "] Signalled writable", tconn->pub.conn->id);
3976
3977 trunk_connection_writable(tconn);
3978}
3979
3980/** Signal that a trunk connection is readable
3981 *
3982 * Should be called from the 'read' I/O handler to signal that requests should be dequeued.
3983 *
3984 * @param[in] tconn to signal.
3985 */
3986void trunk_connection_signal_readable(trunk_connection_t *tconn)
3987{
3988 trunk_t *trunk = tconn->pub.trunk;
3989
3990 if (!fr_cond_assert_msg(!IN_HANDLER(tconn->pub.trunk),
3991 "%s cannot be called within a handler", __FUNCTION__)) return;
3992
3993 DEBUG3("[%" PRIu64 "] Signalled readable", tconn->pub.conn->id);
3994
3995 trunk_connection_readable(tconn);
3996}
3997
3998/** Signal a trunk connection cannot accept more requests
3999 *
4000 * @param[in] tconn to signal.
4001 */
4002void trunk_connection_signal_inactive(trunk_connection_t *tconn)
4003{
4004 /* Can be called anywhere */
4005
4006 switch (tconn->pub.state) {
4007 case TRUNK_CONN_ACTIVE:
4008 case TRUNK_CONN_FULL:
4009 trunk_connection_enter_inactive(tconn);
4010 break;
4011
4012 case TRUNK_CONN_DRAINING:
4013 trunk_connection_enter_inactive_draining(tconn);
4014 break;
4015
4016 default:
4017 return;
4018 }
4019}
4020
4021/** Signal a trunk connection is no longer full
4022 *
4023 * @param[in] tconn to signal.
4024 */
4025void trunk_connection_signal_active(trunk_connection_t *tconn)
4026{
4027 switch (tconn->pub.state) {
4028 case TRUNK_CONN_FULL:
4029 trunk_connection_auto_unfull(tconn); /* Mark as active if it should be active */
4030 break;
4031
4032 case TRUNK_CONN_INACTIVE:
4033 /*
4034 * Do the appropriate state transition based on
4035 * how many requests the trunk connection is
4036 * currently servicing.
4037 */
4038 if (trunk_connection_is_full(tconn)) {
4039 trunk_connection_enter_full(tconn);
4040 break;
4041 }
4042 trunk_connection_enter_active(tconn);
4043 break;
4044
4045 /*
4046 * Unsetting the active flag just moves
4047 * the connection back to the normal
4048 * draining state.
4049 */
4050 case TRUNK_CONN_INACTIVE_DRAINING: /* Only an external signal can trigger this transition */
4051 trunk_connection_enter_draining(tconn);
4052 break;
4053
4054 default:
4055 return;
4056 }
4057}
4058
4059/** Signal a trunk connection is no longer viable
4060 *
4061 * @param[in] tconn to signal.
4062 * @param[in] reason the connection is being reconnected.
4063 */
4068
4069/** Standard I/O read function
4070 *
4071 * Underlying FD in now readable, so call the trunk to read any pending requests
4072 * from this connection.
4073 *
4074 * @param[in] el The event list signalling.
4075 * @param[in] fd that's now readable.
4076 * @param[in] flags describing the read event.
4077 * @param[in] uctx The trunk connection handle (tconn).
4078 */
4079void trunk_connection_callback_readable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
4080{
4081 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
4082
4083 trunk_connection_signal_readable(tconn);
4084}
4085
4086/** Standard I/O write function
4087 *
4088 * Underlying FD is now writable, so call the trunk to write any pending requests
4089 * to this connection.
4090 *
4091 * @param[in] el The event list signalling.
4092 * @param[in] fd that's now writable.
4093 * @param[in] flags describing the write event.
4094 * @param[in] uctx The trunk connection handle (tcon).
4095 */
4096void trunk_connection_callback_writable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
4097{
4098 trunk_connection_t *tconn = talloc_get_type_abort(uctx, trunk_connection_t);
4099
4100 trunk_connection_signal_writable(tconn);
4101}
4102
4103
4104/** Returns true if the trunk connection is in one of the specified states
4105 *
4106 * @param[in] tconn To check state for.
4107 * @param[in] state to check
4108 * @return
4109 * - True if trunk connection is in a particular state.
4110 * - False if trunk connection is not in a particular state.
4111 */
4112bool trunk_connection_in_state(trunk_connection_t *tconn, int state)
4113{
4114 return (bool)(tconn->pub.state & state);
4115}
4116
4117/** Close connections in a particular connection list if they have no requests associated with them
4118 *
4119 * @param[in] trunk containing connections we want to close.
4120 * @param[in] head of list of connections to examine.
4121 */
4122static void trunk_connection_close_if_empty(trunk_t *trunk, fr_dlist_head_t *head)
4123{
4124 trunk_connection_t *tconn = NULL;
4125
4126 while ((tconn = fr_dlist_next(head, tconn))) {
4127 trunk_connection_t *prev;
4128
4129 if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) != 0) continue;
4130
4131 prev = fr_dlist_prev(head, tconn);
4132
4133 DEBUG3("Closing %s connection with no requests",
4134 fr_table_str_by_value(trunk_connection_states, tconn->pub.state, "<INVALID>"));
4135 /*
4136 * Close the connection as gracefully
4137 * as possible by signalling it should
4138 * shutdown.
4139 *
4140 * The connection, should, if serviced
4141 * correctly by the underlying library,
4142 * automatically transition to halted after
4143 * all pending reads/writes are
4144 * complete at which point we'll be informed
4145 * and free our tconn wrapper.
4146 */
4147 connection_signal_shutdown(tconn->pub.conn);
4148 tconn = prev;
4149 }
4150}
4151
4152/** Rebalance connections across active trunk members when a new connection becomes active
4153 *
4154 * We don't have any visibility into the connection prioritisation algorithm
4155 * it's essentially a black box.
4156 *
4157 * We can however determine when the correct level of requests per connection
4158 * has been reached, by dequeuing and requeing requests up until the point
4159 * where the connection that just had a request dequeued, receives the same
4160 * request back.
4161 *
4162 * @param[in] trunk The trunk to rebalance.
4163 */
4164static void trunk_rebalance(trunk_t *trunk)
4165{
4166 trunk_connection_t *head;
4167
4168 head = fr_minmax_heap_min_peek(trunk->active);
4169
4170 /*
4171 * Only rebalance if the top and bottom of
4172 * the heap are not equal.
4173 */
4174 if (trunk->funcs.connection_prioritise(fr_minmax_heap_max_peek(trunk->active), head) == 0) return;
4175
4176 DEBUG3("Rebalancing requests");
4177
4178 /*
4179 * Keep requeuing requests from the connection
4180 * at the bottom of the heap until the
4181 * connection at the top is shifted from that
4182 * position.
4183 */
4184 while ((fr_minmax_heap_min_peek(trunk->active) == head) &&
4185 trunk_connection_requests_requeue_priv(fr_minmax_heap_max_peek(trunk->active),
4186 TRUNK_REQUEST_STATE_PENDING, 1, false));
4187}
4188
4189/** Implements the algorithm we use to manage requests per connection levels
4190 *
4191 * This is executed periodically using a timer event, and opens/closes
4192 * connections.
4193 *
4194 * The aim is to try and keep the request per connection level in a sweet spot,
4195 * where there's enough outstanding work for the connection/pipelining to work
4196 * efficiently, but not so much so that we encounter increased latency.
4197 *
4198 * In the request enqueue and dequeue functions we record every time the
4199 * average number of requests per connection goes above the target count
4200 * and record every time the average number of requests per connection goes
4201 * below the target count.
4202 *
4203 * This may sound expensive, but in all cases we're just summing counters.
4204 * CPU time required does not increase with additional requests, only with
4205 * large numbers of connections.
4206 *
4207 * If we do encounter scaling issues, we can always maintain the counters
4208 * as aggregates as an optimisation later.
4209 *
4210 * If when the management function runs, the trunk was above the target
4211 * most recently, we:
4212 * - Return if we've been in this state for a shorter period than 'open_delay'.
4213 * - Return if we're at max.
4214 * - Return if opening a new connection will take us below the load target.
4215 * - Return if we last opened a connection within 'open_delay'.
4216 * - Otherwise we attempt to open a new connection.
4217 *
4218 * If the trunk we below the target most recently, we:
4219 * - Return if we've been in this state for a shorter period than 'close_delay'.
4220 * - Return if we're at min.
4221 * - Return if we have no connections.
4222 * - Close a connection if min is 0, and we have no outstanding
4223 * requests. Then return.
4224 * - Return if closing a new connection will take us above the load target.
4225 * - Return if we last closed a connection within 'closed_delay'.
4226 * - Otherwise we move a connection to draining state.
4227 */
4228static void trunk_manage(trunk_t *trunk, fr_time_t now)
4229{
4230 trunk_connection_t *tconn = NULL;
4231 trunk_request_t *treq;
4232 uint32_t average = 0;
4233 uint32_t req_count;
4234 uint16_t conn_count;
4235 trunk_state_t new_state;
4236
4237 DEBUG4("Managing trunk");
4238
4239 /*
4240 * Cleanup requests in our request cache which
4241 * have been reapable for too long.
4242 */
4243 while ((treq = trunk_list_free_requests_peek(trunk)) &&
4244 fr_time_lteq(fr_time_add(treq->last_freed, trunk->conf.req_cleanup_delay), now)) talloc_free(treq);
4245
4246 /*
4247 * If we have idle connections, then close them, maintaining "min" connections.
4248 */
4249 if (fr_time_delta_ispos(trunk->conf.idle_timeout) &&
4250 (fr_minmax_heap_num_elements(trunk->active) > trunk->conf.min)) {
4251 fr_minmax_heap_iter_t iter;
4252 fr_time_t idle_cutoff = fr_time_sub(now, trunk->conf.idle_timeout);
4253
4254 for (tconn = fr_minmax_heap_iter_init(trunk->active, &iter);
4255 tconn;
4256 tconn = fr_minmax_heap_iter_next(trunk->active, &iter)) {
4257 /*
4258 * The connection has outstanding requests without replies, don't do anything.
4259 */
4260 if (fr_heap_num_elements(tconn->pending) > 0) continue;
4261
4262 /*
4263 * The connection was last active after the idle cutoff time, don't do anything.
4264 */
4265 if (fr_time_gt(tconn->pub.last_write_success, idle_cutoff)) continue;
4266
4267 /*
4268 * This connection has been inactive since before the idle timeout. Drain it,
4269 * and free it.
4270 *
4271 * This also extracts the connection from the minmax heap, which invalidates the
4272 * iterator, so we stop iterating over it.
4273 */
4274 trunk_connection_enter_draining_to_free(tconn);
4275 break;
4276 }
4277 }
4278
4279 /*
4280 * Free any connections which have drained
4281 * and we didn't reactivate during the last
4282 * round of management.
4283 */
4284 trunk_connection_close_if_empty(trunk, &trunk->inactive_draining);
4285 trunk_connection_close_if_empty(trunk, &trunk->draining);
4286 trunk_connection_close_if_empty(trunk, &trunk->draining_to_free);
4287
4288 /*
4289 * Process deferred connection freeing
4290 */
4291 if (!trunk->in_handler) fr_dlist_talloc_free(&trunk->to_free);
4292
4293 /*
4294 * Update the state of the trunk
4295 */
4296 if (trunk_connection_count_by_state(trunk, TRUNK_CONN_ACTIVE)) {
4297 new_state = TRUNK_STATE_ACTIVE;
4298 } else {
4299 /*
4300 * INIT / CONNECTING / FULL mean connections will become active
4301 * so the trunk is PENDING
4302 */
4303 new_state = trunk_connection_count_by_state(trunk, TRUNK_CONN_INIT |
4304 TRUNK_CONN_CONNECTING |
4305 TRUNK_CONN_FULL) ?
4306 TRUNK_STATE_PENDING : TRUNK_STATE_IDLE;
4307 }
4308
4309 if (new_state != trunk->pub.state) TRUNK_STATE_TRANSITION(new_state);
4310
4311 /*
4312 * A trunk can be signalled to not proactively
4313 * manage connections if a destination is known
4314 * to be unreachable, and doing so would result
4315 * in spurious connections still being opened.
4316 *
4317 * We still run other connection management
4318 * functions and just short circuit the function
4319 * here.
4320 */
4321 if (!trunk->managing_connections) return;
4322
4323 /*
4324 * We're above the target requests per connection
4325 * spawn more connections!
4326 */
4327 if (fr_time_gteq(trunk->pub.last_above_target, trunk->pub.last_below_target)) {
4328 /*
4329 * If connecting is provided, check we
4330 * wouldn't have too many connections in
4331 * the connecting state.
4332 *
4333 * This is a throttle in the case of transitory
4334 * load spikes, or a backend becoming
4335 * unavailable.
4336 */
4337 if ((trunk->conf.connecting > 0) &&
4338 (trunk_connection_count_by_state(trunk, TRUNK_CONN_CONNECTING) >=
4339 trunk->conf.connecting)) {
4340 DEBUG4("Not opening connection - Too many (%u) connections in the connecting state",
4341 trunk->conf.connecting);
4342 return;
4343 }
4344
4345 trunk_requests_per_connection(&conn_count, &req_count, trunk, now, true);
4346
4347 /*
4348 * Only apply hysteresis if we have at least
4349 * one available connection.
4350 */
4351 if (conn_count && fr_time_gt(fr_time_add(trunk->pub.last_above_target, trunk->conf.open_delay), now)) {
4352 DEBUG4("Not opening connection - Need to be above target for %pVs. It's been %pVs",
4353 fr_box_time_delta(trunk->conf.open_delay),
4354 fr_box_time_delta(fr_time_sub(now, trunk->pub.last_above_target)));
4355 return; /* too soon */
4356 }
4357
4358 /*
4359 * We don't consider 'draining' connections
4360 * in the max calculation, as if we do
4361 * determine that we need to spawn a new
4362 * request, then we'd move all 'draining'
4363 * connections to active before spawning
4364 * any new connections.
4365 */
4366 if ((trunk->conf.max > 0) && (conn_count >= trunk->conf.max)) {
4367 DEBUG4("Not opening connection - Have %u connections, need %u or below",
4368 conn_count, trunk->conf.max);
4369 return;
4370 }
4371
4372 /*
4373 * We consider requests pending on all connections
4374 * and the trunk's backlog as that's the current count
4375 * load.
4376 */
4377 if (!req_count) {
4378 DEBUG4("Not opening connection - No outstanding requests");
4379 return;
4380 }
4381
4382 /*
4383 * Do the n+1 check, i.e. if we open one connection
4384 * will that take us below our target threshold.
4385 */
4386 if (conn_count > 0) {
4387 average = ROUND_UP_DIV(req_count, (conn_count + 1));
4388 if (average < trunk->conf.target_req_per_conn) {
4389 DEBUG4("Not opening connection - Would leave us below our target requests "
4390 "per connection (now %u, after open %u)",
4391 ROUND_UP_DIV(req_count, conn_count), average);
4392 return;
4393 }
4394 } else {
4395 (void)trunk_connection_spawn(trunk, now);
4396 return;
4397 }
4398
4399 /*
4400 * If we've got a connection in the draining list
4401 * move it back into the active list if we've
4402 * been requested to add a connection back in.
4403 */
4404 tconn = fr_dlist_head(&trunk->draining);
4405 if (tconn) {
4406 if (trunk_connection_is_full(tconn)) {
4407 trunk_connection_enter_full(tconn);
4408 } else {
4409 trunk_connection_enter_active(tconn);
4410 }
4411 return;
4412 }
4413
4414 /*
4415 * Implement delay if there's no connections that
4416 * could be immediately re-activated.
4417 */
4418 if (fr_time_gt(fr_time_add(trunk->pub.last_open, trunk->conf.open_delay), now)) {
4419 DEBUG4("Not opening connection - Need to wait %pVs before opening another connection. "
4420 "It's been %pVs",
4421 fr_box_time_delta(trunk->conf.open_delay),
4422 fr_box_time_delta(fr_time_sub(now, trunk->pub.last_open)));
4423 return;
4424 }
4425
4426 DEBUG4("Opening connection - Above target requests per connection (now %u, target %u)",
4427 ROUND_UP_DIV(req_count, conn_count), trunk->conf.target_req_per_conn);
4428 /* last_open set by trunk_connection_spawn */
4429 (void)trunk_connection_spawn(trunk, now);
4430 }
4431
4432 /*
4433 * We're below the target requests per connection.
4434 * Free some connections...
4435 */
4436 else if (fr_time_gt(trunk->pub.last_below_target, trunk->pub.last_above_target)) {
4437 if (fr_time_gt(fr_time_add(trunk->pub.last_below_target, trunk->conf.close_delay), now)) {
4438 DEBUG4("Not closing connection - Need to be below target for %pVs. It's been %pVs",
4439 fr_box_time_delta(trunk->conf.close_delay),
4440 fr_box_time_delta(fr_time_sub(now, trunk->pub.last_below_target)));
4441 return; /* too soon */
4442 }
4443
4444 trunk_requests_per_connection(&conn_count, &req_count, trunk, now, true);
4445
4446 if (!conn_count) {
4447 DEBUG4("Not closing connection - No connections to close!");
4448 return;
4449 }
4450
4451 if ((trunk->conf.min > 0) && ((conn_count - 1) < trunk->conf.min)) {
4452 DEBUG4("Not closing connection - Have %u connections, need %u or above",
4453 conn_count, trunk->conf.min);
4454 return;
4455 }
4456
4457 if (!req_count) {
4458 DEBUG4("Closing connection - No outstanding requests");
4459 goto close;
4460 }
4461
4462 /*
4463 * The minimum number of connections must be set
4464 * to zero for this to work.
4465 * min == 0, no requests, close all the connections.
4466 * This is useful for backup databases, when
4467 * maintaining the connection would lead to lots of
4468 * log file churn.
4469 */
4470 if (conn_count == 1) {
4471 DEBUG4("Not closing connection - Would leave connections "
4472 "and there are still %u outstanding requests", req_count);
4473 return;
4474 }
4475
4476 /*
4477 * Do the n-1 check, i.e. if we close one connection
4478 * will that take us above our target threshold.
4479 */
4480 average = ROUND_UP_DIV(req_count, (conn_count - 1));
4481 if (average > trunk->conf.target_req_per_conn) {
4482 DEBUG4("Not closing connection - Would leave us above our target requests per connection "
4483 "(now %u, after close %u)", ROUND_UP_DIV(req_count, conn_count), average);
4484 return;
4485 }
4486
4487 DEBUG4("Closing connection - Below target requests per connection (now %u, target %u)",
4488 ROUND_UP_DIV(req_count, conn_count), trunk->conf.target_req_per_conn);
4489
4490 close:
4491 if (fr_time_gt(fr_time_add(trunk->pub.last_closed, trunk->conf.close_delay), now)) {
4492 DEBUG4("Not closing connection - Need to wait %pVs before closing another connection. "
4493 "It's been %pVs",
4494 fr_box_time_delta(trunk->conf.close_delay),
4495 fr_box_time_delta(fr_time_sub(now, trunk->pub.last_closed)));
4496 return;
4497 }
4498
4499 /*
4500 * If the last event on the trunk was a connection failure and
4501 * there is only one connection, this may well be a reconnect
4502 * attempt after a failure - and needs to persist otherwise
4503 * the last event will be a failure and no new connection will
4504 * be made, leading to no new requests being enqueued.
4505 */
4506 if (fr_time_gt(trunk->pub.last_failed, fr_time_wrap(0)) &&
4507 fr_time_lt(trunk->pub.last_connected, trunk->pub.last_failed) && (conn_count == 1)) {
4508 DEBUG4("Not closing remaining connection - last event was a failure");
4509 return;
4510 }
4511
4512 /*
4513 * Inactive connections get counted in the
4514 * set of viable connections, but are likely
4515 * to be congested or dead, so we drain
4516 * (and possibly eventually free) those first.
4517 */
4518 if ((tconn = trunk_list_inactive_peek(trunk))) {
4519 /*
4520 * If the connection has no requests associated
4521 * with it then immediately free.
4522 */
4523 if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0) {
4524 connection_signal_halt(tconn->pub.conn); /* Also frees the tconn */
4525 } else {
4526 trunk_connection_enter_inactive_draining(tconn);
4527 }
4528 /*
4529 * It is possible to have too may connecting
4530 * connections when the connections are
4531 * taking a while to open and the number
4532 * of requests decreases.
4533 */
4534 } else if ((tconn = fr_dlist_tail(&trunk->connecting))) {
4535 connection_signal_halt(tconn->pub.conn); /* Also frees the tconn */
4536
4537 /*
4538 * Finally if there are no "connecting"
4539 * connections to close, and no "inactive"
4540 * connections, start draining "active"
4541 * connections.
4542 */
4543 } else if ((tconn = fr_minmax_heap_max_peek(trunk->active))) {
4544 /*
4545 * If the connection has no requests associated
4546 * with it then immediately free.
4547 */
4548 if (trunk_request_count_by_connection(tconn, TRUNK_REQUEST_STATE_ALL) == 0) {
4549 connection_signal_halt(tconn->pub.conn); /* Also frees the tconn */
4550 } else {
4551 trunk_connection_enter_draining(tconn);
4552 }
4553 }
4554
4555 trunk->pub.last_closed = now;
4556
4557 return;
4558 }
4559}
4560
4561/** Event to periodically call the connection management function
4562 *
4563 * @param[in] tl this event belongs to.
4564 * @param[in] now current time.
4565 * @param[in] uctx The trunk.
4566 */
4567static void _trunk_timer(fr_timer_list_t *tl, fr_time_t now, void *uctx)
4568{
4569 trunk_t *trunk = talloc_get_type_abort(uctx, trunk_t);
4570
4571 trunk_manage(trunk, now);
4572
4573 if (fr_time_delta_ispos(trunk->conf.manage_interval)) {
4574 if (fr_timer_in(trunk, tl, &trunk->manage_ev, trunk->conf.manage_interval,
4575 false, _trunk_timer, trunk) < 0) {
4576 PERROR("Failed inserting trunk management event");
4577 /* Not much we can do, hopefully the trunk will be freed soon */
4578 }
4579 }
4580}
4581
4582/** Return a count of requests on a connection in a specific state
4583 *
4584 * @param[in] trunk to retrieve counts for.
4585 * @param[in] conn_state One or more connection states or'd together.
4586 * @param[in] req_state One or more request states or'd together.
4587 * @return The number of requests in a particular state, on connection in a particular state.
4588 */
4589uint64_t trunk_request_count_by_state(trunk_t *trunk, int conn_state, int req_state)
4590{
4591 uint64_t count = 0;
4592 trunk_connection_t *tconn = NULL;
4593 fr_minmax_heap_iter_t iter;
4594
4595#define COUNT_BY_STATE(_state, _list) \
4596do { \
4597 if (conn_state & (_state)) { \
4598 tconn = NULL; \
4599 while ((tconn = fr_dlist_next(&trunk->_list, tconn))) { \
4600 count += trunk_request_count_by_connection(tconn, req_state); \
4601 } \
4602 } \
4603} while (0)
4604
4605 if (conn_state & TRUNK_CONN_ACTIVE) {
4606 for (tconn = fr_minmax_heap_iter_init(trunk->active, &iter);
4607 tconn;
4608 tconn = fr_minmax_heap_iter_next(trunk->active, &iter)) {
4609 count += trunk_request_count_by_connection(tconn, req_state);
4610 }
4611 }
4612
4613 COUNT_BY_STATE(TRUNK_CONN_FULL, full);
4614 COUNT_BY_STATE(TRUNK_CONN_INACTIVE, inactive);
4615 COUNT_BY_STATE(TRUNK_CONN_INACTIVE_DRAINING, inactive_draining);
4616 COUNT_BY_STATE(TRUNK_CONN_DRAINING, draining);
4617 COUNT_BY_STATE(TRUNK_CONN_DRAINING_TO_FREE, draining_to_free);
4618
4619 if (req_state & TRUNK_REQUEST_STATE_BACKLOG) count += fr_heap_num_elements(trunk->backlog);
4620
4621 return count;
4622}
4623
4624/** Update timestamps for when we last had a transition from above target to below target or vice versa
4625 *
4626 * Should be called on every time a connection or request is allocated or freed.
4627 *
4628 * @param[out] conn_count_out How many connections we considered.
4629 * @param[out] req_count_out How many requests we considered.
4630 * @param[in] trunk to operate on.
4631 * @param[in] now The current time.
4632 * @param[in] verify if true (and this is a debug build), then assert if req_per_conn
4633 * has changed.
4634 * @return
4635 * - 0 if the average couldn't be calculated (no requests or no connections).
4636 * - The average number of requests per connection.
4637 */
4638static uint64_t trunk_requests_per_connection(uint16_t *conn_count_out, uint32_t *req_count_out,
4639 trunk_t *trunk, fr_time_t now,
4640 NDEBUG_UNUSED bool verify)
4641{
4642 uint32_t req_count = 0;
4643 uint16_t conn_count = 0;
4644 uint64_t req_per_conn = 0;
4645
4646 fr_assert(fr_time_gt(now, fr_time_wrap(0)));
4647
4648 /*
4649 * No need to update these as the trunk is being freed
4650 */
4651 if (trunk->freeing) goto done;
4652
4653 /*
4654 * Count all connections except draining and draining to free.
4655 *
4656 * Omitting these connection states artificially raises the
4657 * request to connection ratio, so that we can preemptively spawn
4658 * new connections.
4659 *
4660 * In the case of TRUNK_CONN_DRAINING | TRUNK_CONN_INACTIVE_DRAINING
4661 * the trunk management code has enough hysteresis to not
4662 * immediately reactivate the connection.
4663 *
4664 * In the case of TRUNK_CONN_DRAINING_TO_FREE the trunk
4665 * management code should spawn a new connection to takes its place.
4666 *
4667 * Connections placed in the DRAINING_TO_FREE state are being
4668 * closed preemptively to deal with bugs on the server we're
4669 * talking to, or misconfigured firewalls which are trashing
4670 * TCP/UDP connection states.
4671 */
4672 conn_count = trunk_connection_count_by_state(trunk, TRUNK_CONN_ALL ^
4673 (TRUNK_CONN_DRAINING |
4674 TRUNK_CONN_INACTIVE_DRAINING |
4675 TRUNK_CONN_DRAINING_TO_FREE));
4676
4677 /*
4678 * Requests on all connections
4679 */
4680 req_count = trunk_request_count_by_state(trunk,
4681 TRUNK_CONN_ALL ^
4682 TRUNK_CONN_DRAINING_TO_FREE, TRUNK_REQUEST_STATE_ALL);
4683
4684 /*
4685 * No connections, but we do have requests
4686 */
4687 if (conn_count == 0) {
4688 if ((req_count > 0) && (trunk->conf.target_req_per_conn > 0)) goto above_target;
4689 goto done;
4690 }
4691
4692 if (req_count == 0) {
4693 if (trunk->conf.target_req_per_conn > 0) goto below_target;
4694 goto done;
4695 }
4696
4697 /*
4698 * Calculate the req_per_conn
4699 */
4700 req_per_conn = ROUND_UP_DIV(req_count, conn_count);
4701 if (req_per_conn > trunk->conf.target_req_per_conn) {
4702 above_target:
4703 /*
4704 * Edge - Below target to above target (too many requests per conn - spawn more)
4705 *
4706 * The equality check is correct here as both values start at 0.
4707 */
4708 if (fr_time_lteq(trunk->pub.last_above_target, trunk->pub.last_below_target)) trunk->pub.last_above_target = now;
4709 } else if (req_per_conn < trunk->conf.target_req_per_conn) {
4710 below_target:
4711 /*
4712 * Edge - Above target to below target (too few requests per conn - close some)
4713 *
4714 * The equality check is correct here as both values start at 0.
4715 */
4716 if (fr_time_lteq(trunk->pub.last_below_target, trunk->pub.last_above_target)) trunk->pub.last_below_target = now;
4717 }
4718
4719done:
4720 if (conn_count_out) *conn_count_out = conn_count;
4721 if (req_count_out) *req_count_out = req_count;
4722
4723 /*
4724 * Check we haven't missed a call to trunk_requests_per_connection
4725 */
4726 fr_assert(!verify || (trunk->last_req_per_conn == 0) || (req_per_conn == trunk->last_req_per_conn));
4727
4728 trunk->last_req_per_conn = req_per_conn;
4729
4730 return req_per_conn;
4731}
4732
4733/** Drain the backlog of as many requests as possible
4734 *
4735 * @param[in] trunk To drain backlog requests for.
4736 */
4737static void trunk_backlog_drain(trunk_t *trunk)
4738{
4739 trunk_request_t *treq;
4740
4741 if (fr_heap_num_elements(trunk->backlog) == 0) return;
4742
4743 /*
4744 * If it's always writable, this isn't
4745 * really a noteworthy event.
4746 */
4747 if (!trunk->conf.always_writable) DEBUG3("Draining backlog of requests");
4748
4749 /*
4750 * Do *NOT* add an artificial limit
4751 * here. We rely on all available
4752 * connections entering the full
4753 * state and transitioning back to
4754 * active in order to drain the
4755 * backlog.
4756 */
4757 while ((treq = fr_heap_peek(trunk->backlog))) {
4758 switch (trunk_request_enqueue_existing(treq)) {
4759 case TRUNK_ENQUEUE_OK:
4760 continue;
4761
4762 /*
4763 * Signal to stop
4764 */
4765 case TRUNK_ENQUEUE_IN_BACKLOG:
4766 break;
4767
4768 /*
4769 * Failed enqueueing the request,
4770 * have it enter the failed state
4771 * which will free it and
4772 * re-enliven the yielded request.
4773 */
4774 case TRUNK_ENQUEUE_DST_UNAVAILABLE:
4775 case TRUNK_ENQUEUE_FAIL:
4776 trunk_request_enter_failed(treq);
4777 continue;
4778
4779 case TRUNK_ENQUEUE_NO_CAPACITY:
4780 fr_assert(fr_minmax_heap_num_elements(trunk->active) == 0);
4781 return;
4782 }
4783 }
4784}
4785
4786/** Force the trunk to re-establish its connections
4787 *
4788 * @param[in] trunk to signal.
4789 * @param[in] states One or more states or'd together.
4790 * @param[in] reason Why the connections are being signalled to reconnect.
4791 */
4792void trunk_reconnect(trunk_t *trunk, int states, connection_reason_t reason)
4793{
4794
4795#define RECONNECT_BY_STATE(_state, _list) \
4796do { \
4797 if (states & (_state)) { \
4798 size_t i; \
4799 for (i = fr_dlist_num_elements(&trunk->_list); i > 0; i--) { \
4800 connection_signal_reconnect(((trunk_connection_t *)fr_dlist_tail(&trunk->_list))->pub.conn, reason); \
4801 } \
4802 } \
4803} while (0)
4804
4805 /*
4806 * Connections in the 'connecting' state
4807 * may re-enter that state, so we need to
4808 * be careful not to enter an infinite
4809 * loop, as we iterate over the list
4810 * again and again.
4811 */
4812 RECONNECT_BY_STATE(TRUNK_CONN_CONNECTING, connecting);
4813
4814 if (states & TRUNK_CONN_ACTIVE) {
4815 trunk_connection_t *tconn;
4816 while ((tconn = fr_minmax_heap_min_peek(trunk->active))) connection_signal_reconnect(tconn->pub.conn, reason);
4817 }
4818
4819 RECONNECT_BY_STATE(TRUNK_CONN_INIT, init);
4820 RECONNECT_BY_STATE(TRUNK_CONN_FULL, full);
4821 RECONNECT_BY_STATE(TRUNK_CONN_INACTIVE, inactive);
4822 RECONNECT_BY_STATE(TRUNK_CONN_INACTIVE_DRAINING, inactive_draining);
4823 RECONNECT_BY_STATE(TRUNK_CONN_CLOSED, closed);
4824 RECONNECT_BY_STATE(TRUNK_CONN_DRAINING, draining);
4825 RECONNECT_BY_STATE(TRUNK_CONN_DRAINING_TO_FREE, draining_to_free);
4826}
4827
4828/** Start the trunk running
4829 *
4830 */
4831int trunk_start(trunk_t *trunk)
4832{
4833 uint16_t i;
4834
4835 if (unlikely(trunk->started)) return 0;
4836
4837 /*
4838 * Spawn the initial set of connections
4839 */
4840 for (i = 0; i < trunk->conf.start; i++) {
4841 DEBUG("[%i] Starting initial connection", i);
4842 if (trunk_connection_spawn(trunk, fr_time()) != 0) return -1;
4843 }
4844
4845 /*
4846 * If the idle timeout is set, AND there's no management interval, OR the management interval is
4847 * less than the idle timeout, update the management interval.
4848 */
4849 if (fr_time_delta_ispos(trunk->conf.idle_timeout) &&
4850 (!fr_time_delta_ispos(trunk->conf.manage_interval) ||
4851 fr_time_delta_gt(trunk->conf.manage_interval, trunk->conf.idle_timeout))) {
4852 trunk->conf.manage_interval = trunk->conf.idle_timeout;
4853 }
4854
4855 if (fr_time_delta_ispos(trunk->conf.manage_interval)) {
4856 /*
4857 * Insert the event timer to manage
4858 * the interval between managing connections.
4859 */
4860 if (fr_timer_in(trunk, trunk->el->tl, &trunk->manage_ev, trunk->conf.manage_interval,
4861 false, _trunk_timer, trunk) < 0) {
4862 PERROR("Failed inserting trunk management event");
4863 return -1;
4864 }
4865 }
4866 trunk->started = true;
4867 trunk->managing_connections = true;
4868
4869 return 0;
4870}
4871
4872/** Allow the trunk to open and close connections in response to load
4873 *
4874 */
4875void trunk_connection_manage_start(trunk_t *trunk)
4876{
4877 if (!trunk->started || trunk->managing_connections) return;
4878
4879 DEBUG3("Connection management enabled");
4880 trunk->managing_connections = true;
4881}
4882
4883/** Stop the trunk from opening and closing connections in response to load
4884 *
4885 */
4886void trunk_connection_manage_stop(trunk_t *trunk)
4887{
4888 if (!trunk->started || !trunk->managing_connections) return;
4889
4890 DEBUG3("Connection management disabled");
4891 trunk->managing_connections = false;
4892}
4893
4894/** Schedule a trunk management event for the next time the event loop is executed
4895 */
4896int trunk_connection_manage_schedule(trunk_t *trunk)
4897{
4898 if (!trunk->started || !trunk->managing_connections) return 0;
4899
4900 if (fr_timer_in(trunk, trunk->el->tl, &trunk->manage_ev, fr_time_delta_wrap(0),
4901 false, _trunk_timer, trunk) < 0) {
4902 PERROR("Failed inserting trunk management event");
4903 return -1;
4904 }
4905
4906 return 0;
4907}
4908
4909/** Order connections by queue depth
4910 *
4911 */
4912static int8_t _trunk_connection_order_by_shortest_queue(void const *one, void const *two)
4913{
4914 trunk_connection_t const *a = talloc_get_type_abort_const(one, trunk_connection_t);
4915 trunk_connection_t const *b = talloc_get_type_abort_const(two, trunk_connection_t);
4916
4917 uint32_t a_count = trunk_request_count_by_connection(a, TRUNK_REQUEST_STATE_ALL);
4918 uint32_t b_count = trunk_request_count_by_connection(b, TRUNK_REQUEST_STATE_ALL);
4919
4920 /*
4921 * Add a fudge factor of 1 to reduce spurious rebalancing
4922 */
4923 return ((a_count > b_count) && ((a_count - b_count) > 1)) - ((b_count > a_count) && ((b_count - a_count) > 1));
4924}
4925
4926/** Free a trunk, gracefully closing all connections.
4927 *
4928 */
4929static int _trunk_free(trunk_t *trunk)
4930{
4931 trunk_connection_t *tconn;
4932 trunk_request_t *treq;
4933 trunk_watch_entry_t *watch;
4934 size_t i;
4935
4936 DEBUG4("Trunk free %p", trunk);
4937
4938 trunk->freeing = true; /* Prevent re-enqueuing */
4939
4940 /*
4941 * We really don't want this firing after
4942 * we've freed everything.
4943 */
4944 FR_TIMER_DELETE_RETURN(&trunk->manage_ev);
4945
4946 /*
4947 * Now free the connections in each of the lists.
4948 *
4949 * Each time a connection is freed it removes itself from the list
4950 * its in, which means the head should keep advancing automatically.
4951 */
4952 while ((tconn = fr_minmax_heap_min_peek(trunk->active))) connection_signal_halt(tconn->pub.conn);
4953 while ((tconn = fr_dlist_head(&trunk->init))) connection_signal_halt(tconn->pub.conn);
4954 while ((tconn = fr_dlist_head(&trunk->connecting))) connection_signal_halt(tconn->pub.conn);
4955 while ((tconn = fr_dlist_head(&trunk->full))) connection_signal_halt(tconn->pub.conn);
4956 while ((tconn = fr_dlist_head(&trunk->inactive))) connection_signal_halt(tconn->pub.conn);
4957 while ((tconn = fr_dlist_head(&trunk->inactive_draining))) connection_signal_halt(tconn->pub.conn);
4958 while ((tconn = fr_dlist_head(&trunk->closed))) connection_signal_halt(tconn->pub.conn);
4959 while ((tconn = fr_dlist_head(&trunk->draining))) connection_signal_halt(tconn->pub.conn);
4960 while ((tconn = fr_dlist_head(&trunk->draining_to_free))) connection_signal_halt(tconn->pub.conn);
4961
4962 /*
4963 * Process any deferred connection frees
4964 */
4965 fr_dlist_talloc_free(&trunk->to_free);
4966
4967 /*
4968 * Free any requests left in the backlog
4969 */
4970 while ((treq = fr_heap_peek(trunk->backlog))) trunk_request_enter_failed(treq);
4971
4972 /*
4973 * Free any requests in our request cache
4974 */
4975 while ((treq = trunk_list_free_requests_peek(trunk))) talloc_free(treq);
4976
4977 /*
4978 * Free any entries in the watch lists
4979 */
4980 for (i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
4981 while ((watch = fr_dlist_pop_head(&trunk->watch[i]))) talloc_free(watch);
4982 }
4983
4984 return 0;
4985}
4986
4987/** Allocate a new collection of connections
4988 *
4989 * This function should be called first to allocate a new trunk connection.
4990 *
4991 * After the trunk has been allocated, #trunk_request_alloc and
4992 * #trunk_request_enqueue should be used to allocate memory for trunk
4993 * requests, and pass a preq (protocol request) to the trunk for
4994 * processing.
4995 *
4996 * The trunk will then asynchronously process the request, writing the result
4997 * to a specified rctx. See #trunk_request_enqueue for more details.
4998 *
4999 * @note Trunks may not be shared between multiple threads under any circumstances.
5000 *
5001 * @param[in] ctx To use for any memory allocations. Must be thread local.
5002 * @param[in] el to use for I/O and timer events.
5003 * @param[in] funcs Callback functions.
5004 * @param[in] conf Common user configurable parameters.
5005 * @param[in] log_prefix To prepend to global messages.
5006 * @param[in] uctx User data to pass to the alloc function.
5007 * @param[in] delay_start If true, then we will not spawn any connections
5008 * until the first request is enqueued.
5009 * @param[in] trigger_args Pairs to pass to trigger requests, if triggers are enabled.
5010 * @return
5011 * - New trunk handle on success.
5012 * - NULL on error.
5013 */
5014trunk_t *trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el,
5015 trunk_io_funcs_t const *funcs, trunk_conf_t const *conf,
5016 char const *log_prefix, void const *uctx, bool delay_start, fr_pair_list_t *trigger_args)
5017{
5018 trunk_t *trunk;
5019 size_t i;
5020
5021 /*
5022 * Check we have the functions we need
5023 */
5024 if (!fr_cond_assert(funcs->connection_alloc)) return NULL;
5025
5026 MEM(trunk = talloc_zero(ctx, trunk_t));
5027 trunk->el = el;
5028 trunk->log_prefix = talloc_strdup(trunk, log_prefix);
5029 trunk->trigger_args = trigger_args;
5030
5031 memcpy(&trunk->funcs, funcs, sizeof(trunk->funcs));
5032 if (!trunk->funcs.connection_prioritise) {
5033 trunk->funcs.connection_prioritise = _trunk_connection_order_by_shortest_queue;
5034 }
5035 if (!trunk->funcs.request_prioritise) trunk->funcs.request_prioritise = fr_pointer_cmp;
5036
5037 memcpy(&trunk->conf, conf, sizeof(trunk->conf));
5038
5039 memcpy(&trunk->uctx, &uctx, sizeof(trunk->uctx));
5040 talloc_set_destructor(trunk, _trunk_free);
5041
5042 /*
5043 * Free request list...
5044 */
5045 fr_dlist_talloc_init(&trunk->free_requests, trunk_request_t, entry);
5046
5047 /*
5048 * Request backlog queue
5049 */
5050 MEM(trunk->backlog = fr_heap_talloc_alloc(trunk, _trunk_request_prioritise,
5051 trunk_request_t, heap_id, 0));
5052
5053 /*
5054 * Connection queues and trees
5055 */
5056 MEM(trunk->active = fr_minmax_heap_talloc_alloc(trunk, trunk->funcs.connection_prioritise,
5057 trunk_connection_t, heap_id, 0));
5058 fr_dlist_talloc_init(&trunk->init, trunk_connection_t, entry);
5059 fr_dlist_talloc_init(&trunk->connecting, trunk_connection_t, entry);
5060 fr_dlist_talloc_init(&trunk->full, trunk_connection_t, entry);
5061 fr_dlist_talloc_init(&trunk->inactive, trunk_connection_t, entry);
5062 fr_dlist_talloc_init(&trunk->inactive_draining, trunk_connection_t, entry);
5063 fr_dlist_talloc_init(&trunk->closed, trunk_connection_t, entry);
5064 fr_dlist_talloc_init(&trunk->draining, trunk_connection_t, entry);
5065 fr_dlist_talloc_init(&trunk->draining_to_free, trunk_connection_t, entry);
5066 fr_dlist_talloc_init(&trunk->to_free, trunk_connection_t, entry);
5067
5068 /*
5069 * Watch lists
5070 */
5071 for (i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
5072 fr_dlist_talloc_init(&trunk->watch[i], trunk_watch_entry_t, entry);
5073 }
5074
5075 DEBUG4("Trunk allocated %p", trunk);
5076
5077 if (!delay_start) {
5078 if (trunk_start(trunk) < 0) {
5079 talloc_free(trunk);
5080 return NULL;
5081 }
5082 }
5083
5084 return trunk;
5085}
5086
5087/** Check for a module trigger section when parsing the `triggers` option.
5088 *
5089 */
5090int trunk_trigger_cf_parse(TALLOC_CTX *ctx, void *out, void *parent, CONF_ITEM *ci, conf_parser_t const *rule)
5091{
5092 trunk_conf_t *conf = parent;
5093 CONF_SECTION *cs = cf_item_to_section(cf_parent(ci));
5094
5095 if (cf_pair_parse_value(ctx, out, parent, ci, rule)< 0) return -1;
5096
5097 /*
5098 * If the parent section of the `triggers` option contains a trigger
5099 * section then store it as the module CONF SECTION for the appropriate
5100 * trigger group.
5101 */
5102 if (cf_section_find(cs, "trigger", NULL)) {
5103 if (strcmp(cf_section_name(cs), "request") == 0) {
5104 conf->req_trigger_cs = cs;
5105 } else {
5106 conf->conn_trigger_cs = cs;
5107 }
5108 }
5109
5110 return 0;
5111}
5112
5113#ifndef TALLOC_GET_TYPE_ABORT_NOOP
5114/** Verify a trunk
5115 *
5116 * A trunk has some number of connections, which each have some number of requests. The connections and
5117 * requests are in differing kinds of containers depending on their state and how they are used, and may
5118 * have fields that can only be validated by comparison with a parent. We had planned on passing a "context"
5119 * down with the ancestral values, but that breaks the foo_verify() API. Each foo_verify() will only verify the
5120 * foo's children.
5121 */
5122void trunk_verify(char const *file, int line, trunk_t *trunk)
5123{
5124 fr_fatal_assert_msg(trunk, "CONSISTENCY CHECK FAILED %s[%i]: trunk_t pointer was NULL", file, line);
5125 (void) talloc_get_type_abort(trunk, trunk_t);
5126
5127 for (size_t i = 0; i < NUM_ELEMENTS(trunk->watch); i++) {
5128 _fr_dlist_verify(file, line, &trunk->watch[i]);
5129 }
5130
5131#define IO_FUNC_VERIFY(_func) \
5132 fr_fatal_assert_msg(trunk->funcs._func, "CONSISTENCY_CHECK_FAILED %s[%i}: " #_func " was NULL", file, line)
5133
5134 /*
5135 * Only a few of the function pointers *must* be non-NULL..
5136 */
5137 IO_FUNC_VERIFY(connection_alloc);
5138 IO_FUNC_VERIFY(connection_prioritise);
5139 IO_FUNC_VERIFY(request_prioritise);
5140
5141#define TRUNK_TCONN_CHECKS(_tconn, _state) \
5142do { \
5143 fr_fatal_assert_msg(trunk == _tconn->pub.trunk, \
5144 "CONSISTENCY_CHECK_FAILED %s[%i}: connection-trunk mismatch", file, line); \
5145 fr_fatal_assert_msg(_state == _tconn->pub.state, \
5146 "CONSISTENCY_CHECK_FAILED %s[%i}: connection-state mismatch", file, line); \
5147} while (0)
5148
5149#define TCONN_DLIST_VERIFY(_dlist, _state) \
5150do { \
5151 _fr_dlist_verify(file, line, &(trunk->_dlist)); \
5152 fr_dlist_foreach(&(trunk->_dlist), trunk_connection_t, tconn) { \
5153 trunk_connection_verify(file, line, tconn); \
5154 TRUNK_TCONN_CHECKS(tconn, _state); \
5155 } \
5156} while (0)
5157
5158#define TCONN_MINMAX_HEAP_VERIFY(_heap, _state) \
5159do {\
5160 fr_minmax_heap_verify(file, line, trunk->_heap); \
5161 fr_minmax_heap_foreach(trunk->_heap, trunk_connection_t, tconn) { \
5162 trunk_connection_verify(file, line, tconn); \
5163 TRUNK_TCONN_CHECKS(tconn, _state); \
5164 }} \
5165} while (0)
5166
5167 fr_dlist_verify(&(trunk->free_requests));
5168 FR_HEAP_VERIFY(trunk->backlog);
5169
5170 TCONN_DLIST_VERIFY(init, TRUNK_CONN_INIT);
5171 TCONN_DLIST_VERIFY(connecting, TRUNK_CONN_CONNECTING);
5172 TCONN_MINMAX_HEAP_VERIFY(active, TRUNK_CONN_ACTIVE);
5173 TCONN_DLIST_VERIFY(full, TRUNK_CONN_FULL);
5174 TCONN_DLIST_VERIFY(inactive, TRUNK_CONN_INACTIVE);
5175 TCONN_DLIST_VERIFY(inactive_draining, TRUNK_CONN_INACTIVE_DRAINING);
5176 /* TCONN_DLIST_VERIFY(failed, ???); */
5177 TCONN_DLIST_VERIFY(closed, TRUNK_CONN_CLOSED);
5178 TCONN_DLIST_VERIFY(draining, TRUNK_CONN_DRAINING);
5179 TCONN_DLIST_VERIFY(draining_to_free, TRUNK_CONN_DRAINING_TO_FREE);
5180 TCONN_DLIST_VERIFY(to_free, TRUNK_CONN_HALTED);
5181}
5182
5183void trunk_connection_verify(char const *file, int line, trunk_connection_t *tconn)
5184{
5185 fr_fatal_assert_msg(tconn, "CONSISTENCY CHECK FAILED %s[%i]: trunk_connection_t pointer was NULL", file, line);
5186 (void) talloc_get_type_abort(tconn, trunk_connection_t);
5187
5188 (void) talloc_get_type_abort(tconn->pub.trunk, trunk_t);
5189
5190 /*
5191 * shouldn't be both in heap and on list--but it doesn't look like moves
5192 * to active heap wipe the dlist pointers.
5193 */
5194
5195#define TCONN_TREQ_CHECKS(_treq, _state) \
5196do { \
5197 fr_fatal_assert_msg(tconn == _treq->pub.tconn, \
5198 "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-tconn mismatch", file, line); \
5199 fr_fatal_assert_msg(tconn->pub.trunk == _treq->pub.trunk, \
5200 "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-trunk mismatch", file, line); \
5201 fr_fatal_assert_msg(_state == _treq->pub.state, \
5202 "CONSISTENCY_CHECK_FAILED %s[%i}: trunk request-state mismatch", file, line); \
5203} while (0)
5204
5205#define TREQ_DLIST_VERIFY(_dlist, _state) \
5206do { \
5207 _fr_dlist_verify(file, line, &(tconn->_dlist)); \
5208 fr_dlist_foreach(&(tconn->_dlist), trunk_request_t, treq) { \
5209 trunk_request_verify(file, line, treq); \
5210 TCONN_TREQ_CHECKS(treq, _state); \
5211 } \
5212} while (0)
5213
5214#define TREQ_HEAP_VERIFY(_heap, _state) \
5215do { \
5216 fr_heap_iter_t _iter; \
5217 fr_heap_verify(file, line, tconn->_heap); \
5218 for (trunk_request_t *treq = fr_heap_iter_init(tconn->_heap, &_iter); \
5219 treq; \
5220 treq = fr_heap_iter_next(tconn->_heap, &_iter)) { \
5221 trunk_request_verify(file, line, treq); \
5222 TCONN_TREQ_CHECKS(treq, _state); \
5223 } \
5224} while (0)
5225
5226#define TREQ_OPTION_VERIFY(_option, _state) \
5227do { \
5228 if (tconn->_option) { \
5229 trunk_request_verify(file, line, tconn->_option); \
5230 TCONN_TREQ_CHECKS(tconn->_option, _state); \
5231 } \
5232} while (0)
5233
5234 /* verify associated requests */
5235 TREQ_HEAP_VERIFY(pending, TRUNK_REQUEST_STATE_PENDING);
5236 TREQ_DLIST_VERIFY(sent, TRUNK_REQUEST_STATE_SENT);
5237 TREQ_DLIST_VERIFY(cancel, TRUNK_REQUEST_STATE_CANCEL);
5238 TREQ_DLIST_VERIFY(cancel_sent, TRUNK_REQUEST_STATE_CANCEL_SENT);
5239 TREQ_OPTION_VERIFY(partial, TRUNK_REQUEST_STATE_PARTIAL);
5240 TREQ_OPTION_VERIFY(cancel_partial, TRUNK_REQUEST_STATE_CANCEL_PARTIAL);
5241}
5242
5243void trunk_request_verify(char const *file, int line, trunk_request_t *treq)
5244{
5245 fr_fatal_assert_msg(treq, "CONSISTENCY CHECK FAILED %s[%i]: trunk_request_t pointer was NULL", file, line);
5246 (void) talloc_get_type_abort(treq, trunk_request_t);
5247
5248#ifdef WITH_VERIFY_PTR
5249 if (treq->pub.request) request_verify(file, line, treq->pub.request);
5250#endif
5251}
5252
5253
5254bool trunk_search(trunk_t *trunk, void *ptr)
5255{
5256#define TCONN_DLIST_SEARCH(_dlist) \
5257do { \
5258 fr_dlist_foreach(&(trunk->_dlist), trunk_connection_t, tconn) { \
5259 if (ptr == tconn) { \
5260 fr_fprintf(stderr, "trunk_search: tconn %p on " #_dlist "\n", ptr); \
5261 return true; \
5262 } \
5263 if (trunk_connection_search(tconn, ptr)) { \
5264 fr_fprintf(stderr, " in tconn %p on " #_dlist "\n", tconn); \
5265 return true; \
5266 } \
5267 } \
5268} while (0)
5269
5270#define TCONN_MINMAX_HEAP_SEARCH(_heap) \
5271do { \
5272 fr_minmax_heap_foreach(trunk->_heap, trunk_connection_t, tconn) { \
5273 if (ptr == tconn) { \
5274 fr_fprintf(stderr, "trunk_search: tconn %p on " #_heap "\n", ptr); \
5275 return true; \
5276 } \
5277 if (trunk_connection_search(tconn, ptr)) { \
5278 fr_fprintf(stderr, " on tconn %p on " #_heap "\n", tconn); \
5279 return true; \
5280 } \
5281 }}\
5282} while (0)
5283
5284 TCONN_DLIST_SEARCH(init);
5285 TCONN_DLIST_SEARCH(connecting);
5286 TCONN_MINMAX_HEAP_SEARCH(active);
5287 TCONN_DLIST_SEARCH(full);
5288 TCONN_DLIST_SEARCH(inactive);
5289 TCONN_DLIST_SEARCH(inactive_draining);
5290 TCONN_DLIST_SEARCH(failed);
5291 TCONN_DLIST_SEARCH(closed);
5292 TCONN_DLIST_SEARCH(draining);
5293 TCONN_DLIST_SEARCH(draining_to_free);
5294 TCONN_DLIST_SEARCH(to_free);
5295
5296 return false;
5297}
5298
5299bool trunk_connection_search(trunk_connection_t *tconn, void *ptr)
5300{
5301#define TREQ_DLIST_SEARCH(_dlist) \
5302do { \
5303 fr_dlist_foreach(&(tconn->_dlist), trunk_request_t, treq) { \
5304 if (ptr == treq) { \
5305 fr_fprintf(stderr, "trunk_search: treq %p on " #_dlist "\n", ptr); \
5306 return true; \
5307 } \
5308 if (trunk_request_search(treq, ptr)) { \
5309 fr_fprintf(stderr, "trunk_search: preq %p found on " #_dlist, ptr); \
5310 return true; \
5311 } \
5312 } \
5313} while (0)
5314
5315#define TREQ_HEAP_SEARCH(_heap) \
5316do { \
5317 fr_heap_iter_t _iter; \
5318 for (trunk_request_t *treq = fr_heap_iter_init(tconn->_heap, &_iter); \
5319 treq; \
5320 treq = fr_heap_iter_next(tconn->_heap, &_iter)) { \
5321 if (ptr == treq) { \
5322 fr_fprintf(stderr, "trunk_search: treq %p in " #_heap "\n", ptr); \
5323 return true; \
5324 } \
5325 if (trunk_request_search(treq, ptr)) { \
5326 fr_fprintf(stderr, "trunk_search: preq %p found in " #_heap, ptr); \
5327 return true; \
5328 } \
5329 } \
5330} while (0)
5331
5332#define TREQ_OPTION_SEARCH(_option) \
5333do { \
5334 if (tconn->_option) { \
5335 if (ptr == tconn->_option) { \
5336 fr_fprintf(stderr, "trunk_search: treq %p is " #_option "\n", ptr); \
5337 return true; \
5338 } \
5339 if (trunk_request_search(tconn->_option, ptr)) { \
5340 fr_fprintf(stderr, "trunk_search: preq %p found in " #_option, ptr); \
5341 return true; \
5342 } \
5343 } \
5344} while (0)
5345
5346 /* search associated requests */
5347 TREQ_HEAP_SEARCH(pending);
5348 TREQ_DLIST_SEARCH(sent);
5349 TREQ_DLIST_SEARCH(cancel);
5350 TREQ_DLIST_SEARCH(cancel_sent);
5351 TREQ_OPTION_SEARCH(partial);
5352 TREQ_OPTION_SEARCH(cancel_partial);
5353
5354 return false;
5355}
5356
5357bool trunk_request_search(trunk_request_t *treq, void *ptr)
5358{
5359 return treq->pub.preq == ptr;
5360}
5361#endif
file
int const char * file
Definition acutest.h:702
line
int const char int line
Definition acutest.h:702
request_verify
void request_verify(UNUSED char const *file, UNUSED int line, UNUSED request_t *request)
Definition atomic_queue_test.c:43
L
#define L(_str)
Helper for initialising arrays of string literals.
Definition build.h:228
NDEBUG_UNUSED
#define NDEBUG_UNUSED
Definition build.h:347
FALL_THROUGH
#define FALL_THROUGH
clang 10 doesn't recognised the FALL-THROUGH comment anymore
Definition build.h:343
DIAG_ON
#define DIAG_ON(_x)
Definition build.h:487
unlikely
#define unlikely(_x)
Definition build.h:407
UNUSED
#define UNUSED
Definition build.h:336
NUM_ELEMENTS
#define NUM_ELEMENTS(_t)
Definition build.h:358
DIAG_OFF
#define DIAG_OFF(_x)
Definition build.h:486
cf_pair_parse_value
int cf_pair_parse_value(TALLOC_CTX *ctx, void *out, UNUSED void *base, CONF_ITEM *ci, conf_parser_t const *rule)
Parses a CONF_PAIR into a C data type.
Definition cf_parse.c:213
CONF_PARSER_TERMINATOR
#define CONF_PARSER_TERMINATOR
Definition cf_parse.h:669
conf_parser_s::func
cf_parse_t func
Override default parsing behaviour for the specified type with a custom parsing function.
Definition cf_parse.h:623
FR_CONF_OFFSET
#define FR_CONF_OFFSET(_name, _struct, _field)
conf_parser_t which parses a single CONF_PAIR, writing the result to a field in a struct
Definition cf_parse.h:280
FR_CONF_POINTER
#define FR_CONF_POINTER(_name, _type, _flags, _res_p)
conf_parser_t which parses a single CONF_PAIR producing a single global result
Definition cf_parse.h:334
FR_CONF_OFFSET_SUBSECTION
#define FR_CONF_OFFSET_SUBSECTION(_name, _flags, _struct, _field, _subcs)
conf_parser_t which populates a sub-struct using a CONF_SECTION
Definition cf_parse.h:309
CONF_FLAG_SUBSECTION
@ CONF_FLAG_SUBSECTION
Instead of putting the information into a configuration structure, the configuration file routines MA...
Definition cf_parse.h:423
conf_parser_s
Defines a CONF_PAIR to C data type mapping.
Definition cf_parse.h:606
cf_item
Common header for all CONF_* types.
Definition cf_priv.h:54
cf_pair
Configuration AVP similar to a fr_pair_t.
Definition cf_priv.h:77
cf_section
A section grouping multiple CONF_PAIR.
Definition cf_priv.h:106
cf_section_find
CONF_SECTION * cf_section_find(CONF_SECTION const *cs, char const *name1, char const *name2)
Find a CONF_SECTION with name1 and optionally name2.
Definition cf_util.c:1194
cf_item_to_section
CONF_SECTION * cf_item_to_section(CONF_ITEM const *ci)
Cast a CONF_ITEM to a CONF_SECTION.
Definition cf_util.c:692
cf_section_name
char const * cf_section_name(CONF_SECTION const *cs)
Return name2 if set, else name1.
Definition cf_util.c:1364
cf_parent
#define cf_parent(_cf)
Definition cf_util.h:118
connection_state_t
connection_state_t
Definition connection.h:47
CONNECTION_STATE_FAILED
@ CONNECTION_STATE_FAILED
Connection has failed.
Definition connection.h:56
CONNECTION_STATE_HALTED
@ CONNECTION_STATE_HALTED
The connection is in a halted stat.
Definition connection.h:48
CONNECTION_STATE_CLOSED
@ CONNECTION_STATE_CLOSED
Connection has been closed.
Definition connection.h:57
CONNECTION_STATE_CONNECTED
@ CONNECTION_STATE_CONNECTED
File descriptor is open (ready for writing).
Definition connection.h:54
CONNECTION_STATE_INIT
@ CONNECTION_STATE_INIT
Init state, sets up connection.
Definition connection.h:51
CONNECTION_STATE_CONNECTING
@ CONNECTION_STATE_CONNECTING
Waiting for connection to establish.
Definition connection.h:52
CONNECTION_STATE_SHUTDOWN
@ CONNECTION_STATE_SHUTDOWN
Connection is shutting down.
Definition connection.h:55
connection_reason_t
connection_reason_t
Definition connection.h:84
connection_conf_t
Definition connection.h:92
min
static size_t min(size_t x, size_t y)
Definition dbuff.c:66
fr_cond_assert
#define fr_cond_assert(_x)
Calls panic_action ifndef NDEBUG, else logs error and evaluates to value of _x.
Definition debug.h:131
fr_assert_msg
#define fr_assert_msg(_x, _msg,...)
Calls panic_action ifndef NDEBUG, else logs error and causes the server to exit immediately with code...
Definition debug.h:202
fr_cond_assert_msg
#define fr_cond_assert_msg(_x, _fmt,...)
Calls panic_action ifndef NDEBUG, else logs error and evaluates to value of _x.
Definition debug.h:148
fr_fatal_assert_msg
#define fr_fatal_assert_msg(_x, _fmt,...)
Calls panic_action ifndef NDEBUG, else logs error and causes the server to exit immediately with code...
Definition debug.h:176
MEM
#define MEM(x)
Definition debug.h:36
DEBUG
#define DEBUG(fmt,...)
Definition dhcpclient.c:38
fr_dlist_init
#define fr_dlist_init(_head, _type, _field)
Initialise the head structure of a doubly linked list.
Definition dlist.h:242
fr_dlist_head
static void * fr_dlist_head(fr_dlist_head_t const *list_head)
Return the HEAD item of a list or NULL if the list is empty.
Definition dlist.h:468
_fr_dlist_verify
static void _fr_dlist_verify(char const *file, int line, fr_dlist_head_t const *list_head)
Check all items in the list are valid.
Definition dlist.h:717
fr_dlist_remove
static void * fr_dlist_remove(fr_dlist_head_t *list_head, void *ptr)
Remove an item from the list.
Definition dlist.h:620
fr_dlist_entry_in_list
static bool fr_dlist_entry_in_list(fr_dlist_t const *entry)
Check if a list entry is part of a list.
Definition dlist.h:145
fr_dlist_talloc_free
static void fr_dlist_talloc_free(fr_dlist_head_t *head)
Free all items in a doubly linked list (with talloc)
Definition dlist.h:892
fr_dlist_prev
static void * fr_dlist_prev(fr_dlist_head_t const *list_head, void const *ptr)
Get the previous item in a list.
Definition dlist.h:570
fr_dlist_num_elements
static unsigned int fr_dlist_num_elements(fr_dlist_head_t const *head)
Return the number of elements in the dlist.
Definition dlist.h:921
fr_dlist_pop_head
static void * fr_dlist_pop_head(fr_dlist_head_t *list_head)
Remove the head item in a list.
Definition dlist.h:654
fr_dlist_tail
static void * fr_dlist_tail(fr_dlist_head_t const *list_head)
Return the TAIL item of a list or NULL if the list is empty.
Definition dlist.h:513
fr_dlist_insert_tail
static int fr_dlist_insert_tail(fr_dlist_head_t *list_head, void *ptr)
Insert an item into the tail of a list.
Definition dlist.h:360
fr_dlist_verify
#define fr_dlist_verify(_head)
Definition dlist.h:737
fr_dlist_talloc_init
#define fr_dlist_talloc_init(_head, _type, _field)
Initialise the head structure of a doubly linked list.
Definition dlist.h:257
fr_dlist_insert_head
static int fr_dlist_insert_head(fr_dlist_head_t *list_head, void *ptr)
Insert an item into the head of a list.
Definition dlist.h:320
fr_dlist_next
static void * fr_dlist_next(fr_dlist_head_t const *list_head, void const *ptr)
Get the next item in a list.
Definition dlist.h:537
fr_dlist_head_t
Head of a doubly linked list.
Definition dlist.h:51
fr_dlist_s
Entry in a doubly linked list.
Definition dlist.h:41
fr_heap_insert
int fr_heap_insert(fr_heap_t **hp, void *data)
Insert a new element into the heap.
Definition heap.c:146
fr_heap_index_t
unsigned int fr_heap_index_t
Definition heap.h:80
fr_heap_peek
static void * fr_heap_peek(fr_heap_t *h)
Return the item from the top of the heap but don't pop it.
Definition heap.h:136
FR_HEAP_VERIFY
#define FR_HEAP_VERIFY(_heap)
Definition heap.h:212
fr_heap_num_elements
static unsigned int fr_heap_num_elements(fr_heap_t *h)
Return the number of elements in the heap.
Definition heap.h:179
fr_heap_talloc_alloc
#define fr_heap_talloc_alloc(_ctx, _cmp, _talloc_type, _field, _init)
Creates a heap that verifies elements are of a specific talloc type.
Definition heap.h:115
fr_heap_t
The main heap structure.
Definition heap.h:66
talloc_free
talloc_free(hp)
PERROR
#define PERROR(_fmt,...)
Definition log.h:228
DEBUG3
#define DEBUG3(_fmt,...)
Definition log.h:266
ROPTIONAL
#define ROPTIONAL(_l_request, _l_global, _fmt,...)
Use different logging functions depending on whether request is NULL or not.
Definition log.h:540
RDEBUG3
#define RDEBUG3(fmt,...)
Definition log.h:355
RWARN
#define RWARN(fmt,...)
Definition log.h:309
DEBUG4
#define DEBUG4(_fmt,...)
Definition log.h:267
RATE_LIMIT_LOCAL_ROPTIONAL
#define RATE_LIMIT_LOCAL_ROPTIONAL(_entry, _l_request, _l_global, _fmt,...)
Rate limit messages using a local limiting entry.
Definition log.h:618
fr_rate_limit_t
Track when a log message was last repeated.
Definition log.h:559
fr_time
#define fr_time()
Definition event.c:60
fr_event_list_s
Stores all information relating to an event list.
Definition event.c:377
fr_log
void fr_log(fr_log_t const *log, fr_log_type_t type, char const *file, int line, char const *fmt,...)
Send a server log message to its destination.
Definition log.c:616
fr_log_type_t
fr_log_type_t
Definition log.h:51
ROUND_UP_DIV
#define ROUND_UP_DIV(_x, _y)
Get the ceiling value of integer division.
Definition math.h:211
uint16_t
unsigned short uint16_t
Definition merged_model.c:31
uint32_t
unsigned int uint32_t
Definition merged_model.c:33
request_t
Definition merged_model.c:210
fr_minmax_heap_insert
int fr_minmax_heap_insert(fr_minmax_heap_t *hp, void *data)
Definition minmax_heap.c:424
fr_minmax_heap_iter_next
void * fr_minmax_heap_iter_next(fr_minmax_heap_t *hp, fr_minmax_heap_iter_t *iter)
Get the next entry in a minmax heap.
Definition minmax_heap.c:573
fr_minmax_heap_min_peek
void * fr_minmax_heap_min_peek(fr_minmax_heap_t *hp)
Definition minmax_heap.c:449
fr_minmax_heap_max_peek
void * fr_minmax_heap_max_peek(fr_minmax_heap_t *hp)
Definition minmax_heap.c:466
fr_minmax_heap_num_elements
unsigned int fr_minmax_heap_num_elements(fr_minmax_heap_t *hp)
Return the number of elements in the minmax heap.
Definition minmax_heap.c:533
fr_minmax_heap_iter_init
void * fr_minmax_heap_iter_init(fr_minmax_heap_t *hp, fr_minmax_heap_iter_t *iter)
Iterate over entries in a minmax heap.
Definition minmax_heap.c:551
fr_minmax_heap_extract
int fr_minmax_heap_extract(fr_minmax_heap_t *hp, void *data)
Definition minmax_heap.c:486
fr_minmax_heap_s
Definition minmax_heap.c:53
fr_minmax_heap_iter_t
unsigned int fr_minmax_heap_iter_t
Definition minmax_heap.h:38
fr_minmax_heap_talloc_alloc
#define fr_minmax_heap_talloc_alloc(_ctx, _cmp, _talloc_type, _field, _init)
Creates a minmax heap that verifies elements are of a specific talloc type.
Definition minmax_heap.h:85
fr_pointer_cmp
int8_t fr_pointer_cmp(void const *a, void const *b)
Compares two pointers.
Definition misc.c:449
request_prioritise
static int8_t request_prioritise(void const *one, void const *two)
Definition bio.c:1153
fr_assert
#define fr_assert(_expr)
Definition rad_assert.h:37
RDEBUG
#define RDEBUG(fmt,...)
Definition radclient-ng.h:53
DEBUG2
#define DEBUG2(fmt,...)
Definition radclient-ng.h:43
WARN
#define WARN(fmt,...)
Definition radclient-ng.h:47
done
static bool done
Definition radclient.c:80
INFO
#define INFO(fmt,...)
Definition radict.c:63
events
static fr_event_list_t * events
Definition radsniff.c:58
conf
static rs_t * conf
Definition radsniff.c:52
connection_signal_shutdown
void connection_signal_shutdown(connection_t *conn)
Shuts down a connection gracefully.
Definition connection.c:1286
connection_del_watch_post
int connection_del_watch_post(connection_t *conn, connection_state_t state, connection_watch_t watch)
Remove a watch function from a post list.
Definition connection.c:483
connection_signal_halt
void connection_signal_halt(connection_t *conn)
Shuts down a connection ungracefully.
Definition connection.c:1349
connection_signals_resume
void connection_signals_resume(connection_t *conn)
Resume processing of deferred signals.
Definition connection.c:330
connection_signal_reconnect
void connection_signal_reconnect(connection_t *conn, connection_reason_t reason)
Asynchronously signal the connection should be reconnected.
Definition connection.c:1223
connection_signal_init
void connection_signal_init(connection_t *conn)
Asynchronously signal a halted connection to start.
Definition connection.c:1163
connection_alloc
connection_t * connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, connection_funcs_t const *funcs, connection_conf_t const *conf, char const *log_prefix, void const *uctx)
Allocate a new connection.
Definition connection.c:1571
connection_add_watch_pre
connection_watch_entry_t * connection_add_watch_pre(connection_t *conn, connection_state_t state, connection_watch_t watch, bool oneshot, void const *uctx)
Add a callback to be executed before a state function has been called.
Definition connection.c:521
connection_add_watch_post
connection_watch_entry_t * connection_add_watch_post(connection_t *conn, connection_state_t state, connection_watch_t watch, bool oneshot, void const *uctx)
Add a callback to be executed after a state function has been called.
Definition connection.c:543
connection_del_watch_pre
int connection_del_watch_pre(connection_t *conn, connection_state_t state, connection_watch_t watch)
Remove a watch function from a pre list.
Definition connection.c:466
connection_signals_pause
void connection_signals_pause(connection_t *conn)
Pause processing of deferred signals.
Definition connection.c:321
connection_s
Definition connection.c:87
test_time
static fr_time_t test_time(void)
Definition slab_tests.c:43
test_time_base
static fr_time_t test_time_base
Definition slab_tests.c:42
count
return count
Definition module.c:155
init
init
Enter the EAP-IDENTITY state.
Definition state_machine.c:99
memory_order_relaxed
@ memory_order_relaxed
Definition stdatomic.h:127
atomic_fetch_add_explicit
#define atomic_fetch_add_explicit(object, operand, order)
Definition stdatomic.h:302
ATOMIC_VAR_INIT
#define ATOMIC_VAR_INIT(value)
Definition stdatomic.h:88
fr_log_s
Definition log.h:93
pair_list_s
Definition pair.h:52
fr_table_str_by_value
#define fr_table_str_by_value(_table, _number, _def)
Convert an integer to a string.
Definition table.h:804
fr_table_num_indexed_bit_pos_t
An element in a table indexed by bit position.
Definition table.h:83
fr_table_num_ordered_t
An element in an arbitrarily ordered array of name to num mappings.
Definition table.h:57
talloc_get_type_abort_const
#define talloc_get_type_abort_const
Definition talloc.h:117
talloc_pooled_object
#define talloc_pooled_object(_ctx, _type, _num_subobjects, _total_subobjects_size)
Definition talloc.h:211
talloc_strdup
#define talloc_strdup(_ctx, _str)
Definition talloc.h:149
fr_time_gteq
#define fr_time_gteq(_a, _b)
Definition time.h:238
fr_time_delta_wrap
#define fr_time_delta_wrap(_time)
Definition time.h:152
fr_time_wrap
#define fr_time_wrap(_time)
Definition time.h:145
fr_time_lteq
#define fr_time_lteq(_a, _b)
Definition time.h:240
fr_time_delta_ispos
#define fr_time_delta_ispos(_a)
Definition time.h:290
fr_time_add
#define fr_time_add(_a, _b)
Add a time/time delta together.
Definition time.h:196
fr_time_gt
#define fr_time_gt(_a, _b)
Definition time.h:237
fr_time_sub
#define fr_time_sub(_a, _b)
Subtract one time from another.
Definition time.h:229
fr_time_lt
#define fr_time_lt(_a, _b)
Definition time.h:239
fr_time_delta_gt
#define fr_time_delta_gt(_a, _b)
Definition time.h:283
fr_time_s
"server local" time.
Definition time.h:69
fr_timer_list_s
An event timer list.
Definition timer.c:49
fr_timer_s
A timer event.
Definition timer.c:83
FR_TIMER_DELETE
#define FR_TIMER_DELETE(_ev_p)
Definition timer.h:103
FR_TIMER_DELETE_RETURN
#define FR_TIMER_DELETE_RETURN(_ev_p)
Definition timer.h:110
fr_timer_in
#define fr_timer_in(...)
Definition timer.h:87
FR_TIMER_DISARM
#define FR_TIMER_DISARM(_ev)
Definition timer.h:91
trunk_search
bool trunk_search(trunk_t *trunk, void *ptr)
Definition trunk.c:5254
request_counter
static atomic_uint_fast64_t request_counter
Definition trunk.c:54
trunk_s::trigger_cp
CONF_PAIR * trigger_cp[NUM_ELEMENTS(trunk_conn_trigger_names)]
Cached trigger CONF_PAIRs.
Definition trunk.c:319
trunk_connection_enter_active
static void trunk_connection_enter_active(trunk_connection_t *tconn)
Transition a connection back to the active state.
Definition trunk.c:3301
CONN_REORDER
#define CONN_REORDER(_tconn)
Reorder the connections in the active heap.
Definition trunk.c:792
trunk_req_trigger_names_len
static size_t trunk_req_trigger_names_len
Definition trunk.c:388
trunk_connection_pop_cancellation
int trunk_connection_pop_cancellation(trunk_request_t **treq_out, trunk_connection_t *tconn)
Pop a cancellation request off a connection's cancellation queue.
Definition trunk.c:3900
trunk_connection_s::cancel
fr_dlist_head_t cancel
Requests in the cancel state.
Definition trunk.c:161
trunk_connection_manage_schedule
int trunk_connection_manage_schedule(trunk_t *trunk)
Schedule a trunk management event for the next time the event loop is executed.
Definition trunk.c:4896
REQUEST_EXTRACT_SENT
#define REQUEST_EXTRACT_SENT(_treq)
Remove the current request from the sent list.
Definition trunk.c:762
_trunk_connection_on_shutdown
static void _trunk_connection_on_shutdown(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection transitioned to the shutdown state.
Definition trunk.c:3451
trunk_watch_entry_t
struct trunk_watch_entry_s trunk_watch_entry_t
An entry in a trunk watch function list.
trunk_connection_s::reapable
fr_dlist_head_t reapable
Idle request.
Definition trunk.c:159
trunk_connection_s::pending
fr_heap_t * pending
Requests waiting to be sent.
Definition trunk.c:153
trunk_s::conf
trunk_conf_t conf
Trunk common configuration.
Definition trunk.c:224
trunk_connection_states_len
static size_t trunk_connection_states_len
Definition trunk.c:427
REQUEST_EXTRACT_REAPABLE
#define REQUEST_EXTRACT_REAPABLE(_treq)
Remove the current request from the reapable list.
Definition trunk.c:767
trunk_request_state_log_t::tconn
trunk_connection_t * tconn
The request was associated with.
Definition trunk.c:82
trunk_connection_callback_readable
void trunk_connection_callback_readable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
Standard I/O read function.
Definition trunk.c:4079
trunk_s::limit_last_failure_log
fr_rate_limit_t limit_last_failure_log
Rate limit on "Refusing to enqueue requests - No active conns".
Definition trunk.c:298
trunk_verify
void trunk_verify(char const *file, int line, trunk_t *trunk)
Verify a trunk.
Definition trunk.c:5122
trunk_s::manage_ev
fr_timer_t * manage_ev
Periodic connection management event.
Definition trunk.c:290
IN_HANDLER
#define IN_HANDLER(_trunk)
Definition trunk.c:722
trunk_connection_states
static fr_table_num_ordered_t const trunk_connection_states[]
Definition trunk.c:415
trunk_reconnect
void trunk_reconnect(trunk_t *trunk, int states, connection_reason_t reason)
Force the trunk to re-establish its connections.
Definition trunk.c:4792
trunk_connection_callback_writable
void trunk_connection_callback_writable(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx)
Standard I/O write function.
Definition trunk.c:4096
trunk_watch_entry_s::uctx
void * uctx
User data to pass to the function.
Definition trunk.c:191
trunk_request_enter_pending
static void trunk_request_enter_pending(trunk_request_t *treq, trunk_connection_t *tconn, bool new)
Transition a request to the pending state, adding it to the backlog of an active connection.
Definition trunk.c:1198
trunk_request_remove_from_conn
static void trunk_request_remove_from_conn(trunk_request_t *treq)
Remove a request from all connection lists.
Definition trunk.c:1007
trunk_s::limit_max_requests_alloc_log
fr_rate_limit_t limit_max_requests_alloc_log
Rate limit on "Refusing to alloc requests - Limit of * requests reached".
Definition trunk.c:296
trunk_request_state_log_t::to
trunk_request_state_t to
What state we transitioned to.
Definition trunk.c:80
_trunk_request_prioritise
static int8_t _trunk_request_prioritise(void const *a, void const *b)
Compare two protocol requests.
Definition trunk.c:983
trunk_manage
static void trunk_manage(trunk_t *trunk, fr_time_t now)
Implements the algorithm we use to manage requests per connection levels.
Definition trunk.c:4228
_trunk_connection_free
static int _trunk_connection_free(trunk_connection_t *tconn)
Free a connection.
Definition trunk.c:3745
trunk_s::funcs
trunk_io_funcs_t funcs
I/O functions.
Definition trunk.c:276
trunk_s::draining
fr_dlist_head_t draining
Connections that will be freed once all their requests are complete, but can be reactivated.
Definition trunk.c:261
REQUEST_EXTRACT_CANCEL_PARTIAL
#define REQUEST_EXTRACT_CANCEL_PARTIAL(_treq)
Remove the current request from the cancel_partial slot.
Definition trunk.c:777
trunk_trigger_cf_parse
int trunk_trigger_cf_parse(TALLOC_CTX *ctx, void *out, void *parent, CONF_ITEM *ci, conf_parser_t const *rule)
Check for a module trigger section when parsing the triggers option.
Definition trunk.c:5090
trunk_start
int trunk_start(trunk_t *trunk)
Start the trunk running.
Definition trunk.c:4831
trunk_request_signal_partial
void trunk_request_signal_partial(trunk_request_t *treq)
Signal a partial write.
Definition trunk.c:2075
trunk_request_signal_fail
void trunk_request_signal_fail(trunk_request_t *treq)
Signal that a trunk request failed.
Definition trunk.c:2178
TREQ_OPTION_SEARCH
#define TREQ_OPTION_SEARCH(_option)
trunk_request_signal_cancel_sent
void trunk_request_signal_cancel_sent(trunk_request_t *treq)
Signal that a remote server has been notified of the cancellation.
Definition trunk.c:2306
trunk_connection_enter_draining_to_free
static void trunk_connection_enter_draining_to_free(trunk_connection_t *tconn)
Transition a connection to the draining-to-reconnect state.
Definition trunk.c:3265
trunk_watch_entry_s::func
trunk_watch_t func
Function to call when a trunk enters the state this list belongs to.
Definition trunk.c:187
trunk_connection_signal_readable
void trunk_connection_signal_readable(trunk_connection_t *tconn)
Signal that a trunk connection is readable.
Definition trunk.c:3986
DO_REQUEST_FREE
#define DO_REQUEST_FREE(_treq)
Call the free callback (if set)
Definition trunk.c:614
trunk_request_alloc
trunk_request_t * trunk_request_alloc(trunk_t *trunk, request_t *request)
(Pre-)Allocate a new trunk request
Definition trunk.c:2524
_trunk_connection_on_halted
static void _trunk_connection_on_halted(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection transitioned to the halted state.
Definition trunk.c:3697
REQUEST_EXTRACT_BACKLOG
#define REQUEST_EXTRACT_BACKLOG(_treq)
Remove the current request from the backlog.
Definition trunk.c:733
trunk_connection_s::heap_id
fr_heap_index_t heap_id
Used to track the connection in the connected heap.
Definition trunk.c:138
trunk_s::closed
fr_dlist_head_t closed
Connections that have closed.
Definition trunk.c:258
trunk_s::watch
fr_dlist_head_t watch[TRUNK_STATE_MAX]
To be called when trunk changes state.
Definition trunk.c:282
trunk_watch_call
static void trunk_watch_call(trunk_t *trunk, fr_dlist_head_t *list, trunk_state_t state)
Call a list of watch functions associated with a state.
Definition trunk.c:833
trunk_request_enter_cancel_complete
static void trunk_request_enter_cancel_complete(trunk_request_t *treq)
Cancellation was acked, the request is complete, free it.
Definition trunk.c:1550
trunk_request_state_log_t::line
int line
Line change occurred on.
Definition trunk.c:92
trunk_connection_enter_inactive_draining
static void trunk_connection_enter_inactive_draining(trunk_connection_t *tconn)
Transition a connection to the inactive-draining state.
Definition trunk.c:3203
CONN_STATE_TRANSITION
#define CONN_STATE_TRANSITION(_new, _log)
Definition trunk.c:457
trunk_requests_per_connection
static uint64_t trunk_requests_per_connection(uint16_t *conn_count_out, uint32_t *req_conn_out, trunk_t *trunk, fr_time_t now, NDEBUG_UNUSED bool verify)
Update timestamps for when we last had a transition from above target to below target or vice versa.
Definition trunk.c:4638
trunk_connection_events_len
static size_t trunk_connection_events_len
Definition trunk.c:443
_trunk_connection_on_failed
static void _trunk_connection_on_failed(connection_t *conn, connection_state_t prev, connection_state_t state, void *uctx)
Connection failed.
Definition trunk.c:3645
trunk_watch_entry_s::oneshot
bool oneshot
Remove the function after it's called once.
Definition trunk.c:189
trunk_s::started
bool started
Has the trunk been started.
Definition trunk.c:307
trunk_states_len
static size_t trunk_states_len
Definition trunk.c:413
TCONN_DLIST_VERIFY
#define TCONN_DLIST_VERIFY(_dlist, _state)
IO_FUNC_VERIFY
#define IO_FUNC_VERIFY(_func)
trunk_request_count_by_connection
uint32_t trunk_request_count_by_connection(trunk_connection_t const *tconn, int req_state)
Return the count number of requests associated with a trunk connection.
Definition trunk.c:2932
trunk_s::last_req_per_conn
uint64_t last_req_per_conn
The last request to connection ratio we calculated.
Definition trunk.c:312
DO_REQUEST_COMPLETE
#define DO_REQUEST_COMPLETE(_treq)
Call the complete callback (if set)
Definition trunk.c:575
trunk_connection_auto_full
static void trunk_connection_auto_full(trunk_connection_t *tconn)
Automatically mark a connection as full.
Definition trunk.c:2951
trunk_connection_remove
static void trunk_connection_remove(trunk_connection_t *tconn)
Remove a trunk connection from whichever list it's currently in.
Definition trunk.c:3102
TRUNK_REQUEST_STATE_LOG_MAX
#define TRUNK_REQUEST_STATE_LOG_MAX
The maximum number of state logs to record per request.
Definition trunk.c:71
trunk_connection_writable
static void trunk_connection_writable(trunk_connection_t *tconn)
A connection is writable.
Definition trunk.c:3015
OVER_MAX_CHECK
#define OVER_MAX_CHECK
trunk_connection_s::events
trunk_connection_event_t events
The current events we expect to be notified on.
Definition trunk.c:147
trunk_add_watch
trunk_watch_entry_t * trunk_add_watch(trunk_t *trunk, trunk_state_t state, trunk_watch_t watch, bool oneshot, void const *uctx)
Add a watch entry to the trunk state list.
Definition trunk.c:909
_trunk_free
static int _trunk_free(trunk_t *trunk)
Free a trunk, gracefully closing all connections.
Definition trunk.c:4929
trunk_s::failed
fr_dlist_head_t failed
Connections that'll be reconnected shortly.
Definition trunk.c:256
trunk_rebalance
static void trunk_rebalance(trunk_t *trunk)
Rebalance connections across active trunk members when a new connection becomes active.
Definition trunk.c:4164
trunk_backlog_drain
static void trunk_backlog_drain(trunk_t *trunk)
Drain the backlog of as many requests as possible.
Definition trunk.c:4737
DO_REQUEST_CANCEL
#define DO_REQUEST_CANCEL(_treq, _reason)
Call the cancel callback if set.
Definition trunk.c:536
FR_TRUNK_LIST_FUNC
#define FR_TRUNK_LIST_FUNC(_list, _type)
Definition trunk.c:804
_trunk_connection_order_by_shortest_queue
static int8_t _trunk_connection_order_by_shortest_queue(void const *one, void const *two)
Order connections by queue depth.
Definition trunk.c:4912
trunk_request_s::pub
struct trunk_request_pub_s pub
Public fields in the trunk request.
Definition trunk.c:100
TCONN_MINMAX_HEAP_VERIFY
#define TCONN_MINMAX_HEAP_VERIFY(_heap, _state)
trunk_connection_s::cancel_partial
trunk_request_t * cancel_partial
Partially written cancellation request.
Definition trunk.c:163
TCONN_MINMAX_HEAP_SEARCH
#define TCONN_MINMAX_HEAP_SEARCH(_heap)
trunk_connection_requests_requeue
uint64_t trunk_connection_requests_requeue(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
Move requests off of a connection and requeue elsewhere.
Definition trunk.c:2056
trunk_watch_entry_s::enabled
bool enabled
Whether the watch entry is enabled.
Definition trunk.c:190
trunk_request_s::last_freed
fr_time_t last_freed
Last time this request was freed.
Definition trunk.c:113
DO_REQUEST_CONN_RELEASE
#define DO_REQUEST_CONN_RELEASE(_treq)
Call the "conn_release" callback (if set)
Definition trunk.c:557
TREQ_DLIST_SEARCH
#define TREQ_DLIST_SEARCH(_dlist)
REQUEST_EXTRACT_CANCEL
#define REQUEST_EXTRACT_CANCEL(_treq)
Remove the current request from the cancel list.
Definition trunk.c:772
trunk_connection_is_full
static bool trunk_connection_is_full(trunk_connection_t *tconn)
Return whether a trunk connection should currently be considered full.
Definition trunk.c:2974
trunk_s::pub
struct trunk_pub_s pub
Public fields in the trunk connection.
Definition trunk.c:216
trunk_request_s::cancel_reason
trunk_cancel_reason_t cancel_reason
Why this request was cancelled.
Definition trunk.c:111
REQUEST_BAD_STATE_TRANSITION
#define REQUEST_BAD_STATE_TRANSITION(_new)
Definition trunk.c:502
trunk_request_enqueue_on_conn
trunk_enqueue_t trunk_request_enqueue_on_conn(trunk_request_t **treq_out, trunk_connection_t *tconn, request_t *request, void *preq, void *rctx, bool ignore_limits)
Enqueue additional requests on a specific connection.
Definition trunk.c:2793
_trunk_connection_on_closed
static void _trunk_connection_on_closed(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection failed after it was connected.
Definition trunk.c:3574
trunk_connection_events
static fr_table_num_ordered_t const trunk_connection_events[]
Definition trunk.c:437
trunk_request_enqueue
trunk_enqueue_t trunk_request_enqueue(trunk_request_t **treq_out, trunk_t *trunk, request_t *request, void *preq, void *rctx)
Enqueue a request that needs data written to the trunk.
Definition trunk.c:2639
TCONN_DLIST_SEARCH
#define TCONN_DLIST_SEARCH(_dlist)
trunk_request_enter_unassigned
static void trunk_request_enter_unassigned(trunk_request_t *treq)
Transition a request to the unassigned state, in preparation for re-assignment.
Definition trunk.c:1101
trunk_request_t
struct trunk_request_s trunk_request_t
Definition trunk.c:33
trunk_s::in_handler
void * in_handler
Which handler we're inside.
Definition trunk.c:278
trunk_s::freeing
bool freeing
Trunk is being freed, don't spawn new connections or re-enqueue.
Definition trunk.c:304
trunk_states
static fr_table_num_ordered_t const trunk_states[]
Definition trunk.c:408
trunk_connection_readable
static void trunk_connection_readable(trunk_connection_t *tconn)
A connection is readable.
Definition trunk.c:3005
IS_SERVICEABLE
#define IS_SERVICEABLE(_tconn)
Definition trunk.c:727
trunk_request_requeue
trunk_enqueue_t trunk_request_requeue(trunk_request_t *treq)
Re-enqueue a request on the same connection.
Definition trunk.c:2728
IS_PROCESSING
#define IS_PROCESSING(_tconn)
Definition trunk.c:728
RECONNECT_BY_STATE
#define RECONNECT_BY_STATE(_state, _list)
trunk_connection_enter_draining
static void trunk_connection_enter_draining(trunk_connection_t *tconn)
Transition a connection to the draining state.
Definition trunk.c:3233
trunk_req_trigger_names
static fr_table_num_indexed_bit_pos_t const trunk_req_trigger_names[]
Map request states to trigger names.
Definition trunk.c:373
trunk_request_s::entry
fr_dlist_t entry
Used to track the trunk request in the conn->sent or trunk->backlog request.
Definition trunk.c:108
trunk_connection_close_if_empty
static void trunk_connection_close_if_empty(trunk_t *trunk, fr_dlist_head_t *head)
Close connections in a particular connection list if they have no requests associated with them.
Definition trunk.c:4122
trunk_request_signal_cancel_complete
void trunk_request_signal_cancel_complete(trunk_request_t *treq)
Signal that a remote server acked our cancellation.
Definition trunk.c:2330
trunk_request_check_enqueue
static trunk_enqueue_t trunk_request_check_enqueue(trunk_connection_t **tconn_out, trunk_t *trunk, request_t *request)
Check to see if a trunk request can be enqueued.
Definition trunk.c:1648
DO_REQUEST_MUX
#define DO_REQUEST_MUX(_tconn)
Write one or more requests to a connection.
Definition trunk.c:632
REQUEST_EXTRACT_PARTIAL
#define REQUEST_EXTRACT_PARTIAL(_treq)
Remove the current request from the partial slot.
Definition trunk.c:753
trunk_connection_s::sent
fr_dlist_head_t sent
Sent request.
Definition trunk.c:157
trunk_request_enter_partial
static void trunk_request_enter_partial(trunk_request_t *treq)
Transition a request to the partial state, indicating that is has been partially sent.
Definition trunk.c:1269
trunk_connection_s::lifetime_ev
fr_timer_t * lifetime_ev
Maximum time this connection can be open.
Definition trunk.c:178
trunk_connection_pop_request
int trunk_connection_pop_request(trunk_request_t **treq_out, trunk_connection_t *tconn)
Pop a request off a connection's pending queue.
Definition trunk.c:3948
trunk_s::connecting
fr_dlist_head_t connecting
Connections which are not yet in the open state.
Definition trunk.c:242
TRUNK_STATE_TRANSITION
#define TRUNK_STATE_TRANSITION(_new)
Definition trunk.c:929
trunk_request_signal_cancel
void trunk_request_signal_cancel(trunk_request_t *treq)
Cancel a trunk request.
Definition trunk.c:2198
trunk_request_state_log_entry_add
void trunk_request_state_log_entry_add(char const *function, int line, trunk_request_t *treq, trunk_request_state_t new)
Definition trunk.c:2850
trunk_connection_spawn
static int trunk_connection_spawn(trunk_t *trunk, fr_time_t now)
Attempt to spawn a new connection.
Definition trunk.c:3813
trunk_del_watch
int trunk_del_watch(trunk_t *trunk, trunk_state_t state, trunk_watch_t watch)
Remove a watch function from a trunk state list.
Definition trunk.c:875
_trunk_timer
static void _trunk_timer(fr_timer_list_t *tl, fr_time_t now, void *uctx)
Event to periodically call the connection management function.
Definition trunk.c:4567
trunk_connection_s::pub
struct trunk_connection_pub_s pub
Public fields in the trunk connection.
Definition trunk.c:134
trunk_request_enter_reapable
static void trunk_request_enter_reapable(trunk_request_t *treq)
Transition a request to the reapable state, indicating that it's been sent in its entirety,...
Definition trunk.c:1356
trunk_connection_count_by_state
uint16_t trunk_connection_count_by_state(trunk_t *trunk, int conn_state)
Return the count number of connections in the specified states.
Definition trunk.c:2908
IN_REQUEST_DEMUX
#define IN_REQUEST_DEMUX(_trunk)
Definition trunk.c:724
DO_REQUEST_FAIL
#define DO_REQUEST_FAIL(_treq, _prev_state)
Call the fail callback (if set)
Definition trunk.c:594
trunk_request_enter_cancel
static void trunk_request_enter_cancel(trunk_request_t *treq, trunk_cancel_reason_t reason)
Transition a request to the cancel state, placing it in a connection's cancellation list.
Definition trunk.c:1419
trunk_request_enqueue_existing
static trunk_enqueue_t trunk_request_enqueue_existing(trunk_request_t *treq)
Enqueue a request which has never been assigned to a connection or was previously cancelled.
Definition trunk.c:1726
trunk_s::managing_connections
bool managing_connections
Whether the trunk is allowed to manage (open/close) connections.
Definition trunk.c:309
DO_CONNECTION_ALLOC
#define DO_CONNECTION_ALLOC(_tconn)
Allocate a new connection.
Definition trunk.c:683
trunk_request_state_log_t::function
char const * function
State change occurred in.
Definition trunk.c:91
trunk_request_states_len
static size_t trunk_request_states_len
Definition trunk.c:406
trunk_s::init
fr_dlist_head_t init
Connections which have not yet started connecting.
Definition trunk.c:239
trunk_request_state_log_t::log_head
fr_dlist_head_t * log_head
To allow the log entry to remove itself on free.
Definition trunk.c:77
trunk_request_enter_cancel_partial
static void trunk_request_enter_cancel_partial(trunk_request_t *treq)
Transition a request to the cancel_partial state, placing it in a connection's cancel_partial slot.
Definition trunk.c:1470
_trunk_connection_on_connected
static void _trunk_connection_on_connected(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection transitioned to the connected state.
Definition trunk.c:3508
trunk_alloc
trunk_t * trunk_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, trunk_io_funcs_t const *funcs, trunk_conf_t const *conf, char const *log_prefix, void const *uctx, bool delay_start, fr_pair_list_t *trigger_args)
Allocate a new collection of connections.
Definition trunk.c:5014
trunk_s::to_free
fr_dlist_head_t to_free
Connections we're done with and will free on the next call to trunk_manage.
Definition trunk.c:267
trunk_connection_s::partial
trunk_request_t * partial
Partially written request.
Definition trunk.c:155
trunk_request_enter_failed
static void trunk_request_enter_failed(trunk_request_t *treq)
Request failed, inform the API client and free the request.
Definition trunk.c:1612
trunk_s::active
fr_minmax_heap_t * active
Connections which can service requests.
Definition trunk.c:244
trunk_config
conf_parser_t const trunk_config[]
Config parser definitions to populate a trunk_conf_t.
Definition trunk.c:341
trunk_request_enter_complete
static void trunk_request_enter_complete(trunk_request_t *treq)
Request completed successfully, inform the API client and free the request.
Definition trunk.c:1581
trunk_request_enter_sent
static void trunk_request_enter_sent(trunk_request_t *treq)
Transition a request to the sent state, indicating that it's been sent in its entirety.
Definition trunk.c:1299
DO_REQUEST_CANCEL_MUX
#define DO_REQUEST_CANCEL_MUX(_tconn)
Write one or more cancellation requests to a connection.
Definition trunk.c:665
trunk_connection_enter_full
static void trunk_connection_enter_full(trunk_connection_t *tconn)
Transition a connection to the full state.
Definition trunk.c:3158
trunk_request_free
void trunk_request_free(trunk_request_t **treq_to_free)
If the trunk request is freed then update the target requests.
Definition trunk.c:2368
DO_REQUEST_DEMUX
#define DO_REQUEST_DEMUX(_tconn)
Read one or more requests from a connection.
Definition trunk.c:649
trunk_connection_requests_dequeue
static uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, trunk_connection_t *tconn, int states, uint64_t max)
Shift requests in the specified states onto new connections.
Definition trunk.c:1783
_trunk_request_free
static int _trunk_request_free(trunk_request_t *treq)
Actually free the trunk request.
Definition trunk.c:2491
trunk_s::log_prefix
char const * log_prefix
What to prepend to messages.
Definition trunk.c:220
REQUEST_EXTRACT_PENDING
#define REQUEST_EXTRACT_PENDING(_treq)
Remove the current request from the pending list.
Definition trunk.c:743
_trunk_connection_lifetime_expire
static void _trunk_connection_lifetime_expire(UNUSED fr_timer_list_t *tl, UNUSED fr_time_t now, void *uctx)
Trigger a reconnection of the trunk connection.
Definition trunk.c:3490
trunk_connection_event_update
static void trunk_connection_event_update(trunk_connection_t *tconn)
Update the registrations for I/O events we're interested in.
Definition trunk.c:3038
trunk_config_request
static conf_parser_t const trunk_config_request[]
Definition trunk.c:324
trunk_s::full
fr_dlist_head_t full
Connections which have too many outstanding requests.
Definition trunk.c:246
DEQUEUE_ALL
#define DEQUEUE_ALL(_src_list, _state)
trunk_request_enter_backlog
static void trunk_request_enter_backlog(trunk_request_t *treq, bool new)
Transition a request to the backlog state, adding it to the backlog of the trunk.
Definition trunk.c:1136
trunk_request_states
static fr_table_num_ordered_t const trunk_request_states[]
Definition trunk.c:391
_trunk_connection_on_connecting
static void _trunk_connection_on_connecting(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection transitioned to the connecting state.
Definition trunk.c:3407
trunk_conn_trigger_names
static fr_table_num_indexed_bit_pos_t const trunk_conn_trigger_names[]
Map connection states to trigger names.
Definition trunk.c:198
trunk_s::draining_to_free
fr_dlist_head_t draining_to_free
Connections that will be freed once all their requests are complete.
Definition trunk.c:264
trunk_request_s::id
uint64_t id
Trunk request ID.
Definition trunk.c:104
trunk_connection_s::sent_count
uint64_t sent_count
The number of requests that have been sent using this connection.
Definition trunk.c:171
_trunk_connection_on_init
static void _trunk_connection_on_init(UNUSED connection_t *conn, UNUSED connection_state_t prev, UNUSED connection_state_t state, void *uctx)
Connection transitioned to the init state.
Definition trunk.c:3372
DO_CONNECTION_NOTIFY
#define DO_CONNECTION_NOTIFY(_tconn, _events)
Change what events the connection should be notified about.
Definition trunk.c:705
TREQ_DLIST_VERIFY
#define TREQ_DLIST_VERIFY(_dlist, _state)
trunk_s::inactive
fr_dlist_head_t inactive
Connections which have been signalled to be inactive by the API client.
Definition trunk.c:249
trunk_s::trigger_undef
bool trigger_undef[NUM_ELEMENTS(trunk_conn_trigger_names)]
Record that a specific trigger is undefined.
Definition trunk.c:317
trunk_connection_manage_stop
void trunk_connection_manage_stop(trunk_t *trunk)
Stop the trunk from opening and closing connections in response to load.
Definition trunk.c:4886
TREQ_HEAP_VERIFY
#define TREQ_HEAP_VERIFY(_heap, _state)
trunk_connection_signal_active
void trunk_connection_signal_active(trunk_connection_t *tconn)
Signal a trunk connection is no longer full.
Definition trunk.c:4025
trunk_request_s::log
fr_dlist_head_t log
State change log.
Definition trunk.c:123
trunk_request_state_log_t::tconn_id
uint64_t tconn_id
If the treq was associated with a connection the connection ID.
Definition trunk.c:85
trunk_connection_s::entry
fr_dlist_t entry
Used to track the connection in the connecting, full and failed lists.
Definition trunk.c:141
trunk_request_enter_cancel_sent
static void trunk_request_enter_cancel_sent(trunk_request_t *treq)
Transition a request to the cancel_sent state, placing it in a connection's cancel_sent list.
Definition trunk.c:1505
trunk_connection_enter_inactive
static void trunk_connection_enter_inactive(trunk_connection_t *tconn)
Transition a connection to the inactive state.
Definition trunk.c:3180
trunk_request_state_log_t::from
trunk_request_state_t from
What state we transitioned from.
Definition trunk.c:79
trunk_s::trigger_args
fr_pair_list_t * trigger_args
Passed to trigger.
Definition trunk.c:315
trunk_connection_s::cancel_sent
fr_dlist_head_t cancel_sent
Sent cancellation request.
Definition trunk.c:165
trunk_connection_manage_start
void trunk_connection_manage_start(trunk_t *trunk)
Allow the trunk to open and close connections in response to load.
Definition trunk.c:4875
trunk_s::inactive_draining
fr_dlist_head_t inactive_draining
Connections which have been signalled to be inactive by the API client, which the trunk manager is dr...
Definition trunk.c:252
trunk_connection_signal_inactive
void trunk_connection_signal_inactive(trunk_connection_t *tconn)
Signal a trunk connection cannot accept more requests.
Definition trunk.c:4002
_state_log_entry_free
static int _state_log_entry_free(trunk_request_state_log_t *slog)
Used for sanity checks to ensure all log entries have been freed.
Definition trunk.c:2843
trunk_connection_verify
void trunk_connection_verify(char const *file, int line, trunk_connection_t *tconn)
Definition trunk.c:5183
trunk_s::backlog
fr_heap_t * backlog
The request backlog.
Definition trunk.c:229
IN_REQUEST_CANCEL_MUX
#define IN_REQUEST_CANCEL_MUX(_trunk)
Definition trunk.c:725
trunk_request_verify
void trunk_request_verify(char const *file, int line, trunk_request_t *treq)
Definition trunk.c:5243
trunk_request_count_by_state
uint64_t trunk_request_count_by_state(trunk_t *trunk, int conn_state, int req_state)
Return a count of requests on a connection in a specific state.
Definition trunk.c:4589
trunk_request_signal_cancel_partial
void trunk_request_signal_cancel_partial(trunk_request_t *treq)
Signal a partial cancel write.
Definition trunk.c:2282
trunk_request_signal_sent
void trunk_request_signal_sent(trunk_request_t *treq)
Signal that the request was written to a connection successfully.
Definition trunk.c:2096
COUNT_BY_STATE
#define COUNT_BY_STATE(_state, _list)
trunk_s::uctx
void * uctx
Uctx data to pass to alloc.
Definition trunk.c:280
TREQ_OPTION_VERIFY
#define TREQ_OPTION_VERIFY(_option, _state)
trunk_connection_search
bool trunk_connection_search(trunk_connection_t *tconn, void *ptr)
Definition trunk.c:5299
CONN_BAD_STATE_TRANSITION
#define CONN_BAD_STATE_TRANSITION(_new)
Definition trunk.c:468
trunk_request_s::heap_id
fr_heap_index_t heap_id
Used to track the request conn->pending heap.
Definition trunk.c:106
REQUEST_STATE_TRANSITION
#define REQUEST_STATE_TRANSITION(_new)
Record a request state transition and log appropriate output.
Definition trunk.c:491
trunk_s::next_watcher
trunk_watch_entry_t * next_watcher
Watcher about to be run. Used to prevent nested watchers.
Definition trunk.c:284
trunk_connection_requests_requeue_priv
static uint64_t trunk_connection_requests_requeue_priv(trunk_connection_t *tconn, int states, uint64_t max, bool fail_bound)
Remove requests in specified states from a connection, attempting to distribute them to new connectio...
Definition trunk.c:1894
trunk_request_s::sent
bool sent
Trunk request has been sent at least once.
Definition trunk.c:118
trunk_request_signal_complete
void trunk_request_signal_complete(trunk_request_t *treq)
Signal that a trunk request is complete.
Definition trunk.c:2140
trunk_connection_auto_unfull
static void trunk_connection_auto_unfull(trunk_connection_t *tconn)
Automatically mark a connection as active or reconnect it.
Definition trunk.c:2992
trunk_connection_signal_reconnect
void trunk_connection_signal_reconnect(trunk_connection_t *tconn, connection_reason_t reason)
Signal a trunk connection is no longer viable.
Definition trunk.c:4064
trunk_connection_signal_writable
void trunk_connection_signal_writable(trunk_connection_t *tconn)
Signal that a trunk connection is writable.
Definition trunk.c:3968
trunk_request_search
bool trunk_request_search(trunk_request_t *treq, void *ptr)
Definition trunk.c:5357
trunk_watch_entry_s::entry
fr_dlist_t entry
List entry.
Definition trunk.c:186
trunk_config_connection
static conf_parser_t const trunk_config_connection[]
Definition trunk.c:333
trunk_request_state_log_t::tconn_state
trunk_connection_state_t tconn_state
If the treq was associated with a connection the connection state at the time of the state transition...
Definition trunk.c:87
trunk_request_s::bound_to_conn
bool bound_to_conn
Fail the request if there's an attempt to re-enqueue it.
Definition trunk.c:115
trunk_cancellation_reasons_len
static size_t trunk_cancellation_reasons_len
Definition trunk.c:435
trunk_cancellation_reasons
static fr_table_num_ordered_t const trunk_cancellation_reasons[]
Definition trunk.c:429
trunk_conn_trigger_names_len
static size_t trunk_conn_trigger_names_len
Definition trunk.c:210
trunk_s::el
fr_event_list_t * el
Event list used by this trunk and the connection.
Definition trunk.c:222
trunk_request_state_log
void trunk_request_state_log(fr_log_t const *log, fr_log_type_t log_type, char const *file, int line, trunk_request_t const *treq)
Definition trunk.c:2881
IN_REQUEST_MUX
#define IN_REQUEST_MUX(_trunk)
Definition trunk.c:723
trunk_s::free_requests
fr_dlist_head_t free_requests
Requests in the unassigned state.
Definition trunk.c:226
trunk_connection_in_state
bool trunk_connection_in_state(trunk_connection_t *tconn, int state)
Returns true if the trunk connection is in one of the specified states.
Definition trunk.c:4112
TREQ_HEAP_SEARCH
#define TREQ_HEAP_SEARCH(_heap)
REQUEST_EXTRACT_CANCEL_SENT
#define REQUEST_EXTRACT_CANCEL_SENT(_treq)
Remove the current request from the cancel sent list.
Definition trunk.c:786
trunk_request_state_log_t::entry
fr_dlist_t entry
Entry in the linked list.
Definition trunk.c:78
trunk_request_signal_reapable
void trunk_request_signal_reapable(trunk_request_t *treq)
Signal that the request was written to a connection successfully, but no response is expected.
Definition trunk.c:2118
trunk_connection_s
Associates request queues with a connection.
Definition trunk.c:133
trunk_request_s
Wraps a normal request.
Definition trunk.c:99
trunk_request_state_log_t
Trace state machine changes for a particular request.
Definition trunk.c:76
trunk_s
Main trunk management handle.
Definition trunk.c:215
trunk_watch_entry_s
An entry in a trunk watch function list.
Definition trunk.c:185
trunk_conf_t::max
uint16_t max
Maximum number of connections in the trunk.
Definition trunk.h:232
trunk_conf_t::max_req_per_conn
uint32_t max_req_per_conn
Maximum requests per connection.
Definition trunk.h:241
trunk_pub_s::last_write_success
fr_time_t _CONST last_write_success
Last time we wrote to the connection.
Definition trunk.h:321
trunk_request_pub_s::trunk
trunk_t *_CONST trunk
Trunk this request belongs to.
Definition trunk.h:352
trunk_conf_t::backlog_on_failed_conn
bool backlog_on_failed_conn
Assign requests to the backlog when there are no available connections and the last connection event ...
Definition trunk.h:282
trunk_conf_t::min
uint16_t min
Shouldn't let connections drop below this number.
Definition trunk.h:230
TRUNK_REQUEST_STATE_ALL
#define TRUNK_REQUEST_STATE_ALL
All request states.
Definition trunk.h:196
trunk_request_pub_s::rctx
void *_CONST rctx
Resume ctx of the module.
Definition trunk.h:358
trunk_connection_pub_s::trunk
trunk_t *_CONST trunk
Trunk this connection belongs to.
Definition trunk.h:380
trunk_io_funcs_t::connection_prioritise
fr_heap_cmp_t connection_prioritise
Ordering function for connections.
Definition trunk.h:742
trunk_connection_state_t
trunk_connection_state_t
Used for sanity checks and to track which list the connection is in.
Definition trunk.h:87
TRUNK_CONN_FULL
@ TRUNK_CONN_FULL
Connection is full and can't accept any more requests.
Definition trunk.h:95
TRUNK_CONN_CONNECTING
@ TRUNK_CONN_CONNECTING
Connection is connecting.
Definition trunk.h:90
TRUNK_CONN_DRAINING
@ TRUNK_CONN_DRAINING
Connection will be closed once it has no more outstanding requests, if it's not reactivated.
Definition trunk.h:101
TRUNK_CONN_INACTIVE_DRAINING
@ TRUNK_CONN_INACTIVE_DRAINING
Connection is inactive, can't accept any more requests, and will be closed once it has no more outsta...
Definition trunk.h:97
TRUNK_CONN_INACTIVE
@ TRUNK_CONN_INACTIVE
Connection is inactive and can't accept any more requests.
Definition trunk.h:96
TRUNK_CONN_HALTED
@ TRUNK_CONN_HALTED
Halted, ready to be freed.
Definition trunk.h:88
TRUNK_CONN_CLOSED
@ TRUNK_CONN_CLOSED
Connection was closed, either explicitly or due to failure.
Definition trunk.h:94
TRUNK_CONN_INIT
@ TRUNK_CONN_INIT
In the initial state.
Definition trunk.h:89
TRUNK_CONN_DRAINING_TO_FREE
@ TRUNK_CONN_DRAINING_TO_FREE
Connection will be closed once it has no more outstanding requests.
Definition trunk.h:103
TRUNK_CONN_ACTIVE
@ TRUNK_CONN_ACTIVE
Connection is connected and ready to service requests.
Definition trunk.h:91
trunk_conf_t::req_pool_headers
unsigned req_pool_headers
How many chunk headers the talloc pool allocated with the treq should contain.
Definition trunk.h:267
trunk_request_pub_s::request
request_t *_CONST request
The request that we're writing the data on behalf of.
Definition trunk.h:360
trunk_pub_s::last_open
fr_time_t _CONST last_open
Last time the connection management function opened a connection.
Definition trunk.h:311
trunk_conf_t::idle_timeout
fr_time_delta_t idle_timeout
how long a connection can remain idle for
Definition trunk.h:251
trunk_connection_pub_s::state
trunk_connection_state_t _CONST state
What state the connection is in.
Definition trunk.h:372
trunk_conf_t::req_pool_size
size_t req_pool_size
The size of the talloc pool allocated with the treq.
Definition trunk.h:270
trunk_conf_t::max_uses
uint64_t max_uses
The maximum time a connection can be used.
Definition trunk.h:247
trunk_conf_t::lifetime
fr_time_delta_t lifetime
Time between reconnects.
Definition trunk.h:249
trunk_conf_t::connecting
uint16_t connecting
Maximum number of connections that can be in the connecting state.
Definition trunk.h:234
trunk_pub_s::req_alloc_reused
uint64_t _CONST req_alloc_reused
How many requests were reused.
Definition trunk.h:335
trunk_conf_t::max_backlog
uint32_t max_backlog
Maximum number of requests that can be in the backlog.
Definition trunk.h:245
trunk_pub_s::last_failed
fr_time_t _CONST last_failed
Last time a connection failed.
Definition trunk.h:319
trunk_request_pub_s::state
trunk_request_state_t _CONST state
Which list the request is now located in.
Definition trunk.h:350
trunk_connection_pub_s::last_write_success
fr_time_t _CONST last_write_success
Last time we wrote to the connection.
Definition trunk.h:376
trunk_request_pub_s::tconn
trunk_connection_t *_CONST tconn
Connection this request belongs to.
Definition trunk.h:354
trunk_io_funcs_t::connection_alloc
trunk_connection_alloc_t connection_alloc
Allocate a new connection_t.
Definition trunk.h:738
trunk_pub_s::last_read_success
fr_time_t _CONST last_read_success
Last time we read a response.
Definition trunk.h:323
trunk_pub_s::last_below_target
fr_time_t _CONST last_below_target
Last time average utilisation went below the target value.
Definition trunk.h:308
trunk_connection_pub_s::last_read_success
fr_time_t _CONST last_read_success
Last time we read from the connection.
Definition trunk.h:378
trunk_conf_t::close_delay
fr_time_delta_t close_delay
How long we must be below target utilisation to close an existing connection.
Definition trunk.h:256
trunk_conf_t::start
uint16_t start
How many connections to start.
Definition trunk.h:228
trunk_conf_t::req_cleanup_delay
fr_time_delta_t req_cleanup_delay
How long must a request in the unassigned (free) list not have been used for before it's cleaned up a...
Definition trunk.h:260
TRUNK_REQUEST_STATE_CANCEL_ALL
#define TRUNK_REQUEST_STATE_CANCEL_ALL
All requests in various cancellation states.
Definition trunk.h:214
trunk_conf_t::always_writable
bool always_writable
Set to true if our ability to write requests to a connection handle is not dependent on the state of ...
Definition trunk.h:272
trunk_connection_event_t
trunk_connection_event_t
What type of I/O events the trunk connection is currently interested in receiving.
Definition trunk.h:72
TRUNK_CONN_EVENT_BOTH
@ TRUNK_CONN_EVENT_BOTH
Trunk should be notified if a connection is readable or writable.
Definition trunk.h:79
TRUNK_CONN_EVENT_WRITE
@ TRUNK_CONN_EVENT_WRITE
Trunk should be notified if a connection is writable.
Definition trunk.h:77
TRUNK_CONN_EVENT_NONE
@ TRUNK_CONN_EVENT_NONE
Don't notify the trunk on connection state changes.
Definition trunk.h:73
TRUNK_CONN_EVENT_READ
@ TRUNK_CONN_EVENT_READ
Trunk should be notified if a connection is readable.
Definition trunk.h:75
TRUNK_CONN_ALL
#define TRUNK_CONN_ALL
All connection states.
Definition trunk.h:111
trunk_io_funcs_t::request_prioritise
fr_heap_cmp_t request_prioritise
Ordering function for requests.
Definition trunk.h:744
trunk_pub_s::req_alloc
uint64_t _CONST req_alloc
The number of requests currently allocated that have not been freed or returned to the free list.
Definition trunk.h:329
trunk_cancel_reason_t
trunk_cancel_reason_t
Reasons for a request being cancelled.
Definition trunk.h:55
TRUNK_CANCEL_REASON_NONE
@ TRUNK_CANCEL_REASON_NONE
Request has not been cancelled.
Definition trunk.h:56
TRUNK_CANCEL_REASON_SIGNAL
@ TRUNK_CANCEL_REASON_SIGNAL
Request cancelled due to a signal.
Definition trunk.h:57
TRUNK_CANCEL_REASON_REQUEUE
@ TRUNK_CANCEL_REASON_REQUEUE
A previously sent request is being requeued.
Definition trunk.h:59
TRUNK_CANCEL_REASON_MOVE
@ TRUNK_CANCEL_REASON_MOVE
Request cancelled because it's being moved.
Definition trunk.h:58
trunk_pub_s::req_alloc_new
uint64_t _CONST req_alloc_new
How many requests we've allocated.
Definition trunk.h:333
trunk_conf_t::open_delay
fr_time_delta_t open_delay
How long we must be above target utilisation to spawn a new connection.
Definition trunk.h:253
trunk_connection_pub_s::conn
connection_t *_CONST conn
The underlying connection.
Definition trunk.h:374
trunk_state_t
trunk_state_t
Definition trunk.h:62
TRUNK_STATE_MAX
@ TRUNK_STATE_MAX
Definition trunk.h:66
TRUNK_STATE_PENDING
@ TRUNK_STATE_PENDING
Trunk has connections, but none are active.
Definition trunk.h:65
TRUNK_STATE_ACTIVE
@ TRUNK_STATE_ACTIVE
Trunk has active connections.
Definition trunk.h:64
TRUNK_STATE_IDLE
@ TRUNK_STATE_IDLE
Trunk has no connections.
Definition trunk.h:63
trunk_pub_s::last_closed
fr_time_t _CONST last_closed
Last time the connection management function closed a connection.
Definition trunk.h:314
trunk_watch_t
void(* trunk_watch_t)(trunk_t *trunk, trunk_state_t prev, trunk_state_t state, void *uctx)
Receive a notification when a trunk enters a particular state.
Definition trunk.h:729
trunk_conf_t::manage_interval
fr_time_delta_t manage_interval
How often we run the management algorithm to open/close connections.
Definition trunk.h:264
trunk_enqueue_t
trunk_enqueue_t
Definition trunk.h:149
TRUNK_ENQUEUE_DST_UNAVAILABLE
@ TRUNK_ENQUEUE_DST_UNAVAILABLE
Destination is down.
Definition trunk.h:154
TRUNK_ENQUEUE_FAIL
@ TRUNK_ENQUEUE_FAIL
General failure.
Definition trunk.h:155
TRUNK_ENQUEUE_OK
@ TRUNK_ENQUEUE_OK
Operation was successful.
Definition trunk.h:151
TRUNK_ENQUEUE_NO_CAPACITY
@ TRUNK_ENQUEUE_NO_CAPACITY
At maximum number of connections, and no connection has capacity.
Definition trunk.h:152
TRUNK_ENQUEUE_IN_BACKLOG
@ TRUNK_ENQUEUE_IN_BACKLOG
Request should be enqueued in backlog.
Definition trunk.h:150
trunk_request_pub_s::preq
void *_CONST preq
Data for the muxer to write to the connection.
Definition trunk.h:356
trunk_conf_t::target_req_per_conn
uint32_t target_req_per_conn
How many pending requests should ideally be running on each connection.
Definition trunk.h:237
trunk_pub_s::last_connected
fr_time_t _CONST last_connected
Last time a connection connected.
Definition trunk.h:317
trunk_io_funcs_t::request_cancel_mux
trunk_request_cancel_mux_t request_cancel_mux
!< Read one or more requests from a connection.
Definition trunk.h:751
trunk_request_state_t
trunk_request_state_t
Used for sanity checks and to simplify freeing.
Definition trunk.h:162
TRUNK_REQUEST_STATE_PARTIAL
@ TRUNK_REQUEST_STATE_PARTIAL
Some of the request was written to the socket, more of it should be written later.
Definition trunk.h:171
TRUNK_REQUEST_STATE_REAPABLE
@ TRUNK_REQUEST_STATE_REAPABLE
Request has been written, needs to persist, but we are not currently waiting for any response.
Definition trunk.h:174
TRUNK_REQUEST_STATE_UNASSIGNED
@ TRUNK_REQUEST_STATE_UNASSIGNED
Transition state - Request currently not assigned to any connection.
Definition trunk.h:166
TRUNK_REQUEST_STATE_INIT
@ TRUNK_REQUEST_STATE_INIT
Initial state.
Definition trunk.h:163
TRUNK_REQUEST_STATE_CANCEL_SENT
@ TRUNK_REQUEST_STATE_CANCEL_SENT
We've informed the remote server that the request has been cancelled.
Definition trunk.h:186
TRUNK_REQUEST_STATE_COMPLETE
@ TRUNK_REQUEST_STATE_COMPLETE
The request is complete.
Definition trunk.h:183
TRUNK_REQUEST_STATE_FAILED
@ TRUNK_REQUEST_STATE_FAILED
The request failed.
Definition trunk.h:184
TRUNK_REQUEST_STATE_CANCEL
@ TRUNK_REQUEST_STATE_CANCEL
A request on a particular socket was cancel.
Definition trunk.h:185
TRUNK_REQUEST_STATE_CANCEL_PARTIAL
@ TRUNK_REQUEST_STATE_CANCEL_PARTIAL
We partially wrote a cancellation request.
Definition trunk.h:188
TRUNK_REQUEST_STATE_BACKLOG
@ TRUNK_REQUEST_STATE_BACKLOG
In the backlog.
Definition trunk.h:168
TRUNK_REQUEST_STATE_CANCEL_COMPLETE
@ TRUNK_REQUEST_STATE_CANCEL_COMPLETE
Remote server has acknowledged our cancellation.
Definition trunk.h:189
TRUNK_REQUEST_STATE_PENDING
@ TRUNK_REQUEST_STATE_PENDING
In the queue of a connection and is pending writing.
Definition trunk.h:169
TRUNK_REQUEST_STATE_SENT
@ TRUNK_REQUEST_STATE_SENT
Was written to a socket. Waiting for a response.
Definition trunk.h:173
trunk_pub_s::state
trunk_state_t _CONST state
Current state of the trunk.
Definition trunk.h:338
trunk_pub_s::last_above_target
fr_time_t _CONST last_above_target
Last time average utilisation went above the target value.
Definition trunk.h:305
trunk_conf_t
Common configuration parameters for a trunk.
Definition trunk.h:225
trunk_connection_pub_s
Public fields for the trunk connection.
Definition trunk.h:371
trunk_io_funcs_t
I/O functions to pass to trunk_alloc.
Definition trunk.h:737
trunk_pub_s
Public fields for the trunk.
Definition trunk.h:301
trunk_request_pub_s
Public fields for the trunk request.
Definition trunk.h:349
el
static fr_event_list_t * el
Definition unit_test_attribute.c:264
head
static fr_slen_t head
Definition xlat.h:420
parent
static fr_slen_t parent
Definition pair.h:858
fr_strerror
char const * fr_strerror(void)
Get the last library error.
Definition strerror.c:558
fr_box_time_delta
#define fr_box_time_delta(_val)
Definition value.h:366
nonnull
int nonnull(2, 5))
out
static size_t char ** out
Definition value.h:1030
Generated by   doxygen 1.9.8