You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
231 lines
14 KiB
231 lines
14 KiB
// etcp_router.h — Сервисный слой маршрутизации поверх ETCP |
|
// Маршрутизирует сервисные пакеты до целевой ноды, на промежуточных нодах ретранслирует |
|
// Упрощённый TCP поверх ETCP: восстановление порядка, дедупликация, без переповторов |
|
// |
|
// Формат SVC_ROUTE пакета: |
|
// [cmd:1] [dst_node_id:8] [src_node_id:8] [seq:4] [svc_id:1] [flags:1] [payload...] |
|
// flags: bit7=START, bit6=RST, bit5-4=sess_id(0-3) |
|
// ACK-пакет: тот же заголовок, seq=rx_seq, payload_len=0 |
|
#ifndef ETCP_ROUTER_H |
|
#define ETCP_ROUTER_H |
|
|
|
#ifdef __cplusplus |
|
extern "C" { |
|
#endif |
|
|
|
|
|
#include <stdint.h> |
|
#include "etcp_api.h" |
|
|
|
#define TRANSIT_QUEUE_HASH 256 |
|
|
|
#pragma pack(push, 1) |
|
struct SVC_ROUTE_HDR { |
|
uint8_t cmd; // ETCP_ID_SVC_ROUTE (0x03) / ETCP_RT_ID_SVC_ROUTE |
|
uint64_t dst_node_id; |
|
uint64_t src_node_id; |
|
uint32_t seq; // data: tx_seq; ACK: rx_seq (ожидаемый seq) |
|
uint8_t svc_id; // идентификатор сервиса |
|
uint8_t flags; // bit7=START, bit6=RST, bit5-4=sess_id |
|
uint16_t timestamp; // часы отправителя (0.1ms), для RTT; ACK: эхо последнего data-пакета |
|
}; |
|
#pragma pack(pop) |
|
#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 25 |
|
#define SVC_ROUTE_MAX_BINDINGS 256 |
|
|
|
// Биты в flags |
|
#define ROUTER_FLAG_START 0x80 |
|
#define ROUTER_FLAG_RST 0x40 |
|
#define ROUTER_FLAG_SIGNED 0x08 // пакет содержит Ed25519 подпись (64 байта после payload) |
|
#define ROUTER_FLAG_CLOSE 0x02 // нормальное закрытие conn |
|
#define ROUTER_SESS_ID_SHIFT 4 |
|
#define ROUTER_SESS_ID_MASK 0x30 |
|
|
|
// Очередь транзитных пакетов для пары (src_node_id, dst_node_id) |
|
// Создаётся когда не получается отправить пакет в ETCP напрямую, |
|
// удаляется когда очередь становится пустой. |
|
// ll.data[0..15] = src:8 + dst:8 — совпадает с полями ниже по памяти |
|
struct TRANSIT_QUEUE { |
|
struct ll_entry ll; // data[0..15] = src_node_id:8 + dst_node_id:8 (16-byte hash index) |
|
uint64_t src_node_id; // = data[0..7] |
|
uint64_t dst_node_id; // = data[8..15] |
|
struct ll_queue* q; // FIFO транзитных пакетов (ll_entry с dgram), начиная с data[16] |
|
struct queue_waiter_handle waiter; // backpressure waiter на bp-очередь (normalizer->input или tx_queue) |
|
struct ETCP_CONN* conn; // next_hop (для etcp_send в drain_cb) |
|
}; |
|
|
|
// Состояние одного логического подключения (remote_node_id + svc_id) |
|
struct ETCP_ROUTER_CONN { |
|
struct ll_entry ll; // data[0..7]=remote_node_id, data[8]=svc_id — хеш-индекс |
|
uint64_t remote_node_id; // = data[0..7] |
|
uint8_t svc_id; // = data[8] |
|
uint16_t last_dgram_ts; // последний timestamp датаграммы (rx или tx, timebase 0.1ms) |
|
uint16_t rtt; // измеренный RTT (0.1ms) |
|
int32_t rtt_jitter; // EMA jitter, fixed-point *65536 (как в ETCP) |
|
uint16_t max_inflight; // настраиваемый рабочий максимум (default=ROUTER_MAX_INFLIGHT) |
|
uint16_t inflight_limit; // текущий эффективный лимит (=max_inflight или MINRTT_PROBE_MAX_INFLIGHT) |
|
uint16_t minrtt; // скользящее среднее minRTT (0.1ms), default=100 (10ms) |
|
uint16_t minrtt_window[10]; // окно 10 последних замеров minRTT |
|
uint64_t minrtt_window_tb[10];// таймстемпы замеров (timebase 0.1ms) |
|
uint8_t minrtt_window_idx; // cyclic index 0..9 |
|
uint8_t minrtt_window_count; // активных записей в окне (0..10) |
|
uint8_t minrtt_probe; // режим измерения (inflight capped at 4) |
|
uint64_t last_inflight_loaded_tb; // когда inflight >= threshold |
|
uint64_t last_inflight_unloaded_tb; // когда inflight < threshold |
|
uint8_t inflight_was_loaded; // 1=канал загружен |
|
uint16_t last_recv_pkt_ts; // timestamp пира из последнего data-пакета |
|
uint64_t last_recv_pkt_local_tb; // наше время приёма последнего data-пакета (0.1ms timebase) |
|
uint8_t last_recv_updated; // флаг: есть данные для эха в ACK |
|
|
|
uint32_t tx_seq; // следующий seq для отправки |
|
uint32_t rx_seq; // ожидаемый seq для сборки (next expected) |
|
uint32_t tx_acked; // сколько наших пакетов подтвердил remote (для inflight) |
|
uint32_t last_sent_ack_seq; // последний отправленный ACK (= rx_seq на момент отправки) |
|
uint64_t last_ack_sent_tb; // время отправки последнего ACK (timebase 0.1ms) |
|
|
|
struct UTUN_INSTANCE* inst; |
|
struct ll_queue* recv_q; // reorder очередь: hash по seq (4 байта, offset 0) |
|
void* ack_timer; // периодический таймер (10ms при consumer_ack) |
|
void* idle_ack_timer; // idle таймер (дослать последний ack) |
|
|
|
struct ll_queue* send_q; // очередь ожидающих отправки (inflight полон) |
|
void* send_resume_timer; // таймер возобновления отправки |
|
uint8_t send_blocked; // 1 = inflight полон, ждём ack/таймер |
|
uint8_t no_route; // 1 = нет BGP-маршрута, передача приостановлена |
|
void* no_route_timer; // таймер 20ms проверки появления маршрута |
|
uint8_t no_ack_count; // счётчик последовательных ретрансмиссий без ACK |
|
uint8_t closed; // 1 = в процессе закрытия, таймеры игнорируют |
|
void* close_callback; // пользовательский коллбэк завершения закрытия |
|
void* close_callback_arg; // аргумент для close_callback |
|
uint8_t sess_id; // наш session id (0-3) |
|
uint8_t peer_sess_id; // последний sess_id от peer'а |
|
uint8_t start_sent; // 0 = нужно отправить START в первом data |
|
uint8_t peer_sync_done; // 0 = rx_seq ещё не синхронизирован с удалённым seq (авто-создание/рестарт) |
|
|
|
// Ретрансмиты: inflight очередь |
|
struct ll_queue* inflight_q; // хеш по seq (4 байта), непрерывный блок без SACK |
|
void* retrans_timer; // таймер проверки застоя ACK |
|
uint64_t last_ack_changed_tb; // когда tx_acked последний раз менялся (0.1ms) |
|
|
|
// Очередь приёма входящего трафика (между сетью и recv_q) |
|
struct ll_queue* incoming_q; // FIFO очередь входящих пакетов |
|
uint8_t incoming_data_ready; // 1 = данные в incoming_q и callback был вызван |
|
|
|
uint32_t c_pkts_sent; // успешные отправки данных |
|
uint32_t c_pkts_send_err; // ошибки отправки |
|
uint32_t c_pkts_rcvd; // получено и доставлено данных |
|
uint32_t c_ack_sent; // отправлено ACK |
|
uint32_t c_ack_recv; // получено ACK (последовательных) |
|
uint32_t c_retrans_done; // число выполненых ретрансмитов |
|
uint32_t c_dup_dropped; // дропнуто дубликатов seq |
|
uint32_t c_oob_dropped; // дропнуто out-of-bounds seq |
|
uint32_t c_stale_ack; // устаревших ACK |
|
uint32_t c_sign_fail; // signature verification failures |
|
}; |
|
|
|
#define ROUTER_CONN_HASH_SIZE 256 |
|
#define ROUTER_RECVQ_HASH_SIZE 1024 |
|
#define ROUTER_MAX_INFLIGHT 256 // макс пакетов в полёте (для контроля inflight) |
|
#define ROUTER_ACK_INTERVAL_TB 100 // интервал ACK: 10ms в timebase (0.1ms) |
|
#define ROUTER_ACK_IDLE_TB 5000 // idle таймаут: 500ms |
|
#define ROUTER_SEND_RESUME_TB 500 // retry интервал send_q: 50ms |
|
#define ROUTER_MAX_SEND_Q_PACKETS 64 // порог backpressure на send_q |
|
|
|
// Ретрансмиты |
|
#define ROUTER_RETRANS_TIMEOUT_TB 3000 // 300ms в timebase (0.1ms) |
|
#define ROUTER_INFLIGHT_HASH_SIZE 1024 |
|
#define ROUTER_NO_ROUTE_RETRY_TB 200 // 20ms проверка появления маршрута |
|
#define ROUTER_NO_ACK_MAX_RETRANS 17 // 17 × 300ms ≈ 5s таймаут без ACK |
|
|
|
// minRTT measurement |
|
#define MINRTT_UNLOADED_MIN_TB 500 // 50ms минимальное время разгрузки канала |
|
#define MINRTT_PROBE_TIMEOUT_TB 100000 // 10 сек без свежих замеров → probe |
|
#define MINRTT_PROBE_MAX_INFLIGHT 4 // лимит inflight в режиме probe |
|
#define MINRTT_WINDOW_SIZE 10 |
|
#define MINRTT_DEFAULT_TB 100 // default minRTT = 10ms (заменяется первым RTT) |
|
|
|
// Inflight запись — копия отправленного пакета для возможного ретрансмита |
|
struct ROUTER_INFLIGHT { |
|
struct ll_entry ll; // индекс по seq (4 байта, offset 0) |
|
uint32_t seq; |
|
uint64_t last_sent_tb; // время последней отправки (0.1ms) |
|
uint8_t send_count; // число переотправок |
|
uint8_t* payload; // копия данных |
|
size_t payload_len; |
|
}; |
|
|
|
// Bindings для сервисов внутри etcp_router (аналогично ETCP_BINDINGS) |
|
struct ETCP_ROUTER_BINDINGS { |
|
etcp_recv_fn callbacks[SVC_ROUTE_MAX_BINDINGS]; |
|
}; |
|
|
|
// Инициализация: etcp_bind(ETCP_ID_SVC_ROUTE) + очистка bindings + создание router_conns |
|
int etcp_router_init(struct UTUN_INSTANCE* inst); |
|
|
|
// Деинициализация |
|
void etcp_router_destroy(struct UTUN_INSTANCE* inst); |
|
|
|
// Зарегистрировать обработчик сервиса |
|
int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback); |
|
int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id); |
|
|
|
// Отправить сервисный пакет (авто-conn, seq, inflight-контроль через send_q) |
|
int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_entry* entry, int force); |
|
|
|
// Найти/создать состояние seq-подключения по (remote_node_id, svc_id) |
|
struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, |
|
uint64_t remote_node_id, uint8_t svc_id); |
|
|
|
// Отправить данные с авто-seq и контролем inflight |
|
// data: payload без svc_id, flags: битовые флаги (зарезервировано) |
|
int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, |
|
const uint8_t* data, size_t len); |
|
|
|
// Отправить данные с Ed25519 подписью (аналог etcp_router_conn_send с флагом ROUTER_FLAG_SIGNED) |
|
int etcp_router_conn_send_signed(struct ETCP_ROUTER_CONN* rconn, |
|
const uint8_t* data, size_t len); |
|
|
|
// Закрыть seq-подключение |
|
void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn); |
|
|
|
// Установить рабочий max_inflight, пересчитывает inflight_limit и state machine |
|
void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max); |
|
|
|
// Асинхронное закрытие: коллбэк вызывается после освобождения памяти rconn |
|
void etcp_router_conn_close_async(struct ETCP_ROUTER_CONN* rconn, |
|
void (*on_close_done)(void* arg), |
|
void* close_arg); |
|
|
|
// Закрыть все router_conn для указанного remote_node_id (peer умер) |
|
void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id); |
|
|
|
// Возвращает количество пакетов в очереди normalizer->input для узла node_id |
|
int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t node_id); |
|
|
|
// Backpressure: зарегистрировать/отменить waiter на normalizer->input очереди |
|
void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, |
|
struct queue_waiter_handle* h, |
|
queue_threshold_callback_fn callback, void* arg); |
|
void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, |
|
struct queue_waiter_handle* h); |
|
|
|
// Backpressure: зарегистрировать/отменить waiter на send_q очереди |
|
// h — handle из структуры сервиса, callback вызывается когда send_q.count <= threshold (128) |
|
void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, |
|
struct queue_waiter_handle* h, |
|
queue_threshold_callback_fn callback, void* arg); |
|
void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, |
|
struct queue_waiter_handle* h); |
|
|
|
// Сбросить состояние роутера для конкретного peer+svc (перезапуск удалённой стороны). |
|
// Очищает send_q/recv_q, уведомляет сервис через cb(NULL, entry) с remote_node_id. |
|
void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id); |
|
|
|
// Удалить все транзитные очереди соединения (вызывается при закрытии ETCP_CONN) |
|
void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn); |
|
|
|
|
|
#ifdef __cplusplus |
|
} |
|
#endif |
|
#endif // ETCP_ROUTER_H
|
|
|