track requests and listeners

so that when a listener goes away, we cancel all requests
associated with it.

diff --git a/src/lib/io/worker.c b/src/lib/io/worker.c
index 7d51a6a0f8cae8cc7f86088695af2e0fecdd4783..4cc547544746d0cc1259a3c8c584cefe77b62361 100644 (file)
--- a/src/lib/io/worker.c
+++ b/src/lib/io/worker.c
@@ -102,6 +102,8 @@ struct fr_worker_s {
        fr_minmax_heap_t        *time_order;    //!< time ordered heap of requests
        fr_rb_tree_t            *dedup;         //!< de-dup tree
 
+       fr_rb_tree_t            *listeners;     //!< so we can cancel requests when a listener goes away
+
        fr_io_stats_t           stats;          //!< input / output stats
        fr_time_elapsed_t       cpu_time;       //!< histogram of total CPU time per request
        fr_time_elapsed_t       wall_clock;     //!< histogram of wall clock time per request
@@ -122,6 +124,27 @@ struct fr_worker_s {
        fr_channel_t            **channel;      //!< list of channels
 };
 
+typedef struct {
+       fr_listen_t const       *listener;      //!< incoming packets
+
+       fr_rb_node_t            node;           //!< in tree of listeners
+
+       /*
+        *      To save time, we don't care about num_elements here.  Which means that we don't
+        *      need to cache or lookup the fr_worker_listen_t when we free a request.
+        */
+       fr_dlist_head_t         dlist;          //!< of requests associated with this listener.
+} fr_worker_listen_t;
+
+
+static int8_t worker_listener_cmp(void const *one, void const *two)
+{
+       fr_worker_listen_t const *a = one, *b = two;
+
+       return CMP(a->listener, b->listener);
+}
+
+
 /*
  *     Explicitly cleanup the memory allocated to the ring buffer,
  *     just in case valgrind complains about it.
@@ -308,8 +331,21 @@ static void worker_channel_callback(void *ctx, void const *data, size_t data_siz
        }
 }
 
-static int fr_worker_listen_cancel_self(UNUSED fr_worker_t *worker, UNUSED fr_listen_t const *li)
+static int fr_worker_listen_cancel_self(fr_worker_t *worker, fr_listen_t const *li)
 {
+       fr_worker_listen_t *wl;
+       request_t *request;
+
+       wl = fr_rb_find(worker->listeners, &(fr_worker_listen_t) { .listener = li });
+       if (!wl) return -1;
+
+       while ((request = fr_dlist_pop_head(&wl->dlist)) != NULL) {
+               RDEBUG("Canceling request due to socket being closed");
+               unlang_interpret_signal(request, FR_SIGNAL_CANCEL);
+       }
+
+       (void) fr_rb_delete(worker->listeners, wl);
+       talloc_free(wl);
 
        return 0;
 }
@@ -674,6 +710,8 @@ static void worker_send_reply(fr_worker_t *worker, request_t *request, bool send
        fr_assert(!fr_minmax_heap_entry_inserted(request->time_order_id));
        fr_assert(!fr_heap_entry_inserted(request->runnable_id));
 
+       fr_dlist_entry_unlink(&request->listen_entry);
+
 #ifndef NDEBUG
        request->async->el = NULL;
        request->async->process = NULL;
@@ -888,6 +926,21 @@ nak:
        }
 
        worker_request_time_tracking_start(worker, request, now);
+
+       {
+               fr_worker_listen_t *wl;
+
+               wl = fr_rb_find(worker->listeners, &(fr_worker_listen_t) { .listener = listen });
+               if (!wl) {
+                       MEM(wl = talloc_zero(worker, fr_worker_listen_t));
+                       fr_dlist_init(&wl->dlist, request_t, listen_entry);
+                       wl->listener = listen;
+
+                       (void) fr_rb_insert(worker->listeners, wl);
+               }
+
+               fr_dlist_insert_tail(&wl->dlist, request);
+       }
 }
 
 /**
@@ -1358,6 +1411,12 @@ nomem:
                goto fail;
        }
 
+       worker->listeners = fr_rb_inline_talloc_alloc(worker, fr_worker_listen_t, node, worker_listener_cmp, NULL);
+       if (!worker->listeners) {
+               fr_strerror_const("Failed creating listener tree");
+               goto fail;
+       }
+
        worker->intp = unlang_interpret_init(worker, el,
                                             &(unlang_request_func_t){
                                                        .init_internal = _worker_request_internal_init,

diff --git a/src/lib/server/request.c b/src/lib/server/request.c
index aeafb414837839a76bf43a1dd1e8fcd4cc5b6636..2b18c5c40d811a3acaef7829f58273638ab02fc4 100644 (file)
--- a/src/lib/server/request.c
+++ b/src/lib/server/request.c
@@ -467,6 +467,11 @@ request_t *_request_alloc(char const *file, int line, TALLOC_CTX *ctx,
         */
        fr_dlist_entry_init(&request->free_entry);      /* Needs to be initialised properly, else bad things happen */
 
+       /*
+        *      This is only used by src/lib/io/worker.c
+        */
+       fr_dlist_entry_init(&request->listen_entry);
+
        /*
         *      Bind lifetime to a parent.
         *

diff --git a/src/lib/server/request.h b/src/lib/server/request.h
index 39b46acee15ee53b7d921568ec56205ae937035c..b1eabae1d9816cb6c2d484324ba44553d0b7fb3c 100644 (file)
--- a/src/lib/server/request.h
+++ b/src/lib/server/request.h
@@ -233,6 +233,7 @@ struct request_s {
 
        int                     alloc_line;     //!< Line the request was allocated on.
 
+       fr_dlist_t              listen_entry;   //!< request's entry in the list for this listener / socket
        fr_dlist_t              free_entry;     //!< Request's entry in the free list.
 };                             /* request_t typedef */