track duplicate packets based on recv_time.

If the packet_ctx is the same, and the recv_time is the same,
then the new packet is a duplicate of the old one.  Otherwise,
it's not.

And only listeners with "track_duplicates" set will have requests
inserted into the worker dedup tree.

diff --git a/src/lib/io/application.h b/src/lib/io/application.h
index 0c7ac15a7ac0fbc8e22ad4734fd9b4f71d73df1e..00dd80dd32b059e80ed4aa6e8bbb59274c0688d2 100644 (file)
--- a/src/lib/io/application.h
+++ b/src/lib/io/application.h
@@ -96,7 +96,8 @@ typedef struct fr_app_io_t {
        fr_app_event_list_set_t         event_list_set; //!< Called by the network thread to pass an event list
                                                        //!< for use by the app_io_t.
 
-       size_t                          default_message_size;   // Usually minimum message size
+       size_t                          default_message_size;   //!< Usually minimum message size
+       bool                            track_duplicates;       //!< track duplicate packets
 
        fr_io_open_t                    open;           //!< Open a new socket for listening, or accept/connect a new
                                                        //!< connection.

diff --git a/src/lib/io/worker.c b/src/lib/io/worker.c
index cdcc3ba0a72d3a111afca9d51cf4d04b00155682..3b801d973f61b6edd757a5529307e4002385aebf 100644 (file)
--- a/src/lib/io/worker.c
+++ b/src/lib/io/worker.c
@@ -539,7 +539,7 @@ static void worker_stop_request(fr_worker_t *worker, REQUEST *request, fr_time_t
        /*
         *      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.
+        *      MAY be in the dedup list, but if not, no worries.
         */
        (void) fr_heap_extract(worker->time_order, request);
        (void) fr_heap_extract(worker->runnable, request);
@@ -725,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, *old;
+       REQUEST                 *request;
        fr_listen_t const       *listen;
 #ifndef HAVE_TALLOC_POOLED_OBJECT
        TALLOC_CTX              *ctx;
@@ -761,7 +761,7 @@ static REQUEST *fr_worker_get_request(fr_worker_t *worker, fr_time_t now)
                if (cd->request.recv_time && (cd->m.when != *cd->request.recv_time)) {
                        DEBUG("\t%sIGNORING old message: was %zd now %zd", worker->name,
                              *cd->request.recv_time, cd->m.when);
-                       fr_worker_nak(worker, cd, fr_time());
+                       fr_worker_nak(worker, cd, now);
                        cd = NULL;
                }
        } while (!cd);
@@ -819,7 +819,7 @@ static REQUEST *fr_worker_get_request(fr_worker_t *worker, fr_time_t now)
                DEBUG("\t%sFAILED decode of request %"PRIu64, worker->name, request->number);
                talloc_free(ctx);
 nak:
-               fr_worker_nak(worker, cd, fr_time());
+               fr_worker_nak(worker, cd, now);
                return NULL;
        }
 
@@ -831,7 +831,7 @@ nak:
 
        if (!request->async->process) {
                ERROR("Protocol failed to set 'process' function");
-               fr_worker_nak(worker, cd, fr_time());
+               fr_worker_nak(worker, cd, now);
                return NULL;
        }
 
@@ -858,10 +858,55 @@ nak:
         *      @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, fr_time());
+       /*
+        *      Look for conflicting / duplicate packets, but only if
+        *      requested to do so.
+        */
+       if (request->async->listen->app_io->track_duplicates) {
+               REQUEST *old;
+
+               old = rbtree_finddata(worker->dedup, request);
+               if (!old) goto insert_new;
+
+               rad_assert(old->async->listen == request->async->listen);
+               rad_assert(old->async->channel == request->async->channel);
+
+               /*
+                *      There's a new packet.  Do we keep the old one,
+                *      or the new one?  This decision is made by
+                *      checking the recv_time, which is a
+                *      nanosecond-resolution timer.  If the time is
+                *      identical, then the new packet is the same as
+                *      the old one.
+                *
+                *      If the new packet is a duplicate of the old
+                *      one, then we can just discard the new one.  We
+                *      have to tell the channel that we've "eaten"
+                *      this reply, so the sequence number should
+                *      increase.
+                *
+                *      @todo - fix the channel code to do queue
+                *      depth, and not sequence / ack.
+                *
+                *      @todo - send DUP signal to old request!
+                */
+               if (old->async->recv_time == request->async->recv_time) {
+                       fr_channel_null_reply(request->async->channel);
+                       talloc_free(request);
+                       return NULL;
+               }
+
+               /*
+                *      Stop the old request, and decrement the number
+                *      of active requests.
+                */
+               worker_stop_request(worker, old, now);
+               rad_assert(worker->num_active > 0);
+               worker->num_active--;
                talloc_free(old);
+
+       insert_new:
+               (void) rbtree_insert(worker->dedup, request);
        }
 
        /*
@@ -1065,8 +1110,12 @@ static int worker_time_order_cmp(void const *one, void const *two)
  */
 static int worker_dedup_cmp(void const *one, void const *two)
 {
+       int ret;
        REQUEST const *a = one, *b = two;
 
+       ret = (a->async->listen > b->async->listen) - (a->async->listen < b->async->listen);
+       if (ret) return ret;
+
        return (a->async->packet_ctx > b->async->packet_ctx) - (a->async->packet_ctx < b->async->packet_ctx);
 }