add maximum_outstanding limit

for doing many packets in parallel

diff --git a/src/modules/proto_detail/proto_detail.h b/src/modules/proto_detail/proto_detail.h
index dde2aec2cc96e759915383072c0c08454687e378..5c47d1155b81a89b61f185b39aa81c678a627988 100644 (file)
--- a/src/modules/proto_detail/proto_detail.h
+++ b/src/modules/proto_detail/proto_detail.h
@@ -89,6 +89,8 @@ typedef struct proto_detail_work_t {
        uint32_t                        mrt;
        uint32_t                        mrc;
        uint32_t                        mrd;
+       uint32_t                        max_outstanding;        //!< number of packets to run in parallel
+       uint32_t                        outstanding;            //!< number of currently outstanding records;
 
        fr_dlist_t                      list;                   //!< for retransmissions
 
@@ -104,7 +106,6 @@ typedef struct proto_detail_work_t {
 
        int                             mode;                   //!< O_RDWR or O_RDONLY
 
-       int                             outstanding;            //!< number of outstanding records;
        int                             count;                  //!< number of packets we read from this file.
 
        size_t                          last_search;            //!< where we last searched in the buffer

diff --git a/src/modules/proto_detail/proto_detail_work.c b/src/modules/proto_detail/proto_detail_work.c
index f1d4019328e3222cd56b9444e693929533968ce1..fed821e25b53b9ae8dba355d021dfaf4773afc66 100644 (file)
--- a/src/modules/proto_detail/proto_detail_work.c
+++ b/src/modules/proto_detail/proto_detail_work.c
@@ -69,6 +69,7 @@ static CONF_PARSER limit_config[] = {
        { FR_CONF_OFFSET("maximum_retransmission_time", FR_TYPE_UINT32, proto_detail_work_t, mrt), .dflt = STRINGIFY(16) },
        { FR_CONF_OFFSET("maximum_retransmission_count", FR_TYPE_UINT32, proto_detail_work_t, mrc), .dflt = STRINGIFY(5) },
        { FR_CONF_OFFSET("maximum_retransmission_duration", FR_TYPE_UINT32, proto_detail_work_t, mrd), .dflt = STRINGIFY(30) },
+       { FR_CONF_OFFSET("maximum_outstanding", FR_TYPE_UINT32, proto_detail_work_t, max_outstanding), .dflt = STRINGIFY(1) },
        CONF_PARSER_TERMINATOR
 };
 
@@ -437,7 +438,7 @@ done:
        /*
         *      Pause reading until such time as we need more packets.
         */
-       if (!inst->paused) {
+       if (!inst->paused && (inst->outstanding >= inst->max_outstanding)) {
                (void) fr_event_filter_update(inst->el, inst->fd, FR_EVENT_FILTER_IO, pause_read);
                inst->paused = true;
 
@@ -453,7 +454,7 @@ done:
         */
        inst->last_search = 0;
 
-       MPRINT("Returning NUM %d - %.*s", inst->outstanding, (int) packet_len, buffer);
+       MPRINT("Returning NUM %u - %.*s", inst->outstanding, (int) packet_len, buffer);
        return packet_len;
 }
 
@@ -468,7 +469,7 @@ static void work_retransmit(UNUSED fr_event_list_t *el, UNUSED struct timeval *n
 
        fr_dlist_insert_tail(&inst->list, &track->entry);
 
-       if (inst->paused) {
+       if (inst->paused && (inst->outstanding < inst->max_outstanding)) {
                (void) fr_event_filter_update(inst->el, inst->fd, FR_EVENT_FILTER_IO, resume_read);
                inst->paused = false;
        }
@@ -553,7 +554,7 @@ static ssize_t mod_write(void *instance, void *packet_ctx,
                        goto free_track;
                }
 
-               if (!inst->paused) {
+               if (!inst->paused && (inst->outstanding >= inst->max_outstanding)) {
                        (void) fr_event_filter_update(inst->el, inst->fd, FR_EVENT_FILTER_IO, pause_read);
                        inst->paused = true;
                }
@@ -574,10 +575,9 @@ free_track:
        inst->outstanding--;
 
        /*
-        *      There are no outstanding packets, let's go read some
-        *      more.
+        *      If we need to read some more packet, let's do so.
         */
-       if (!inst->outstanding && inst->paused) {
+       if (inst->paused && (inst->outstanding < inst->max_outstanding)) {
                (void) fr_event_filter_update(inst->el, inst->fd, FR_EVENT_FILTER_IO, resume_read);
                inst->paused = false;
        }