jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add bounded NAT backpressure and forward TUN packets without copying

Browse Source
This commit is contained in:
evgeny committed 2026-10-21 22:30:01 +02:00
1 parent 9d6d6454b0
commit d0c83cfff1
2 files changed
+278 -173
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/nat_transport.c
+264 -173
View File
@@ -1,223 +1,314 @@
#include "nat_transport.h"
#include "eim_nat.h"
#include "etcp.h"
#include "utun_instance.h"
#include "config_parser.h"
#include "tun_if.h"
#include "etcp_api.h"
#include "etcp_router.h"
#include "../routing_layer/topo_node.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/ll_queue.h"
#include "../lib/u_async.h"
#include <string.h>
#include <errno.h>
#define NAT_SVC_HDR_SIZE 1 // send: svc_id(1) + ip_data (src/dst добавляет роутер)
#define NAT_MAINTENANCE_TB 1000
#define NAT_WRITE_RETRY_TB 100
static ip_str_t ip_host_to_str(uint32_t ip_host) {
struct in_addr a; a.s_addr = htonl(ip_host);
return ip_to_str(&a, AF_INET);
// Один владелец FIFO и ожидания на каждого адресата. node_id начинается с ll.data для хеш-индекса.
struct nat_peer {
struct ll_entry ll;
uint64_t node_id;
struct UTUN_INSTANCE* inst;
struct ll_queue* packets;
struct ll_entry* pending;
struct queue_waiter_handle waiter;
uint64_t last_activity;
};
static void nat_peer_send_ready(struct ll_queue* q, void* arg);
static void nat_tun_retry(void* arg);
// Освободить пакет с сохранением его allocator/pool metadata.
static void nat_packet_free(struct ll_entry* packet) {
if (packet) { queue_dgram_free(packet); queue_entry_free(packet); }
}
// ==================== Callbacks ====================
// CLIENT: NAT TUN output → encapsulate in ETCP_RT_ID_NAT → send to provider via etcp_router
static void nat_transport_client_tun_out_cb(struct ll_queue* q, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
if (!inst) { queue_resume_callback(q); return; }
// Перегрузка одного получателя не останавливает остальных; диагностировать суммарные потери не чаще раза в секунду.
static void nat_drop(struct UTUN_INSTANCE* inst, struct ll_entry* packet, uint64_t destination, const char* reason) {
struct nat_transport_ctx* tr = &inst->nat_tr;
struct ll_entry* pkt = queue_data_get(q);
if (!pkt) { queue_resume_callback(q); return; }
if (!pkt->dgram || pkt->len < 2) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
size_t ip_len = pkt->len - 1;
size_t total_len = NAT_SVC_HDR_SIZE + ip_len;
uint8_t* new_dgram = u_malloc(total_len);
if (!new_dgram) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
new_dgram[0] = ETCP_RT_ID_NAT;
memcpy(new_dgram + 1, pkt->dgram + 1, ip_len);
struct ll_entry* new_entry = queue_entry_new(0);
if (!new_entry) { u_free(new_dgram); queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
new_entry->dgram = new_dgram;
new_entry->len = total_len;
queue_dgram_free(pkt); queue_entry_free(pkt);
queue_resume_callback(q);
// Роутер принимает владение new_entry и dgram при любом результате, включая ошибку.
int ret = etcp_route_send(inst, TOPO_GROUP_UTUN, tr->nat_via_node_id, new_entry, 0, 0);
if (ret != 0) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT client: etcp_route_send to provider %016llx failed: error=%d ip_bytes=%zu",
(unsigned long long)tr->nat_via_node_id, ret, ip_len);
tr->dropped++;
uint64_t now = get_time_tb();
if (!tr->last_drop_log || now - tr->last_drop_log >= 10000) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT drop: reason=%s destination=%016llx total=%llu pending=%zu",
reason, (unsigned long long)destination, (unsigned long long)tr->dropped, tr->pending_count);
tr->last_drop_log = now;
}
nat_packet_free(packet);
}
// PROVIDER: NAT TUN output (internet response) → ingress NAT → send back via etcp_router
static void nat_transport_provider_tun_out_cb(struct ll_queue* q, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
if (!inst) { queue_resume_callback(q); return; }
struct eim_nat_ctx* ctx = &inst->nat;
struct ll_entry* pkt = queue_data_get(q);
if (!pkt) { queue_resume_callback(q); return; }
if (!pkt->dgram || pkt->len < 2) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
struct eim_nat_entry* entry = NULL;
int ret = eim_nat_ingress(ctx, pkt->dgram + 1, pkt->len - 1, &entry);
if (ret != 0 || !entry || entry->src_node_id == 0) {
if (ret == 0) DEBUG_WARN(DEBUG_CATEGORY_NAT, "Ingress NAT: no matching entry, dropping");
queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q);
// Передать удержанный пакет, когда у роутера есть место; waiter не хранит указатель на NAT mapping.
static void nat_peer_send_ready(struct ll_queue* q, void* arg) {
(void)q;
struct nat_peer* peer = arg;
struct UTUN_INSTANCE* inst = peer->inst;
struct nat_transport_ctx* tr = &inst->nat_tr;
if (!peer->pending) return;
struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT);
if (!route || route->closed) return; // maintenance повторит после завершения close/reset.
if (!etcp_router_send_q_has_room(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT)) {
if (!peer->waiter.internal && !peer->waiter.call_soon_id) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT waiting for router: destination=%016llx pending=%zu",
(unsigned long long)peer->node_id, tr->pending_count);
etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT,
&peer->waiter, nat_peer_send_ready, peer);
}
return;
}
struct ll_entry* packet = peer->pending;
size_t ip_len = packet->len - 1;
peer->pending = NULL;
tr->pending_count--;
peer->last_activity = get_time_tb();
// Роутер синхронно копирует payload и освобождает исходный TUN entry/dgram при любом результате.
int result = etcp_route_send(inst, TOPO_GROUP_UTUN, peer->node_id, packet, 0, 0);
if (result) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT router send failed: destination=%016llx error=%d ip_bytes=%zu",
(unsigned long long)peer->node_id, result, ip_len);
}
queue_resume_callback(peer->packets);
if (tr->blocked_packet && tr->pending_count < NAT_PENDING_TOTAL) queue_resume_callback(tr->nat_tun->output_queue);
}
size_t ip_len = pkt->len - 1;
size_t total_len = NAT_SVC_HDR_SIZE + ip_len;
uint8_t* new_dgram = u_malloc(total_len);
if (!new_dgram) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
// Забрать ровно один пакет из FIFO; до готовности роутера callback остаётся приостановленным.
static void nat_peer_take(struct ll_queue* q, void* arg) {
struct nat_peer* peer = arg;
peer->pending = queue_data_get(q);
if (!peer->pending) { queue_resume_callback(q); return; }
nat_peer_send_ready(NULL, peer);
}
new_dgram[0] = ETCP_RT_ID_NAT;
memcpy(new_dgram + 1, pkt->dgram + 1, ip_len);
// Найти/создать FIFO адресата. Выделения требуются только при первом обращении к этому узлу.
static struct nat_peer* nat_peer_get(struct UTUN_INSTANCE* inst, uint64_t node_id) {
struct nat_transport_ctx* tr = &inst->nat_tr;
struct nat_peer* peer = (struct nat_peer*)queue_find_data_by_index(tr->peers, &node_id);
if (peer) return peer;
peer = (struct nat_peer*)queue_entry_new(sizeof(*peer) - sizeof(peer->ll));
if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer allocation failed node=%016llx", (unsigned long long)node_id); return NULL; }
peer->node_id = node_id; peer->inst = inst; peer->last_activity = get_time_tb();
peer->packets = queue_new(inst->ua, 0, 0, 0, "NAT peer packets");
if (!peer->packets) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer queue allocation failed node=%016llx", (unsigned long long)node_id);
queue_entry_free(&peer->ll);
return NULL;
}
queue_set_callback(peer->packets, nat_peer_take, peer);
if (queue_data_put_with_index(tr->peers, &peer->ll) != 0) {
// put принимает/освобождает peer; отдельная очередь принадлежит вызывающему до успеха.
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer index insertion failed node=%016llx", (unsigned long long)node_id);
return NULL;
}
return peer;
}
struct ll_entry* new_entry = queue_entry_new(0);
if (!new_entry) { u_free(new_dgram); queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
new_entry->dgram = new_dgram;
new_entry->len = total_len;
// 1 — пакет принят/отброшен, исходный TUN callback можно возобновить; 0 — пакет возвращён в TUN при общем лимите.
static int nat_forward(struct UTUN_INSTANCE* inst, struct ll_entry* packet, uint64_t destination) {
struct nat_transport_ctx* tr = &inst->nat_tr;
packet->dgram[0] = ETCP_RT_ID_NAT;
if (tr->pending_count >= NAT_PENDING_TOTAL) {
tr->blocked_packet = packet; tr->blocked_destination = destination;
if (queue_data_put_first(tr->nat_tun->output_queue, packet) == 0) return 0;
tr->blocked_packet = NULL;
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT failed to preserve packet at global backpressure limit");
return 1;
}
struct nat_peer* peer = nat_peer_get(inst, destination);
if (!peer) { nat_drop(inst, packet, destination, "peer allocation"); return 1; }
if (queue_entry_count(peer->packets) + (peer->pending != NULL) >= NAT_PENDING_PER_PEER) {
nat_drop(inst, packet, destination, "per-peer pending limit");
return 1;
}
peer->last_activity = get_time_tb();
tr->pending_count++;
if (queue_data_put(peer->packets, packet) != 0) {
tr->pending_count--;
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT pending queue rejected packet destination=%016llx", (unsigned long long)destination);
}
return 1;
}
queue_dgram_free(pkt); queue_entry_free(pkt);
queue_resume_callback(q);
// Обе роли получают [prefix][IP]; преобразование ответа выполняется один раз до возможного ожидания.
static void nat_tun_output(struct ll_queue* q, void* arg) {
struct UTUN_INSTANCE* inst = arg;
struct nat_transport_ctx* tr = &inst->nat_tr;
struct ll_entry* packet = queue_data_get(q);
if (!packet) { queue_resume_callback(q); return; }
if (!packet->dgram || packet->len < 2) {
nat_drop(inst, packet, 0, "invalid TUN packet");
queue_resume_callback(q);
return;
}
uint64_t destination = tr->nat_via_node_id;
if (packet == tr->blocked_packet) {
destination = tr->blocked_destination; tr->blocked_packet = NULL;
} else if (!destination) {
struct eim_nat_entry* mapping = NULL;
int result = eim_nat_ingress(&inst->nat, packet->dgram + 1, packet->len - 1, &mapping);
if (result != EIM_NAT_OK || !mapping || !mapping->src_node_id) {
nat_drop(inst, packet, 0, "no ingress mapping or invalid packet");
queue_resume_callback(q);
return;
}
destination = mapping->src_node_id;
}
if (nat_forward(inst, packet, destination)) queue_resume_callback(q);
}
// После передачи роутеру new_entry и dgram больше не принадлежат NAT, даже при ошибке.
int send_ret = etcp_route_send(inst, TOPO_GROUP_UTUN, entry->src_node_id, new_entry, 0, 0);
if (send_ret != 0) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT provider: etcp_route_send back to node %016llx failed: error=%d ip_bytes=%zu",
(unsigned long long)entry->src_node_id, send_ret, ip_len);
// Пока один пакет ожидает TUN, последующие остаются в router recv_q без ACK.
static int nat_receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* data, size_t len) {
(void)group; (void)peer; (void)data; (void)len;
return inst->nat_tr.initialized && !inst->nat_tr.tun_pending;
}
// Повторить запись того же уже преобразованного пакета; mapping мог истечь, повторной трансляции нет.
static void nat_tun_flush(struct UTUN_INSTANCE* inst) {
struct nat_transport_ctx* tr = &inst->nat_tr;
struct ll_entry* entry = tr->tun_pending;
if (!entry) return;
size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (tun_write(tr->nat_tun, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, len) < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
if (!tr->tun_retry_timer) tr->tun_retry_timer = uasync_set_timeout(inst->ua, NAT_WRITE_RETRY_TB, inst, nat_tun_retry, "nat_tun_retry");
if (!tr->tun_retry_timer) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT TUN retry allocation failed; maintenance will retry");
return;
}
nat_packet_free(entry); tr->tun_pending = NULL;
if (inst->router_conns) {
for (struct ll_entry* e = inst->router_conns->head; e; e = e->next) {
struct ETCP_ROUTER_CONN* route = (struct ETCP_ROUTER_CONN*)e;
if (route->svc_id == ETCP_RT_ID_NAT && route->group_id == TOPO_GROUP_UTUN)
etcp_router_resume_receive(inst, route->group_id, route->remote_node_id, ETCP_RT_ID_NAT);
}
}
}
// ETCP_RT_ID_NAT receive via etcp_router: CLIENT gets response, PROVIDER gets request
static void nat_transport_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!conn || !entry || !entry->dgram || entry->len < ROUTER_SVC_HDR_SIZE) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
static void nat_tun_retry(void* arg) {
struct UTUN_INSTANCE* inst = arg;
inst->nat_tr.tun_retry_timer = NULL;
nat_tun_flush(inst);
}
// Router передаёт владение entry; CLOSE/reset не содержит IP и не меняет таблицу.
static void nat_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!conn || !conn->instance || !entry || !entry->dgram || entry->len < ROUTER_SVC_HDR_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT malformed router delivery");
nat_packet_free(entry);
return;
}
struct UTUN_INSTANCE* inst = conn->instance;
if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return; }
struct nat_transport_ctx* tr = &inst->nat_tr;
struct eim_nat_ctx* ctx = &inst->nat;
if (!tr->initialized || !ctx->initialized) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t src_node_id;
memcpy(&src_node_id, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
uint64_t group_id;
memcpy(&group_id, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
if (group_id != TOPO_GROUP_UTUN || (tr->nat_via_node_id && src_node_id != tr->nat_via_node_id)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT receive rejected: group=%016llx source=%016llx configured_provider=%016llx",
(unsigned long long)group_id, (unsigned long long)src_node_id, (unsigned long long)tr->nat_via_node_id);
queue_dgram_free(entry); queue_entry_free(entry);
if (!tr->initialized || entry->len == ROUTER_SVC_HDR_SIZE) { nat_packet_free(entry); return; }
uint64_t source, group;
memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
if (group != TOPO_GROUP_UTUN || (tr->nat_via_node_id && source != tr->nat_via_node_id)) {
nat_drop(inst, entry, source, "unexpected source/group");
return;
}
uint8_t* ip_data = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
size_t ip_len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (!ip_len) { queue_dgram_free(entry); queue_entry_free(entry); return; } // Router CLOSE/reset, не IP-пакет.
if (tr->nat_via_node_id != 0) {
if (tr->nat_tun) {
if (tun_write(tr->nat_tun, ip_data, ip_len) == (ssize_t)ip_len)
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT client: wrote %zu bytes from provider to TUN", ip_len);
else DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT client: TUN delivery failed bytes=%zu", ip_len);
}
} else {
int ret = eim_nat_egress(ctx, ip_data, ip_len, src_node_id, conn);
if (ret < 0) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "Egress NAT failed");
} else if (ret == EIM_NAT_SKIP) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT provider: unsupported IP packet dropped bytes=%zu", ip_len);
} else if (ret == 0 && tr->nat_tun) {
if (tun_write(tr->nat_tun, ip_data, ip_len) == (ssize_t)ip_len)
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT provider: wrote %zu bytes to internet TUN", ip_len);
else DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT provider: TUN delivery failed bytes=%zu", ip_len);
}
size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (!tr->nat_via_node_id) {
int result = eim_nat_egress(&inst->nat, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, len, source, conn);
if (result != EIM_NAT_OK) { nat_drop(inst, entry, source, "egress invalid or unsupported packet"); return; }
}
queue_dgram_free(entry); queue_entry_free(entry);
if (tr->tun_pending) { nat_drop(inst, entry, source, "concurrent TUN delivery"); return; } // Возможен только loopback.
tr->tun_pending = entry;
nat_tun_flush(inst);
}
// ==================== Init / Destroy ====================
// Освободить FIFO и waiter до уничтожения TUN-пула; вызывается только вне callbacks этого peer.
static void nat_peer_free(struct nat_peer* peer) {
struct UTUN_INSTANCE* inst = peer->inst;
etcp_router_cancel_send_ready(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT, &peer->waiter);
if (peer->pending) { nat_packet_free(peer->pending); inst->nat_tr.pending_count--; }
struct ll_entry* packet;
while ((packet = queue_data_get(peer->packets))) { nat_packet_free(packet); inst->nat_tr.pending_count--; }
queue_free(peer->packets);
queue_entry_free(&peer->ll);
}
int nat_transport_init(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->config) return -1;
// Периодическая работа не входит в обработку пакета. Отсутствующие после reset waiters восстанавливаются.
static void nat_maintenance(void* arg) {
struct UTUN_INSTANCE* inst = arg;
struct nat_transport_ctx* tr = &inst->nat_tr;
struct global_config* g = &inst->config->global;
memset(tr, 0, sizeof(*tr));
if (!g->nat_enabled) return 0;
tr->self_node_id = inst->node_id;
tr->nat_via_node_id = g->nat_via_node_id;
if (tr->nat_via_node_id == 0) {
if (eim_nat_init_ctx(&inst->nat, g) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to init NAT engine");
return -1;
tr->maintenance_timer = NULL;
uint64_t now = get_time_tb();
if (!tr->nat_via_node_id) eim_nat_expire(&inst->nat, now, 128);
for (struct ll_entry* e = tr->peers->head; e;) {
struct ll_entry* next = e->next;
struct nat_peer* peer = (struct nat_peer*)e;
if (peer->pending && !peer->waiter.internal && !peer->waiter.call_soon_id) nat_peer_send_ready(NULL, peer);
if (!peer->pending && !queue_entry_count(peer->packets) && now >= peer->last_activity && now - peer->last_activity >= 600000) {
queue_remove_data(tr->peers, e);
nat_peer_free(peer);
}
} else {
inst->nat.initialized = 1;
e = next;
}
const char* tun_name = g->nat_tun_ifname[0] ? g->nat_tun_ifname : "tun_nat";
char ip_str[64] = "";
if (g->nat_tun_ip.family == AF_INET) {
snprintf(ip_str, sizeof(ip_str), "%s", ip_to_str(&g->nat_tun_ip.addr.v4, AF_INET).str);
} else if (g->tun_ip.family == AF_INET) {
snprintf(ip_str, sizeof(ip_str), "%s", ip_to_str(&g->tun_ip.addr.v4, AF_INET).str);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "No NAT TUN IP");
eim_nat_destroy_ctx(&inst->nat);
return -1;
}
tr->nat_tun = tun_init_nat(inst->ua, tun_name, ip_str, TUN_MTU_DEFAULT, g->tun_test_mode);
if (!tr->nat_tun) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to create NAT TUN %s ip=%s", tun_name, ip_str);
eim_nat_destroy_ctx(&inst->nat);
return -1;
}
if (etcp_router_bind(inst, ETCP_RT_ID_NAT, nat_transport_etcp_recv_cb) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to bind ETCP_RT_ID_NAT via etcp_router");
tun_close(tr->nat_tun); tr->nat_tun = NULL;
eim_nat_destroy_ctx(&inst->nat);
return -1;
}
struct eim_nat_ctx* ctx = &inst->nat;
if (tr->nat_via_node_id != 0) {
if (tr->nat_tun->output_queue) queue_set_callback(tr->nat_tun->output_queue, nat_transport_client_tun_out_cb, inst);
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT client via node %016llx, TUN=%s gw=%s ports=%u-%u",
(unsigned long long)tr->nat_via_node_id, tun_name,
ip_host_to_str(ctx->gateway_ip).str, ctx->port_start, ctx->port_end);
} else {
if (tr->nat_tun->output_queue) queue_set_callback(tr->nat_tun->output_queue, nat_transport_provider_tun_out_cb, inst);
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT provider TUN=%s gw=%s ports=%u-%u",
tun_name, ip_host_to_str(ctx->gateway_ip).str, ctx->port_start, ctx->port_end);
}
tr->initialized = 1;
return 0;
nat_tun_flush(inst);
tr->maintenance_timer = uasync_set_timeout(inst->ua, NAT_MAINTENANCE_TB, inst, nat_maintenance, "nat_maintenance");
if (!tr->maintenance_timer) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT maintenance scheduling failed");
}
// Все ресурсы создаются в потоке ядра; при любой ошибке инициализация откатывает только свои ресурсы.
int nat_transport_init(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->config) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT init requires instance/config"); return -1; }
struct nat_transport_ctx* tr = &inst->nat_tr;
struct global_config* config = &inst->config->global;
memset(tr, 0, sizeof(*tr));
if (!config->nat_enabled) return 0;
tr->self_node_id = inst->node_id; tr->nat_via_node_id = config->nat_via_node_id;
if (!tr->nat_via_node_id) {
if (eim_nat_init_ctx(&inst->nat, config)) return -1;
} else inst->nat.initialized = 1;
const struct IP* ip = config->nat_tun_ip.family == AF_INET ? &config->nat_tun_ip : &config->tun_ip;
if (ip->family != AF_INET) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT requires IPv4 TUN address"); goto failed; }
const char* name = config->nat_tun_ifname[0] ? config->nat_tun_ifname : "tun_nat";
tr->nat_tun = tun_init_nat(inst->ua, name, ip_to_str(&ip->addr.v4, AF_INET).str, TUN_MTU_DEFAULT, config->tun_test_mode);
if (!tr->nat_tun) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT TUN creation failed name=%s", name); goto failed; }
tr->peers = queue_new(inst->ua, 256, 0, sizeof(uint64_t), "NAT peers");
if (!tr->peers) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer index allocation failed"); goto failed; }
if (etcp_router_bind(inst, ETCP_RT_ID_NAT, nat_receive)) goto failed;
tr->initialized = 1;
etcp_router_set_receive_ready(inst, ETCP_RT_ID_NAT, nat_receive_ready);
queue_set_callback(tr->nat_tun->output_queue, nat_tun_output, inst);
tr->maintenance_timer = uasync_set_timeout(inst->ua, NAT_MAINTENANCE_TB, inst, nat_maintenance, "nat_maintenance");
if (!tr->maintenance_timer) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT maintenance timer allocation failed");
nat_transport_destroy(inst);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT %s initialized TUN=%s peer_limit=%u total_limit=%u",
tr->nat_via_node_id ? "client" : "provider", name, NAT_PENDING_PER_PEER, NAT_PENDING_TOTAL);
return 0;
failed:
if (tr->peers) queue_free(tr->peers);
if (tr->nat_tun) tun_close(tr->nat_tun);
eim_nat_destroy_ctx(&inst->nat);
memset(tr, 0, sizeof(*tr));
return -1;
}
// Отменить производителей/таймеры и вернуть удержанные TUN-пакеты до уничтожения их пула.
void nat_transport_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->nat_tr.initialized) return;
struct nat_transport_ctx* tr = &inst->nat_tr;
if (tr->maintenance_timer) uasync_cancel_timeout(inst->ua, tr->maintenance_timer);
if (tr->tun_retry_timer) uasync_cancel_timeout(inst->ua, tr->tun_retry_timer);
etcp_router_unbind(inst, ETCP_RT_ID_NAT);
if (tr->nat_tun) { tun_close(tr->nat_tun); tr->nat_tun = NULL; }
size_t cancelled = tr->pending_count;
struct ll_entry* peer;
while ((peer = queue_data_get(tr->peers))) nat_peer_free((struct nat_peer*)peer);
queue_free(tr->peers);
nat_packet_free(tr->tun_pending);
tun_close(tr->nat_tun);
eim_nat_destroy_ctx(&inst->nat);
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT destroyed: cancelled=%zu dropped=%llu", cancelled, (unsigned long long)tr->dropped);
memset(tr, 0, sizeof(*tr));
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT transport destroyed");
}
src/nat_transport.h
+14
View File
@@ -16,13 +16,27 @@ extern "C" {
#include <stdint.h>
#include <stddef.h>
#define NAT_PENDING_PER_PEER 64 // Дополнительный буфер одного получателя; переполнение локально отбрасывается с диагностикой.
#define NAT_PENDING_TOTAL 1024 // Общий предел: остановить чтение TUN до освобождения места.
struct tun_if;
struct ll_entry;
struct ll_queue;
struct nat_transport_ctx {
uint64_t self_node_id; // Копия локального ID, установленная при init.
uint64_t nat_via_node_id; // 0 — провайдер; иначе ID единственного NAT-провайдера клиента.
struct tun_if* nat_tun; // Отдельный принадлежащий модулю TUN, MTU=TUN_MTU_DEFAULT.
struct ll_queue* peers; // Внутренний хеш получателей с раздельными FIFO и router waiters.
struct ll_entry* blocked_packet; // При общем backpressure остаётся в output_queue TUN.
uint64_t blocked_destination; // Зафиксирован до ожидания/истечения mapping.
size_t pending_count; // Пакеты в FIFO получателей и их активных ожиданиях, <= NAT_PENDING_TOTAL.
struct ll_entry* tun_pending; // Единственный принятый IP-пакет, ожидающий доступности TUN для записи.
void* maintenance_timer; // Порционная очистка mapping, восстановление ожиданий и уборка получателей.
void* tun_retry_timer; // Повтор записи TUN при EAGAIN; отменяется до tun_close.
uint64_t dropped, last_drop_log; // Счётчик отбрасываний и ограничение частоты диагностики перегрузки.
int initialized; // TUN, binding и callback созданы; 0 при отключённом NAT/после destroy.
};