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