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