Give cancelled requests extremely high priority

This ensures they're cleaned ASAP and release any resources.

diff --git a/src/lib/io/worker.c b/src/lib/io/worker.c
index 60a59604c956a44c5757cacf7cf9c23698f181ae..4db64deb02fcb0bf16310f80647bb7feb74eface 100644 (file)
--- a/src/lib/io/worker.c
+++ b/src/lib/io/worker.c
@@ -1291,6 +1291,22 @@ static bool _worker_request_scheduled(request_t const *request, UNUSED void *uct
        return fr_heap_entry_inserted(request->runnable);
 }
 
+/** Update a request's priority
+ *
+ */
+static void _worker_request_prioritise(request_t *request, void *uctx)
+{
+       fr_worker_t *worker = talloc_get_type_abort(uctx, fr_worker_t);
+
+       RDEBUG3("Request priority changed");
+
+       /* Extract the request from the runnable queue _if_ it's in the runnable queue */
+       if (fr_heap_extract(&worker->runnable, request) < 0) return;
+
+       /* Reinsert it to re-evaluate its new priority */
+       fr_heap_insert(&worker->runnable, request);
+}
+
 /** Run a request
  *
  *  Until it either yields, or is done.
@@ -1471,7 +1487,8 @@ nomem:
                                                        .resume = _worker_request_resume,
                                                        .mark_runnable = _worker_request_runnable,
 
-                                                       .scheduled = _worker_request_scheduled
+                                                       .scheduled = _worker_request_scheduled,
+                                                       .prioritise = _worker_request_prioritise
                                             },
                                             worker);
        if (!worker->intp){

diff --git a/src/lib/unlang/interpret.c b/src/lib/unlang/interpret.c
index cbf5b602dc017bcaec3a28025bf65365c666b46a..c88f1c9034aa966eb241cd8afc71f68d016378c3 100644 (file)
--- a/src/lib/unlang/interpret.c
+++ b/src/lib/unlang/interpret.c
@@ -1036,6 +1036,20 @@ void unlang_interpret_request_detach(request_t *request)
        intp->funcs.detach(request, intp->uctx);
 }
 
+void unlang_interpret_request_prioritise(request_t *request, uint32_t priority)
+{
+       unlang_stack_t          *stack = request->stack;
+       unlang_interpret_t      *intp;
+
+       if (!fr_cond_assert(stack != NULL)) return;
+
+       intp = stack->intp;
+
+       request->async->priority = priority;
+
+       if (intp->funcs.prioritise) intp->funcs.prioritise(request, intp->uctx);
+}
+
 /** Delivers a frame to one or more frames in the stack
  *
  * This is typically called via an "async" action, i.e. an action outside
@@ -1145,6 +1159,12 @@ void unlang_interpret_signal(request_t *request, fr_signal_t action)
                 */
                request->master_state = REQUEST_STOP_PROCESSING;
 
+               /*
+                *      Give cancelled requests the highest priority
+                *      to get them to release resources ASAP.
+                */
+               unlang_interpret_request_prioritise(request, UINT32_MAX);
+
                /*
                 *      If the request is yielded, mark it as runnable
                 *

diff --git a/src/lib/unlang/interpret.h b/src/lib/unlang/interpret.h
index 8cd68a07b98e57796a399621296e1644fa0a2d39..676652643c5efd0ba1adb6cf64dc6225b9cda3ab 100644 (file)
--- a/src/lib/unlang/interpret.h
+++ b/src/lib/unlang/interpret.h
@@ -88,6 +88,12 @@ typedef void (*unlang_request_runnable_t)(request_t *request, void *uctx);
  */
 typedef bool (*unlang_request_scheduled_t)(request_t const *request, void *uctx);
 
+/** Re-priotise the request in the runnable queue
+ *
+ * The new priority will be available in request->async->priority.
+ */
+typedef void (*unlang_request_prioritise_t)(request_t *request, void *uctx);
+
 /** External functions provided by the owner of the interpret
  *
  * These functions allow the event loop to signal the caller when a given
@@ -119,6 +125,8 @@ typedef struct {
                                                        ///< added back to the runnable queue.
        unlang_request_scheduled_t      scheduled;      //!< Function to check if a request is already
                                                        ///< scheduled.
+       unlang_request_prioritise_t     prioritise;     //!< Function to re-priotise a request in the
+                                                       ///< runnable queue.
 } unlang_request_func_t;
 
 int                    unlang_interpret_push_section(request_t *request, CONF_SECTION *cs,
@@ -160,6 +168,8 @@ bool                        unlang_request_is_done(request_t const *request);
 
 void                   unlang_interpret_request_done(request_t *request);
 
+void                   unlang_interpret_request_prioritise(request_t *request, uint32_t priority);
+
 void                   unlang_interpret_mark_runnable(request_t *request);
 
 bool                   unlang_interpret_is_resumable(request_t *request);

diff --git a/src/lib/unlang/interpret_synchronous.c b/src/lib/unlang/interpret_synchronous.c
index 80bce035bdb92132ef4d2c14190d9156c0cfd182..f7f6345f0bcabeed2d3c608cc5053dddc85b3bd1 100644 (file)
--- a/src/lib/unlang/interpret_synchronous.c
+++ b/src/lib/unlang/interpret_synchronous.c
@@ -169,7 +169,7 @@ static unlang_interpret_synchronous_t *unlang_interpret_synchronous_alloc(TALLOC
                                                        .yield = _request_yield,
                                                        .resume = _request_resume,
                                                        .mark_runnable = _request_runnable,
-                                                       .scheduled = _request_scheduled
+                                                       .scheduled = _request_scheduled,
                                                },
                                                intps));