From 36c096c4e90ce86a0e473a28f46747e8dc9dcdc3 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 17:32:13 +0200 Subject: [PATCH] Make UDP dummynet configuration and resource failures recoverable --- src/transport_layer/dummynet.c | 101 ++++++++++++++++++++------------- 1 file changed, 60 insertions(+), 41 deletions(-) diff --git a/src/transport_layer/dummynet.c b/src/transport_layer/dummynet.c index 536248ca..09929a83 100644 --- a/src/transport_layer/dummynet.c +++ b/src/transport_layer/dummynet.c @@ -36,8 +36,6 @@ static void dummynet_shaper_callback(void* user_arg); struct dummynet_pkt { uint8_t data[DUMMYNET_MAX_PKT_SIZE]; size_t len; - struct sockaddr_storage src_addr; - socklen_t src_len; uint64_t recv_time_tb; uint32_t delay_ms; int direction; @@ -80,8 +78,7 @@ struct dummynet_dir { struct dummynet { struct UASYNC* ua; socket_t sock; - struct sockaddr_storage bind_addr; - socklen_t bind_len; + int socket_registered; uint16_t listen_port; struct dummynet_dir dirs[DUMMYNET_DIR_COUNT]; @@ -137,8 +134,12 @@ static void dummynet_process_queue(struct dummynet* dn, int dir_idx) { 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(); + DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: shaper timer failed bytes=%zu queued=%u wait_tb=%llu; dropping head", + dir_idx, pkt->len, dir->stats.queue_size, (unsigned long long)wait_tb); + struct ll_entry* failed = queue_data_get(dir->queue); + queue_entry_free(failed); dir->stats.dropped++; dir->stats.queue_size--; + dir->tx_started = 0; dir->tx_finish_ns = now_ns; + continue; } dir->shaper_timer_set++; return; @@ -174,7 +175,7 @@ static void dummynet_shaper_callback(void* user_arg) { /* Callback чтения из UDP сокета */ static void dummynet_read_callback(socket_t sock, void* user_arg) { struct dummynet* dn = (struct dummynet*)user_arg; - uint8_t buf[DUMMYNET_MAX_PKT_SIZE]; + uint8_t buf[65536]; /* Вся IPv4 UDP-датаграмма: слишком большой пакет отбрасывается без усечения. */ struct sockaddr_storage src_addr; socklen_t src_len = sizeof(src_addr); @@ -186,8 +187,6 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) { return; } - if (n == 0) return; - int dir_idx = dummynet_get_direction(dn, &src_addr); if (dir_idx < 0) { DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Packet from unknown endpoint %s, expected %s or %s", @@ -204,6 +203,11 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) { dir->stats.recv++; dir->stats.recv_bytes += n; + if (n > DUMMYNET_MAX_PKT_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Dir %d: oversized UDP packet bytes=%zd limit=%d source=%s", + dir_idx, n, DUMMYNET_MAX_PKT_SIZE, sockaddr_storage_to_str(&src_addr).str); + dir->stats.dropped++; return; + } if (dir->mode == DUMMYNET_DROP) { dir->stats.lost++; dir->stats.blocked++; return; @@ -214,7 +218,7 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) { uint32_t roll = (uint32_t)rand() % 1000; if (roll < dir->loss_permille) { dir->stats.lost++; - DEBUG_INFO(DEBUG_CATEGORY_DUMMYNET, "Packet lost randomly (roll=%u, threshold=%u)", + DEBUG_DEBUG(DEBUG_CATEGORY_DUMMYNET, "Packet lost randomly (roll=%u, threshold=%u)", roll, dir->loss_permille); return; } @@ -222,7 +226,8 @@ 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); + DEBUG_DEBUG(DEBUG_CATEGORY_DUMMYNET, "Dir %d: queue full size=%u limit=%u bytes=%zd", + dir_idx, dir->stats.queue_size, dir->max_queue_pkts, n); return; } @@ -237,16 +242,11 @@ static void dummynet_read_callback(socket_t sock, void* user_arg) { struct dummynet_pkt* pkt = (struct dummynet_pkt*)entry->data; pkt->len = n; memcpy(pkt->data, buf, n); - pkt->src_addr = src_addr; - pkt->src_len = src_len; pkt->recv_time_tb = get_time_tb(); pkt->direction = dir_idx; /* Вычисляем задержку */ uint32_t delay_ms = dir->delay_fixed_ms + random_range(dir->delay_random_ms); - if (delay_ms > DUMMYNET_MAX_DELAY_MS) { - delay_ms = DUMMYNET_MAX_DELAY_MS; - } int delay_tb = MS_TO_TB(delay_ms); pkt->delay_ms = delay_ms; @@ -337,7 +337,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"); + dn->dirs[i].queue = queue_new(ua, 0, 0, 0, i ? "dummynet-backward" : "dummynet-forward"); if (!dn->dirs[i].queue) { DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Failed to create queue for direction %d", i); dummynet_destroy(dn); @@ -382,7 +382,9 @@ struct dummynet* dummynet_create(struct UASYNC* ua, const char* bind_ip, uint16_ return NULL; } dn->listen_port = socket_get_bound_port(dn->sock); - if (!dn->listen_port) { dummynet_destroy(dn); return NULL; } + if (!dn->listen_port) { + DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Cannot query bound dummynet UDP port"); dummynet_destroy(dn); return NULL; + } /* Добавляем сокет в uasync */ void* sock_id = uasync_add_socket_t(ua, dn->sock, dummynet_read_callback, NULL, NULL, "dummynet", dn); @@ -391,6 +393,7 @@ struct dummynet* dummynet_create(struct UASYNC* ua, const char* bind_ip, uint16_ dummynet_destroy(dn); return NULL; } + dn->socket_registered = 1; DEBUG_INFO(DEBUG_CATEGORY_DUMMYNET, "Dummynet created on %s:%u; configure both peer endpoints before traffic", bind_ip ? bind_ip : "0.0.0.0", dn->listen_port); @@ -402,10 +405,13 @@ struct dummynet* dummynet_create(struct UASYNC* ua, const char* bind_ip, uint16_ void dummynet_destroy(struct dummynet* dn) { if (!dn) return; - DEBUG_INFO(DEBUG_CATEGORY_DUMMYNET, "Destroying dummynet"); + DEBUG_INFO(DEBUG_CATEGORY_DUMMYNET, "Destroying dummynet port=%u queued=%u/%u sent=%llu/%llu dropped=%llu/%llu", + dn->listen_port, dn->dirs[0].stats.queue_size, dn->dirs[1].stats.queue_size, + (unsigned long long)dn->dirs[0].stats.sent, (unsigned long long)dn->dirs[1].stats.sent, + (unsigned long long)dn->dirs[0].stats.dropped, (unsigned long long)dn->dirs[1].stats.dropped); if (dn->sock != SOCKET_INVALID) { - uasync_remove_socket_t(dn->ua, dn->sock); + if (dn->socket_registered) uasync_remove_socket_t(dn->ua, dn->sock); socket_close_wrapper(dn->sock); } @@ -451,6 +457,30 @@ int dummynet_set_direction(struct dummynet* dn, int direction, DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Invalid destination"); return -1; } + if ((uint64_t)delay_fixed_ms + delay_random_ms > DUMMYNET_MAX_DELAY_MS || max_queue_pkts > INT_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: invalid delay=%u+%u ms (max=%d) or queue=%u (max=%d)", + direction, delay_fixed_ms, delay_random_ms, DUMMYNET_MAX_DELAY_MS, max_queue_pkts, INT_MAX); + return -1; + } + + struct sockaddr_storage destination = {0}; + struct sockaddr_in* dest_sin = (struct sockaddr_in*)&destination; + dest_sin->sin_family = AF_INET; dest_sin->sin_port = htons(dest_port); + if (inet_pton(AF_INET, dest_ip, &dest_sin->sin_addr) != 1 || dest_sin->sin_addr.s_addr == INADDR_ANY) { + DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: invalid destination IPv4=%s port=%u", direction, dest_ip, dest_port); return -1; + } + struct dummynet_dir* dir = &dn->dirs[direction]; + struct dummynet_dir* reverse = &dn->dirs[1-direction]; + const struct sockaddr_in* opposite = (const struct sockaddr_in*)&reverse->dest_addr; + if (reverse->configured && dest_sin->sin_addr.s_addr == opposite->sin_addr.s_addr && dest_sin->sin_port == opposite->sin_port) { + DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Dir %d: endpoints must differ, destination=%s:%u", direction, dest_ip, dest_port); return -1; + } + const struct sockaddr_in* previous = (const struct sockaddr_in*)&dir->dest_addr; + if ((!dir->configured || previous->sin_addr.s_addr != dest_sin->sin_addr.s_addr || previous->sin_port != dest_sin->sin_port) && + (dir->stats.queue_size || reverse->stats.queue_size)) { + DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Dir %d: endpoint change rejected while queued=%u/%u destination=%s:%u", + direction, dir->stats.queue_size, reverse->stats.queue_size, dest_ip, dest_port); return -1; + } if (loss_permille > DUMMYNET_MAX_LOSS_PERMILLE) { DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Loss permille %u clamped to %d", @@ -458,23 +488,13 @@ int dummynet_set_direction(struct dummynet* dn, int direction, loss_permille = DUMMYNET_MAX_LOSS_PERMILLE; } - struct dummynet_dir* dir = &dn->dirs[direction]; - dir->delay_fixed_ms = delay_fixed_ms; dir->delay_random_ms = delay_random_ms; dir->bandwidth_kbps = bandwidth_kbps; dir->max_queue_pkts = max_queue_pkts; dir->loss_permille = loss_permille; - memset(&dir->dest_addr, 0, sizeof(dir->dest_addr)); - struct sockaddr_in* dest_sin = (struct sockaddr_in*)&dir->dest_addr; - dest_sin->sin_family = AF_INET; - dest_sin->sin_port = htons(dest_port); - - if (inet_pton(AF_INET, dest_ip, &dest_sin->sin_addr) != 1) { - DEBUG_ERROR(DEBUG_CATEGORY_DUMMYNET, "Invalid destination IP: %s", dest_ip); - return -1; - } + dir->dest_addr = destination; dir->dest_len = sizeof(struct sockaddr_in); dir->configured = 1; @@ -532,7 +552,9 @@ const struct dummynet_stats* dummynet_get_stats(struct dummynet* dn, int directi /* Сбрасывает статистику */ void dummynet_reset_stats(struct dummynet* dn, int direction) { - if (!dn) return; + if (!dn || direction < -1 || direction >= DUMMYNET_DIR_COUNT) { + DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Invalid statistics reset direction=%d context=%p", direction, (void*)dn); return; + } if (direction < 0) { for (int i = 0; i < DUMMYNET_DIR_COUNT; i++) { @@ -577,15 +599,12 @@ int dummynet_get_queue_size(struct dummynet *dn, int direction) { void dummynet_get_debug_counters(struct dummynet *dn, int direction, uint64_t *delay_set, uint64_t *delay_fire, uint64_t *shaper_set, uint64_t *shaper_fire) { - if (!dn || direction < 0 || direction >= DUMMYNET_DIR_COUNT) { - *delay_set = *delay_fire = *shaper_set = *shaper_fire = 0; - return; - } - struct dummynet_dir *dir = &dn->dirs[direction]; - *delay_set = dir->delay_timer_set; - *delay_fire = dir->delay_timer_fire; - *shaper_set = dir->shaper_timer_set; - *shaper_fire = dir->shaper_timer_fire; + struct dummynet_dir* dir = dn && direction >= 0 && direction < DUMMYNET_DIR_COUNT ? &dn->dirs[direction] : NULL; + if (!dir) DEBUG_WARN(DEBUG_CATEGORY_DUMMYNET, "Invalid timer counter query direction=%d context=%p", direction, (void*)dn); + if (delay_set) *delay_set = dir ? dir->delay_timer_set : 0; + if (delay_fire) *delay_fire = dir ? dir->delay_timer_fire : 0; + if (shaper_set) *shaper_set = dir ? dir->shaper_timer_set : 0; + if (shaper_fire) *shaper_fire = dir ? dir->shaper_timer_fire : 0; } // === Inline filter: send_hook для etcp_udp_send ===