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 bd4807f6b86dd2a48f1e59e38a6ee0255aa60f6f..dce2b98ae7df273ecb13bbb2a19d4c2a3f109fd1 100644 (file)
--- 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 d3de9821a7a78330e937dc0dd231bd08f934f5c8..a73b7ef3ec5b0b829f956ceba480b9f83e645354 100644 (file)
--- 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 fa308773e9da9e9bf4340a4e6428e1da4ff85af1..ca5a9e69bd514f508623aea6b78add46552a74e8 100644 (file)
--- 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, "<INVALID>"))) return;
-
+                                       fr_table_str_by_value(fr_trunk_request_states, treq->pub.state, "<INVALID>"))) {
+                       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