Make dummynet bandwidth timing cumulative and add bounded DROP/HOLD modes

src/transport_layer/dummynet.c

@@ -16,6 +16,7 @@
 #include <string.h>
 #include <stdio.h>
 #include <errno.h>
+#include <limits.h>
 #include "../lib/mem.h"
 
 /* Timebase: 0.1 мс */
@@ -26,9 +27,7 @@
 #define SHAPER_BURST_TB 10
 
 /* Debug category */
-#ifndef DEBUG_CATEGORY_DUMMYNET
-#define DEBUG_CATEGORY_DUMMYNET ((debug_category_t)1 << 14)
-#endif
+#define DEBUG_CATEGORY_DUMMYNET DEBUG_CATEGORY_ETCP
 
 /* Forward declarations */
 static void dummynet_read_callback(socket_t sock, void* user_arg);
@@ -51,6 +50,9 @@ struct dummynet_pkt {
  * @brief Направление передачи
  */
 struct dummynet_dir {
+    struct dummynet* owner;
+    int index;
+    enum dummynet_mode mode;
     uint32_t delay_fixed_ms;
     uint32_t delay_random_ms;
     uint32_t bandwidth_kbps;
@@ -63,7 +65,8 @@ struct dummynet_dir {
     struct ll_queue* queue;
 
     void* shaper_timer;
-    int shaper_pending;
+    uint64_t tx_finish_ns;        // абсолютный срок; дробные миллисекунды не теряются между пакетами
+    int tx_started;
 
     /* Отладочные счетчики */
     uint64_t delay_timer_set;    /* Число установок delay таймеров */
@@ -86,6 +89,7 @@ struct dummynet {
 
     struct dummynet_dir dirs[DUMMYNET_DIR_COUNT];
     struct memory_pool* pkt_pool;
+    struct delay_cb_arg* delays;  // владение аргументами/пакетами всех ожидающих delay-таймеров
 };
 
 /**
@@ -94,14 +98,8 @@ struct dummynet {
 struct delay_cb_arg {
     struct dummynet* dn;
     struct ll_entry* entry;
-};
-
-/**
- * @brief Аргумент для shaper callback
- */
-struct shaper_cb_arg {
-    struct dummynet* dn;
-    int dir_idx;
+    void* timer;
+    struct delay_cb_arg *next, *prev;
 };
 
 /* Генерация случайного числа 0..max */
@@ -127,64 +125,49 @@ static int dummynet_get_direction(struct dummynet* dn, const struct sockaddr_sto
 static void dummynet_process_queue(struct dummynet* dn, int dir_idx) {
     struct dummynet_dir* dir = &dn->dirs[dir_idx];
 
-    struct ll_entry* entry = queue_data_get(dir->queue);
-    if (!entry) {
-        dir->shaper_pending = 0;
-        return;
-    }
-
-    struct dummynet_pkt* pkt = (struct dummynet_pkt*)entry->data;
-    ssize_t n = socket_sendto(dn->sock, pkt->data, pkt->len,
-                              (struct sockaddr*)&dir->dest_addr, dir->dest_len);
-    if (n < 0) {
-        DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "sendto failed: %s", socket_strerror(socket_get_error()));
-    } else {
-        dir->stats.sent++;
-        DEBUG_DEBUG(DEBUG_CATEGORY_DUMMYNET, "Dir %d: sent packet (%zd bytes)", dir_idx, n);
-    }
-
-    size_t bytes_sent = (n > 0) ? (size_t)n : 0;
-    dir->stats.queue_size = queue_entry_count(dir->queue);
-    queue_entry_free(entry);
-
-    /* Планируем следующую отправку */
-    if (dir->stats.queue_size > 0) {
-        int delay_tb = 0;
-
-        if (dir->bandwidth_kbps > 0) {
-            uint64_t delay_calc = (uint64_t)bytes_sent * 80ULL / (uint64_t)dir->bandwidth_kbps;
-            delay_tb = (int)delay_calc;
-            if (delay_tb == 0) delay_tb = 1;
+    if (dir->mode != DUMMYNET_RUN || dir->shaper_timer) return;
+    uint64_t now_ns = get_time_us() * 1000;
+    if (!dir->tx_started) dir->tx_finish_ns = now_ns;
+    while (dir->queue->head) {
+        struct dummynet_pkt* pkt = (struct dummynet_pkt*)dir->queue->head->data;
+        if (!dir->tx_started) {
+            if (dir->bandwidth_kbps) dir->tx_finish_ns += ((uint64_t)pkt->len * 8000000 + dir->bandwidth_kbps - 1) /
+                dir->bandwidth_kbps;
+            dir->tx_started = 1;
         }
-
-        struct shaper_cb_arg* sarg = (struct shaper_cb_arg*)u_malloc(sizeof(struct shaper_cb_arg));
-        if (sarg) {
-            sarg->dn = dn;
-            sarg->dir_idx = dir_idx;
-            dir->shaper_timer = uasync_set_timeout(dn->ua, delay_tb, sarg, dummynet_shaper_callback, "dummynet_shaper");
-            if (dir->shaper_timer) {
-                dir->shaper_timer_set++;
-            } else {
-                u_free(sarg);
-                dir->shaper_pending = 0;
+        if (dir->tx_finish_ns > now_ns) {
+            uint64_t wait_tb = (dir->tx_finish_ns - now_ns + 99999) / 100000;
+            if (wait_tb > INT_MAX) wait_tb = INT_MAX;
+            dir->shaper_timer = uasync_set_timeout(dn->ua, (int)wait_tb, dir, dummynet_shaper_callback, "dummynet_shaper");
+            if (!dir->shaper_timer) {
+                DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: cannot schedule serializer", dir_idx);
+                abort();
             }
-        } else {
-            dir->shaper_pending = 0;
+            dir->shaper_timer_set++;
+            return;
         }
-    } else {
-        dir->shaper_pending = 0;
+        struct ll_entry* entry = queue_data_get(dir->queue);
+        ssize_t n = socket_sendto(dn->sock, pkt->data, pkt->len, (struct sockaddr*)&dir->dest_addr, dir->dest_len);
+        if (n != (ssize_t)pkt->len) {
+            DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: sendto bytes=%zd expected=%zu error=%s", dir_idx, n,
+                pkt->len, socket_strerror(socket_get_error()));
+            dir->stats.dropped++;
+        } else {
+            dir->stats.sent++; dir->stats.sent_bytes += n;
+        }
+        dir->stats.queue_size--;
+        queue_entry_free(entry);
+        dir->tx_started = 0;
     }
+    dir->tx_finish_ns = 0;
 }
 
 /* Callback шейпера */
 static void dummynet_shaper_callback(void* user_arg) {
-    struct shaper_cb_arg* arg = (struct shaper_cb_arg*)user_arg;
-    struct dummynet* dn = arg->dn;
-    int dir_idx = arg->dir_idx;
-    u_free(arg);
-
-    dn->dirs[dir_idx].shaper_timer_fire++;
-    dummynet_process_queue(dn, dir_idx);
+    struct dummynet_dir* dir = user_arg;
+    dir->shaper_timer = NULL;
+    dir->shaper_timer_fire++;
+    dummynet_process_queue(dir->owner, dir->index);
 }
 
 /* Callback чтения из UDP сокета */
@@ -219,6 +202,11 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) {
     }
 
     dir->stats.recv++;
+    dir->stats.recv_bytes += n;
+    if (dir->mode == DUMMYNET_DROP) {
+        dir->stats.lost++; dir->stats.blocked++;
+        return;
+    }
 
     /* Random loss check */
     if (dir->loss_permille > 0) {
@@ -231,6 +219,12 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) {
         }
     }
 
+    if (dir->stats.queue_size >= dir->max_queue_pkts) {
+        dir->stats.dropped++;
+        DEBUG_TRACE(DEBUG_CATEGORY_DUMMYNET, "Dir %d: queue full (%u)", dir_idx, dir->stats.queue_size);
+        return;
+    }
+
     /* Выделяем пакет */
     struct ll_entry* entry = queue_entry_new_from_pool(dn->pkt_pool);
     if (!entry) {
@@ -266,16 +260,23 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) {
     }
     targ->dn = dn;
     targ->entry = entry;
+    targ->prev = NULL; targ->next = dn->delays;
+    if (dn->delays) dn->delays->prev = targ;
+    dn->delays = targ;
 
-    void* timer = uasync_set_timeout(dn->ua, delay_tb, targ, dummynet_delay_callback, "dummynet_delay");
-    if (!timer) {
+    targ->timer = uasync_set_timeout(dn->ua, delay_tb, targ, dummynet_delay_callback, "dummynet_delay");
+    if (!targ->timer) {
         DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Failed to set delay timer");
         dir->stats.dropped++;
+        dn->delays = targ->next;
+        if (targ->next) targ->next->prev = NULL;
         u_free(targ);
         queue_entry_free(entry);
         return;
     }
     dir->delay_timer_set++;
+    dir->stats.queue_size++;
+    if (dir->stats.queue_size > dir->stats.queue_max) dir->stats.queue_max = dir->stats.queue_size;
 }
 
 /* Callback таймера задержки */
@@ -283,6 +284,9 @@ static void dummynet_delay_callback(void* user_arg) {
     struct delay_cb_arg* arg = (struct delay_cb_arg*)user_arg;
     struct dummynet* dn = arg->dn;
     struct ll_entry* entry = arg->entry;
+    if (arg->prev) arg->prev->next = arg->next;
+    else dn->delays = arg->next;
+    if (arg->next) arg->next->prev = arg->prev;
     u_free(arg);
 
     struct dummynet_pkt* pkt = (struct dummynet_pkt*)entry->data;
@@ -290,49 +294,16 @@ static void dummynet_delay_callback(void* user_arg) {
     struct dummynet_dir* dir = &dn->dirs[dir_idx];
     dir->delay_timer_fire++;
 
-    /* Проверяем размер очереди */
-    int queue_count = queue_entry_count(dir->queue);
-    if ((uint32_t)queue_count >= dir->max_queue_pkts) {
-        dir->stats.dropped++;
-//        DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Dir %d: queue full (%d packets), dropping packet", dir_idx, queue_count);
-        queue_entry_free(entry);
-        return;
-    }
-
     /* Добавляем в очередь */
-    uint32_t id = (uint32_t)(uintptr_t)entry;
     // При провале queue_data_put сам освобождает entry, повторный free дал бы double-free.
     if (queue_data_put(dir->queue, entry) != 0) {
         dir->stats.dropped++;
+        dir->stats.queue_size--;
         DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Dir %d: queue_data_put failed", dir_idx);
         return;
     }
 
-    dir->stats.queue_size = queue_count + 1;
-    if ((uint32_t)(queue_count + 1) > dir->stats.queue_max) {
-        dir->stats.queue_max = queue_count + 1;
-    }
-
-    /* Запускаем шейпер если не активен */
-    if (!dir->shaper_pending) {
-        dir->shaper_pending = 1;
-        struct shaper_cb_arg* sarg = (struct shaper_cb_arg*)u_malloc(sizeof(struct shaper_cb_arg));
-        if (sarg) {
-            sarg->dn = dn;
-            sarg->dir_idx = dir_idx;
-            int shaper_delay_tb = 0;
-            if (dir->bandwidth_kbps > 0) {
-                uint64_t delay_calc = (uint64_t)pkt->len * 80ULL / (uint64_t)dir->bandwidth_kbps;
-                shaper_delay_tb = (int)delay_calc;
-                if (shaper_delay_tb == 0) shaper_delay_tb = 1;
-            }
-            dir->shaper_timer = uasync_set_timeout(dn->ua, shaper_delay_tb, sarg, dummynet_shaper_callback, "dummynet_shaper0");
-            if (dir->shaper_timer) dir->shaper_timer_set++;
-            else { u_free(sarg); dir->shaper_pending = 0; }
-        } else {
-            dir->shaper_pending = 0;
-        }
-    }
+    dummynet_process_queue(dn, dir_idx);
 }
 
 /* Создаёт dummynet контекст */
@@ -349,6 +320,7 @@ struct dummynet* dummynet_create(struct UASYNC* ua, const char* bind_ip, uint16_
     }
 
     dn->ua = ua;
+    dn->sock = SOCKET_INVALID;
     dn->listen_port = listen_port;
 
     /* Создаём пул для пакетов */
@@ -361,6 +333,7 @@ struct dummynet* dummynet_create(struct UASYNC* ua, const char* bind_ip, uint16_
 
     /* Создаём очереди для направлений */
     for (int i = 0; i < DUMMYNET_DIR_COUNT; i++) {
+        dn->dirs[i].owner = dn; dn->dirs[i].index = i;
         dn->dirs[i].queue = queue_new(ua, 0, 0, 0, "dummynet1");
         if (!dn->dirs[i].queue) {
             DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Failed to create queue for direction %d", i);
@@ -446,6 +419,13 @@ void dummynet_destroy(struct dummynet* dn) {
         }
     }
 
+    while (dn->delays) {
+        struct delay_cb_arg* arg = dn->delays;
+        dn->delays = arg->next;
+        uasync_cancel_timeout(dn->ua, arg->timer);
+        queue_entry_free(arg->entry); u_free(arg);
+    }
+
     if (dn->pkt_pool) {
         memory_pool_destroy(dn->pkt_pool);
     }
@@ -503,6 +483,41 @@ int dummynet_set_direction(struct dummynet* dn, int direction,
     return 0;
 }
 
+/* HOLD сохраняет пакеты; DROP освобождает также delay-пакеты, не оставляя callbacks на удалённые записи. */
+int dummynet_set_mode(struct dummynet* dn, int direction, enum dummynet_mode mode) {
+    if (!dn || direction < 0 || direction >= DUMMYNET_DIR_COUNT || mode < DUMMYNET_RUN || mode > DUMMYNET_HOLD) {
+        DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Invalid mode: direction=%d mode=%d", direction, mode);
+        return -1;
+    }
+    struct dummynet_dir* dir = &dn->dirs[direction];
+    if (dir->mode == mode) return 0;
+    DEBUG_INFO(DEBUG_CATEGORY_DUMMYNET, "Dir %d: mode %d -> %d queued=%u", direction, dir->mode, mode, dir->stats.queue_size);
+    dir->mode = mode;
+    if (dir->shaper_timer) { uasync_cancel_timeout(dn->ua, dir->shaper_timer); dir->shaper_timer = NULL; }
+    dir->tx_started = 0; dir->tx_finish_ns = 0;
+    if (mode == DUMMYNET_DROP) {
+        struct ll_entry* entry;
+        while ((entry = queue_data_get(dir->queue))) {
+            queue_entry_free(entry); dir->stats.lost++; dir->stats.blocked++; dir->stats.queue_size--;
+        }
+        struct delay_cb_arg* arg = dn->delays;
+        while (arg) {
+            struct delay_cb_arg* next = arg->next;
+            if (((struct dummynet_pkt*)arg->entry->data)->direction == direction) {
+                if (arg->prev) arg->prev->next = arg->next;
+                else dn->delays = arg->next;
+                if (arg->next) arg->next->prev = arg->prev;
+                uasync_cancel_timeout(dn->ua, arg->timer);
+                queue_entry_free(arg->entry); u_free(arg);
+                dir->stats.lost++; dir->stats.blocked++; dir->stats.queue_size--;
+            }
+            arg = next;
+        }
+    }
+    if (mode == DUMMYNET_RUN) dummynet_process_queue(dn, direction);
+    return 0;
+}
+
 /* Получает статистику направления */
 const struct dummynet_stats* dummynet_get_stats(struct dummynet* dn, int direction) {
     if (!dn || direction < 0 || direction >= DUMMYNET_DIR_COUNT) {
@@ -517,10 +532,14 @@ void dummynet_reset_stats(struct dummynet* dn, int direction) {
 
     if (direction < 0) {
         for (int i = 0; i < DUMMYNET_DIR_COUNT; i++) {
+            uint32_t queued = dn->dirs[i].stats.queue_size;
             memset(&dn->dirs[i].stats, 0, sizeof(struct dummynet_stats));
+            dn->dirs[i].stats.queue_size = dn->dirs[i].stats.queue_max = queued;
         }
     } else if (direction >= 0 && direction < DUMMYNET_DIR_COUNT) {
+        uint32_t queued = dn->dirs[direction].stats.queue_size;
         memset(&dn->dirs[direction].stats, 0, sizeof(struct dummynet_stats));
+        dn->dirs[direction].stats.queue_size = dn->dirs[direction].stats.queue_max = queued;
     }
 }
 
@@ -547,7 +566,7 @@ int dummynet_get_queue_size(struct dummynet *dn, int direction) {
     if (!dn || direction < 0 || direction >= DUMMYNET_DIR_COUNT) {
         return -1;
     }
-    return queue_entry_count(dn->dirs[direction].queue);
+    return (int)dn->dirs[direction].stats.queue_size;
 }
 
 /* Получает отладочные счетчики направления */

src/transport_layer/dummynet.h

@@ -7,7 +7,7 @@
  * listen_port и пересылает между двумя явно настроенными endpoint с эмуляцией условий канала.
  *
  * Архитектура:
- *   [UDP recv] -> [delay timer] -> [check queue/drop] -> [ll_queue] -> [shaper timer] -> [UDP send]
+ *   [UDP recv] -> [check queue/drop] -> [delay timer] -> [ll_queue] -> [shaper timer] -> [UDP send]
  *                                       ^
  *                                   loss check
  *
@@ -50,6 +50,12 @@ struct ll_entry;
 /* Максимальное значение loss_permille (1000 = 100%) */
 #define DUMMYNET_MAX_LOSS_PERMILLE 1000
 
+enum dummynet_mode {
+    DUMMYNET_RUN,            // обычная передача
+    DUMMYNET_DROP,           // потеря новых и уже задержанных пакетов
+    DUMMYNET_HOLD            // задержанные пакеты сохраняются; очередь ограничена max_queue_pkts
+};
+
 /**
  * @brief Статистика направления
  */
@@ -58,7 +64,10 @@ struct dummynet_stats {
     uint64_t sent;           // Отправлено пакетов
     uint64_t dropped;        // Дропнуто (queue full)
     uint64_t lost;           // Потеряно случайно (random loss)
-    uint32_t queue_size;     // Текущий размер очереди
+    uint64_t recv_bytes;     // Принято UDP-байт, без IP/UDP заголовков
+    uint64_t sent_bytes;     // Передано UDP-байт, включая протокольные накладные расходы ETCP
+    uint64_t blocked;        // Пакетов отброшено режимом DROP (включены в lost)
+    uint32_t queue_size;     // Все ожидающие пакеты, включая delay и HOLD
     uint32_t queue_max;      // Максимальный размер очереди (пиковый)
 };
 
@@ -106,6 +115,11 @@ int dummynet_set_direction(struct dummynet* dn, int direction,
                            uint32_t loss_permille,
                            const char* dest_ip, uint16_t dest_port);
 
+/* Меняет режим направления; 0 / -1. HOLD отменяет текущую сериализацию; RUN начинает её заново
+ * для головы очереди и выпускает накопленное с BWlim, без мгновенного сброса всей очереди.
+ * DROP удаляет также пакеты в delay. Все вызовы — в потоке ua. BWlim=0 означает unlimited. */
+int dummynet_set_mode(struct dummynet* dn, int direction, enum dummynet_mode mode);
+
 /**
  * @brief Получает статистику направления
  *