gnunet

commit d2d63aa2365db76f1171950eb9322a32537d7be7
parent 37564fc6b8417ebed6fbf3f4ab6eeb31dd5dc28a
Author: Martin Schanzenbach <schanzen@gnunet.org>
Date:   Thu, 30 Jul 2026 07:48:29 +0200

communicator(tcp): fix multiple queues starving each other out

Diffstat:
M
src/service/transport/gnunet-service-transport.c
 | 
64
++++++++++++++++++++++++++++++++++++++++++++++++++++++----------

1 file changed, 54 insertions(+), 10 deletions(-)
diff --git a/src/service/transport/gnunet-service-transport.c b/src/service/transport/gnunet-service-transport.c
@@ -4149,6 +4149,16 @@ check_for_queue_with_higher_prio (struct Queue *queue, struct Queue *queue_head)
 
 
 /**
+ * Make sure #free_timedout_queue_entry() will run for @a tc, so that
+ * `struct QueueEntry's the communicator never acknowledged are reclaimed.
+ *
+ * @param tc communicator whose queue entries need sweeping
+ */
+static void
+arm_free_queue_entry_task (struct TransportClient *tc);
+
+
+/**
  * Called whenever something changed that might effect when we
  * try to do the next transmission on @a queue using #transmit_on_queue().
  *
@@ -4182,6 +4192,17 @@ schedule_transmit_on_queue (struct GNUNET_TIME_Relative delay,
       1,
       GNUNET_NO);
     queue->idle = GNUNET_NO;
+    /* @e total_queue_length is shared by every peer this communicator serves,
+       so at the limit we can send to none of them, and the only things that
+       lower it again are an ACK from the communicator, #free_queue() and
+       #free_timedout_queue_entry().  If we got here because a peer stopped
+       acknowledging, no ACK is coming; and we just declined to arm
+       @e transmit_task, so #queue_send_msg() -- the *only* place that arms
+       the sweeper -- will not run either.  That leaves nothing at all to
+       recover the count, and the communicator stays mute towards every peer
+       until some unrelated queue happens to go down.  Keep the sweeper
+       going. */
+    arm_free_queue_entry_task (queue->tc);
     return;
   }
   if (queue->queue_length >= QUEUE_LENGTH_LIMIT)
@@ -4195,6 +4216,8 @@ schedule_transmit_on_queue (struct GNUNET_TIME_Relative delay,
                               1,
                               GNUNET_NO);
     queue->idle = GNUNET_NO;
+    /* Same reasoning as above, for the per-queue limit. */
+    arm_free_queue_entry_task (queue->tc);
     return;
   }
   if (0 == queue->q_capacity)
@@ -5170,10 +5193,40 @@ free_timedout_queue_entry (void *cls)
                     " and QID %u\n",
                     pos->mid,
                     queue->qid);
+        GNUNET_STATISTICS_update (GST_stats,
+                                  "# QueueEntries timed out",
+                                  1,
+                                  GNUNET_NO);
         free_queue_entry (pos, tc);
       }
     }
   }
+  /* Entries younger than #QUEUE_ENTRY_TIMEOUT are still to be reclaimed, and
+     this task used to be a one-shot armed only by #queue_send_msg().  So the
+     sweep that was supposed to bound how long an unacknowledged entry holds
+     its slot only ever ran while *new* messages were being queued -- exactly
+     not the case once the throttles in #schedule_transmit_on_queue() have
+     stopped transmission.  Keep going as long as anything is outstanding. */
+  arm_free_queue_entry_task (tc);
+}
+
+
+static void
+arm_free_queue_entry_task (struct TransportClient *tc)
+{
+  if (NULL != tc->details.communicator.free_queue_entry_task)
+    return;
+  for (struct Queue *queue = tc->details.communicator.queue_head; NULL != queue;
+       queue = queue->next_client)
+  {
+    if (NULL == queue->queue_head)
+      continue;
+    tc->details.communicator.free_queue_entry_task =
+      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+                                    &free_timedout_queue_entry,
+                                    tc);
+    return;
+  }
 }
 
 
@@ -5279,16 +5332,7 @@ queue_send_msg (struct Queue *queue,
       queue->idle = GNUNET_NO;
 
     if (GNUNET_NO == queue->idle)
-    {
-      struct TransportClient *tc = queue->tc;
-
-      if (NULL == tc->details.communicator.free_queue_entry_task)
-        tc->details.communicator.free_queue_entry_task =
-          GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
-                                        &
-                                        free_timedout_queue_entry,
-                                        tc);
-    }
+      arm_free_queue_entry_task (queue->tc);
     if (NULL != pm && NULL != (pa = pm->pa_head))
     {
       while (pm != pa->pm)