From: Arran Cudbard-Bell Date: Thu, 9 Apr 2020 02:36:10 +0000 (-0500) Subject: connection: Allow nested signal pausing, and prevent duplicates X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ead0a189fc99721fecd5ea081af8a54f42454ca3;p=thirdparty%2Ffreeradius-server.git connection: Allow nested signal pausing, and prevent duplicates trunk: Pause connection signal delivery in more places --- diff --git a/src/lib/server/connection.c b/src/lib/server/connection.c index bd4807f6b86..dce2b98ae7d 100644 --- a/src/lib/server/connection.c +++ b/src/lib/server/connection.c @@ -105,7 +105,7 @@ struct fr_connection_s { ///< if a function deeper in the call stack freed ///< the connection. - bool signals_pause; //!< Temporarily stop processing of signals. + unsigned int signals_pause; //!< Temporarily stop processing of signals. }; #define STATE_TRANSITION(_new) \ @@ -184,7 +184,10 @@ static void connection_state_enter_init(fr_connection_t *conn); */ static inline void connection_deferred_signal_add(fr_connection_t *conn, connection_dsignal_t signal) { - connection_dsignal_entry_t *dsignal; + connection_dsignal_entry_t *dsignal, *prev; + + prev = fr_dlist_tail(&conn->deferred_signals); + if (prev && (prev->signal == signal)) return; /* Don't insert duplicates */ dsignal = talloc_zero(conn, connection_dsignal_entry_t); dsignal->signal = signal; @@ -280,25 +283,26 @@ static void connection_deferred_signal_process(fr_connection_t *conn) */ void fr_connection_signals_pause(fr_connection_t *conn) { - conn->signals_pause = true; + conn->signals_pause++; } /** Resume processing of deferred signals * * @param[in] conn to resume signal processing for. */ -void fr_connection_deferred_signals_resume(fr_connection_t *conn) +void fr_connection_signals_resume(fr_connection_t *conn) { + if (conn->signals_pause > 0) conn->signals_pause--; + if (conn->signals_pause > 0) return; + /* - * If we were paused, and we're not in a handler - * then process the signals now. + * If we're not in a handler process the + * deferred signals now. */ - if (!conn->in_handler && conn->signals_pause) { - conn->signals_pause = false; + if (!conn->in_handler) { connection_deferred_signal_process(conn); return; } - conn->signals_pause = false; } /** Called when we enter a handler diff --git a/src/lib/server/connection.h b/src/lib/server/connection.h index d3de9821a7a..a73b7ef3ec5 100644 --- a/src/lib/server/connection.h +++ b/src/lib/server/connection.h @@ -246,7 +246,7 @@ void fr_connection_signal_halt(fr_connection_t *conn); void fr_connection_signals_pause(fr_connection_t *conn); -void fr_connection_deferred_signals_resume(fr_connection_t *conn); +void fr_connection_signals_resume(fr_connection_t *conn); /** @} */ /** @name Install generic I/O events on an FD to signal state changes diff --git a/src/lib/server/trunk.c b/src/lib/server/trunk.c index fa308773e9d..ca5a9e69bd5 100644 --- a/src/lib/server/trunk.c +++ b/src/lib/server/trunk.c @@ -737,6 +737,10 @@ static void trunk_request_remove_from_conn(fr_trunk_request_t *treq) } /** Transition a request to the unassigned state, in preparation for re-assignment + * + * @note treq->tconn may be inviable after calling + * if treq->conn and fr_connection_signals_pause are not used. + * This is due to call to trunk_request_remove_from_conn. * * @param[in] treq to trigger a state change for. */ @@ -767,6 +771,10 @@ static void trunk_request_enter_unassigned(fr_trunk_request_t *treq) } /** Transition a request to the backlog state, adding it to the backlog of the trunk + * + * @note treq->tconn and treq may be inviable after calling + * if treq->conn and fr_connection_signals_pause are not used. + * This is due to call to trunk_manage. * * @param[in] treq to trigger a state change for. */ @@ -825,6 +833,10 @@ static void trunk_request_enter_backlog(fr_trunk_request_t *treq) * All trunk requests being added to a connection get passed to this function. * All trunk requests being removed from a connection get passed to #trunk_request_remove_from_conn. * + * @note treq->tconn and treq may be inviable after calling + * if treq->conn and fr_connection_signals_pause is not used. + * This is due to call to trunk_connection_event_update. + * * @param[in] treq to trigger a state change for. * @param[in] tconn to enqueue the request on. */ @@ -903,6 +915,10 @@ static void trunk_request_enter_partial(fr_trunk_request_t *treq) } /** Transition a request to the sent state, indicating that it's been sent in its entirety + * + * @note treq->tconn and treq may be inviable after calling + * if treq->conn and fr_connection_signals_pause is not used. + * This is due to call to trunk_connection_event_update. * * @param[in] treq to trigger a state change for. */ @@ -961,6 +977,10 @@ static void trunk_request_enter_sent(fr_trunk_request_t *treq) * A request must enter this state before being added to the backlog * of another connection if it's been sent or partially sent. * + * @note treq->tconn and treq may be inviable after calling + * if treq->conn and fr_connection_signals_pause is not used. + * This is due to call to trunk_connection_event_update. + * * @param[in] treq to trigger a state change for. * @param[in] reason Why the request was cancelled. * Should be one of: @@ -1048,6 +1068,10 @@ static void trunk_request_enter_cancel_partial(fr_trunk_request_t *treq) * that the cancel request is complete when the remote server * acknowledges the cancellation request. * + * @note treq->tconn and treq may be inviable after calling + * if treq->conn and fr_connection_signals_pause is not used. + * This is due to call to trunk_connection_event_update. + * * @param[in] treq to trigger a state change for. */ static void trunk_request_enter_cancel_sent(fr_trunk_request_t *treq) @@ -1088,6 +1112,11 @@ static void trunk_request_enter_cancel_sent(fr_trunk_request_t *treq) * The API client will not be informed, as the original REQUEST * * will likely have been freed by this point. * + * @note treq will be inviable after a call to this function. + * treq->tconn may be inviable after calling + * if treq->conn and fr_connection_signals_pause is not used. + * This is due to call to trunk_request_remove_from_conn. + * * @param[in] treq to mark as complete. */ static void trunk_request_enter_cancel_complete(fr_trunk_request_t *treq) @@ -1114,6 +1143,10 @@ static void trunk_request_enter_cancel_complete(fr_trunk_request_t *treq) } /** Request completed successfully, inform the API client and free the request + * + * @note treq will be inviable after a call to this function. + * treq->tconn may also be inviable due to call to + * trunk_request_remove_from_conn. * * @param[in] treq to mark as complete. */ @@ -1140,6 +1173,10 @@ static void trunk_request_enter_complete(fr_trunk_request_t *treq) } /** Request failed, inform the API client and free the request + * + * @note treq will be inviable after a call to this function. + * treq->tconn may also be inviable due to call to + * trunk_request_remove_from_conn. * * @param[in] treq to mark as failed. */ @@ -1156,6 +1193,15 @@ static void trunk_request_enter_failed(fr_trunk_request_t *treq) REQUEST_EXTRACT_BACKLOG(treq); break; + /* + * These two states should be dealt with + * separately. + */ + case FR_TRUNK_REQUEST_STATE_CANCEL_SENT: + case FR_TRUNK_REQUEST_STATE_SENT: + REQUEST_BAD_STATE_TRANSITION(FR_TRUNK_REQUEST_STATE_FAILED); + break; + default: trunk_request_remove_from_conn(treq); break; @@ -1262,8 +1308,14 @@ static fr_trunk_enqueue_t trunk_request_enqueue_existing(fr_trunk_request_t *tre rcode = trunk_request_check_enqueue(&tconn, trunk, treq->pub.request); switch (rcode) { case FR_TRUNK_ENQUEUE_OK: - trunk_request_enter_pending(treq, tconn); - if (trunk->conf.always_writable) trunk_connection_writable(tconn); + if (trunk->conf.always_writable) { + fr_connection_signals_pause(tconn->pub.conn); + trunk_request_enter_pending(treq, tconn); + trunk_connection_writable(tconn); + fr_connection_signals_resume(tconn->pub.conn); + } else { + trunk_request_enter_pending(treq, tconn); + } break; case FR_TRUNK_ENQUEUE_IN_BACKLOG: @@ -1363,9 +1415,16 @@ static uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk treq = tconn->partial; if (treq) { rad_assert(treq->pub.state == FR_TRUNK_REQUEST_STATE_PARTIAL); + + /* + * Don't allow the connection to change state whilst + * we're draining requests from it. + */ + fr_connection_signals_pause(tconn->pub.conn); trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_MOVE); trunk_request_enter_unassigned(treq); fr_dlist_insert_tail(out, treq); + fr_connection_signals_resume(tconn->pub.conn); } } @@ -1373,13 +1432,20 @@ static uint64_t trunk_connection_requests_dequeue(fr_dlist_head_t *out, fr_trunk * Cancel sent requests */ if (states & FR_TRUNK_REQUEST_STATE_SENT) { + /* + * Don't allow the connection to change state whilst + * we're draining requests from it. + */ + fr_connection_signals_pause(tconn->pub.conn); while ((treq = fr_dlist_head(&tconn->sent))) { OVER_MAX_CHECK; rad_assert(treq->pub.state == FR_TRUNK_REQUEST_STATE_SENT); + trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_MOVE); trunk_request_enter_unassigned(treq); fr_dlist_insert_tail(out, treq); } + fr_connection_signals_resume(tconn->pub.conn); } return count; @@ -1647,6 +1713,11 @@ void fr_trunk_request_signal_cancel(fr_trunk_request_t *treq) { fr_trunk_t *trunk; + /* + * Ensure treq hasn't been freed + */ + (void)talloc_get_type_abort(treq, fr_trunk_request_t); + if (!fr_cond_assert_msg(treq->pub.trunk, "treq not associated with trunk")) return; if (!fr_cond_assert_msg(!IN_HANDLER(treq->pub.trunk), @@ -1661,12 +1732,20 @@ void fr_trunk_request_signal_cancel(fr_trunk_request_t *treq) */ case FR_TRUNK_REQUEST_STATE_PARTIAL: case FR_TRUNK_REQUEST_STATE_SENT: - trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_SIGNAL); + { + fr_trunk_connection_t *tconn = treq->pub.tconn; + /* + * Don't allow connection state changes + */ + fr_connection_signals_pause(tconn->pub.conn); + trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_SIGNAL); if (!fr_cond_assert_msg(treq->pub.state == FR_TRUNK_REQUEST_STATE_CANCEL, "Bad state %s after cancellation", - fr_table_str_by_value(fr_trunk_request_states, treq->pub.state, ""))) return; - + fr_table_str_by_value(fr_trunk_request_states, treq->pub.state, ""))) { + fr_connection_signals_resume(tconn->pub.conn); + return; + } /* * No cancel muxer. We're done. * @@ -1679,6 +1758,8 @@ void fr_trunk_request_signal_cancel(fr_trunk_request_t *treq) trunk_request_enter_unassigned(treq); fr_trunk_request_free(&treq); } + fr_connection_signals_resume(tconn->pub.conn); + } break; /* @@ -2019,8 +2100,14 @@ fr_trunk_enqueue_t fr_trunk_request_enqueue(fr_trunk_request_t **treq_out, fr_tr } treq->pub.preq = preq; treq->pub.rctx = rctx; - trunk_request_enter_pending(treq, tconn); - if (trunk->conf.always_writable) trunk_connection_writable(tconn); + if (trunk->conf.always_writable) { + fr_connection_signals_pause(tconn->pub.conn); + trunk_request_enter_pending(treq, tconn); + trunk_connection_writable(tconn); + fr_connection_signals_resume(tconn->pub.conn); + } else { + trunk_request_enter_pending(treq, tconn); + } break; case FR_TRUNK_ENQUEUE_IN_BACKLOG: @@ -2085,8 +2172,10 @@ fr_trunk_enqueue_t fr_trunk_request_requeue(fr_trunk_request_t *treq) switch (treq->pub.state) { case FR_TRUNK_REQUEST_STATE_PARTIAL: case FR_TRUNK_REQUEST_STATE_SENT: + fr_connection_signals_pause(tconn->pub.conn); trunk_request_enter_cancel(treq, FR_TRUNK_CANCEL_REASON_REQUEUE); trunk_request_enter_pending(treq, tconn); + fr_connection_signals_resume(tconn->pub.conn); break; case FR_TRUNK_REQUEST_STATE_BACKLOG: /* Do nothing.... */ @@ -2163,8 +2252,14 @@ fr_trunk_enqueue_t fr_trunk_request_enqueue_on_conn(fr_trunk_request_t **treq_ou treq->pub.rctx = rctx; treq->bound_to_conn = true; /* Don't let the request be transferred */ - trunk_request_enter_pending(treq, tconn); - if (trunk->conf.always_writable) trunk_connection_writable(tconn); + if (trunk->conf.always_writable) { + fr_connection_signals_pause(tconn->pub.conn); + trunk_request_enter_pending(treq, tconn); + trunk_connection_writable(tconn); + fr_connection_signals_resume(tconn->pub.conn); + } else { + trunk_request_enter_pending(treq, tconn); + } return FR_TRUNK_ENQUEUE_OK; } @@ -2381,7 +2476,7 @@ static void trunk_connection_event_update(fr_trunk_connection_t *tconn) fr_connection_signals_pause(tconn->pub.conn); DO_CONNECTION_NOTIFY(tconn, events); tconn->events = events; - fr_connection_deferred_signals_resume(tconn->pub.conn); + fr_connection_signals_resume(tconn->pub.conn); } } @@ -2890,14 +2985,14 @@ static void _trunk_connection_on_closed(UNUSED fr_connection_t *conn, UNUSED fr_ tconn->sent_count = 0; /* - * Remove the I/O events + * Remove the reconnect event */ - trunk_connection_event_update(tconn); + if (trunk->conf.lifetime > 0) fr_event_timer_delete(&tconn->lifetime_ev); /* - * Remove the reconnect event + * Remove the I/O events */ - if (trunk->conf.lifetime > 0) fr_event_timer_delete(&tconn->lifetime_ev); + trunk_connection_event_update(tconn); } /** Connection transitioned to the halted state