start of dedup / conflicting packet detection

diff --git a/src/lib/io/worker.c b/src/lib/io/worker.c
index 162ae6e1231dcf5dc531c661b6d3c2719c890469..2235a127b9df884f6a88caae4d29ecda40f9742a 100644 (file)
--- a/src/lib/io/worker.c
+++ b/src/lib/io/worker.c
@@ -130,6 +130,7 @@ struct fr_worker_t {
 
        fr_heap_t               *runnable;      //!< current runnable requests which we've spent time processing
        fr_heap_t               *time_order;    //!< time ordered heap of requests
+       rbtree_t                *dedup;         //!< de-dup tree
 
        int                     num_requests;   //!< number of requests processed by this worker
        int                     num_decoded;    //!< number of messages which have been decoded
@@ -520,11 +521,30 @@ static void fr_worker_send_reply(fr_worker_t *worker, REQUEST *request, size_t s
         *      and insert it back into a slab allocator.
         */
        (void) fr_heap_extract(worker->time_order, request);
+       (void) rbtree_deletebydata(worker->dedup, request);
        talloc_free(request);
 
        if (!worker->num_active) worker_reset_timer(worker);
 }
 
+
+/**  Tell a request that it's stopped.
+ *
+ */
+static void worker_stop_request(fr_worker_t *worker, REQUEST *request)
+{
+       (void) request->async->process(request, FR_IO_ACTION_DONE);
+
+       /*
+        *      The request is ALWAYS in the time_order list.  It MAY
+        *      be in the runnable list, but if not, no worries.  It
+        *      ALWAYS is in the dedup list.
+        */
+       (void) fr_heap_extract(worker->time_order, request);
+       (void) fr_heap_extract(worker->runnable, request);
+       (void) rbtree_deletebydata(worker->dedup, request);
+}
+
 /** Enforce max_request_time
  *
  *  Run periodically, and tries to clean up old requests.  In the
@@ -556,12 +576,9 @@ static void fr_worker_max_request_time(UNUSED fr_event_list_t *el, UNUSED struct
                /*
                 *      Waiting too long, delete it.
                 */
-               (void) fr_heap_extract(worker->time_order, request);
-               (void) fr_heap_extract(worker->runnable, request);
                fr_time_tracking_resume(&request->async->tracking, now);
-
                DEBUG("(%"PRIu64") taking too long, stopping it", request->number);
-               (void) request->async->process(request, FR_IO_ACTION_DONE);
+               worker_stop_request(worker, request);
 
                /*
                 *      Tell the network side that this request is done.
@@ -708,7 +725,7 @@ static REQUEST *fr_worker_get_request(fr_worker_t *worker, fr_time_t now)
 {
        int                     ret = -1;
        fr_channel_data_t       *cd;
-       REQUEST                 *request;
+       REQUEST                 *request, *old;
        fr_listen_t const       *listen;
 #ifndef HAVE_TALLOC_POOLED_OBJECT
        TALLOC_CTX              *ctx;
@@ -836,16 +853,16 @@ nak:
        (void) fr_heap_insert(worker->time_order, request);
 
        /*
-        *      @todo - create an RBtree based on comparing request->async->packet_ctx?
-        *      so that we can detect duplicates...
-        *
-        *      add ASYNC API to do handle (old, &FR_IO_DONE, new, &FR_IO_DONE)
-        *      so the async process function can choose either (or both) of the packets
-        *      and do any async signalling as necessary
+        *      Look for conflicting / duplicate packets.
         *
-        *      which probably also menas moving to queue-depth instead of seq/ack for network/worker,
-        *      as we don't want to send NAKs for conflicing / discarded packets
+        *      @todo - somehow figure out if the packet is a DUP, or
+        *      a conflicting one?
         */
+       old = rbtree_finddata(worker->dedup, request);
+       if (old) {
+               worker_stop_request(worker, old);
+               talloc_free(old);
+       }
 
        /*
         *      Bootstrap the async state machine with the initial
@@ -1043,6 +1060,16 @@ static int worker_time_order_cmp(void const *one, void const *two)
        return (a->async->recv_time > b->async->recv_time) - (a->async->recv_time < b->async->recv_time);
 }
 
+/**
+ *  Track a REQUEST in the "dedup" tree
+ */
+static int worker_dedup_cmp(void const *one, void const *two)
+{
+       REQUEST const *a = one, *b = two;
+
+       return (a->async->packet_ctx > b->async->packet_ctx) - (a->async->packet_ctx < b->async->packet_ctx);
+}
+
 /** Destroy a worker.
  *
  *  The input channels are signaled, and local messages are cleaned up.
@@ -1074,33 +1101,22 @@ void fr_worker_destroy(fr_worker_t *worker)
                fr_message_done(&cd->m);
        }
 
-       /*
-        *      Remove the requests from the "runnable" queue.
-        *
-        *      @todo - set a destructor for the REQUEST which cleans
-        *      it all up.
-        */
-       while ((request = fr_heap_pop(worker->runnable)) != NULL) {
-               (void) fr_heap_extract(worker->time_order, request);
-               fr_time_tracking_resume(&request->async->tracking, now);
-               talloc_free(request);
-       }
-       talloc_free(worker->runnable);
-
        /*
         *      Destroy all of the active requests.  These are ones
         *      which are still waiting for timers or file descriptor
         *      events.
         */
        while ((request = fr_heap_peek(worker->time_order)) != NULL) {
-               (void) request->async->process(request, FR_IO_ACTION_DONE);
-
-               (void) fr_heap_extract(worker->time_order, request);
                fr_time_tracking_resume(&request->async->tracking, now);
+
+               worker_stop_request(worker, request);
                talloc_free(request);
        }
        talloc_free(worker->time_order);
 
+       rad_assert(fr_heap_num_elements(worker->runnable) == 0);
+       talloc_free(worker->runnable);
+
 #if 0
        /*
         *      Signal the channels that we're closing.
@@ -1226,6 +1242,12 @@ nomem:
                goto fail;
        }
 
+       worker->dedup = rbtree_create(worker, worker_dedup_cmp, NULL, RBTREE_FLAG_NONE);
+       if (!worker->dedup) {
+               fr_strerror_printf("Failed creating de_dup tree");
+               goto fail;
+       }
+
        if (fr_event_post_insert(worker->el, fr_worker_post_event, worker) < 0) {
                fr_strerror_printf("Failed inserting post-processing event");
                talloc_free(worker->runnable);
@@ -1451,6 +1473,9 @@ static void fr_worker_verify(fr_worker_t *worker)
        rad_assert(worker->runnable != NULL);
        (void) talloc_get_type_abort(worker->runnable, fr_heap_t);
 
+       rad_assert(worker->dedup != NULL);
+       (void) talloc_get_type_abort(worker->dedup, rbtree_t);
+
        for (i = 0; i < worker->max_channels; i++) {
                if (!worker->channel[i]) continue;