From: Arran Cudbard-Bell Date: Wed, 11 Dec 2019 00:54:56 +0000 (+0700) Subject: Add support for shutting down connections gracefully X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ed9fdb27fa6166e40a39df9648fe2ced8cb11345;p=thirdparty%2Ffreeradius-server.git Add support for shutting down connections gracefully --- diff --git a/src/lib/server/connection.c b/src/lib/server/connection.c index f0ead1aec5c..6cc59bedecc 100644 --- a/src/lib/server/connection.c +++ b/src/lib/server/connection.c @@ -45,8 +45,9 @@ fr_table_num_ordered_t const fr_connection_states[] = { { "CONNECTING", FR_CONNECTION_STATE_CONNECTING }, { "TIMEOUT", FR_CONNECTION_STATE_TIMEOUT }, { "CONNECTED", FR_CONNECTION_STATE_CONNECTED }, - { "FAILED", FR_CONNECTION_STATE_FAILED }, + { "SHUTDOWN", FR_CONNECTION_STATE_SHUTDOWN }, { "CLOSED", FR_CONNECTION_STATE_CLOSED }, + { "FAILED", FR_CONNECTION_STATE_FAILED }, { "HALTED", FR_CONNECTION_STATE_HALTED }, }; size_t fr_connection_states_len = NUM_ELEMENTS(fr_connection_states); @@ -88,6 +89,7 @@ struct fr_conn { fr_connection_init_t init; //!< Callback for initialising a connection. fr_connection_open_t open; //!< Callback for 'open' notification. fr_connection_close_t close; //!< Callback to close a connection. + fr_connection_shutdown_t shutdown; //!< Signal the connection handle to start shutting down. fr_connection_failed_t failed; //!< Callback for 'failed' notification. fr_event_timer_t const *connection_timer; //!< Timer to prevent connections going on indefinitely. @@ -172,6 +174,7 @@ static void connection_state_closed_enter(fr_connection_t *conn); static void connection_state_failed_enter(fr_connection_t *conn); static void connection_state_timeout_enter(fr_connection_t *conn); static void connection_state_connected_enter(fr_connection_t *conn); +static void connection_state_shutdown_enter(fr_connection_t *conn); static void connection_state_connecting_enter(fr_connection_t *conn); static void connection_state_halted_enter(fr_connection_t *conn); static void connection_state_init_enter(fr_connection_t *conn); @@ -218,34 +221,24 @@ uint64_t fr_connection_get_id(fr_connection_t const *conn) return conn->id; } -/** Get the handle associated with a connection - * - * @param[in] conn to retrieve fd from. - * @return the active connection handle. - */ -void *fr_connection_get_handle(fr_connection_t const *conn) -{ - return conn->h; -} - -/** Set the handle associated with a connection - * - * Will not free the previous handle. This must be done manually. +/** Get the log_prefix, useful for debugging messages * - * @param[in] conn to set fd for. - * @param[in] handle to set. + * @param[in] conn to retrieve number from. + * @return the log prefix set for the connection. */ -void fr_connection_set_handle(fr_connection_t *conn, void *handle) +char const *fr_connection_get_log_prefix(fr_connection_t const *conn) { - conn->h = handle; + return conn->log_prefix; } -/** Set an (optional) callback to be called on connection timeout/failure +/** Get the handle associated with a connection * + * @param[in] conn to retrieve fd from. + * @return the active connection handle. */ -void fr_connection_set_failed_func(fr_connection_t *conn, fr_connection_failed_t func) +void *fr_connection_get_handle(fr_connection_t const *conn) { - conn->failed = func; + return conn->h; } /** Remove a watch function from a pre/post[state] list @@ -350,111 +343,6 @@ void fr_connection_add_watch_post(fr_connection_t *conn, fr_connection_state_t s connection_add_watch(conn, &conn->watch_post[state], watch, oneshot, uctx); } -/** Close a connection if it's freed - * - * @param[in] conn to free. - * @return - * - 0 connection was freed immediately. - * - 1 connection free was deferred. - */ -static int _connection_free(fr_connection_t *conn) -{ - /* - * Don't allow the connection to be - * arbitrarily freed by a callback. - * - * Set the deferred free flag, and - * free the connection afterwards. - */ - if (conn->in_handler) { - conn->deferred_free = true; - return 1; - } - - switch (conn->state) { - case FR_CONNECTION_STATE_HALTED: - break; - - - /* - * Need to close the connection first - */ - case FR_CONNECTION_STATE_CONNECTING: - case FR_CONNECTION_STATE_CONNECTED: - connection_state_closed_enter(conn); - /* FALL-THROUGH */ - - default: - connection_state_halted_enter(conn); - break; - } - return 0; -} - -/** Allocate a new connection - * - * After the connection has been allocated, it should be started with a call to #fr_connection_signal_init. - * - * The connection state machine can detect when the connection is open in one of two ways. - * - You can install a generic socket open/fail callback, using fr_connection_signal_on_fd. - * - You can call either #fr_connection_signal_connected or fr_connection_signal_recommend. - * This allows the connection state machine to work with more difficult library APIs, - * which may not return control to the caller as connections are opened. - * - * @param[in] ctx to allocate connection handle in. If the connection - * handle is freed, and the #fr_connection_state_t is - * #FR_CONNECTION_STATE_CONNECTING or #FR_CONNECTION_STATE_CONNECTED the - * close callback will be called. - * @param[in] el to use for timer events, and to pass to the #fr_connection_open_t callback. - * @param[in] connection_timeout (optional) how long to wait for a connection to open. - * @param[in] reconnection_delay How long to wait on connection failure before retrying. - * @param[in] init (optional) callback to initialise a new file descriptor. - * @param[in] open (optional) callback to receive notifications that the connection is open. - * @param[in] close (optional) Callback to close the connection. - * @param[in] log_prefix To prepend to log messages. - * @param[in] uctx User context to pass to callbacks. - * @return - * - A new #fr_connection_t on success. - * - NULL on failure. - */ -fr_connection_t *fr_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, - fr_time_delta_t connection_timeout, - fr_time_delta_t reconnection_delay, - fr_connection_init_t init, fr_connection_open_t open, fr_connection_close_t close, - char const *log_prefix, - void const *uctx) -{ - size_t i; - fr_connection_t *conn; - - rad_assert(el); - - conn = talloc_zero(ctx, fr_connection_t); - if (!conn) return NULL; - talloc_set_destructor(conn, _connection_free); - - conn->id = atomic_fetch_add_explicit(&connection_counter, 1, memory_order_relaxed); - conn->state = FR_CONNECTION_STATE_HALTED; - conn->el = el; - conn->h = NULL; - conn->reconnection_delay = reconnection_delay; - conn->connection_timeout = connection_timeout; - conn->init = init; - conn->open = open; - conn->close = close; - conn->log_prefix = talloc_typed_strdup(conn, log_prefix); - memcpy(&conn->uctx, &uctx, sizeof(conn->uctx)); - - for (i = 0; i < NUM_ELEMENTS(conn->watch_pre); i++) { - fr_dlist_talloc_init(&conn->watch_pre[i], fr_connection_watch_entry_t, list); - } - for (i = 0; i < NUM_ELEMENTS(conn->watch_post); i++) { - fr_dlist_talloc_init(&conn->watch_post[i], fr_connection_watch_entry_t, list); - } - - return conn; -} - /** The requisite period of time has passed, try and re-open the connection * * @param[in] el the time event ocurred on. @@ -496,8 +384,8 @@ static void connection_state_closed_enter(fr_connection_t *conn) WATCH_PRE(conn); if (conn->close && !conn->is_closed) { HANDLER_BEGIN(conn); - DEBUG4("Calling close(h=%p, uctx=%p)", conn->h, conn->uctx); - conn->close(conn->h, conn->uctx); + DEBUG4("Calling close(el=%p, h=%p, uctx=%p)", conn->el, conn->h, conn->uctx); + conn->close(conn->el, conn->h, conn->uctx); conn->is_closed = true; /* Ensure close doesn't get called twice if the connection is freed */ HANDLER_END(conn); } else { @@ -506,6 +394,62 @@ static void connection_state_closed_enter(fr_connection_t *conn) WATCH_POST(conn); } +/** Connection timeout + * + * Connection wasn't opened within the configured period of time + * + * @param[in] el the time event ocurred on. + * @param[in] now The current time. + * @param[in] uctx The #fr_connection_t the fd is associated with. + */ +static void _connection_timeout(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx) +{ + fr_connection_t *conn = talloc_get_type_abort(uctx, fr_connection_t); + + connection_state_timeout_enter(conn); +} + +/** Gracefully shutdown the handle + * + */ +static void connection_state_shutdown_enter(fr_connection_t *conn) +{ + switch (conn->state) { + case FR_CONNECTION_STATE_CONNECTED: + break; + + default: + rad_assert(0); + return; + } + + STATE_TRANSITION(FR_CONNECTION_STATE_SHUTDOWN); + + WATCH_PRE(conn); + HANDLER_BEGIN(conn); + DEBUG4("Calling shutdown(el=%p, h=%p, uctx=%p)", conn->el, conn->h, conn->uctx); + conn->shutdown(conn->el, conn->h, conn->uctx); + HANDLER_END(conn); + WATCH_POST(conn); + + /* + * If there's a connection timeout, + * set, then add the timer. + * + * The connection may be bad, in which + * case we want to automatically fail + * if it doesn't shutdown within the + * timeout period. + */ + if (conn->connection_timeout) { + if (fr_event_timer_in(conn, conn->el, &conn->connection_timer, + conn->connection_timeout, _connection_timeout, conn) < 0) { + PERROR("Failed inserting connection timeout event"); + rad_assert(0); + } + } +} + /** Connection failed * * Transition to the FR_CONNECTION_STATE_FAILED state. @@ -633,7 +577,14 @@ static void connection_state_failed_enter(fr_connection_t *conn) */ static void connection_state_timeout_enter(fr_connection_t *conn) { - rad_assert(conn->state == FR_CONNECTION_STATE_CONNECTING); + switch (conn->state) { + case FR_CONNECTION_STATE_CONNECTING: + case FR_CONNECTION_STATE_SHUTDOWN: + break; + + default: + rad_assert(0); + } ERROR("Connection failed - timed out after %pVs", fr_box_time_delta(conn->connection_timeout)); @@ -709,21 +660,6 @@ static void connection_state_connected_enter(fr_connection_t *conn) } } -/** Connection timeout - * - * Connection wasn't opened within the configured period of time - * - * @param[in] el the time event ocurred on. - * @param[in] now The current time. - * @param[in] uctx The #fr_connection_t the fd is associated with. - */ -static void _connection_timeout(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx) -{ - fr_connection_t *conn = talloc_get_type_abort(uctx, fr_connection_t); - - connection_state_timeout_enter(conn); -} - /** Enter the connecting state * * After this function returns we wait to be signalled with fr_connection_singal_connected @@ -872,8 +808,9 @@ void fr_connection_signal_connected(fr_connection_t *conn) * and should be reconnected. * * @param[in] conn to reconnect. + * @param[in] reason Why the connection was signalled to reconnect. */ -void fr_connection_signal_reconnect(fr_connection_t *conn) +void fr_connection_signal_reconnect(fr_connection_t *conn, fr_connection_reason_t reason) { DEBUG2("Signalled to reconnect"); @@ -886,8 +823,16 @@ void fr_connection_signal_reconnect(fr_connection_t *conn) fr_connection_signal_init(conn); return; - case FR_CONNECTION_STATE_CONNECTING: + case FR_CONNECTION_STATE_SHUTDOWN: + if (reason == FR_CONNECTION_EXPIRED) return; + connection_state_failed_enter(conn); + break; + case FR_CONNECTION_STATE_CONNECTED: + if ((reason == FR_CONNECTION_EXPIRED) && conn->shutdown) connection_state_shutdown_enter(conn); + /* FALL-THROUGH */ + + case FR_CONNECTION_STATE_CONNECTING: case FR_CONNECTION_STATE_TIMEOUT: case FR_CONNECTION_STATE_CLOSED: connection_state_failed_enter(conn); @@ -1006,3 +951,136 @@ int fr_connection_signal_on_fd(fr_connection_t *conn, int fd) _connection_signal_on_fd_cleanup, true, fd_s); return 0; } + +/** Set the handle associated with a connection + * + * Will not free the previous handle. This must be done manually. + * + * @param[in] conn to set fd for. + * @param[in] handle to set. + */ +void fr_connection_set_handle(fr_connection_t *conn, void *handle) +{ + conn->h = handle; +} + +/** Set an (optional) callback to be called on connection timeout/failure + * + */ +void fr_connection_set_failed_func(fr_connection_t *conn, fr_connection_failed_t func) +{ + conn->failed = func; +} + +/** Set an (optional) callback to trigger a graceful shutdown + * + */ +void fr_connection_set_shutdown_func(fr_connection_t *conn, fr_connection_shutdown_t func) +{ + conn->shutdown = func; +} + +/** Close a connection if it's freed + * + * @param[in] conn to free. + * @return + * - 0 connection was freed immediately. + * - 1 connection free was deferred. + */ +static int _connection_free(fr_connection_t *conn) +{ + /* + * Don't allow the connection to be + * arbitrarily freed by a callback. + * + * Set the deferred free flag, and + * free the connection afterwards. + */ + if (conn->in_handler) { + conn->deferred_free = true; + return 1; + } + + switch (conn->state) { + case FR_CONNECTION_STATE_HALTED: + break; + + + /* + * Need to close the connection first + */ + case FR_CONNECTION_STATE_CONNECTING: + case FR_CONNECTION_STATE_CONNECTED: + connection_state_closed_enter(conn); + /* FALL-THROUGH */ + + default: + connection_state_halted_enter(conn); + break; + } + return 0; +} + +/** Allocate a new connection + * + * After the connection has been allocated, it should be started with a call to #fr_connection_signal_init. + * + * The connection state machine can detect when the connection is open in one of two ways. + * - You can install a generic socket open/fail callback, using fr_connection_signal_on_fd. + * - You can call either #fr_connection_signal_connected or fr_connection_signal_recommend. + * This allows the connection state machine to work with more difficult library APIs, + * which may not return control to the caller as connections are opened. + * + * @param[in] ctx to allocate connection handle in. If the connection + * handle is freed, and the #fr_connection_state_t is + * #FR_CONNECTION_STATE_CONNECTING or #FR_CONNECTION_STATE_CONNECTED the + * close callback will be called. + * @param[in] el to use for timer events, and to pass to the #fr_connection_open_t callback. + * @param[in] connection_timeout (optional) how long to wait for a connection to open. + * @param[in] reconnection_delay How long to wait on connection failure before retrying. + * @param[in] init (optional) callback to initialise a new file descriptor. + * @param[in] open (optional) callback to receive notifications that the connection is open. + * @param[in] close (optional) Callback to close the connection. + * @param[in] log_prefix To prepend to log messages. + * @param[in] uctx User context to pass to callbacks. + * @return + * - A new #fr_connection_t on success. + * - NULL on failure. + */ +fr_connection_t *fr_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, + fr_time_delta_t connection_timeout, + fr_time_delta_t reconnection_delay, + fr_connection_init_t init, fr_connection_open_t open, fr_connection_close_t close, + char const *log_prefix, + void const *uctx) +{ + size_t i; + fr_connection_t *conn; + + rad_assert(el); + + conn = talloc_zero(ctx, fr_connection_t); + if (!conn) return NULL; + talloc_set_destructor(conn, _connection_free); + + conn->id = atomic_fetch_add_explicit(&connection_counter, 1, memory_order_relaxed); + conn->state = FR_CONNECTION_STATE_HALTED; + conn->el = el; + conn->h = NULL; + conn->reconnection_delay = reconnection_delay; + conn->connection_timeout = connection_timeout; + conn->init = init; + conn->open = open; + conn->close = close; + conn->log_prefix = talloc_typed_strdup(conn, log_prefix); + memcpy(&conn->uctx, &uctx, sizeof(conn->uctx)); + + for (i = 0; i < NUM_ELEMENTS(conn->watch_pre); i++) { + fr_dlist_talloc_init(&conn->watch_pre[i], fr_connection_watch_entry_t, list); + } + for (i = 0; i < NUM_ELEMENTS(conn->watch_post); i++) { + fr_dlist_talloc_init(&conn->watch_post[i], fr_connection_watch_entry_t, list); + } + + return conn; +} diff --git a/src/lib/server/connection.h b/src/lib/server/connection.h index 6aca156a4a7..11cd8973349 100644 --- a/src/lib/server/connection.h +++ b/src/lib/server/connection.h @@ -43,11 +43,20 @@ typedef enum { FR_CONNECTION_STATE_CONNECTING, //!< Waiting for connection to establish. FR_CONNECTION_STATE_TIMEOUT, //!< Timeout during #FR_CONNECTION_STATE_CONNECTING. FR_CONNECTION_STATE_CONNECTED, //!< File descriptor is open (ready for writing). + FR_CONNECTION_STATE_SHUTDOWN, //!< Connection is shutting down. FR_CONNECTION_STATE_CLOSED, //!< Connection has been closed. FR_CONNECTION_STATE_FAILED, //!< Connection failed and is waiting to reconnect. FR_CONNECTION_STATE_MAX } fr_connection_state_t; +typedef enum { + FR_CONNECTION_FAILED = 0, //!< Connection is being reconnected because it failed. + FR_CONNECTION_EXPIRED //!< Connection is being reconnected because it's at + ///< the end of its life. In this case we enter the + ///< closing state and try and close the connection + ///< gracefully. +} fr_connection_reason_t; + extern fr_table_num_ordered_t const fr_connection_states[]; extern size_t fr_connection_states_len; @@ -81,6 +90,25 @@ typedef fr_connection_state_t (*fr_connection_init_t)(void **h_out, fr_connectio */ typedef fr_connection_state_t (*fr_connection_open_t)(fr_event_list_t *el, void *h, void *uctx); +/** Start the process of gracefully shutting down the connection + * + * This function is called when the connection is signalled to gracefully + * disconnect. It should place the connection in a state where pending + * I/O operations complete, and buffers are flushed. + * + * After all pending events are complete, the connection should be signalled + * that the handle is in the closed state. + * + * @param[in] el to use for inserting I/O events. + * @param[in] h Handle that needs to be closed. + * @param[in] uctx User context. + * @return + * - #FR_CONNECTION_STATE_CLOSING if the handle is shutting down. + * - #FR_CONNECTION_STATE_FAILED if the handle is unusable, and we + * should just transition directly to failed. + */ +typedef fr_connection_state_t (*fr_connection_shutdown_t)(fr_event_list_t *el, void *h, void *uctx); + /** Notification that a connection attempt has failed * * @note If the callback frees the connection, it must return #FR_CONNECTION_STATE_HALTED. @@ -107,10 +135,11 @@ typedef fr_connection_state_t (*fr_connection_failed_t)(void *h, fr_connection_s * If this callback does not close the file descriptor, the server will leak * file descriptors. * + * @param[in] el to use for inserting I/O events. * @param[in] h Handle to close. * @param[in] uctx User context. */ -typedef void (*fr_connection_close_t)(void *h, void *uctx); +typedef void (*fr_connection_close_t)(fr_event_list_t *el, void *h, void *uctx); /** Receive a notification when a connection enters a particular state * @@ -123,41 +152,90 @@ typedef void (*fr_connection_close_t)(void *h, void *uctx); */ typedef void(*fr_connection_watch_t)(fr_connection_t *conn, fr_connection_state_t state, void *uctx); +/** @name Statistics + * @{ + */ uint64_t fr_connection_get_num_reconnected(fr_connection_t const *conn); uint64_t fr_connection_get_num_timed_out(fr_connection_t const *conn); +/** @} */ +/** @name Retrieve read only values from the connection + * @{ + */ fr_event_list_t *fr_connection_get_el(fr_connection_t const *conn); uint64_t fr_connection_get_id(fr_connection_t const *conn); -void *fr_connection_get_handle(fr_connection_t const *conn); -void fr_connection_set_handle(fr_connection_t *conn, void *handle); +char const *fr_connection_get_log_prefix(fr_connection_t const *conn); -void fr_connection_set_failed_func(fr_connection_t *conn, fr_connection_failed_t func); +void *fr_connection_get_handle(fr_connection_t const *conn); +/** @} */ +/** @name Add watcher functions that get called before (pre) the state callback and after (post) + * @{ + */ void fr_connection_add_watch_pre(fr_connection_t *conn, fr_connection_state_t state, fr_connection_watch_t watch, bool oneshot, void const *uctx); + void fr_connection_add_watch_post(fr_connection_t *conn, fr_connection_state_t state, fr_connection_watch_t watch, bool oneshot, void const *uctx); int fr_connection_del_watch_pre(fr_connection_t *conn, fr_connection_state_t state, fr_connection_watch_t watch); + int fr_connection_del_watch_post(fr_connection_t *conn, fr_connection_state_t state, fr_connection_watch_t watch); +/** @} */ + +/** @name Signal the connection to change states + * @{ + */ +void fr_connection_signal_init(fr_connection_t *conn); + +void fr_connection_signal_connected(fr_connection_t *conn); + +void fr_connection_signal_reconnect(fr_connection_t *conn, fr_connection_reason_t reason); +/** @} */ + +/** @name Install generic I/O events on an FD to signal state changes + * @{ + */ +int fr_connection_signal_on_fd(fr_connection_t *conn, int fd); +/** @} */ + +/** @name Modify the connection handle + * @{ + */ +void fr_connection_set_handle(fr_connection_t *conn, void *handle); +/** @} */ +/** @name Set a callback for additional states + * + * These are not needed for every connection. + * - Failed is needed if you want to deal with failure conditions specifically. + * Otherwise most of the same work can be done in close(). + * - Shutdown is needed if the API you're working with supports graceful + * shutdown, or if you're working with raw TCP connections and need to call + * shutdown(). + * @{ + */ +void fr_connection_set_failed_func(fr_connection_t *conn, fr_connection_failed_t func); + +void fr_connection_set_shutdown_func(fr_connection_t *conn, fr_connection_shutdown_t func); +/** @} */ + +/** @name Allocate a new connection + * @{ + */ fr_connection_t *fr_connection_alloc(TALLOC_CTX *ctx, fr_event_list_t *el, fr_time_delta_t connection_timeout, fr_time_delta_t reconnection_delay, fr_connection_init_t init, fr_connection_open_t open, fr_connection_close_t close, char const *log_prefix, void const *uctx); +/** @} */ -void fr_connection_signal_init(fr_connection_t *conn); -void fr_connection_signal_connected(fr_connection_t *conn); -void fr_connection_signal_reconnect(fr_connection_t *conn); - -int fr_connection_signal_on_fd(fr_connection_t *conn, int fd); #ifdef __cplusplus } #endif diff --git a/src/lib/server/trunk.c b/src/lib/server/trunk.c index 696163f75e8..1539b834d44 100644 --- a/src/lib/server/trunk.c +++ b/src/lib/server/trunk.c @@ -203,6 +203,8 @@ struct fr_trunk_connection_s { ///< the number of requests associated with it ///< falls below max_req_per_conn. + bool free_on_close; //!< When the connection closes, free it. + bool freeing; //!< Conn is being freed, cancel_sent state should ///< be skipped. /** @} */ @@ -2278,6 +2280,40 @@ static void _trunk_connection_on_connecting(UNUSED fr_connection_t *conn, UNUSED fr_dlist_insert_head(&trunk->connecting, tconn); /* MUST remain a head insertion for reconnect logic */ } +/** Connection transitioned to the shutdown state + * + * If we're not already in the draining-to-free state, transition there now. + * + * The idea is that if something signalled the connection to shutdown, we need + * to reflect that by dequeuing any pending requests, not accepting new ones, + * and waiting for the existing requests to complete. + * + * @note This function is only called from the connection API as a watcher. + * + * @param[in] conn The connection which changes state. + * @param[in] state The connection is now in. + * @param[in] uctx The fr_trunk_connection_t wrapping the connection. + */ +static void _trunk_connection_on_shutdown(UNUSED fr_connection_t *conn, UNUSED fr_connection_state_t state, void *uctx) +{ + fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t); + + switch (tconn->state) { + case FR_TRUNK_CONN_DRAINING_TO_FREE: /* Do Nothing */ + return; + + case FR_TRUNK_CONN_ACTIVE: /* Transition to draining-to-free */ + case FR_TRUNK_CONN_INACTIVE: + case FR_TRUNK_CONN_DRAINING: + break; + + default: + CONN_BAD_STATE_TRANSITION(FR_TRUNK_CONN_DRAINING_TO_FREE); + } + + trunk_connection_enter_draining_to_free(tconn); +} + /** Trigger a reconnection of the trunk connection * * @param[in] el Event list the timer was inserted into. @@ -2288,7 +2324,7 @@ static void _trunk_connection_lifetime_expire(UNUSED fr_event_list_t *el, UNUSE { fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t); - fr_trunk_connection_signal_reconnect(tconn); + trunk_connection_enter_draining_to_free(tconn); } /** Connection transitioned to the connected state @@ -2417,6 +2453,20 @@ static void _trunk_connection_on_closed(UNUSED fr_connection_t *conn, UNUSED fr_ if (trunk->conf->lifetime > 0) fr_event_timer_delete(trunk->el, &tconn->lifetime_ev); } +/** Give the trunk API an opportunity to free the connection if it closes + * + */ +static void _trunk_connection_on_closed_post(UNUSED fr_connection_t *conn, + UNUSED fr_connection_state_t state, UNUSED void *uctx) +{ + fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t); + + /* + * Something wanted the connection freed when it finally closed... + */ + if (tconn->free_on_close) talloc_free(tconn); +} + /** Connection failed to connect before it was connected * */ @@ -2613,6 +2663,12 @@ static int trunk_connection_spawn(fr_trunk_t *trunk, fr_time_t now) fr_connection_add_watch_pre(tconn->conn, FR_CONNECTION_STATE_CLOSED, _trunk_connection_on_closed, false, tconn); /* Before close() has been called */ + fr_connection_add_watch_post(tconn->conn, FR_CONNECTION_STATE_CLOSED, + _trunk_connection_on_closed_post, false, tconn); /* After close() has been called */ + + fr_connection_add_watch_post(tconn->conn, FR_CONNECTION_STATE_SHUTDOWN, /* After shutdown() has been called */ + _trunk_connection_on_shutdown, false, tconn); + fr_connection_add_watch_pre(tconn->conn, FR_CONNECTION_STATE_FAILED, _trunk_connection_on_failed, false, tconn); @@ -2785,9 +2841,9 @@ void fr_trunk_connection_signal_active(fr_trunk_connection_t *tconn) * * @param[in] tconn to signal. */ -void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn) +void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn, fr_connection_reason_t reason) { - fr_connection_signal_reconnect(tconn->conn); + fr_connection_signal_reconnect(tconn->conn, reason); } /** Rebalance connections across active trunk members when a new connection becomes active @@ -3069,7 +3125,14 @@ done: fr_trunk_connection_t *prev; prev = fr_dlist_prev(&trunk->draining, tconn); - talloc_free(tconn); + + /* + * Close the connection as gracefully + * as possible, by invoking its shutdown + * method. + */ + fr_connection_signal_reconnect(tconn->conn, FR_CONNECTION_EXPIRED); + tconn->free_on_close = true; tconn = prev; } } @@ -3082,8 +3145,14 @@ done: if (fr_trunk_request_count_by_connection(tconn, FR_TRUNK_REQUEST_ALL) == 0) { fr_trunk_connection_t *prev; + /* + * Close the connection as gracefully + * as possible, by invoking its shutdown + * method. + */ prev = fr_dlist_prev(&trunk->draining_to_free, tconn); - talloc_free(tconn); + fr_connection_signal_reconnect(tconn->conn, FR_CONNECTION_EXPIRED); + tconn->free_on_close = true; tconn = prev; } } @@ -3303,8 +3372,9 @@ static void trunk_backlog_drain(fr_trunk_t *trunk) * * @param[in] trunk to signal. * @param[in] states One or more states or'd together. + * @param[in] reason Why the connections are being signalled to reconnect. */ -void fr_trunk_reconnect(fr_trunk_t *trunk, int states) +void fr_trunk_reconnect(fr_trunk_t *trunk, int states, fr_connection_reason_t reason) { size_t i; @@ -3321,7 +3391,7 @@ void fr_trunk_reconnect(fr_trunk_t *trunk, int states) * the head so this should be ok. */ for (i = fr_dlist_num_elements(&trunk->connecting); i > 0; i--) { - fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->connecting))->conn); + fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->connecting))->conn, reason); } } @@ -3329,31 +3399,31 @@ void fr_trunk_reconnect(fr_trunk_t *trunk, int states) fr_trunk_connection_t *tconn; while ((tconn = fr_heap_peek(trunk->active))) { - fr_connection_signal_reconnect(tconn->conn); + fr_connection_signal_reconnect(tconn->conn, reason); } } if (states & FR_TRUNK_CONN_INACTIVE) { for (i = fr_dlist_num_elements(&trunk->inactive); i > 0; i--) { - fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->inactive))->conn); + fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->inactive))->conn, reason); } } if (states & FR_TRUNK_CONN_CLOSED) { for (i = fr_dlist_num_elements(&trunk->failed); i > 0; i--) { - fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->failed))->conn); + fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->failed))->conn, reason); } } if (states & FR_TRUNK_CONN_DRAINING) { for (i = fr_dlist_num_elements(&trunk->draining); i > 0; i--) { - fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->draining))->conn); + fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->draining))->conn, reason); } } if (states & FR_TRUNK_CONN_DRAINING_TO_FREE) { for (i = fr_dlist_num_elements(&trunk->draining_to_free); i > 0; i--) { - fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->draining_to_free))->conn); + fr_connection_signal_reconnect(((fr_trunk_connection_t *)fr_dlist_tail(&trunk->draining_to_free))->conn, reason); } } } diff --git a/src/lib/server/trunk.h b/src/lib/server/trunk.h index 2aaadc97cb8..a506b5c9963 100644 --- a/src/lib/server/trunk.h +++ b/src/lib/server/trunk.h @@ -454,13 +454,13 @@ void fr_trunk_connection_signal_inactive(fr_trunk_connection_t *tconn); void fr_trunk_connection_signal_active(fr_trunk_connection_t *tconn); -void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn); +void fr_trunk_connection_signal_reconnect(fr_trunk_connection_t *tconn, fr_connection_reason_t reason); /** @} */ /** @name Connection management * @{ */ -void fr_trunk_reconnect(fr_trunk_t *trunk, int state); +void fr_trunk_reconnect(fr_trunk_t *trunk, int state, fr_connection_reason_t reason); /** @} */ /** @name Trunk allocation diff --git a/src/lib/server/trunk_tests.c b/src/lib/server/trunk_tests.c index c2ff2d87567..b3d764004f3 100644 --- a/src/lib/server/trunk_tests.c +++ b/src/lib/server/trunk_tests.c @@ -138,7 +138,7 @@ static void _conn_io_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t); - fr_trunk_connection_signal_reconnect(tconn); + fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED); } static void _conn_io_read(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, void *uctx) @@ -264,7 +264,7 @@ static void _conn_io_loopback(fr_event_list_t *el, int fd, int flags, void *uctx } } -static void _conn_close(void *h, UNUSED void *uctx) +static void _conn_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx) { int *our_h = talloc_get_type_abort(h, int); @@ -552,7 +552,7 @@ static void test_socket_pair_alloc_then_reconnect_check_delay(void) /* * Trigger reconnection */ - fr_connection_signal_reconnect(tconn->conn); + fr_connection_signal_reconnect(tconn->conn, FR_CONNECTION_FAILED); test_time_base += NSEC * 0.5; events = fr_event_corral(el, test_time_base, false); @@ -1053,7 +1053,7 @@ static void test_requeue_on_reconnect(void) tconn = treq->tconn; /* Store the conn the request was assigned to */ TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 1); - fr_trunk_connection_signal_reconnect(tconn); + fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED); /* * Should be reassigned to the other connection @@ -1064,7 +1064,7 @@ static void test_requeue_on_reconnect(void) /* * Should be reassigned to the backlog */ - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_BACKLOG) == 1); TEST_CHECK(!treq->tconn); @@ -1095,7 +1095,7 @@ static void test_requeue_on_reconnect(void) * then be re-assigned. */ tconn = treq->tconn; - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(preq->completed == false); TEST_CHECK(preq->failed == false); @@ -1118,7 +1118,7 @@ static void test_requeue_on_reconnect(void) TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_SENT) == 1); tconn = treq->tconn; - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(fr_trunk_request_count_by_state(trunk, FR_TRUNK_CONN_ALL, FR_TRUNK_REQUEST_PENDING) == 1); @@ -1150,7 +1150,7 @@ static void test_requeue_on_reconnect(void) * freed instead of being moved between * connections. */ - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(preq->completed == false); TEST_CHECK(preq->failed == false); @@ -1201,7 +1201,7 @@ static void test_requeue_on_reconnect(void) /* * Trigger a reconnection */ - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(preq->completed == false); TEST_CHECK(preq->failed == false); @@ -1251,7 +1251,7 @@ static void test_requeue_on_reconnect(void) /* * Trigger a reconnection */ - fr_trunk_connection_signal_reconnect(treq->tconn); + fr_trunk_connection_signal_reconnect(treq->tconn, FR_CONNECTION_FAILED); TEST_CHECK(preq->completed == false); TEST_CHECK(preq->failed == false); diff --git a/src/modules/rlm_logtee/rlm_logtee.c b/src/modules/rlm_logtee/rlm_logtee.c index 91eb7cf188c..371edae16cf 100644 --- a/src/modules/rlm_logtee/rlm_logtee.c +++ b/src/modules/rlm_logtee/rlm_logtee.c @@ -219,7 +219,7 @@ static void _logtee_conn_error(UNUSED fr_event_list_t *el, int sock, UNUSED int /* * Something bad happened... Fix it... */ - fr_connection_signal_reconnect(t->conn); + fr_connection_signal_reconnect(t->conn, FR_CONNECTION_FAILED); } /** Drain any data we received @@ -244,7 +244,7 @@ static void _logtee_conn_read(UNUSED fr_event_list_t *el, int sock, UNUSED int f case ETIMEDOUT: case EIO: case ENXIO: - fr_connection_signal_reconnect(t->conn); + fr_connection_signal_reconnect(t->conn, FR_CONNECTION_FAILED); return; /* @@ -289,7 +289,7 @@ static void _logtee_conn_writable(UNUSED fr_event_list_t *el, int sock, UNUSED i case ENXIO: case EPIPE: case ENETDOWN: - fr_connection_signal_reconnect(t->conn); + fr_connection_signal_reconnect(t->conn, FR_CONNECTION_FAILED); return; /* @@ -351,7 +351,7 @@ static void logtee_fd_active(rlm_logtee_thread_t *t) /** Shutdown/close a file descriptor * */ -static void _logtee_conn_close(void *h, UNUSED void *uctx) +static void _logtee_conn_close(UNUSED fr_event_list_t *el, void *h, UNUSED void *uctx) { int fd = *((int *)h); diff --git a/src/modules/rlm_radius/rlm_radius_udp.c b/src/modules/rlm_radius/rlm_radius_udp.c index 2590ad85eac..3d36c6b96e0 100644 --- a/src/modules/rlm_radius/rlm_radius_udp.c +++ b/src/modules/rlm_radius/rlm_radius_udp.c @@ -453,7 +453,7 @@ static void fd_idle(fr_io_connection_t *c) conn_error, c) < 0) { PERROR("Failed inserting FD event"); - fr_connection_signal_reconnect(c->conn); + fr_connection_signal_reconnect(c->conn, FR_CONNECTION_FAILED); } } @@ -482,7 +482,7 @@ static void fd_active(fr_io_connection_t *c) /* * May free the connection! */ - fr_connection_signal_reconnect(c->conn); + fr_connection_signal_reconnect(c->conn, FR_CONNECTION_FAILED); } } @@ -498,14 +498,14 @@ static void conn_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int fla /* * Something bad happened... Fix it... */ - fr_connection_signal_reconnect(c->conn); + fr_connection_signal_reconnect(c->conn, FR_CONNECTION_FAILED); } /** Shutdown/close a file descriptor * */ -static void _conn_close(void *h, void *uctx) +static void _conn_close(UNUSED fr_event_list_t *el, void *h, void *uctx) { int fd = *((int *)h); fr_io_connection_t *c = talloc_get_type_abort(uctx, fr_io_connection_t);