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

Unify UDP and STCP packet link API with bounded backpressure

Browse Source
This commit is contained in:
evgeny committed 2026-10-29 12:53:57 +01:00
1 parent 3203947698
commit 43b14ece78
24 files changed
+881 -61
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/packet_link.txt
+77
View File
@@ -0,0 +1,77 @@
Единый пакетный интерфейс транспорта
==================================
Публичный контракт: lib/packet_link.h. Интерфейс реализации: lib/packet_link_backend.h.
Привязка ETCP к UDP/STCP: src/transport_layer/etcp_link_io.h/c.
Все операции и callbacks выполняются в одном uasync-потоке.
Владение и события
-----------------
create копирует ops, callbacks и начальное состояние; arg остаётся заимствованным.
При успехе оболочка владеет impl; при отказе impl освобождает вызывающий.
Первое уведомление состояния отложено; RX первоначально выключен.
После установки своего владельца вызвать packet_link_set_rx_enabled(link, 1).
send принимает целый ll_entry, не состоящий в очереди. OK передаёт владение entry
и dgram модулю. Любой отказ сохраняет владение вызывающему. Частичного принятия нет.
Успех означает принятие ответственности, а не доставку удалённому приложению.
Ошибка после принятия сообщается состоянием TERMINATED. Подтверждения ETCP остаются.
tx_ready означает свободное место в данном модуле для пакета до max_packet.
Адаптер со своей очередью считает только её заполнение. Если своей очереди нет,
он использует публичную готовность следующего модуля. Чужие внутренние буферы
не инспектируются и не суммируются. Проверка готовности не заменяет результат send.
Из send верхние callbacks не вызываются. Состояние меняется немедленно,
state_changed доставляется через call_soon и объединяет накопленные изменения.
Обработчик читает актуальный get_state; он может отправлять и закрывать линк.
backend_receive передаёт владение получателю только при OK. При отключённом RX,
CONNECTING или активном send возвращает WOULD_BLOCK, не забирая пакет.
Оболочка очередей данных не создаёт. Каждый backend сам удерживает свой лимит RX.
close сразу отменяет уведомления и вызывает stop реализации. destroy выполняется
после выхода активных callbacks, через call_soon. Это защищает также стек backend,
из которого вызван backend_receive. После close handle недействителен; новых
событий от реализации быть не должно. До уничтожения uasync освободить все линки
и выполнить его штатный drain. TERMINATED необратим; OPEN может перейти в
CONNECTING у адаптера пути, который сохраняет владельца при native-реконнекте.
ETCP
----
ETCP_LINK владеет etcp_link_io. Общий packet_link из get заимствован: отдельно
не закрывать, его освобождает etcp_link_close до native-ресурсов.
UDP передаёт существующие зашифрованные датаграммы, STCP — существующие сообщения
в его потоковое шифрование/framing. Wire-форматы, INIT, session handshake,
SOCKS, REALITY и правила реконнекта сохранены.
Неизвестный UDP endpoint проходит существующий bootstrap до создания ETCP_LINK;
последующие пакеты привязанного пути проходят общий receive callback.
UDP сначала пробует sendto. EAGAIN сохраняет одну датаграмму в очереди данного
адаптера и включает writable общего сокета. При SOCKS, ещё не готовом к передаче,
адаптер ждёт ready_cb relay, не крутит EPOLLOUT. writable обслуживает ожидающие
датаграммы путей сокета. Ошибка досылки закрывает только соответствующий путь.
STCP-адаптер разрешает один ожидающий пакет в своей TX-очереди. Модуль STCP
извлекает следующий только после освобождения собственной досылки. Освобождение
очереди адаптера уведомляет ETCP через packet_link. Входная очередь STCP сохраняет
свои прежние пороги и приостанавливает чтение TCP; отключение RX останавливает
её потребление. У ETCP-адаптера RX ограничен 64 пакетами, обычный приём синхронный.
Выбор DATA проверяет готовность packet_link вместе с ETCP-окном и pacing.
ACK и измерительные серии также проверяют транспортную готовность.
ACK, служебные признаки и номер пакета измерительной серии фиксируются только
после принятого send. При WOULD_BLOCK остаются исходные очереди и признаки;
новый проход отправителя вызывает событие готовности, не периодический polling.
Проверки
--------
test_packet_link: владение TX/RX, oversized, blocked/resume, объединённые события,
RX pause, отправка из state callback, закрытие из state/receive, отмена событий,
освобождение всех учитываемых выделений.
test_etcp_packet_link: настоящий UDP-адаптер с управляемым send hook, EAGAIN,
два пути общего сокета, выбор доступного пути, запрет обхода занятости burst,
ACK commit/rollback, ошибка досылки и независимое закрытие.
Существующие test_transport/test_etcp_transport, test_etcp_windows,
test_etcp_session/test_etcp_multipath и сетевые интеграционные тесты сохраняют
прежние ожидания. В синтетических fixtures освобождается добавленный I/O handle.
Диагностика: sys=debug для состояния packet_link, etcp=debug для привязки/ETCP.
lib/Makefile.am
+3
View File
@@ -5,6 +5,9 @@ libuasync_a_SOURCES = \
u_async.h \
ll_queue.h \
ll_queue.c \
packet_link.c \
packet_link.h \
packet_link_backend.h \
debug_config.c \
debug_config.h \
timeout_heap.c \
lib/packet_link.c
+129
View File
@@ -0,0 +1,129 @@
#include "packet_link_backend.h"
#include "u_async.h"
#include "ll_queue.h"
#include "mem.h"
#include "debug_config.h"
#include <stdlib.h>
struct packet_link {
struct UASYNC* ua;
struct packet_link_ops ops;
struct packet_link_callbacks callbacks;
struct packet_link_state state;
void *impl, *arg, *notification;
unsigned active;
int closing, rx_enabled, sending;
};
static void destroy_deferred(void* arg) {
struct packet_link* link = arg;
link->ops.destroy(link->impl);
u_free(link);
}
static void release(struct packet_link* link) {
if (--link->active || !link->closing) return;
/* Backend мог вызвать receive из своего I/O callback: его стек тоже должен завершиться. */
if (!uasync_call_soon(link->ua, link, destroy_deferred)) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: cannot defer destruction"); abort();
}
}
static void notify(void* arg) {
struct packet_link* link = arg;
link->notification = NULL;
link->active++;
if (link->callbacks.state_changed) link->callbacks.state_changed(link, link->arg);
release(link);
}
static int schedule(struct packet_link* link) {
if (link->notification || link->closing) return 0;
link->notification = uasync_call_soon(link->ua, link, notify);
if (link->notification) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: cannot schedule state notification");
return -1;
}
static int valid_state(const struct packet_link_state* s) {
return s && s->phase >= PACKET_LINK_CONNECTING && s->phase <= PACKET_LINK_TERMINATED &&
s->max_packet && s->max_packet <= UINT16_MAX && (!s->tx_ready || s->phase == PACKET_LINK_OPEN);
}
struct packet_link* packet_link_create(struct UASYNC* ua, const struct packet_link_ops* ops, void* impl,
const struct packet_link_state* initial, const struct packet_link_callbacks* callbacks, void* arg) {
if (!ua || !ops || !ops->send || !ops->stop || !ops->destroy || !ops->set_rx_enabled || !valid_state(initial) ||
!callbacks || !callbacks->received) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: invalid create arguments"); return NULL;
}
struct packet_link* link = u_calloc(1, sizeof(*link));
if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: allocation failed"); return NULL; }
link->ua = ua; link->ops = *ops; link->impl = impl; link->state = *initial; link->callbacks = *callbacks; link->arg = arg;
if (schedule(link)) { u_free(link); return NULL; }
DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "packet link=%p created phase=%d max=%zu ready=%d",
(void*)link, initial->phase, initial->max_packet, initial->tx_ready);
return link;
}
struct packet_link_state packet_link_get_state(const struct packet_link* link) {
if (!link || link->closing) return (struct packet_link_state){PACKET_LINK_TERMINATED, 0, 0, PACKET_LINK_CLOSED};
return link->state;
}
void packet_link_backend_set_state(struct packet_link* link, const struct packet_link_state* state) {
if (!link || link->closing) return;
if (!valid_state(state) || (link->state.phase == PACKET_LINK_TERMINATED && state->phase != PACKET_LINK_TERMINATED)) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: invalid state transition"); return;
}
struct packet_link_state old = link->state;
if (old.phase == state->phase && old.max_packet == state->max_packet && old.tx_ready == state->tx_ready && old.error == state->error) return;
link->state = *state;
DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "packet link=%p phase=%d ready=%d max=%zu error=%d",
(void*)link, state->phase, state->tx_ready, state->max_packet, state->error);
/* Потерять обязательное событие нельзя; синхронный callback нарушил бы контракт send. */
if (schedule(link)) abort();
}
int packet_link_send(struct packet_link* link, struct ll_entry* packet) {
if (!link || link->closing || link->state.phase == PACKET_LINK_TERMINATED) return PACKET_LINK_CLOSED;
if (!packet || packet->owner || packet->next || packet->prev || !packet->dgram || !packet->len) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: invalid TX packet"); return PACKET_LINK_INVALID;
}
if (packet->len > link->state.max_packet) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: oversized TX size=%u max=%zu", packet->len, link->state.max_packet);
return PACKET_LINK_TOO_LARGE;
}
if (!link->state.tx_ready || link->sending) return PACKET_LINK_WOULD_BLOCK;
link->active++; link->sending = 1;
int result = link->ops.send(link->impl, packet);
link->sending = 0;
release(link);
return result;
}
void packet_link_set_rx_enabled(struct packet_link* link, int enabled) {
if (!link || link->closing || link->rx_enabled == !!enabled) return;
link->active++; link->rx_enabled = !!enabled;
link->ops.set_rx_enabled(link->impl, !!enabled);
release(link);
}
int packet_link_backend_receive(struct packet_link* link, struct ll_entry* packet) {
if (!link || link->closing || link->state.phase == PACKET_LINK_TERMINATED) return PACKET_LINK_CLOSED;
if (!packet || packet->owner || !packet->dgram || !packet->len || packet->len > link->state.max_packet) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "packet link: invalid RX packet"); return PACKET_LINK_INVALID;
}
if (!link->rx_enabled || link->sending || link->state.phase != PACKET_LINK_OPEN) return PACKET_LINK_WOULD_BLOCK;
link->active++;
link->callbacks.received(link, packet, link->arg);
release(link);
return PACKET_LINK_OK;
}
void packet_link_close(struct packet_link* link) {
if (!link || link->closing) return;
link->active++; link->closing = 1;
if (link->notification) { uasync_call_soon_cancel(link->ua, link->notification); link->notification = NULL; }
link->ops.stop(link->impl);
release(link);
}
lib/packet_link.h
+39
View File
@@ -0,0 +1,39 @@
/* Пакетный канал одного uasync-потока: state → send/received → close.
* Каждый модуль сообщает готовность только собственных ресурсов. Формат байтов задаёт пользователь.
* send не вызывает верхние callbacks; received передаёт владение entry и dgram получателю.
* close допустим из callback, отменяет следующие события; после close handle недействителен. */
#ifndef PACKET_LINK_H
#define PACKET_LINK_H
#include <stddef.h>
#ifdef __cplusplus
extern "C" {
#endif
struct packet_link;
struct ll_entry;
enum packet_link_result {
PACKET_LINK_OK = 0, PACKET_LINK_WOULD_BLOCK = -1, PACKET_LINK_CLOSED = -2,
PACKET_LINK_TOO_LARGE = -3, PACKET_LINK_INVALID = -4, PACKET_LINK_ERROR = -5
};
enum packet_link_phase { PACKET_LINK_CONNECTING, PACKET_LINK_OPEN, PACKET_LINK_TERMINATED };
struct packet_link_state {
enum packet_link_phase phase; /* OPEN — готов транспорт, не ETCP-сессия. */
size_t max_packet; /* Максимальный пакет, включая байты формата пользователя. */
int tx_ready; /* Есть место для одного пакета до max_packet. */
int error; /* 0 — штатно; иначе packet_link_result. Подробности — в логе реализации. */
};
struct packet_link_callbacks {
void (*state_changed)(struct packet_link* link, void* arg); /* События объединяются; читать актуальный get_state. */
void (*received)(struct packet_link* link, struct ll_entry* packet, void* arg); /* Получатель владеет пакетом. */
};
/* Снимок без callbacks/изменения очередей; NULL возвращает TERMINATED/CLOSED. */
struct packet_link_state packet_link_get_state(const struct packet_link* link);
/* Целый извлечённый из очереди entry: OK передаёт владение, любой отказ оставляет его вызывающему. */
int packet_link_send(struct packet_link* link, struct ll_entry* packet);
/* Включить/приостановить received; нижний модуль сам ограничивает накопление. Изначально RX выключен. */
void packet_link_set_rx_enabled(struct packet_link* link, int enabled);
/* Отменить события и остановить реализацию; освобождение ждёт активные обработчики. NULL допустим. */
void packet_link_close(struct packet_link* link);
#ifdef __cplusplus
}
#endif
#endif
lib/packet_link_backend.h
+29
View File
@@ -0,0 +1,29 @@
/* Интерфейс реализации packet_link. После create реализация публикует состояние и целые RX-пакеты.
* Оболочка не создаёт очередей данных. Все вызовы — в одном uasync-потоке.
* stop отменяет источники событий, destroy освобождает impl после выхода обработчиков оболочки. */
#ifndef PACKET_LINK_BACKEND_H
#define PACKET_LINK_BACKEND_H
#include "packet_link.h"
#ifdef __cplusplus
extern "C" {
#endif
struct UASYNC;
struct packet_link_ops {
int (*send)(void* impl, struct ll_entry* packet); /* Отказ только до принятия; после принятия ошибки через state. */
void (*set_rx_enabled)(void* impl, int enabled); /* При возобновлении допустима синхронная выдача RX. */
void (*stop)(void* impl); /* Снять callbacks/таймеры, не освобождать impl. */
void (*destroy)(void* impl); /* Освободить остановленную реализацию. */
};
/* Копирует ops/state/callbacks. Успех передаёт impl оболочке, отказ оставляет вызывающему.
* arg заимствован до close. Первое state_changed отложено до возврата create. */
struct packet_link* packet_link_create(struct UASYNC* ua, const struct packet_link_ops* ops, void* impl,
const struct packet_link_state* initial, const struct packet_link_callbacks* callbacks, void* arg);
/* Обновить снимок сразу; уведомление отложено/объединено. TERMINATED необратим. */
void packet_link_backend_set_state(struct packet_link* link, const struct packet_link_state* state);
/* OK передаёт пакет получателю. CLOSED/WOULD_BLOCK/INVALID оставляет пакет реализации.
* Во время send выдача RX запрещена: реализация должна дождаться своего следующего события. */
int packet_link_backend_receive(struct packet_link* link, struct ll_entry* packet);
#ifdef __cplusplus
}
#endif
#endif
src/Makefile.am
+2
View File
@@ -83,6 +83,7 @@ utun_CORE_SOURCES = \
transport_layer/etcp_tx.c \
transport_layer/etcp_tx.h \
transport_layer/etcp_loadbalancer.c \
transport_layer/etcp_link_io.c \
transport_layer/etcp_debug.c \
transport_layer/etcp_dump.c \
transport_layer/secure_channel.c \
@@ -275,6 +276,7 @@ libutun_a_SOURCES = \
transport_layer/etcp_tx.c \
transport_layer/etcp_tx.h \
transport_layer/etcp_loadbalancer.c \
transport_layer/etcp_link_io.c \
transport_layer/etcp_debug.c \
transport_layer/etcp_dump.c \
transport_layer/secure_channel.c \
src/transport_layer/etcp.c
+13 -10
View File
@@ -1,5 +1,6 @@
// etcp.c - ETCP Protocol Implementation (refactored and expanded based on etcp_protocol.txt)
#include "etcp_link_io.h"
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_session.h"
@@ -1063,7 +1064,8 @@ static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp) {// вызыв
etcp->callbacks_running++;
struct ETCP_DGRAM* dgram;
while (!etcp->close_requested && (dgram = etcp_request_pkt(etcp))) {
if (etcp_loadbalancer_send(dgram) < 0) { etcp_schedule_send(etcp, 100); break; }
int sent = etcp_loadbalancer_send(dgram);
if (sent < 0) { if (sent != -2) etcp_schedule_send(etcp, 100); break; }
}
etcp->callbacks_running--;
etcp->tx_running = 0;
@@ -1093,7 +1095,7 @@ static void write_le(uint8_t* data, uint64_t value, unsigned size) {
static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct ETCP_LINK** chosen) {
struct ETCP_LINK* burst = NULL;
for (struct ETCP_LINK* l = conn->links; l; l = l->next)
if (l->burst_active && l->initialized && l->link_status) { burst = l; break; }
if (l->burst_active && l->initialized && l->link_status && etcp_link_io_ready(l)) { burst = l; break; }
struct INFLIGHT_PACKET* fresh = NULL;
struct ETCP_LINK* fresh_link = NULL;
unsigned examined = 0, retry_elsewhere = 0;
@@ -1150,7 +1152,7 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
struct ETCP_LINK* ack_link = NULL;
uint64_t first_due = UINT64_MAX;
for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next) {
if (!candidate->initialized || !candidate->link_status) continue;
if (!candidate->initialized || !candidate->link_status || !etcp_link_io_ready(candidate)) continue;
const struct ACK_PACKET* first = candidate->ack_q ? (struct ACK_PACKET*)candidate->ack_q->head : NULL;
if (!candidate->ack_reply_pending && !first) continue;
uint64_t due = candidate->ack_reply_pending ? 0 : first->received_at_us + ETCP_ACK_DELAY_US;
@@ -1189,6 +1191,7 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
dgram->payload_len = packet ? packet->ll.len : 0;
dgram->timestamp = get_current_timestamp(); dgram->burst_packet = link->burst_active;
dgram->ack_request = conn->ack_request_pending;
dgram->ack_count = dgram->commit_timestamp = dgram->commit_burst_response = 0;
if (packet) {
dgram->attempt = etcp_tx_prepare(link, packet);
if (!dgram->attempt) {
@@ -1200,14 +1203,15 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
write_le(data + 2, conn->last_delivered_id, 4); write_le(data + 6, conn->rx_dup_count, 2);
int ptr = ETCP_ACK_BASE_SIZE, remaining = max_enc - ptr - payload_size - dgram->ack_request;
int ack_space = remaining - ETCP_ACK_RX_POINT_SIZE;
while (ack_space >= ETCP_ACK_ENTRY_SIZE && data[1] != UINT8_MAX) {
struct ACK_PACKET* ack = (struct ACK_PACKET*)queue_data_get(link->ack_q);
if (!ack) break;
struct ll_entry* ack_entry = link->ack_q->head;
while (ack_entry && ack_space >= ETCP_ACK_ENTRY_SIZE && data[1] != UINT8_MAX) {
struct ACK_PACKET* ack = (struct ACK_PACKET*)ack_entry;
ack_entry = ack_entry->next;
write_le(data + ptr, ack->seq, 4); write_le(data + ptr + 4, (uint16_t)ack->attempt_id, 2);
uint64_t delay = get_time_us() - ack->received_at_us;
write_le(data + ptr + 6, delay >= UINT32_MAX ? UINT32_MAX : delay, 4);
ptr += ETCP_ACK_ENTRY_SIZE; remaining -= ETCP_ACK_ENTRY_SIZE; ack_space -= ETCP_ACK_ENTRY_SIZE; data[1]++;
queue_entry_free(&ack->ll);
dgram->ack_count++;
}
if (data[1]) {
write_le(data + ptr, link->rx_data_bytes, 8); write_le(data + ptr + 8, link->rx_data_time_us, 8);
@@ -1223,7 +1227,7 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
write_le(data + ptr, link->burst_resp_gap_avg, 4); ptr += 4;
write_le(data + ptr, link->burst_resp_gap_min, 4); ptr += 4;
data[ptr++] = link->burst_resp_pkt_count;
remaining -= MEAS_RESP_SECTION_SIZE; link->burst_resp_pending = 0;
remaining -= MEAS_RESP_SECTION_SIZE; dgram->commit_burst_response = 1;
}
int pkt_size_offset = 0;
if (dgram->burst_packet && remaining >= MEAS_TS_SECTION_SIZE) {
@@ -1238,7 +1242,7 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
}
if (link->last_recv_updated && remaining >= 5) {
uint64_t dt = get_time_tb() - link->last_recv_local_time;
link->last_recv_updated = 0;
dgram->commit_timestamp = 1;
if (dt < 1000000) {
data[ptr++] = ETCP_SECTION_TIMESTAMP;
write_le(data + ptr, link->last_recv_timestamp + dt, 2); ptr += 2;
@@ -1259,7 +1263,6 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
dgram->data_len = ptr;
if (pkt_size_offset) {
write_le(data + pkt_size_offset, ptr, 2); link->burst_pkt_size = ptr;
if (++link->burst_seq >= link->burst_count) etcp_link_burst_finish(link);
}
etcp_dump_pkt_sections(dgram, link, 1);
return dgram;
src/transport_layer/etcp_connections.c
+43 -17
View File
@@ -1,3 +1,4 @@
#include "etcp_link_io.h"
#include "etcp_connections.h"
#include "etcp_api.h"
#include "call_ring.h"
@@ -76,6 +77,7 @@ static void etcp_link_init_timer_cbk(void* arg);
static void burst_resp_timeout_cb(void* arg);
static int etcp_tcp_send(struct ETCP_DGRAM* dgram);
static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data, ssize_t recv_len, struct sockaddr_storage addr);
static void etcp_process_packet_direct(struct ETCP_SOCKET* e_sock, uint8_t* data, ssize_t recv_len, struct sockaddr_storage addr);
static void socks_etcp_read_callback(socket_t sock, void* arg);
static void socks_relay_ready_cb(struct socks_udp* s, int err, void* arg);
@@ -85,6 +87,7 @@ static void socks_relay_ready_cb(struct socks_udp* s, int err, void* arg) {
(void)s;
if (err != 0) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "socks UDP associate error for socket %s err=%d", e_sock->name, err); return; }
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "socks UDP relay ready for socket %s", e_sock->name);
etcp_link_io_socket_writable(e_sock->fd, e_sock);
}
// Чтение с локального UDP-сокета SOCKS5 UDP ASSOCIATE: снять SOCKS-заголовок → реальный src,
@@ -879,10 +882,10 @@ struct ETCP_SOCKET* etcp_socket_add(struct UTUN_INSTANCE* instance, struct CFG_S
u_free(e_sock);
return NULL;
}
e_sock->socket_id = uasync_add_socket_t(instance->ua, e_sock->fd, socks_etcp_read_callback, NULL, NULL, "etcp_sock_socks", e_sock);
e_sock->socket_id = uasync_add_socket_t(instance->ua, e_sock->fd, socks_etcp_read_callback, etcp_link_io_socket_writable, NULL, "etcp_sock_socks", e_sock);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "socket %s tunneled via SOCKS5 UDP (proxy=%s:%u)", e_sock->name, gc->socks_host, gc->socks_port);
} else {
e_sock->socket_id = uasync_add_socket_t(instance->ua, e_sock->fd, etcp_connections_read_callback_socket, NULL, NULL, "etcp_sock", e_sock);
e_sock->socket_id = uasync_add_socket_t(instance->ua, e_sock->fd, etcp_connections_read_callback_socket, etcp_link_io_socket_writable, NULL, "etcp_sock", e_sock);
}
if (!e_sock->socket_id) {
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "Failed to register socket with uasync");
@@ -1171,6 +1174,7 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn
etcp->log_name, link, conn ? conn->name : "tcp", link->local_link_id, link->is_server, link->mtu, link->is_tcp,
etcp->normalizer->frag_size, link->inflight_lim_bytes);
link->active_users++;
if (etcp_link_io_create(link)) { etcp_link_close(link); etcp_link_release(link); return NULL; }
etcp_fire_link_status_cbk(link, -1, 0);
if (!link->closing && !etcp->close_requested && !is_tcp && is_server == 0) {
@@ -1237,6 +1241,7 @@ void etcp_link_close(struct ETCP_LINK* link) {
if (!link || link->closing) return;
link->active_users++;
link->closing = 1;
etcp_link_io_destroy(link);
struct ETCP_CONN *etcp = link->etcp;
etcp->callbacks_running++;
void (*release_hook)(struct ETCP_LINK*, void*) = link->send_hook_release;
@@ -1347,6 +1352,7 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
if (!link || !link->etcp || link->closing || link->etcp->close_requested) return;
struct ETCP_CONN *etcp = link->etcp;
struct stcp_link *transport = link->tcp_link;
etcp_link_io_update(link);
etcp->callbacks_running++;
link->active_users++;
@@ -1462,6 +1468,7 @@ static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg) {
etcp->callbacks_running++;
stcp_link_close(sl);
link->tcp_link = NULL;
etcp_link_io_update(link);
if (is_server) {
/* Серверный линк: etcp_link_close сам снимает link из списка conn, затем
@@ -1505,10 +1512,8 @@ static int etcp_tcp_send(struct ETCP_DGRAM* dgram) {
dgram->timestamp = get_current_timestamp();
uint8_t buf[PACKET_DATA_SIZE + ETCP_SESSION_HEADER_SIZE + 3];
size_t total_len = etcp_encode_plain(dgram, buf);
int rc;
if (link->send_hook) rc = (int)link->send_hook(0, buf, total_len, NULL, 0, link, link->send_hook_ctx);
else if (link->tcp_link) rc = stcp_link_send(link->tcp_link, buf, total_len) == 0 ? (int)dgram->data_len : -1;
else rc = -1;
int rc = etcp_link_io_send(link, buf, total_len);
if (rc >= 0 && !link->send_hook) rc = (int)dgram->data_len;
if (rc > 0) link->total_encrypted += (size_t)rc;
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TCP send: link=%d total=%zu dlen=%d rc=%d via=%s",
link->etcp->log_name, link->local_link_id, total_len, dgram->data_len, rc,
@@ -1565,16 +1570,9 @@ int etcp_encrypt_send(struct ETCP_DGRAM* dgram) {
if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO, "Encrypted", enc_buf, enc_buf_len + dgram->noencrypt_len);
struct sockaddr_storage* addr=&dgram->link->remote_addr;
socklen_t addr_len = (addr->ss_family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);
ssize_t sent = etcp_udp_send(dgram->link, dgram->link->conn->fd, enc_buf, enc_buf_len + dgram->noencrypt_len,
(struct sockaddr*)addr, addr_len);
if (sent < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "sendto failed, sock_err=%d dst=%s fd=%d",
socket_get_error(), sockaddr_storage_to_str(addr).str, dgram->link->conn->fd);
dgram->link->send_errors++; errcode=4; goto es_err;
}
int sent = etcp_link_io_send(dgram->link, enc_buf, enc_buf_len + dgram->noencrypt_len);
if (sent == -2) return -2;
if (sent < 0) { dgram->link->send_errors++; errcode=4; goto es_err; }
return (int)sent;
es_err:
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "error %d", errcode);
@@ -2288,6 +2286,34 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
etcp_process_packet(e_sock, data, recv_len, addr);
}
/* Неизвестный UDP endpoint проходит прежний INIT/bootstrap; известный — общий packet_link. */
static void etcp_process_packet(struct ETCP_SOCKET* sock, uint8_t* data, ssize_t size, struct sockaddr_storage addr) {
struct ETCP_LINK* link = etcp_link_find_by_addr(sock, &addr, 0);
if (!link || !link->io) { etcp_process_packet_direct(sock, data, size, addr); return; }
struct ll_entry* p = ll_alloc_lldgram(size);
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "UDP packet RX allocation failed size=%zd", size); return; }
memcpy(p->dgram, data, size); p->len = size;
etcp_link_io_receive(link, p);
}
void etcp_link_receive_packet(struct ETCP_LINK* link, struct ll_entry* entry) {
if (!link->is_tcp) {
etcp_process_packet_direct(link->conn, entry->dgram, entry->len, link->remote_addr);
} else if (entry->len < 3 || entry->len > PACKET_DATA_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP invalid packet size=%u", entry->len);
} else {
struct ETCP_DGRAM* pkt = memory_pool_alloc(link->etcp->instance->pkt_pool);
if (!pkt) DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP packet RX pool exhausted");
else {
memcpy(&pkt->timestamp, entry->dgram, 2); pkt->flag_up = entry->dgram[2] & 1;
memcpy(pkt->data, entry->dgram + 3, entry->len - 3); pkt->link = link;
link->total_decrypted += entry->len;
etcp_packet_decrypted(link->conn, pkt, link, entry->len);
}
}
queue_dgram_free(entry); queue_entry_free(entry);
}
// Обработка принятого сырого UDP-пакета: расшифровка и диспетчеризация.
// Выделена из read_callback, чтобы один путь обслуживал прямой приём и приём через
// SOCKS5 UDP ASSOCIATE (socks_etcp_read_callback подставляет реальный src пира).
@@ -2307,7 +2333,7 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
// 46 - расшифрованный пакет слишком маленький (< 3 байта)
// 55 - не удалось создать подключение
// 66 - не удалось создать линк
static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
static void etcp_process_packet_direct(struct ETCP_SOCKET* e_sock, uint8_t* data,
ssize_t recv_len, struct sockaddr_storage addr) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
src/transport_layer/etcp_connections.h
+3
View File
@@ -74,6 +74,8 @@ struct ETCP_DGRAM {// пакет (незашифрованный)
uint16_t payload_len; // локальные метаданные request_pkt: DATA-байты; 0 — только ACK/control/FILLER, вне DATA-pacing
struct ETCP_TX_ATTEMPT* attempt; // локальный резерв окна; complete снимает/фиксирует его после send
uint8_t burst_packet; // Пакет создан измерительной серией; не зависит от последующей смены burst_active
uint8_t ack_count; // Число ACK, снимаемых из головы ack_q только после принятия send
uint8_t commit_timestamp, commit_burst_response; // Подготовленные служебные секции
uint8_t ack_request; // Пакет содержит запрос границы окна; сбрасывается только после успешного send
uint16_t timestamp;// timestamp отправляющего узла при отправке пакета
uint8_t flag_up:1;// bit0 = up/down (recv_keepalive)
@@ -223,6 +225,7 @@ struct ETCP_LINK {
struct ETCP_CONN* etcp; // подключение (parent)
struct ETCP_SOCKET* conn; // сокет через который работаем
struct etcp_link_io* io; // Владеющий пакетный адаптер; закрывается до native-транспорта
// Путь соединения
struct sockaddr_storage remote_addr; // Удалённый адрес
uint64_t lan_generation; // ненулевое поколение неподтверждённой LAN-пробы; 0 = обычный линк
src/transport_layer/etcp_link_io.c
+195
View File
@@ -0,0 +1,195 @@
#include "etcp_link_io.h"
#include "etcp_connections.h"
#include "etcp.h"
#include "stcp_link.h"
#include "socks_client.h"
#include "../../lib/packet_link_backend.h"
#include "../../lib/u_async.h"
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
#include <string.h>
#include <stdlib.h>
struct etcp_link_io {
struct ETCP_LINK* owner;
struct packet_link* channel;
struct ll_queue *tx, *rx;
int rx_enabled;
};
static void free_packet(struct ll_entry* p) { queue_dgram_free(p); queue_entry_free(p); }
static void rx_drain(struct ll_queue* q, void* arg) {
struct etcp_link_io* io = arg;
if (!io->rx_enabled) return;
struct ll_entry* p = queue_data_get(q);
if (!p) { queue_resume_callback(q); return; }
/* Получатель может закрыть owner и адаптер: продолжение планируем до callback. */
queue_resume_callback(q);
if (packet_link_backend_receive(io->channel, p) != PACKET_LINK_OK) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "packet adapter RX rejected"); free_packet(p);
}
}
static int udp_send(struct etcp_link_io* io, struct ll_entry* p) {
struct ETCP_LINK* link = io->owner;
if (link->conn->socks_udp && !socks_udp_is_ready(link->conn->socks_udp)) return PACKET_LINK_WOULD_BLOCK;
struct sockaddr_storage* addr = &link->remote_addr;
socklen_t size = addr->ss_family == AF_INET6 ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in);
ssize_t n = etcp_udp_send(link, link->conn->fd, p->dgram, p->len, (struct sockaddr*)addr, size);
if (n == p->len) { free_packet(p); return PACKET_LINK_OK; }
int err = socket_get_error();
if (n < 0 && (err == ERR_AGAIN || err == ERR_WOULDBLOCK)) return PACKET_LINK_WOULD_BLOCK;
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] UDP packet send failed path=%u size=%u result=%zd error=%d",
link->etcp->log_name, link->local_link_id, p->len, n, err);
return PACKET_LINK_ERROR;
}
void etcp_link_io_update(struct ETCP_LINK* link) {
if (!link || !link->io) return;
struct etcp_link_io* io = link->io;
struct packet_link_state state = {PACKET_LINK_OPEN, PACKET_DATA_SIZE, io->tx && !io->tx->count, 0};
if (link->is_tcp) {
state.phase = stcp_link_is_ready(link->tcp_link) ? PACKET_LINK_OPEN : PACKET_LINK_CONNECTING;
state.tx_ready = state.phase == PACKET_LINK_OPEN && stcp_link_can_send(link->tcp_link);
}
packet_link_backend_set_state(io->channel, &state);
}
static int adapter_send(void* arg, struct ll_entry* p) {
struct etcp_link_io* io = arg;
struct ETCP_LINK* link = io->owner;
int result;
if (link->is_tcp) {
if (link->send_hook) {
ssize_t n = link->send_hook(0, p->dgram, p->len, NULL, 0, link, link->send_hook_ctx);
result = n >= 0 ? PACKET_LINK_OK : PACKET_LINK_ERROR;
if (result == PACKET_LINK_OK) free_packet(p);
} else result = stcp_link_send_entry(link->tcp_link, p);
} else {
if (io->tx->count) return PACKET_LINK_WOULD_BLOCK;
result = udp_send(io, p);
if (result == PACKET_LINK_WOULD_BLOCK) {
/* Сохраняем целую датаграмму, а не считаем EAGAIN успешной передачей в сокет. */
if (queue_data_put(io->tx, p)) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "UDP TX queue invariant failed"); abort(); }
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] UDP TX waits path=%u bytes=%u", link->etcp->log_name, link->local_link_id, p->len);
if (!link->conn->socks_udp || socks_udp_is_ready(link->conn->socks_udp))
uasync_set_socket_write(link->etcp->instance->ua, link->conn->socket_id, 1);
result = PACKET_LINK_OK;
}
}
etcp_link_io_update(link);
return result;
}
static void rx_enabled(void* arg, int enabled) {
struct etcp_link_io* io = arg;
io->rx_enabled = enabled;
if (io->owner->tcp_link) stcp_link_set_rx_enabled(io->owner->tcp_link, enabled);
if (enabled) queue_resume_callback(io->rx);
}
static void stop(void* arg) {
struct etcp_link_io* io = arg;
queue_set_callback(io->rx, NULL, NULL);
io->rx_enabled = 0;
if (io->owner->tcp_link) stcp_link_set_rx_enabled(io->owner->tcp_link, 0);
}
static void destroy(void* arg) {
struct etcp_link_io* io = arg;
queue_clear(io->tx); queue_free(io->tx);
queue_clear(io->rx); queue_free(io->rx);
u_free(io);
}
static void changed(struct packet_link* channel, void* arg) {
struct etcp_link_io* io = arg;
struct packet_link_state state = packet_link_get_state(channel);
if (state.phase == PACKET_LINK_TERMINATED) { etcp_link_close(io->owner); return; }
if (state.tx_ready) etcp_conn_wake_sender(io->owner->etcp);
}
static void received(struct packet_link* channel, struct ll_entry* p, void* arg) {
(void)channel;
struct etcp_link_io* io = arg;
etcp_link_receive_packet(io->owner, p);
}
int etcp_link_io_create(struct ETCP_LINK* link) {
struct etcp_link_io* io = u_calloc(1, sizeof(*io));
if (!io) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "packet adapter allocation failed"); return -1; }
io->owner = link;
if (!link->is_tcp) io->tx = queue_new(link->etcp->instance->ua, 0, 0, 0, "link_tx");
io->rx = queue_new(link->etcp->instance->ua, 0, 0, 0, "link_rx");
if ((!link->is_tcp && !io->tx) || !io->rx) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "packet adapter queue allocation failed"); destroy(io); return -1;
}
const struct packet_link_ops ops = {adapter_send, rx_enabled, stop, destroy};
const struct packet_link_callbacks callbacks = {changed, received};
struct packet_link_state state = {link->is_tcp ? PACKET_LINK_CONNECTING : PACKET_LINK_OPEN, PACKET_DATA_SIZE, !link->is_tcp, 0};
io->channel = packet_link_create(link->etcp->instance->ua, &ops, io, &state, &callbacks, io);
if (!io->channel) { destroy(io); return -1; }
link->io = io;
queue_set_callback(io->rx, rx_drain, io);
packet_link_set_rx_enabled(io->channel, 1);
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] packet adapter path=%u channel=%p transport=%s",
link->etcp->log_name, link->local_link_id, (void*)io->channel, link->is_tcp ? "STCP" : "UDP");
return 0;
}
void etcp_link_io_destroy(struct ETCP_LINK* link) {
if (!link || !link->io) return;
struct etcp_link_io* io = link->io; link->io = NULL;
packet_link_close(io->channel);
}
struct packet_link* etcp_link_io_get(struct ETCP_LINK* link) { return link && link->io ? link->io->channel : NULL; }
int etcp_link_io_ready(struct ETCP_LINK* link) { return link && (!link->io || packet_link_get_state(link->io->channel).tx_ready); }
int etcp_link_io_send(struct ETCP_LINK* link, const void* bytes, size_t size) {
if (!link || !bytes || !size || size > PACKET_DATA_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "packet adapter invalid TX size=%zu", size); return -1;
}
if (!link->io && etcp_link_io_create(link)) return -1;
if (!etcp_link_io_ready(link)) return -2;
struct ll_entry* p = ll_alloc_lldgram(size);
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "packet adapter TX allocation failed size=%zu", size); return -1; }
memcpy(p->dgram, bytes, size); p->len = size;
int result = packet_link_send(link->io->channel, p);
if (result == PACKET_LINK_OK) return (int)size;
free_packet(p);
return result == PACKET_LINK_WOULD_BLOCK ? -2 : -1;
}
void etcp_link_io_receive(struct ETCP_LINK* link, struct ll_entry* p) {
if (link->io && link->io->rx_enabled && packet_link_backend_receive(link->io->channel, p) == PACKET_LINK_OK) return;
if (!link->io || link->io->rx->count >= 64) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "packet adapter RX full/unavailable path=%u", link->local_link_id); free_packet(p); return;
}
if (queue_data_put(link->io->rx, p)) DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "packet adapter RX insertion failed");
}
void etcp_link_io_socket_writable(socket_t socket, void* arg) {
(void)socket;
struct ETCP_SOCKET* sock = arg;
uasync_set_socket_write(sock->instance->ua, sock->socket_id, 0);
for (struct ll_entry* e = sock->links_queue->head; e; e = e->next) {
struct ETCP_LINK* link = ((struct link_queue_entry*)e->data)->link;
if (link->conn != sock || !link->io || !link->io->tx || !link->io->tx->head) continue;
struct etcp_link_io* io = link->io;
struct ll_entry* p = queue_data_get(io->tx);
int result = udp_send(io, p);
if (result == PACKET_LINK_WOULD_BLOCK) {
queue_data_put(io->tx, p);
if (!sock->socks_udp || socks_udp_is_ready(sock->socks_udp))
uasync_set_socket_write(sock->instance->ua, sock->socket_id, 1);
return;
}
if (result != PACKET_LINK_OK) {
free_packet(p);
struct packet_link_state state = {PACKET_LINK_TERMINATED, PACKET_DATA_SIZE, 0, result};
packet_link_backend_set_state(io->channel, &state);
} else etcp_link_io_update(link);
}
}
src/transport_layer/etcp_link_io.h
+35
View File
@@ -0,0 +1,35 @@
/* Привязка UDP/STCP к packet_link. Все операции — uasync-поток ETCP_LINK.
* Линк владеет адаптером, stop снимает его события до закрытия native-транспорта.
* UDP сохраняет не более одной датаграммы при EAGAIN; STCP сообщает готовность через своё API. */
#ifndef ETCP_LINK_IO_H
#define ETCP_LINK_IO_H
#include "../../lib/packet_link.h"
#include "../../lib/socket_compat.h"
#ifdef __cplusplus
extern "C" {
#endif
struct ETCP_LINK;
struct ETCP_SOCKET;
struct etcp_link_io;
/* Создать адаптер и привязать к link; 0/-1. Не начинает handshake. */
int etcp_link_io_create(struct ETCP_LINK* link);
/* Снять события, освободить очереди; NULL допустим. */
void etcp_link_io_destroy(struct ETCP_LINK* link);
/* Обновить состояние после события native-транспорта. */
void etcp_link_io_update(struct ETCP_LINK* link);
/* Заимствованный общий интерфейс; действителен до закрытия/замены адаптера. Отдельно close не вызывать. */
struct packet_link* etcp_link_io_get(struct ETCP_LINK* link);
/* Готовность для ETCP. До привязки адаптера дополнительного ограничения нет; send создаёт его при необходимости. */
int etcp_link_io_ready(struct ETCP_LINK* link);
/* Копирует краткоживущий wire-буфер в пакет; >=0 — принято, -2 — занят, -1 — ошибка. */
int etcp_link_io_send(struct ETCP_LINK* link, const void* bytes, size_t size);
/* Входящий пакет во владении вызывающего; передаёт callback либо освобождает с диагностикой отказа. */
void etcp_link_io_receive(struct ETCP_LINK* link, struct ll_entry* packet);
/* Общий UDP-сокет получил writable; обслужить ожидающие датаграммы его адаптеров. */
void etcp_link_io_socket_writable(socket_t socket, void* arg);
/* Получатель ETCP: вызывается через общий packet_link, владеет packet. */
void etcp_link_receive_packet(struct ETCP_LINK* link, struct ll_entry* packet);
#ifdef __cplusplus
}
#endif
#endif
src/transport_layer/etcp_loadbalancer.c
+4 -3
View File
@@ -1,5 +1,6 @@
// Выбор линка по индивидуальному окну DATA и pacing; ACK не занимает окно DATA.
#include "etcp_link_io.h"
#include "etcp_loadbalancer.h"
#include "../lib/debug_config.h"
#include "../lib/u_async.h"
@@ -97,7 +98,7 @@ struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t
link->attempts->count >= MAX_INFLIGHT_SIZE || data_len > link->inflight_lim_bytes ||
link->inflight_bytes > link->inflight_lim_bytes - data_len);
if (full) link->cwnd_limited = 1;
else if ((!data_len || loadbalancer_link_can_send(link)) && (retry || !best || link->inflight_bytes < best->inflight_bytes)) {
else if (etcp_link_io_ready(link) && (!data_len || loadbalancer_link_can_send(link)) && (retry || !best || link->inflight_bytes < best->inflight_bytes)) {
best = link;
if (retry) break; // За один круг прежний путь встретится только после всех альтернатив.
}
@@ -129,7 +130,7 @@ int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) {
int send_result = etcp_encrypt_send(dgram);
etcp_tx_complete(dgram, send_result >= 0);
if (send_result < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] encrypt/send failed (%d)", etcp->log_name, send_result);
if (send_result != -2) DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] encrypt/send failed (%d)", etcp->log_name, send_result);
memory_pool_free(etcp->instance->pkt_pool, dgram);
return send_result;
}
@@ -166,7 +167,7 @@ int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) {
}
int loadbalancer_link_can_send(struct ETCP_LINK* link) {
if (!link) return 0;
if (!link || !etcp_link_io_ready(link)) return 0;
if (link->burst_active) return 1; // burst bypasses all limits
if (link->shaper_timer) return 0;
return 1;
src/transport_layer/etcp_tx.c
+8
View File
@@ -195,6 +195,14 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_
void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
struct ETCP_CONN* conn = dgram->link->etcp;
if (success) {
for (unsigned i = 0; i < dgram->ack_count; ++i) {
struct ll_entry* ack = queue_data_get(dgram->link->ack_q);
if (!ack) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX commit: missing prepared ACK"); break; }
queue_entry_free(ack);
}
if (dgram->commit_timestamp) dgram->link->last_recv_updated = 0;
if (dgram->commit_burst_response) dgram->link->burst_resp_pending = 0;
if (dgram->burst_packet && ++dgram->link->burst_seq >= dgram->link->burst_count) etcp_link_burst_finish(dgram->link);
dgram->link->ack_reply_pending = 0;
conn->ack_till_deadline_us = 0;
if (dgram->ack_request) dgram->link->etcp->ack_request_pending = 0;
src/transport_layer/socks_client.c
+2
View File
@@ -496,6 +496,8 @@ void socks_udp_destroy(struct socks_udp *s) {
u_free(s);
}
int socks_udp_is_ready(const struct socks_udp *s) { return s && s->ready; }
ssize_t socks_udp_sendto(struct socks_udp *s, const uint8_t *data, size_t len,
const struct sockaddr_storage *dst) {
if (!s || (!data && len) || !dst || (dst->ss_family != AF_INET && dst->ss_family != AF_INET6)) {
src/transport_layer/socks_client.h
+2
View File
@@ -75,6 +75,8 @@ struct socks_udp *socks_udp_associate(struct UASYNC *ua, const struct socks_cfg
// -1 пока ассоциация не готова или при ошибке. Удалённая доставка не гарантируется.
ssize_t socks_udp_sendto(struct socks_udp *s, const uint8_t *data, size_t len,
const struct sockaddr_storage *dst);
/* Готовность relay; NULL — не готов. После reconnect ready_cb уведомляет о восстановлении. */
int socks_udp_is_ready(const struct socks_udp *s);
// Снять SOCKS-UDP заголовок с входящей датаграммы: вернуть payload и реальный src пира.
// s не используется, может быть NULL; raw/src обязательны. Поддержаны IPv4/IPv6, FRAG=0.
src/transport_layer/stcp.c
+19 -4
View File
@@ -12,6 +12,8 @@
#include <string.h>
#include <errno.h>
static void stcp_schedule_close(struct stcp_conn *c, int err);
void stcp_conn_set_tx_queue(struct stcp_conn *c, struct ll_queue *q) {
if (!c || !q) return;
c->tx_queue = q;
@@ -139,8 +141,15 @@ void stcp_pending_queue(struct stcp_conn *c, const uint8_t *data, size_t len) {
if (!c) return;
if (c->finish_requested) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "DATA rejected during graceful close"); return; }
struct pending_entry *pe = u_malloc(sizeof(*pe));
if (!pe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_pending_queue malloc entry failed"); return; }
if (len && data) { pe->data = u_malloc(len); if (!pe->data) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_pending_queue malloc data(%zu) failed", len); u_free(pe); return; } memcpy(pe->data, data, len); }
if (!pe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_pending_queue malloc entry failed"); stcp_schedule_close(c, ENOMEM); return; }
if (len && data) {
pe->data = u_malloc(len);
if (!pe->data) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_pending_queue malloc data(%zu) failed", len);
u_free(pe); stcp_schedule_close(c, ENOMEM); return;
}
memcpy(pe->data, data, len);
}
else { pe->data = NULL; }
pe->len = len;
pe->next = NULL;
@@ -258,6 +267,7 @@ void stcp_write_cb(socket_t sock, void *arg) {
if (c->close_after_send) { stcp_conn_do_close(c, 0); return; }
if (c->finish_requested) { stcp_conn_finish(c); return; }
stcp_flush_pending(c);
if (!c->send_buf && !c->pending_head) queue_resume_callback(c->tx_queue);
}
}
@@ -270,7 +280,10 @@ void stcp_flush_pending(struct stcp_conn *c) {
size_t enc_len;
uint8_t *enc = u_malloc(2 + pe->len + SC_CRC32_SIZE);
if (!enc) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_flush_pending malloc enc failed len=%zu", pe->len); if (pe->data) u_free(pe->data); u_free(pe); continue; }
if (!enc) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_flush_pending malloc enc failed len=%zu", pe->len);
u_free(pe->data); u_free(pe); stcp_schedule_close(c, ENOMEM); return;
}
if (stcp_frame_encrypt(c, pe->data, pe->len, enc, &enc_len)) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_flush_pending encrypt failed"); u_free(enc); if (pe->data) u_free(pe->data); u_free(pe);
stcp_schedule_close(c, ECANCELED);
@@ -478,14 +491,16 @@ void stcp_conn_do_close(struct stcp_conn *c, int err) {
void stcp_tx_queue_cb(struct ll_queue *q, void *arg) {
struct stcp_conn *c = (struct stcp_conn *)arg;
/* STCP владеет досылкой: не извлекаем следующий пакет в ещё одну очередь при занятом выходе. */
if (c->send_buf || c->pending_head) return;
struct ll_entry *e = queue_data_get(q);
if (!e) { queue_resume_callback(q); return; }
if (c->state == STCP_STATE_DATA && !c->finish_requested && !c->close_after_send && !c->close_err && e->dgram)
stcp_pending_queue(c, e->dgram, e->len);
queue_dgram_free(e);
queue_entry_free(e);
queue_resume_callback(q);
stcp_flush_pending(c);
if (!c->send_buf && !c->pending_head) queue_resume_callback(q);
}
// ====== rx backpressure ======
src/transport_layer/stcp_link.c
+50 -26
View File
@@ -1,5 +1,7 @@
// stcp_link.c — STCP link management implementation
#include "stcp_link.h"
#include "etcp_link_io.h"
#include "../../lib/packet_link.h"
#include "call_ring.h"
#include "stcp.h"
#include "stcp_server.h"
@@ -36,6 +38,7 @@ struct stcp_link {
uint8_t ready;
uint8_t closing;
uint8_t rx_enabled;
stcp_link_cb on_ready_cb;
void *ready_arg;
@@ -54,30 +57,18 @@ struct stcp_link {
static void link_rx_cb(struct ll_queue *q, void *arg) {
struct stcp_link *link = (struct stcp_link *)arg;
if (link->closing) return;
if (link->closing || !link->rx_enabled) return;
struct ll_entry *e = queue_data_get(q);
if (!e) { queue_resume_callback(q); return; }
if (link->etcp_conn && link->etcp_link && link->inst) {
if (e->len > PACKET_DATA_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_link rx oversized pkt: %u > %u, closing link=%p", e->len, (unsigned)PACKET_DATA_SIZE, (void*)link);
queue_dgram_free(e); queue_entry_free(e);
queue_resume_callback(q);
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_link rx oversized pkt: %u > %u", e->len, (unsigned)PACKET_DATA_SIZE);
queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q);
if (link->conn) stcp_conn_do_close(link->conn, EMSGSIZE);
return;
}
if (e->len < 3) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "stcp_link rx too short: %u bytes", e->len); queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; }
struct ETCP_DGRAM *pkt = memory_pool_alloc(link->inst->pkt_pool);
if (!pkt) { queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; }
pkt->timestamp = ((uint16_t*)e->dgram)[0];
pkt->flag_up = e->dgram[2] & 1;
if (e->len > 3) memcpy(pkt->data, e->dgram + 3, e->len - 3);
else memset(pkt->data, 0, 1);
pkt->link = link->etcp_link;
link->etcp_link->total_decrypted += e->len;
struct ETCP_SOCKET *sock = link->etcp_link->conn;
etcp_packet_decrypted(sock, pkt, link->etcp_link, e->len);
queue_dgram_free(e); queue_entry_free(e);
etcp_link_io_receive(link->etcp_link, e);
} else {
struct UTUN_INSTANCE *inst = link->inst;
uint8_t id = (e->dgram && e->len > 0) ? e->dgram[0] : ETCP_ID_DATA;
@@ -103,6 +94,7 @@ static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg)
conn->rx_queue = conn->tx_queue = NULL;
queue_set_callback(link->rx_queue, NULL, NULL);
queue_set_callback(link->tx_queue, NULL, NULL);
queue_set_on_get(link->tx_queue, NULL, NULL);
void (*callback)(struct stcp_link*, int, void*) = link->on_close_cb;
void *callback_arg = link->close_arg;
link->on_close_cb = NULL; link->close_arg = NULL;
@@ -111,7 +103,14 @@ static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg)
/* Callback мог вызвать close; после уведомления адаптер больше не используем. */
}
static void tx_dequeued(struct ll_queue *q, void *arg) {
(void)q;
struct stcp_link *link = arg;
if (!link->closing && link->etcp_link) etcp_link_io_update(link->etcp_link);
}
static int stcp_link_init_queues(struct stcp_link *link) {
link->rx_enabled = 1;
link->rx_queue = queue_new(link->ua, 0, 0, 0, "stcp_rx");
link->tx_queue = queue_new(link->ua, 0, 0, 0, "stcp_tx");
if (!link->rx_queue || !link->tx_queue) {
@@ -121,7 +120,8 @@ static int stcp_link_init_queues(struct stcp_link *link) {
}
queue_set_callback(link->rx_queue, link_rx_cb, link);
queue_set_threshold(link->rx_queue, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES);
queue_set_threshold(link->tx_queue, 0, 0);
queue_set_threshold(link->tx_queue, 1, 0);
queue_set_on_get(link->tx_queue, tx_dequeued, link);
queue_set_waiter_defer(link->tx_queue, 1);
stcp_conn_set_rx_queue(link->conn, link->rx_queue);
stcp_conn_set_tx_queue(link->conn, link->tx_queue);
@@ -333,6 +333,7 @@ void stcp_link_close(struct stcp_link *link) {
link->on_ready_cb = NULL; link->ready_arg = NULL;
queue_set_callback(link->rx_queue, NULL, NULL);
queue_set_callback(link->tx_queue, NULL, NULL);
queue_set_on_get(link->tx_queue, NULL, NULL);
callring_put(CR_SLINK_CLOSE, (uintptr_t)link, (uintptr_t)(link->etcp_link));
/* Обрыв вперёд-ссылки родителя: после отложенного free (close_impl)
* etcp_link->tcp_link не должен остаться висячим. */
@@ -360,16 +361,39 @@ void stcp_link_close(struct stcp_link *link) {
}
}
int stcp_link_can_send(struct stcp_link *link) {
return link && !link->closing && link->ready && link->conn && link->tx_queue && !link->tx_queue->count;
}
int stcp_link_send_entry(struct stcp_link *link, struct ll_entry *packet) {
if (!link || link->closing || !link->ready || !link->conn) return PACKET_LINK_CLOSED;
if (!packet || packet->owner || packet->next || packet->prev || !packet->dgram || !packet->len) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP invalid TX packet"); return PACKET_LINK_INVALID;
}
if (!stcp_link_can_send(link)) return PACKET_LINK_WOULD_BLOCK;
/* Очередь без hash/size_limit: после проверок отказ put означает нарушение инварианта,
* не recoverable отказ send (ll_queue при таком отказе уже освобождает entry). */
if (queue_data_put(link->tx_queue, packet)) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP packet TX queue invariant failed"); abort();
}
return PACKET_LINK_OK;
}
void stcp_link_set_rx_enabled(struct stcp_link *link, int enabled) {
if (!link || link->closing) return;
link->rx_enabled = !!enabled;
if (enabled) queue_resume_callback(link->rx_queue);
}
int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len) {
if (!link || link->closing || !link->ready || !link->conn) return -1;
struct ll_entry *e = queue_entry_new(0);
if (!e) return -1;
e->dgram = u_malloc(len ? len : 1);
if (!e->dgram) { queue_entry_free(e); return -1; }
if (len) memcpy(e->dgram, data, len);
e->len = (uint16_t)len;
queue_data_put(link->tx_queue, e);
return 0;
if (!stcp_link_can_send(link) || !data || !len || len > STCP_MAX_MSG_SIZE) return -1;
struct ll_entry *e = ll_alloc_lldgram(len);
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP TX allocation failed len=%zu", len); return -1; }
memcpy(e->dgram, data, len); e->len = len;
int result = stcp_link_send_entry(link, e);
if (result == PACKET_LINK_OK) return 0;
queue_dgram_free(e); queue_entry_free(e);
return -1;
}
int stcp_link_is_ready(struct stcp_link *link) {
src/transport_layer/stcp_link.h
+6
View File
@@ -59,6 +59,12 @@ struct stcp_link *stcp_link_connect(struct ETCP_LINK *etcp_link,
/* Сразу отменяет работу и отвязывает владельца; освобождает handle через call_soon, без on_close. */
void stcp_link_close(struct stcp_link *link);
int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len);
/* Пакетный путь: готовность собственной TX-очереди адаптера (один ожидающий пакет).
* send_entry: packet_link_result; только OK передаёт владение. Старый send копирует данные. */
struct ll_entry;
int stcp_link_can_send(struct stcp_link *link);
int stcp_link_send_entry(struct stcp_link *link, struct ll_entry *packet);
void stcp_link_set_rx_enabled(struct stcp_link *link, int enabled);
int stcp_link_is_ready(struct stcp_link *link);
struct ETCP_CONN *stcp_link_get_etcp_conn(struct stcp_link *link);
tests/Makefile.am
+8
View File
@@ -29,6 +29,8 @@ check_PROGRAMS = \
test_config_debug \
test_debug_categories \
test_log_rotation \
test_packet_link \
test_etcp_packet_link \
test_etcp_bbr \
test_packet_dump \
test_radio_vad \
@@ -842,6 +844,12 @@ test_debug_categories_LDADD = $(COMMON_LIBS)
test_log_rotation_SOURCES = test_log_rotation.c
test_log_rotation_LDADD = $(COMMON_LIBS)
test_packet_link_SOURCES = test_packet_link.c
test_packet_link_LDADD = $(COMMON_LIBS)
test_etcp_packet_link_SOURCES = test_etcp_packet_link.c
test_etcp_packet_link_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_config_debug_SOURCES = test_config_debug.c
test_config_debug_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_config_debug_LDADD = $(LIBUTUN) $(COMMON_LIBS)
tests/test_etcp_packet_link.c
+99
View File
@@ -0,0 +1,99 @@
/* Настоящий ETCP_LINK/UDP adapter: EAGAIN, два пути на сокете, ACK rollback и аварийный teardown. */
#ifdef NDEBUG
#undef NDEBUG
#endif
#include "etcp.h"
#include "etcp_connections.h"
#include "etcp_loadbalancer.h"
#include "etcp_link_io.h"
#include "../lib/u_async.h"
#include "../lib/mem.h"
#include "../lib/debug_config.h"
#include <assert.h>
#include <stdio.h>
#include <string.h>
struct output { int blocked, failed, sent; unsigned char value; };
static ssize_t wire(socket_t fd, const void* bytes, size_t size, const struct sockaddr* addr,
socklen_t length, struct ETCP_LINK* link, void* arg) {
(void)fd; (void)addr; (void)length; (void)link;
struct output* out = arg;
if (out->blocked || out->failed) {
#ifdef _WIN32
WSASetLastError(out->blocked ? WSAEWOULDBLOCK : WSAECONNRESET);
#else
errno = out->blocked ? EAGAIN : EIO;
#endif
return -1;
}
assert(size == 8 && ((const unsigned char*)bytes)[0] == out->value); out->sent++;
return size;
}
static struct ll_entry* packet(unsigned char value) {
struct ll_entry* p = ll_alloc_lldgram(8); assert(p); p->len = 8; memset(p->dgram, value, 8); return p;
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
size_t baseline = u_get_allocated_count();
struct UASYNC* ua = uasync_create(); assert(ua);
struct utun_config cfg = {0};
struct UTUN_INSTANCE inst = {0};
inst.ua = ua; inst.config = &cfg; inst.node_id = 1; cfg.global.bbr_max_cwnd = INFLIGHT_LIM_MAX;
inst.connections = queue_new(ua, 16, 0, 8, "packet_test");
inst.pkt_pool = memory_pool_init(sizeof(struct ETCP_DGRAM) + PACKET_DATA_SIZE, "packet_test");
inst.data_pool = memory_pool_init(PACKET_DATA_SIZE, "packet_data");
inst.ack_pool = memory_pool_init(sizeof(struct ACK_PACKET), "packet_ack");
assert(inst.connections && inst.pkt_pool && inst.data_pool && inst.ack_pool);
struct ETCP_SOCKET sock = {0}; sock.instance = &inst; sock.mtu = 1280;
sock.fd = socket(AF_INET, SOCK_DGRAM, 0); assert(sock.fd != SOCKET_INVALID);
assert(!socket_set_nonblocking(sock.fd));
sock.links_queue = queue_new(ua, 16, 0, LINK_ADDR_KEY_SIZE, "packet_links"); assert(sock.links_queue);
sock.socket_id = uasync_add_socket_t(ua, sock.fd, NULL, etcp_link_io_socket_writable, NULL, "packet_test", &sock);
assert(sock.socket_id);
struct ETCP_CONN* conn = etcp_connection_create(&inst, "packet_test"); assert(conn);
conn->peer_node_id = 2; etcp_conn_ready(conn); conn->links_up = 1;
struct sockaddr_storage addr = {0}; struct sockaddr_in* ip = (struct sockaddr_in*)&addr;
ip->sin_family = AF_INET; ip->sin_addr.s_addr = htonl(INADDR_LOOPBACK); ip->sin_port = htons(1111);
struct ETCP_LINK* a = etcp_link_new(conn, &sock, &addr, 1); assert(a);
ip->sin_port = htons(2222);
struct ETCP_LINK* b = etcp_link_new(conn, &sock, &addr, 1); assert(b);
a->initialized = a->link_status = b->initialized = b->link_status = 1;
struct output first = {.blocked = 1, .value = 42}, second = {.value = 17};
a->send_hook = wire; a->send_hook_ctx = &first; b->send_hook = wire; b->send_hook_ctx = &second;
assert(packet_link_send(etcp_link_io_get(a), packet(42)) == PACKET_LINK_OK);
assert(!first.sent && !etcp_link_io_ready(a));
struct ll_entry* held = packet(42);
assert(packet_link_send(etcp_link_io_get(a), held) == PACKET_LINK_WOULD_BLOCK && held->dgram[0] == 42);
assert(packet_link_send(etcp_link_io_get(b), packet(17)) == PACKET_LINK_OK && second.sent == 1);
assert(etcp_loadbalancer_select_link(conn, 0, NULL) == b);
a->burst_active = 1; assert(!loadbalancer_link_can_send(a)); a->burst_active = 0;
first.blocked = 0; etcp_link_io_socket_writable(sock.fd, &sock);
assert(first.sent == 1 && etcp_link_io_ready(a));
assert(packet_link_send(etcp_link_io_get(a), held) == PACKET_LINK_OK && first.sent == 2);
/* Подготовка не снимает ACK; отказ не меняет очередь/служебные признаки. */
b->initialized = b->link_status = 0;
struct ACK_PACKET* ack = (struct ACK_PACKET*)queue_entry_new_from_pool(inst.ack_pool); assert(ack);
ack->seq = 9; ack->attempt_id = etcp_tx_attempt_id(9, 1); ack->received_at_us = get_time_us();
assert(!queue_data_put_with_index(a->ack_q, &ack->ll)); a->ack_reply_pending = 1;
a->burst_resp_pending = 1; a->last_recv_updated = 1; a->last_recv_local_time = get_time_tb();
struct ETCP_DGRAM* p = etcp_request_pkt(conn); assert(p && p->ack_count == 1 && a->ack_q->count == 1);
etcp_tx_complete(p, 0); memory_pool_free(inst.pkt_pool, p);
assert(a->ack_q->count == 1 && a->ack_reply_pending && a->burst_resp_pending && a->last_recv_updated);
p = etcp_request_pkt(conn); assert(p && p->ack_count == 1);
etcp_tx_complete(p, 1); memory_pool_free(inst.pkt_pool, p);
assert(!a->ack_q->count && !a->ack_reply_pending && !a->burst_resp_pending && !a->last_recv_updated);
/* Ошибка после принятия ожидающей датаграммы закрывает только её путь. */
first.blocked = 1; assert(packet_link_send(etcp_link_io_get(a), packet(42)) == PACKET_LINK_OK);
first.blocked = 0; first.failed = 1; etcp_link_io_socket_writable(sock.fd, &sock);
assert(packet_link_get_state(etcp_link_io_get(a)).phase == PACKET_LINK_TERMINATED);
uasync_drain_immediate(ua); assert(conn->links == b && !b->next);
etcp_connection_close(conn); uasync_drain_immediate(ua);
uasync_remove_socket(ua, sock.socket_id); socket_close_wrapper(sock.fd); queue_free(sock.links_queue);
queue_free(inst.connections);
memory_pool_destroy(inst.pkt_pool); memory_pool_destroy(inst.data_pool); memory_pool_destroy(inst.ack_pool);
uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline);
puts("PASS ETCP packet adapter: UDP EAGAIN, shared socket paths, ACK commit/rollback, close, no leaks");
return 0;
}
tests/test_etcp_padding.c
+2
View File
@@ -1,3 +1,4 @@
#include "etcp_link_io.h"
/* Padding: реальные таймеры/UDP, проверка расшифрованных пакетов и независимости владельцев. */
#include <stdio.h>
#include <stdlib.h>
@@ -212,6 +213,7 @@ int main(void) {
etcp_link_send_keepalive(&link);
}
CHECK(keepalive_packets == 2);
etcp_link_io_destroy(&link);
struct ETCP_DGRAM* malformed = memory_pool_alloc(sender->pkt_pool); CHECK(malformed);
memset(malformed->data, 0, 7); malformed->data[0] = ETCP_KEEPALIVE;
malformed->data[1] = KA_FLAG_SLEEP; malformed->data_len = 7; malformed->link = &link;
tests/test_etcp_session.c
+6 -1
View File
@@ -1,3 +1,4 @@
#include "etcp_link_io.h"
/* Real ETCP lifecycle, AES-CCM and common ingress; deterministic encrypted wire faults. */
#include <stdio.h>
#include <stdlib.h>
@@ -208,6 +209,7 @@ static void setup(int multi) {
static void teardown(void) {
for (int i=0;i<2;i++) {
for (int j=0;j<2;j++) {
etcp_link_io_destroy(&ep[i].links[j]);
if (ep[i].links[j].attempts) etcp_tx_link_destroy(&ep[i].links[j]);
if (ep[i].links[j].rx_timeout_timer) uasync_cancel_timeout(ua, ep[i].links[j].rx_timeout_timer);
if (ep[i].links[j].keepalive_timer) uasync_cancel_timeout(ua, ep[i].links[j].keepalive_timer);
@@ -226,6 +228,7 @@ static void teardown(void) {
static void fresh_endpoint(int i) {
for (int j = 0; j < 2; j++) {
struct ETCP_LINK* link = &ep[i].links[j];
etcp_link_io_destroy(link);
if (link->attempts) etcp_tx_link_destroy(link);
if (link->keepalive_timer) { uasync_cancel_timeout(ua, link->keepalive_timer); link->keepalive_timer = NULL; }
if (link->rx_timeout_timer) { uasync_cancel_timeout(ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL; }
@@ -519,8 +522,10 @@ int main(void) {
blackhole=1;
uint64_t started = ep[0].c->session_started;
CHECK(etcp_conn_ref_take(ep[0].c)==0);
for (int i = 0; i < 2; i++) for (int j = 0; j < 2; j++)
for (int i = 0; i < 2; i++) for (int j = 0; j < 2; j++) {
etcp_link_io_destroy(&ep[i].links[j]);
if (ep[i].links[j].attempts) etcp_tx_link_destroy(&ep[i].links[j]);
}
ep[0].c->links=NULL; // timeout closes a real connection, links belong to fixture
while (ep[0].c->state!=2 && get_time_tb() < started + 110000) pump();
CHECK(ep[0].c->state == 2 && get_time_tb() - started >= 100000);
tests/test_packet_link.c
+99
View File
@@ -0,0 +1,99 @@
/* Контракт packet_link: владение, занятость, объединение событий, RX pause и teardown внутри callbacks. */
#ifdef NDEBUG
#undef NDEBUG
#endif
#include "../lib/packet_link_backend.h"
#include "../lib/ll_queue.h"
#include "../lib/u_async.h"
#include "../lib/mem.h"
#include "../lib/debug_config.h"
#include <assert.h>
#include <stdio.h>
struct fixture {
struct packet_link* link;
struct ll_entry* pending;
int states, received, stopped, destroyed, in_send, close_state, close_rx, send_state, enabled;
};
static struct ll_entry* packet(void) {
struct ll_entry* p = ll_alloc_lldgram(8); assert(p);
p->len = 8; p->dgram[0] = 42; return p;
}
static void free_packet(struct ll_entry* p) { queue_dgram_free(p); queue_entry_free(p); }
static void publish(struct fixture* f, int ready) {
const struct packet_link_state state = {PACKET_LINK_OPEN, 1500, ready, 0};
packet_link_backend_set_state(f->link, &state);
}
static int accept_packet(void* arg, struct ll_entry* p) {
struct fixture* f = arg; assert(!f->pending && !f->stopped);
f->in_send = 1; f->pending = p; publish(f, 0); f->in_send = 0;
return PACKET_LINK_OK;
}
static void enable(void* arg, int enabled) { ((struct fixture*)arg)->enabled = enabled; }
static void stop(void* arg) {
struct fixture* f = arg; assert(!f->stopped); f->stopped = 1;
if (f->pending) { free_packet(f->pending); f->pending = NULL; }
}
static void destroy(void* arg) {
struct fixture* f = arg; assert(f->stopped && !f->destroyed); f->destroyed = 1;
}
static void changed(struct packet_link* link, void* arg) {
struct fixture* f = arg; assert(!f->in_send && !f->stopped); f->states++;
if (f->close_state) { packet_link_close(link); return; }
if (f->send_state && packet_link_get_state(link).tx_ready) {
f->send_state = 0; assert(packet_link_send(link, packet()) == PACKET_LINK_OK);
}
}
static void received(struct packet_link* link, struct ll_entry* p, void* arg) {
struct fixture* f = arg; assert(f->enabled && !f->in_send && !f->stopped);
assert(p->dgram[0] == 42); f->received++; free_packet(p);
if (f->close_rx) packet_link_close(link);
}
static void create(struct fixture* f, struct UASYNC* ua) {
const struct packet_link_ops ops = {accept_packet, enable, stop, destroy};
const struct packet_link_callbacks callbacks = {changed, received};
const struct packet_link_state initial = {PACKET_LINK_OPEN, 1500, 1, 0};
f->link = packet_link_create(ua, &ops, f, &initial, &callbacks, f); assert(f->link);
assert(!f->states && !f->enabled);
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
size_t baseline = u_get_allocated_count();
struct UASYNC* ua = uasync_create(); assert(ua);
struct fixture f = {0}; create(&f, ua);
struct ll_entry* p = packet();
assert(packet_link_backend_receive(f.link, p) == PACKET_LINK_WOULD_BLOCK);
packet_link_set_rx_enabled(f.link, 1);
assert(packet_link_backend_receive(f.link, p) == PACKET_LINK_OK && f.received == 1);
packet_link_set_rx_enabled(f.link, 0);
p = packet(); assert(packet_link_backend_receive(f.link, p) == PACKET_LINK_WOULD_BLOCK); free_packet(p);
p = ll_alloc_lldgram(1501); assert(p); p->len = 1501;
assert(packet_link_send(f.link, p) == PACKET_LINK_TOO_LARGE); free_packet(p);
p = packet(); assert(packet_link_send(f.link, p) == PACKET_LINK_OK && f.pending == p);
assert(!packet_link_get_state(f.link).tx_ready && !f.states);
p = packet(); assert(packet_link_send(f.link, p) == PACKET_LINK_WOULD_BLOCK && p->dgram[0] == 42);
uasync_drain_immediate(ua); assert(f.states == 1);
free_packet(f.pending); f.pending = NULL; publish(&f, 1);
assert(packet_link_send(f.link, p) == PACKET_LINK_OK);
uasync_drain_immediate(ua); assert(f.states == 2);
packet_link_close(f.link); assert(f.stopped && !f.destroyed);
uasync_drain_immediate(ua); assert(f.destroyed && f.states == 2);
struct fixture cancel = {0}; create(&cancel, ua); packet_link_close(cancel.link);
uasync_drain_immediate(ua); assert(!cancel.states && cancel.destroyed);
struct fixture state = {.close_state = 1}; create(&state, ua);
uasync_drain_immediate(ua); assert(state.states == 1 && state.destroyed);
struct fixture rx = {.close_rx = 1}; create(&rx, ua); packet_link_set_rx_enabled(rx.link, 1);
assert(packet_link_backend_receive(rx.link, packet()) == PACKET_LINK_OK);
assert(rx.stopped && !rx.destroyed); uasync_drain_immediate(ua); assert(rx.destroyed && !rx.states);
struct fixture nested = {.send_state = 1}; create(&nested, ua);
uasync_drain_immediate(ua); assert(nested.states == 2 && nested.pending);
struct packet_link_state closed = {PACKET_LINK_TERMINATED, 1500, 0, PACKET_LINK_ERROR};
packet_link_backend_set_state(nested.link, &closed);
p = packet(); assert(packet_link_send(nested.link, p) == PACKET_LINK_CLOSED); free_packet(p);
packet_link_close(nested.link); uasync_drain_immediate(ua); assert(nested.destroyed);
uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline);
puts("PASS packet_link: ownership, TX blocked/resume, deferred/coalesced state, RX pause, close in callbacks, no leaks");
return 0;
}
tools/chatgui/CMakeLists.txt
+8
View File
@@ -492,6 +492,14 @@ add_executable(test_log_rotation ../../tests/test_log_rotation.c)
target_link_libraries(test_log_rotation PRIVATE uasync pthread)
add_test(NAME test_log_rotation COMMAND test_log_rotation)
add_executable(test_packet_link ../../tests/test_packet_link.c)
target_link_libraries(test_packet_link PRIVATE uasync pthread)
add_test(NAME test_packet_link COMMAND test_packet_link)
add_executable(test_etcp_packet_link ../../tests/test_etcp_packet_link.c)
target_link_libraries(test_etcp_packet_link PRIVATE utun uasync pthread)
add_test(NAME test_etcp_packet_link COMMAND test_etcp_packet_link)
add_executable(test_channel_delegate tests/test_channel_delegate.cpp src/channeldelegate.cpp)
target_link_libraries(test_channel_delegate PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun pthread ${CMAKE_DL_LIBS})
add_test(NAME test_channel_delegate COMMAND test_channel_delegate)