From f5f6ff50af3ca0bd95da5740c9dff59dc01bb880 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 23:18:11 +0200 Subject: [PATCH] Make dummynet bandwidth timing cumulative and add bounded DROP/HOLD modes --- src/transport_layer/dummynet.c | 219 ++++++++++++++++++--------------- src/transport_layer/dummynet.h | 18 ++- 2 files changed, 135 insertions(+), 102 deletions(-) diff --git a/src/transport_layer/dummynet.c b/src/transport_layer/dummynet.c index 86b13c26..233a3ce4 100644 --- a/src/transport_layer/dummynet.c +++ b/src/transport_layer/dummynet.c @@ -16,6 +16,7 @@ #include #include #include +#include #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; } /* Получает отладочные счетчики направления */ diff --git a/src/transport_layer/dummynet.h b/src/transport_layer/dummynet.h index 6b40dc96..8155b2c4 100644 --- a/src/transport_layer/dummynet.h +++ b/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 Получает статистику направления *