diff --git a/doc/packet_link.txt b/doc/packet_link.txt new file mode 100644 index 00000000..6595feb5 --- /dev/null +++ b/doc/packet_link.txt @@ -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. diff --git a/lib/Makefile.am b/lib/Makefile.am index 1fcc7c71..98822eae 100644 --- a/lib/Makefile.am +++ b/lib/Makefile.am @@ -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 \ diff --git a/lib/packet_link.c b/lib/packet_link.c new file mode 100644 index 00000000..dd8eeb27 --- /dev/null +++ b/lib/packet_link.c @@ -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 + +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); +} diff --git a/lib/packet_link.h b/lib/packet_link.h new file mode 100644 index 00000000..705882c0 --- /dev/null +++ b/lib/packet_link.h @@ -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 +#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 diff --git a/lib/packet_link_backend.h b/lib/packet_link_backend.h new file mode 100644 index 00000000..ef5e0b2f --- /dev/null +++ b/lib/packet_link_backend.h @@ -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 diff --git a/src/Makefile.am b/src/Makefile.am index e90b2b35..9269798e 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -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 \ diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 8a4dbc73..05256772 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -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; diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index 9d76cac0..441046f3 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -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, ""); diff --git a/src/transport_layer/etcp_connections.h b/src/transport_layer/etcp_connections.h index c8fcdf20..7e420758 100644 --- a/src/transport_layer/etcp_connections.h +++ b/src/transport_layer/etcp_connections.h @@ -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 = обычный линк diff --git a/src/transport_layer/etcp_link_io.c b/src/transport_layer/etcp_link_io.c new file mode 100644 index 00000000..b5a58b56 --- /dev/null +++ b/src/transport_layer/etcp_link_io.c @@ -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 +#include + +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); + } +} diff --git a/src/transport_layer/etcp_link_io.h b/src/transport_layer/etcp_link_io.h new file mode 100644 index 00000000..7f69f6a8 --- /dev/null +++ b/src/transport_layer/etcp_link_io.h @@ -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 diff --git a/src/transport_layer/etcp_loadbalancer.c b/src/transport_layer/etcp_loadbalancer.c index 7ae29cee..21fc258c 100644 --- a/src/transport_layer/etcp_loadbalancer.c +++ b/src/transport_layer/etcp_loadbalancer.c @@ -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; diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index eec3e212..d632efd8 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -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; diff --git a/src/transport_layer/socks_client.c b/src/transport_layer/socks_client.c index dd5fcdc8..9639ff9a 100644 --- a/src/transport_layer/socks_client.c +++ b/src/transport_layer/socks_client.c @@ -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)) { diff --git a/src/transport_layer/socks_client.h b/src/transport_layer/socks_client.h index 32ccefc8..ce342c6c 100644 --- a/src/transport_layer/socks_client.h +++ b/src/transport_layer/socks_client.h @@ -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. diff --git a/src/transport_layer/stcp.c b/src/transport_layer/stcp.c index 38182505..4b01896b 100644 --- a/src/transport_layer/stcp.c +++ b/src/transport_layer/stcp.c @@ -12,6 +12,8 @@ #include #include +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 ====== diff --git a/src/transport_layer/stcp_link.c b/src/transport_layer/stcp_link.c index 95c5fec9..35c486a2 100644 --- a/src/transport_layer/stcp_link.c +++ b/src/transport_layer/stcp_link.c @@ -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) { diff --git a/src/transport_layer/stcp_link.h b/src/transport_layer/stcp_link.h index 13d17e2d..64ac9c0a 100644 --- a/src/transport_layer/stcp_link.h +++ b/src/transport_layer/stcp_link.h @@ -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); diff --git a/tests/Makefile.am b/tests/Makefile.am index 7f255cd7..774a3fc3 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -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) diff --git a/tests/test_etcp_packet_link.c b/tests/test_etcp_packet_link.c new file mode 100644 index 00000000..5c35413d --- /dev/null +++ b/tests/test_etcp_packet_link.c @@ -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 +#include +#include + +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; +} diff --git a/tests/test_etcp_padding.c b/tests/test_etcp_padding.c index 5edf8d20..9423833a 100644 --- a/tests/test_etcp_padding.c +++ b/tests/test_etcp_padding.c @@ -1,3 +1,4 @@ +#include "etcp_link_io.h" /* Padding: реальные таймеры/UDP, проверка расшифрованных пакетов и независимости владельцев. */ #include #include @@ -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; diff --git a/tests/test_etcp_session.c b/tests/test_etcp_session.c index f2c72a3a..078293f0 100644 --- a/tests/test_etcp_session.c +++ b/tests/test_etcp_session.c @@ -1,3 +1,4 @@ +#include "etcp_link_io.h" /* Real ETCP lifecycle, AES-CCM and common ingress; deterministic encrypted wire faults. */ #include #include @@ -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); diff --git a/tests/test_packet_link.c b/tests/test_packet_link.c new file mode 100644 index 00000000..9a7293f9 --- /dev/null +++ b/tests/test_packet_link.c @@ -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 +#include + +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; +} diff --git a/tools/chatgui/CMakeLists.txt b/tools/chatgui/CMakeLists.txt index 81333153..1ef6b2ce 100644 --- a/tools/chatgui/CMakeLists.txt +++ b/tools/chatgui/CMakeLists.txt @@ -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)