Browse Source
- Implement queue_wait_threshold() with automatic waiter checking in ll_queue - Add pkt_normalizer_flush() to force buffer sending - Enhance test suite with async wait tests, fragmentation verification, and buffer flush tests - Fix test edge cases by flushing packer buffer between subtests - Improve test reliability with increased iteration limits and while loop processing - Add connection module for secure UDP communication with ECC/AES-CCM cryptographyv2_dev
jek
9 months ago
21 changed files with 10664 additions and 388 deletions
@ -0,0 +1,724 @@
|
||||
// connection.c - Реализация минималистичного API для защищенных UDP подключений
|
||||
|
||||
#include "connection.h" |
||||
#include "sc_lib.h" |
||||
#include "pkt_normalizer.h" |
||||
#include "etcp.h" |
||||
#include "u_async.h" |
||||
#include "ll_queue.h" |
||||
#include "settings.h" |
||||
|
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <stdio.h> |
||||
#include <errno.h> |
||||
|
||||
/* Сетевые заголовки */ |
||||
#include <unistd.h> |
||||
#include <sys/socket.h> |
||||
#include <netinet/in.h> |
||||
#include <arpa/inet.h> |
||||
#include <fcntl.h> |
||||
|
||||
/* Внутренняя структура подключения */ |
||||
struct conn_handle { |
||||
/* Сетевая часть */ |
||||
int sockfd; |
||||
struct sockaddr_in local_addr; |
||||
struct sockaddr_in remote_addr; |
||||
conn_mode_t mode; |
||||
uint8_t remote_defined; /* 1 если удаленный адрес определен */ |
||||
uint8_t socket_connected; /* 1 если сокет создан и настроен */ |
||||
|
||||
/* Криптография */ |
||||
sc_context_t crypto_ctx; |
||||
uint8_t keys_set; /* 1 если ключи установлены */ |
||||
uint8_t crypto_ready; /* 1 если сессионный ключ вычислен */ |
||||
|
||||
/* Обработка данных */ |
||||
epkt_t* etcp; |
||||
pkt_normalizer_pair* normalizer; |
||||
ll_queue_t* app_input_queue; /* Данные от приложения для отправки */ |
||||
ll_queue_t* app_output_queue; /* Собранные данные для приложения */ |
||||
|
||||
/* Callback'и */ |
||||
conn_recv_callback_t recv_callback; |
||||
void* recv_callback_user; |
||||
|
||||
/* Состояние */ |
||||
uint8_t is_closing; |
||||
uint8_t is_destroying; |
||||
|
||||
/* Таймеры */ |
||||
void* socket_read_timer; |
||||
|
||||
/* Статистика */ |
||||
conn_stats_t stats; |
||||
}; |
||||
|
||||
/* Внутренние функции */ |
||||
static int create_udp_socket(const char* ip, uint16_t port, struct sockaddr_in* addr); |
||||
static void socket_read_callback(void* arg); |
||||
static void etcp_tx_callback(epkt_t* epkt, uint8_t* data, uint16_t len, void* arg); |
||||
static void app_output_callback(ll_queue_t* q, ll_entry_t* entry, void* arg); |
||||
static void process_received_packet(conn_handle_t* conn, uint8_t* data, size_t len); |
||||
static void update_stats_from_etcp(conn_handle_t* conn); |
||||
|
||||
/* Bridge functions for data flow */ |
||||
static void app_input_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg); |
||||
static void packer_output_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg); |
||||
static void etcp_output_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg); |
||||
|
||||
/* Глобальная инициализация uasync (вызывается один раз) */ |
||||
static int uasync_initialized = 0; |
||||
|
||||
/* ==================== Публичные функции ==================== */ |
||||
|
||||
conn_handle_t* conn_create(void) |
||||
{ |
||||
/* Инициализация uasync глобально */ |
||||
if (!uasync_initialized) { |
||||
uasync_init(); |
||||
uasync_initialized = 1; |
||||
} |
||||
|
||||
/* Выделение памяти */ |
||||
conn_handle_t* conn = calloc(1, sizeof(conn_handle_t)); |
||||
if (!conn) { |
||||
return NULL; |
||||
} |
||||
|
||||
/* Инициализация очередей приложения */ |
||||
conn->app_input_queue = queue_new(); |
||||
conn->app_output_queue = queue_new(); |
||||
if (!conn->app_input_queue || !conn->app_output_queue) { |
||||
if (conn->app_input_queue) queue_free(conn->app_input_queue); |
||||
if (conn->app_output_queue) queue_free(conn->app_output_queue); |
||||
free(conn); |
||||
return NULL; |
||||
} |
||||
|
||||
/* Настройка callback для выходной очереди */ |
||||
queue_set_callback(conn->app_output_queue, app_output_callback, conn); |
||||
|
||||
/* Инициализация статистики */ |
||||
memset(&conn->stats, 0, sizeof(conn->stats)); |
||||
|
||||
/* Начальное состояние */ |
||||
conn->sockfd = -1; |
||||
conn->mode = CONN_MODE_CLIENT; /* по умолчанию клиент */ |
||||
conn->is_closing = 0; |
||||
conn->is_destroying = 0; |
||||
|
||||
return conn; |
||||
} |
||||
|
||||
int conn_set_keys(conn_handle_t* conn, |
||||
const uint8_t* my_pub_key, |
||||
const uint8_t* my_priv_key, |
||||
const uint8_t* peer_pub_key) |
||||
{ |
||||
if (!conn || conn->socket_connected) { |
||||
return -1; /* Ключи нужно устанавливать до подключения */ |
||||
} |
||||
|
||||
sc_status_t status; |
||||
|
||||
if (my_pub_key && my_priv_key) { |
||||
/* Использование предоставленных ключей */ |
||||
status = sc_init_local_keys(&conn->crypto_ctx, my_pub_key, my_priv_key); |
||||
if (status != SC_OK) { |
||||
return -1; |
||||
} |
||||
} else { |
||||
/* Автоматическая генерация ключей */ |
||||
status = sc_generate_keypair(&conn->crypto_ctx); |
||||
if (status != SC_OK) { |
||||
return -1; |
||||
} |
||||
} |
||||
|
||||
/* Установка ключа пира, если предоставлен */ |
||||
if (peer_pub_key) { |
||||
status = sc_set_peer_public_key(&conn->crypto_ctx, peer_pub_key); |
||||
if (status != SC_OK) { |
||||
return -1; |
||||
} |
||||
conn->crypto_ready = 1; |
||||
} else { |
||||
/* Для сервера ключ пира будет получен из первого пакета */ |
||||
conn->crypto_ready = 0; |
||||
} |
||||
|
||||
conn->keys_set = 1; |
||||
return 0; |
||||
} |
||||
|
||||
int conn_connect(conn_handle_t* conn, |
||||
const char* local_ip, |
||||
uint16_t local_port, |
||||
const char* remote_ip, |
||||
uint16_t remote_port, |
||||
conn_mode_t mode) |
||||
{ |
||||
if (!conn || conn->socket_connected) { |
||||
return -1; /* Уже подключен */ |
||||
} |
||||
|
||||
/* Проверка ключей */ |
||||
if (!conn->keys_set) { |
||||
/* Автоматическая установка ключей по умолчанию */ |
||||
if (conn_set_keys(conn, NULL, NULL, NULL) != 0) { |
||||
return -1; |
||||
} |
||||
} |
||||
|
||||
/* Создание UDP сокета */ |
||||
conn->sockfd = create_udp_socket(local_ip, local_port, &conn->local_addr); |
||||
if (conn->sockfd < 0) { |
||||
return -1; |
||||
} |
||||
|
||||
/* Сохранение режима */ |
||||
conn->mode = mode; |
||||
|
||||
/* Настройка удаленного адреса для клиента */ |
||||
if (mode == CONN_MODE_CLIENT && remote_ip) { |
||||
memset(&conn->remote_addr, 0, sizeof(conn->remote_addr)); |
||||
conn->remote_addr.sin_family = AF_INET; |
||||
conn->remote_addr.sin_port = htons(remote_port); |
||||
|
||||
if (inet_pton(AF_INET, remote_ip, &conn->remote_addr.sin_addr) != 1) { |
||||
close(conn->sockfd); |
||||
conn->sockfd = -1; |
||||
return -1; |
||||
} |
||||
|
||||
conn->remote_defined = 1; |
||||
} else { |
||||
/* Сервер или клиент без указанного адреса */ |
||||
conn->remote_defined = 0; |
||||
} |
||||
|
||||
/* Инициализация нормализатора пакетов */ |
||||
conn->normalizer = pkt_normalizer_pair_init(); |
||||
if (!conn->normalizer) { |
||||
close(conn->sockfd); |
||||
conn->sockfd = -1; |
||||
return -1; |
||||
} |
||||
|
||||
/* Инициализация ETCP */ |
||||
conn->etcp = etcp_init(); |
||||
if (!conn->etcp) { |
||||
pkt_normalizer_pair_deinit(conn->normalizer); |
||||
close(conn->sockfd); |
||||
conn->sockfd = -1; |
||||
return -1; |
||||
} |
||||
|
||||
/* Настройка callback'ов ETCP */ |
||||
etcp_set_callback(conn->etcp, etcp_tx_callback, conn); |
||||
|
||||
/* Начальная полоса пропускания: максимальная для uint16_t ≈ 6553 байт/мс */ |
||||
etcp_set_bandwidth(conn->etcp, 65535); /* в единицах 0.1мс: 6553.5 * 10 */ |
||||
|
||||
/* Связывание нормализатора с ETCP */ |
||||
/* Packer: app_input_queue -> normalizer -> etcp_tx_put */ |
||||
/* Unpacker: etcp_output_queue -> normalizer -> app_output_queue */ |
||||
|
||||
/* Set up bridge callbacks */ |
||||
|
||||
/* 1. App input -> packer input */ |
||||
queue_set_callback(conn->app_input_queue, app_input_bridge, conn); |
||||
|
||||
/* 2. Packer output -> ETCP tx */ |
||||
if (conn->normalizer->packer && conn->normalizer->packer->output) { |
||||
queue_set_callback(conn->normalizer->packer->output, packer_output_bridge, conn); |
||||
} |
||||
|
||||
/* 3. ETCP output -> unpacker input */ |
||||
ll_queue_t* etcp_output_q = etcp_get_output_queue(conn->etcp); |
||||
if (etcp_output_q) { |
||||
queue_set_callback(etcp_output_q, etcp_output_bridge, conn); |
||||
} |
||||
|
||||
/* 4. Unpacker output -> app output queue */ |
||||
if (conn->normalizer->unpacker && conn->normalizer->unpacker->output) { |
||||
queue_set_callback(conn->normalizer->unpacker->output, app_output_callback, conn); |
||||
} |
||||
|
||||
/* Запуск таймера чтения сокета */ |
||||
conn->socket_read_timer = uasync_set_timeout(1, conn, socket_read_callback); |
||||
|
||||
conn->socket_connected = 1; |
||||
return 0; |
||||
} |
||||
|
||||
void conn_set_recv_callback(conn_handle_t* conn, |
||||
conn_recv_callback_t callback, |
||||
void* user_data) |
||||
{ |
||||
if (!conn) return; |
||||
|
||||
conn->recv_callback = callback; |
||||
conn->recv_callback_user = user_data; |
||||
} |
||||
|
||||
int conn_send(conn_handle_t* conn, const uint8_t* data, size_t len) |
||||
{ |
||||
if (!conn || !data || len == 0 || !conn->socket_connected || conn->is_closing) { |
||||
return -1; |
||||
} |
||||
|
||||
/* Проверка готовности криптографии (для клиента) */ |
||||
/* TODO: implement proper key exchange */ |
||||
/* if (conn->mode == CONN_MODE_CLIENT && !conn->crypto_ready) {
|
||||
return -1; |
||||
} */ |
||||
|
||||
/* Создание записи очереди */ |
||||
ll_entry_t* entry = queue_entry_new(len); |
||||
if (!entry) { |
||||
return -1; |
||||
} |
||||
|
||||
/* Копирование данных */ |
||||
memcpy(ll_entry_data(entry), data, len); |
||||
|
||||
/* Добавление в очередь ввода приложения */ |
||||
int result = queue_entry_put(conn->app_input_queue, entry); |
||||
if (result != 0) { |
||||
queue_entry_free(entry); |
||||
return -1; |
||||
} |
||||
|
||||
/* Обновление статистики */ |
||||
conn->stats.bytes_sent += len; |
||||
|
||||
/* Активация обработки (через callback очереди) */ |
||||
queue_resume_callback(conn->app_input_queue); |
||||
|
||||
return 0; |
||||
} |
||||
|
||||
void conn_close(conn_handle_t* conn) |
||||
{ |
||||
if (!conn || conn->is_closing) { |
||||
return; |
||||
} |
||||
|
||||
conn->is_closing = 1; |
||||
|
||||
/* Отмена таймеров */ |
||||
if (conn->socket_read_timer) { |
||||
uasync_cancel_timeout(conn->socket_read_timer); |
||||
conn->socket_read_timer = NULL; |
||||
} |
||||
|
||||
/* Закрытие сокета */ |
||||
if (conn->sockfd >= 0) { |
||||
close(conn->sockfd); |
||||
conn->sockfd = -1; |
||||
} |
||||
|
||||
conn->socket_connected = 0; |
||||
} |
||||
|
||||
void conn_destroy(conn_handle_t* conn) |
||||
{ |
||||
if (!conn) return; |
||||
|
||||
conn->is_destroying = 1; |
||||
|
||||
/* Закрытие подключения если активно */ |
||||
conn_close(conn); |
||||
|
||||
/* Освобождение ETCP */ |
||||
if (conn->etcp) { |
||||
etcp_free(conn->etcp); |
||||
conn->etcp = NULL; |
||||
} |
||||
|
||||
/* Освобождение нормализатора */ |
||||
if (conn->normalizer) { |
||||
pkt_normalizer_pair_deinit(conn->normalizer); |
||||
conn->normalizer = NULL; |
||||
} |
||||
|
||||
/* Освобождение очередей */ |
||||
if (conn->app_input_queue) { |
||||
queue_free(conn->app_input_queue); |
||||
conn->app_input_queue = NULL; |
||||
} |
||||
|
||||
if (conn->app_output_queue) { |
||||
queue_free(conn->app_output_queue); |
||||
conn->app_output_queue = NULL; |
||||
} |
||||
|
||||
/* Очистка криптографического контекста */ |
||||
memset(&conn->crypto_ctx, 0, sizeof(conn->crypto_ctx)); |
||||
|
||||
/* Освобождение дескриптора */ |
||||
free(conn); |
||||
} |
||||
|
||||
int conn_get_stats(conn_handle_t* conn, conn_stats_t* stats) |
||||
{ |
||||
if (!conn || !stats) { |
||||
return -1; |
||||
} |
||||
|
||||
/* Обновление статистики из ETCP */ |
||||
update_stats_from_etcp(conn); |
||||
|
||||
/* Копирование статистики */ |
||||
memcpy(stats, &conn->stats, sizeof(conn_stats_t)); |
||||
|
||||
return 0; |
||||
} |
||||
|
||||
/* ==================== Внутренние функции ==================== */ |
||||
|
||||
/* Bridge functions implementation */ |
||||
static void app_input_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg) |
||||
{ |
||||
(void)entry; /* unused parameter - we'll get entry from queue */ |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || !q || !conn->normalizer || !conn->normalizer->packer) { |
||||
return; |
||||
} |
||||
|
||||
/* Remove entry from app_input_queue */ |
||||
ll_entry_t* entry_to_forward = queue_entry_get(q); |
||||
if (!entry_to_forward) { |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
size_t len = ll_entry_size(entry_to_forward); |
||||
printf("[CONN DEBUG] app_input_bridge: forwarding packet len=%zu to packer\n", len); |
||||
|
||||
/* Forward entry from app_input_queue to packer's input queue */ |
||||
queue_entry_put(conn->normalizer->packer->input, entry_to_forward); |
||||
queue_resume_callback(q); |
||||
} |
||||
|
||||
static void packer_output_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg) |
||||
{ |
||||
(void)entry; /* unused parameter - we'll get entry from queue */ |
||||
printf("[CONN DEBUG] packer_output_bridge ENTER, queue count=%d\n", queue_entry_count(q)); |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || !q || !conn->etcp) { |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
/* Remove entry from queue */ |
||||
ll_entry_t* entry_to_process = queue_entry_get(q); |
||||
if (!entry_to_process) { |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
/* Extract data from entry */ |
||||
uint8_t* data = ll_entry_data(entry_to_process); |
||||
size_t len = ll_entry_size(entry_to_process); |
||||
printf("[CONN DEBUG] packer_output_bridge: processing packet len=%zu\n", len); |
||||
|
||||
if (len > 65535) { |
||||
/* Too large for ETCP (max 64KB) */ |
||||
queue_entry_free(entry_to_process); |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
uint8_t* data_to_send = data; |
||||
uint16_t data_len = (uint16_t)len; |
||||
|
||||
/* Encrypt if crypto is ready - TEMPORARILY DISABLED */ |
||||
if (conn->crypto_ready) { |
||||
/* Allocate buffer for ciphertext + tag */ |
||||
uint8_t* ciphertext = malloc(len); |
||||
uint8_t tag[SC_TAG_SIZE]; |
||||
sc_status_t status = sc_encrypt(&conn->crypto_ctx, data, len, ciphertext, tag); |
||||
if (status == SC_OK) { |
||||
/* Combine ciphertext and tag into single buffer */ |
||||
uint8_t* encrypted_data = malloc(len + SC_TAG_SIZE); |
||||
if (encrypted_data) { |
||||
memcpy(encrypted_data, ciphertext, len); |
||||
memcpy(encrypted_data + len, tag, SC_TAG_SIZE); |
||||
data_to_send = encrypted_data; |
||||
data_len = (uint16_t)(len + SC_TAG_SIZE); |
||||
} |
||||
/* Free ciphertext buffer */ |
||||
free(ciphertext); |
||||
} |
||||
/* If encryption failed, fall back to plaintext */ |
||||
} |
||||
|
||||
/* Note: etcp_tx_put copies data, we can free entry after call */ |
||||
int result = etcp_tx_put(conn->etcp, data_to_send, data_len); |
||||
|
||||
/* Free encrypted buffer if it was allocated */ |
||||
if (conn->crypto_ready && data_to_send != data) { |
||||
free(data_to_send); |
||||
} |
||||
|
||||
queue_entry_free(entry_to_process); |
||||
|
||||
(void)result; /* Ignore result for now */ |
||||
queue_resume_callback(q); |
||||
} |
||||
|
||||
static void etcp_output_bridge(ll_queue_t* q, ll_entry_t* entry, void* arg) |
||||
{ |
||||
(void)entry; /* unused parameter - we'll get entry from queue */ |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || !q || !conn->normalizer || !conn->normalizer->unpacker) { |
||||
return; |
||||
} |
||||
|
||||
/* Remove entry from ETCP output queue */ |
||||
ll_entry_t* entry_to_forward = queue_entry_get(q); |
||||
if (!entry_to_forward) { |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
/* Extract data from entry */ |
||||
uint8_t* data = ll_entry_data(entry_to_forward); |
||||
size_t len = ll_entry_size(entry_to_forward); |
||||
|
||||
ll_entry_t* new_entry = NULL; |
||||
|
||||
/* Decrypt if crypto is ready - TEMPORARILY DISABLED */ |
||||
if (conn->crypto_ready) { |
||||
size_t ciphertext_len = len - SC_TAG_SIZE; |
||||
uint8_t* ciphertext = data; |
||||
uint8_t* tag = data + ciphertext_len; |
||||
|
||||
/* Allocate buffer for plaintext */ |
||||
uint8_t* plaintext = malloc(ciphertext_len); |
||||
if (plaintext) { |
||||
sc_status_t status = sc_decrypt(&conn->crypto_ctx, ciphertext, ciphertext_len, tag, plaintext); |
||||
if (status == SC_OK) { |
||||
/* Create new entry with plaintext */ |
||||
new_entry = queue_entry_new(ciphertext_len); |
||||
if (new_entry) { |
||||
memcpy(ll_entry_data(new_entry), plaintext, ciphertext_len); |
||||
} |
||||
} |
||||
free(plaintext); |
||||
} |
||||
} |
||||
|
||||
/* If we created a new entry, replace the old one */ |
||||
if (new_entry) { |
||||
queue_entry_free(entry_to_forward); |
||||
entry_to_forward = new_entry; |
||||
} |
||||
|
||||
/* Forward entry to unpacker's input queue */ |
||||
queue_entry_put(conn->normalizer->unpacker->input, entry_to_forward); |
||||
queue_resume_callback(q); |
||||
} |
||||
|
||||
static int create_udp_socket(const char* ip, uint16_t port, struct sockaddr_in* addr) |
||||
{ |
||||
int sock = socket(AF_INET, SOCK_DGRAM, 0); |
||||
if (sock < 0) { |
||||
return -1; |
||||
} |
||||
|
||||
/* Установка non-blocking режима */ |
||||
int flags = fcntl(sock, F_GETFL, 0); |
||||
if (flags < 0) { |
||||
close(sock); |
||||
return -1; |
||||
} |
||||
|
||||
if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0) { |
||||
close(sock); |
||||
return -1; |
||||
} |
||||
|
||||
/* Настройка адреса */ |
||||
memset(addr, 0, sizeof(*addr)); |
||||
addr->sin_family = AF_INET; |
||||
addr->sin_port = htons(port); |
||||
|
||||
if (ip && ip[0] != '\0') { |
||||
if (inet_pton(AF_INET, ip, &addr->sin_addr) != 1) { |
||||
close(sock); |
||||
return -1; |
||||
} |
||||
} else { |
||||
addr->sin_addr.s_addr = INADDR_ANY; |
||||
} |
||||
|
||||
/* Bind к адресу */ |
||||
if (bind(sock, (struct sockaddr*)addr, sizeof(*addr)) < 0) { |
||||
close(sock); |
||||
return -1; |
||||
} |
||||
|
||||
return sock; |
||||
} |
||||
|
||||
static void socket_read_callback(void* arg) |
||||
{ |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || conn->is_closing || conn->sockfd < 0) { |
||||
return; |
||||
} |
||||
|
||||
uint8_t buffer[2048]; /* Максимальный размер UDP пакета */ |
||||
struct sockaddr_in from_addr; |
||||
socklen_t from_len = sizeof(from_addr); |
||||
|
||||
/* Чтение всех доступных пакетов */ |
||||
while (1) { |
||||
ssize_t received = recvfrom(conn->sockfd, buffer, sizeof(buffer), 0, |
||||
(struct sockaddr*)&from_addr, &from_len); |
||||
|
||||
if (received <= 0) { |
||||
if (errno == EAGAIN || errno == EWOULDBLOCK) { |
||||
/* Нет больше данных */ |
||||
break; |
||||
} |
||||
/* Ошибка чтения */ |
||||
break; |
||||
} |
||||
|
||||
/* Обновление статистики */ |
||||
conn->stats.bytes_received += received; |
||||
conn->stats.packets_received++; |
||||
printf("[CONN DEBUG] socket_read_callback: received packet len=%zd, total_received=%u\n", received, conn->stats.packets_received); |
||||
|
||||
/* Для сервера: установка удаленного адреса при первом пакете */ |
||||
if (!conn->remote_defined && conn->mode == CONN_MODE_SERVER) { |
||||
memcpy(&conn->remote_addr, &from_addr, sizeof(conn->remote_addr)); |
||||
conn->remote_defined = 1; |
||||
} |
||||
|
||||
/* Обработка полученного пакета */ |
||||
process_received_packet(conn, buffer, received); |
||||
} |
||||
|
||||
/* Перезапуск таймера чтения */ |
||||
if (!conn->is_closing) { |
||||
conn->socket_read_timer = uasync_set_timeout(1, conn, socket_read_callback); |
||||
} |
||||
} |
||||
|
||||
static void etcp_tx_callback(epkt_t* epkt, uint8_t* data, uint16_t len, void* arg) |
||||
{ |
||||
(void)epkt; /* unused parameter */ |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || conn->is_closing || conn->sockfd < 0) { |
||||
free(data); /* Данные были выделены в etcp.c */ |
||||
return; |
||||
} |
||||
|
||||
/* Шифрование данных если криптография готова */ |
||||
uint8_t* packet_to_send = data; |
||||
uint16_t packet_len = len; |
||||
|
||||
if (conn->crypto_ready) { |
||||
/* TODO: реализовать шифрование через sc_encrypt */ |
||||
/* Временно пропускаем шифрование для тестирования */ |
||||
} |
||||
|
||||
/* Отправка через UDP сокет */ |
||||
if (conn->remote_defined) { |
||||
printf("[CONN DEBUG] etcp_tx_callback: sending packet len=%u, stats.packets_sent=%u\n", packet_len, conn->stats.packets_sent); |
||||
ssize_t sent = sendto(conn->sockfd, packet_to_send, packet_len, 0, |
||||
(struct sockaddr*)&conn->remote_addr,
|
||||
sizeof(conn->remote_addr)); |
||||
|
||||
if (sent == (ssize_t)packet_len) { |
||||
conn->stats.packets_sent++; |
||||
printf("[CONN DEBUG] etcp_tx_callback: sent successfully, new count=%u\n", conn->stats.packets_sent); |
||||
} else { |
||||
printf("[CONN DEBUG] etcp_tx_callback: send failed, sent=%zd, errno=%d\n", sent, errno); |
||||
} |
||||
} |
||||
|
||||
/* Освобождение данных (выделены в etcp.c) - packet is stored in ETCP sent_list for retransmission */ |
||||
/* DO NOT free here */ |
||||
(void)packet_to_send; /* unused */ |
||||
} |
||||
|
||||
static void app_output_callback(ll_queue_t* q, ll_entry_t* entry, void* arg) |
||||
{ |
||||
(void)entry; /* unused parameter - we'll get entry from queue */ |
||||
conn_handle_t* conn = (conn_handle_t*)arg; |
||||
if (!conn || !q || conn->is_closing) { |
||||
return; |
||||
} |
||||
|
||||
/* Remove entry from unpacker output queue */ |
||||
ll_entry_t* entry_to_process = queue_entry_get(q); |
||||
if (!entry_to_process) { |
||||
queue_resume_callback(q); |
||||
return; |
||||
} |
||||
|
||||
/* Получение данных из записи */ |
||||
uint8_t* data = ll_entry_data(entry_to_process); |
||||
size_t len = ll_entry_size(entry_to_process); |
||||
|
||||
/* Вызов пользовательского callback'а если установлен */ |
||||
if (conn->recv_callback) { |
||||
conn->recv_callback(conn, data, len, conn->recv_callback_user); |
||||
} |
||||
|
||||
/* Освобождение записи */ |
||||
queue_entry_free(entry_to_process); |
||||
|
||||
/* Обновление статистики */ |
||||
conn->stats.fragments_assembled++; |
||||
queue_resume_callback(q); |
||||
} |
||||
|
||||
static void process_received_packet(conn_handle_t* conn, uint8_t* data, size_t len) |
||||
{ |
||||
if (!conn || !data || len == 0) { |
||||
return; |
||||
} |
||||
|
||||
/* Расшифрование если криптография готова */ |
||||
uint8_t* packet_to_process = data; |
||||
uint16_t packet_len = len; |
||||
|
||||
if (conn->crypto_ready) { |
||||
/* TODO: реализовать расшифрование через sc_decrypt */ |
||||
/* Временно пропускаем для тестирования */ |
||||
} else { |
||||
/* Криптография не готова, но ключи должны быть установлены заранее */ |
||||
printf("[CONN] WARNING: Received packet but crypto not ready (keys should be pre-shared)\n"); |
||||
/* Все равно передаем пакет в ETCP для обработки */ |
||||
/* В реальном использовании это ошибка конфигурации */ |
||||
} |
||||
|
||||
/* Передача пакета в ETCP для обработки */ |
||||
printf("[CONN DEBUG] process_received_packet: forwarding to ETCP, len=%u\n", packet_len); |
||||
etcp_rx_input(conn->etcp, packet_to_process, packet_len); |
||||
} |
||||
|
||||
|
||||
|
||||
static void update_stats_from_etcp(conn_handle_t* conn) |
||||
{ |
||||
if (!conn || !conn->etcp) { |
||||
return; |
||||
} |
||||
|
||||
/* Получение метрик из ETCP */ |
||||
conn->stats.current_rtt_ms = etcp_get_rtt(conn->etcp) / 10; /* 0.1ms -> ms */ |
||||
conn->stats.jitter_ms = etcp_get_jitter(conn->etcp) / 10; |
||||
|
||||
/* TODO: получение счетчика ретрансмиссий из ETCP */ |
||||
} |
||||
@ -0,0 +1,118 @@
|
||||
// connection.h - Минималистичный API для защищенных UDP подключений
|
||||
#ifndef CONNECTION_H |
||||
#define CONNECTION_H |
||||
|
||||
#include <stdint.h> |
||||
#include <stddef.h> |
||||
|
||||
#ifdef __cplusplus |
||||
extern "C" { |
||||
#endif |
||||
|
||||
/* Непрозрачный дескриптор подключения */ |
||||
typedef struct conn_handle conn_handle_t; |
||||
|
||||
/* Режим подключения */ |
||||
typedef enum { |
||||
CONN_MODE_CLIENT, /* Инициируем подключение к указанному удаленному адресу */ |
||||
CONN_MODE_SERVER /* Ожидаем входящие подключения */ |
||||
} conn_mode_t; |
||||
|
||||
/* Callback для входящих данных */ |
||||
typedef void (*conn_recv_callback_t)(conn_handle_t* conn,
|
||||
const uint8_t* data,
|
||||
size_t len,
|
||||
void* user_data); |
||||
|
||||
/*
|
||||
* Создание дескриптора подключения (только выделение памяти). |
||||
* Возвращает NULL при ошибке. |
||||
*/ |
||||
conn_handle_t* conn_create(void); |
||||
|
||||
/*
|
||||
* Установка криптографических ключей. |
||||
* Должна быть вызвана до conn_connect(). |
||||
*
|
||||
* @param conn Дескриптор подключения |
||||
* @param my_pub_key Публичный ключ (64 байта для secp256r1), NULL для авто-генерации |
||||
* @param my_priv_key Приватный ключ (32 байта), NULL для авто-генерации |
||||
* @param peer_pub_key Публичный ключ пира (64 байта), NULL для сервера (получит из первого пакета) |
||||
*
|
||||
* @return 0 при успехе, -1 при ошибке |
||||
*/ |
||||
int conn_set_keys(conn_handle_t* conn, |
||||
const uint8_t* my_pub_key, |
||||
const uint8_t* my_priv_key, |
||||
const uint8_t* peer_pub_key); |
||||
|
||||
/*
|
||||
* Подключение к удаленному узлу или начало ожидания входящих подключений. |
||||
*
|
||||
* @param conn Дескриптор подключения |
||||
* @param local_ip Локальный IP для bind (NULL для "0.0.0.0") |
||||
* @param local_port Локальный порт (0 для авто-выбора) |
||||
* @param remote_ip Удаленный IP (NULL для серверного режима) |
||||
* @param remote_port Удаленный порт (игнорируется если remote_ip NULL) |
||||
* @param mode Режим подключения (CONN_MODE_CLIENT/SERVER) |
||||
*
|
||||
* @return 0 при успехе, -1 при ошибке |
||||
*/ |
||||
int conn_connect(conn_handle_t* conn, |
||||
const char* local_ip, |
||||
uint16_t local_port, |
||||
const char* remote_ip, |
||||
uint16_t remote_port, |
||||
conn_mode_t mode); |
||||
|
||||
/*
|
||||
* Установка callback'а для входящих данных. |
||||
* Callback будет вызываться при получении полных собранных пакетов. |
||||
*/ |
||||
void conn_set_recv_callback(conn_handle_t* conn, |
||||
conn_recv_callback_t callback, |
||||
void* user_data); |
||||
|
||||
/*
|
||||
* Отправка данных. |
||||
* Данные будут автоматически фрагментированы, зашифрованы и отправлены. |
||||
*
|
||||
* @return 0 при успехе, -1 при ошибке |
||||
*/ |
||||
int conn_send(conn_handle_t* conn, const uint8_t* data, size_t len); |
||||
|
||||
/*
|
||||
* Закрытие подключения (немедленное, без протокола завершения). |
||||
* После вызова дескриптор можно уничтожить через conn_destroy(). |
||||
*/ |
||||
void conn_close(conn_handle_t* conn); |
||||
|
||||
/*
|
||||
* Полное уничтожение дескриптора подключения и освобождение всех ресурсов. |
||||
* Автоматически вызывает conn_close() если подключение активно. |
||||
*/ |
||||
void conn_destroy(conn_handle_t* conn); |
||||
|
||||
/*
|
||||
* Получение статистики подключения (опционально). |
||||
*
|
||||
* @return 0 при успехе, -1 при ошибке |
||||
*/ |
||||
typedef struct { |
||||
uint64_t bytes_sent; |
||||
uint64_t bytes_received; |
||||
uint32_t packets_sent; |
||||
uint32_t packets_received; |
||||
uint32_t retransmissions; |
||||
uint32_t fragments_assembled; |
||||
uint16_t current_rtt_ms; /* Текущее RTT в миллисекундах */ |
||||
uint16_t jitter_ms; /* Джиттер в миллисекундах */ |
||||
} conn_stats_t; |
||||
|
||||
int conn_get_stats(conn_handle_t* conn, conn_stats_t* stats); |
||||
|
||||
#ifdef __cplusplus |
||||
} |
||||
#endif |
||||
|
||||
#endif /* CONNECTION_H */ |
||||
@ -0,0 +1,179 @@
|
||||
// etcp.h - Extended Transmission Control Protocol |
||||
#ifndef ETCP_H |
||||
#define ETCP_H |
||||
|
||||
#include <stdint.h> |
||||
#include <stddef.h> |
||||
#include "ll_queue.h" |
||||
|
||||
// Debug logging |
||||
#ifdef ETCP_DEBUG |
||||
#include <stdio.h> |
||||
#define ETCP_LOG(fmt, ...) printf("[ETCP] " fmt, ##__VA_ARGS__) |
||||
#else |
||||
#define ETCP_LOG(fmt, ...) ((void)0) |
||||
#endif |
||||
|
||||
#ifdef __cplusplus |
||||
extern "C" { |
||||
#endif |
||||
|
||||
// Forward declarations |
||||
typedef struct epkt epkt_t; |
||||
|
||||
// Callback type for sending packets via UDP |
||||
typedef void (*etcp_tx_callback_t)(epkt_t* epkt, uint8_t* pkt, uint16_t len, void* arg); |
||||
|
||||
// Main ETCP structure |
||||
struct epkt { |
||||
// Queues |
||||
ll_queue_t* tx_queue; // Queue of data to send |
||||
ll_queue_t* output_queue; // Output queue (reassembled data) |
||||
|
||||
// Received packets sorted linked list |
||||
struct rx_packet* rx_list; |
||||
|
||||
// Sent packets (for retransmission) |
||||
struct sent_packet* sent_list; |
||||
|
||||
// Metrics |
||||
uint16_t rtt_last; // Last RTT (timebase 0.1us) |
||||
uint16_t rtt_avg_10; // Average RTT last 10 packets |
||||
uint16_t rtt_avg_100; // Average RTT last 100 packets |
||||
uint16_t jitter; // Jitter (averaged) |
||||
uint16_t bandwidth; // Current bandwidth (bytes per timebase) |
||||
uint32_t bytes_sent_total; // Total bytes sent |
||||
uint16_t last_sent_timestamp; // Timestamp of last sent packet |
||||
uint32_t bytes_allowed; // Calculated bytes allowed to send |
||||
|
||||
// State |
||||
uint16_t next_tx_id; // Next ID for transmission |
||||
uint16_t last_rx_id; // Last received ID (for ACK) |
||||
uint16_t last_delivered_id; // Last delivered to output_queue ID |
||||
|
||||
// Timers |
||||
void* next_tx_timer; // Timer for next transmission |
||||
void* retransmit_timer; // Timer for retransmissions |
||||
|
||||
// Callback |
||||
etcp_tx_callback_t tx_callback; |
||||
void* tx_callback_arg; |
||||
|
||||
// RTT history for averaging |
||||
uint16_t rtt_history[100]; |
||||
uint8_t rtt_history_idx; |
||||
uint8_t rtt_history_count; |
||||
|
||||
// Pending ACKs |
||||
uint16_t pending_ack_ids[32]; |
||||
uint16_t pending_ack_timestamps[32]; |
||||
uint8_t pending_ack_count; |
||||
|
||||
// Pending retransmission requests |
||||
uint16_t pending_retransmit_ids[32]; |
||||
uint8_t pending_retransmit_count; |
||||
|
||||
// Window management |
||||
uint32_t unacked_bytes; // Number of bytes sent but not yet acknowledged |
||||
uint32_t window_size; // Current window size in bytes (calculated) |
||||
uint16_t last_acked_id; // Last acknowledged packet ID |
||||
uint16_t last_rx_ack_id; // Latest received ACK ID from receiver |
||||
uint16_t retrans_timer_period; // Current retransmission timer period (timebase) |
||||
uint16_t next_retrans_time; // Time of next retransmission check |
||||
uint8_t window_blocked; // Flag: transmission blocked by window limit |
||||
}; |
||||
|
||||
// API Functions |
||||
|
||||
/** |
||||
* @brief Initialize new ETCP instance |
||||
* @return Pointer to new instance or NULL on error |
||||
*/ |
||||
epkt_t* etcp_init(void); |
||||
|
||||
/** |
||||
* @brief Free ETCP instance and all associated resources |
||||
* @param epkt Instance to free |
||||
*/ |
||||
void etcp_free(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Set callback for sending packets via UDP |
||||
* @param epkt ETCP instance |
||||
* @param cb Callback function |
||||
* @param arg User argument passed to callback |
||||
*/ |
||||
void etcp_set_callback(epkt_t* epkt, etcp_tx_callback_t cb, void* arg); |
||||
|
||||
/** |
||||
* @brief Process received UDP packet |
||||
* @param epkt ETCP instance |
||||
* @param pkt Packet data |
||||
* @param len Packet length |
||||
* @return 0 on success, -1 on error |
||||
*/ |
||||
int etcp_rx_input(epkt_t* epkt, uint8_t* pkt, uint16_t len); |
||||
|
||||
/** |
||||
* @brief Get total number of packets waiting in transmission queues |
||||
* @param epkt ETCP instance |
||||
* @return Number of packets |
||||
*/ |
||||
int etcp_tx_queue_size(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Put data into transmission queue |
||||
* @param epkt ETCP instance |
||||
* @param data Data to send |
||||
* @param len Data length |
||||
* @return 0 on success, -1 on error |
||||
*/ |
||||
int etcp_tx_put(epkt_t* epkt, uint8_t* data, uint16_t len); |
||||
|
||||
/** |
||||
* @brief Get output queue for reading received data |
||||
* @param epkt ETCP instance |
||||
* @return Pointer to output queue (ll_queue_t*) |
||||
*/ |
||||
ll_queue_t* etcp_get_output_queue(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Set bandwidth limit |
||||
* @param epkt ETCP instance |
||||
* @param bandwidth Bytes per timebase (0.1us) |
||||
*/ |
||||
void etcp_set_bandwidth(epkt_t* epkt, uint16_t bandwidth); |
||||
|
||||
/** |
||||
* @brief Update window size based on current RTT and bandwidth |
||||
* @param epkt ETCP instance |
||||
* Window size = RTT * bandwidth * 2 (bytes in flight) |
||||
*/ |
||||
void etcp_update_window(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Get current RTT |
||||
* @param epkt ETCP instance |
||||
* @return RTT in timebase units |
||||
*/ |
||||
uint16_t etcp_get_rtt(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Get current jitter |
||||
* @param epkt ETCP instance |
||||
* @return Jitter in timebase units |
||||
*/ |
||||
uint16_t etcp_get_jitter(epkt_t* epkt); |
||||
|
||||
/** |
||||
* @brief Reset connection state (clear queues, metrics, timers) |
||||
* @param epkt ETCP instance |
||||
* Note: Keeps bandwidth setting and callback |
||||
*/ |
||||
void etcp_reset(epkt_t* epkt); |
||||
|
||||
#ifdef __cplusplus |
||||
} |
||||
#endif |
||||
|
||||
#endif // ETCP_H |
||||
@ -1,55 +1,282 @@
|
||||
etcp - extended transmission control protocol |
||||
Протокол для передачи-приёма, пободный TCP, реализованый отдельным модулем (etcp.c/h). |
||||
Задача протокола: |
||||
- передать пакеты через UDP (учитывая его особенности - потери, негарантированный порядок), восстанавливая порядок и потери. |
||||
- пакеты уже предварительно подогнаны под размер чтобы вмещались доп. заголовки и служебные фреймы. |
||||
- |
||||
|
||||
|
||||
На приёмной стороне создаются две очереди. |
||||
первая - сортированный linked-list - в нее добавляются принятые пакеты, но отсеиваются дубликаты и вставляются в нужное место. |
||||
и перемещаются в выходную очередь когда все нужные пакеты дошли. |
||||
при получении принятого пакета он сразу парсится, для всех hdr!=0 вызываем static upd_metric_for_transmitter(epkt*, buf*, size) - и он разгребает метрики, обновляя: |
||||
- rtt_last (roud-trip delay) для последнего пакета |
||||
- rtt average last 10 |
||||
- rtt average last 100 |
||||
- jitter как усредненное: jitter+=(abs(rtt_last_10-rtt_last)-jitter)*0.1f |
||||
- для retransmission request - если timestamp последней попытки передачи этого пакета больше rtt_last_10*1.2+jitter*2 то отправляем сейчас |
||||
|
||||
при отправке также ограничиваем полосу пропускания: суммируем сколько байт отправлено всего (uint32_t), обновляем timestamp и расчетное число байт которое может быть отправлено на момент этого timestamp. и формируем таймер для отправки следующего пакета. |
||||
|
||||
вторая - ll_queue - выходная осчередь с пакетами в строгом порядке (строгий инкремент по id). |
||||
|
||||
struct epkt* = etcp_init() - инициализирует новый instance и выделяет под него память |
||||
etcp_free(struct epkt*) |
||||
etcp_rx_input(struct epkt*,uint8_t* pkt, uint16_t len) - принятый пакет на обработку |
||||
rx_output - через механизм ll_queue, функция не нужна. |
||||
|
||||
новый пакет на передачу отправляется в очередь передачи ll_queue. используй callback для обработки очереди. |
||||
для передачи сформированных пакетов по udp: |
||||
etcp_set_callback(epkt*, &cbk) -> etcp_tx_output(struct epkt*,uint8_t* pkt, uint16_t len) - callback в управляющей структуре (отправка пакета в udp сокет) |
||||
|
||||
внутренняя структура передачи: |
||||
при готовности отправить очередной пакет из очереди ll_queue пакеты перемещаются в linked_list (как отправленные но неподтвержденные), и освобождаются при получении подтверждения (ack). |
||||
int etcp_tx_queue_size(epkt*) - должна быть функция которая возвращает общее кол-во пакетов на передачу в очередях (входящей и рабочей) |
||||
|
||||
|
||||
|
||||
Формат udp пакета: |
||||
|
||||
<id> <timestamp> [<hdr> metrics] <hdr=0> <payload> |
||||
|
||||
id - uint16_t циклический порядковый номер пакета (при передаче следующего пакета инкрементируется). при ретрансмиссии передается с этим же id и payload, но с обновленными остальными полями |
||||
timestamp - uint16_t текущий timestamp (циклическое, 16 бит, timebase = 0.1mS) |
||||
hdr - 1 байт: |
||||
|
||||
payload - передаваемые полезные данные |
||||
metrics - опциональное поле для передачи служебных фреймов (ack, retransmission request, statistic reply) |
||||
|
||||
hdr: |
||||
0x00 - hdr для payload (от следующего байта до конца пакета) |
||||
0x01 - hdr для отчета о timestamp (время приёма) принятого пакета, 4 байта: <id> <timestamp> - передаётся при очередной передаче пакета, для формирования статичтики на приёмной стороне. отчеты передаются для всех новых принятых пакетов с момента последней передачи. т.е. накапливаем timestamp-ы и передаём их. если пакет потерялся - не страшно. |
||||
0x10-0x2f - hdr для перезапроса пакетов (передачу каких пакетов надо повторить. значение определяет количество записей (номеров пакетов) от 1 до 32, если больше - 32 самых старых), далее по 2 байта идут ID пакетов. и в конце - 2 байта номер последнего пакета который ушел в выходную очередь (т.е. последний номер для успешно собранной цепочки) |
||||
если что-то еще надо можно добавить. |
||||
|
||||
Если данных нет (очередь на передачу пустая) и нужно передать только метрику, то передаётся пакет с id=0 и без <hdr=0> <payload>. на приёмной стороне он определяется по отсутствию записи с hdr=0 |
||||
ETCP - Extended Transmission Control Protocol |
||||
============================================= |
||||
|
||||
Протокол для надежной передачи данных поверх UDP с восстановлением порядка, |
||||
повторной передачей потерянных пакетов и управлением потоком. |
||||
|
||||
Основные задачи: |
||||
- Передача пакетов через UDP с учетом его особенностей (потери, негарантированный порядок) |
||||
- Восстановление правильного порядка пакетов на приемной стороне |
||||
- Повторная передача потерянных пакетов |
||||
- Управление потоком на основе измерения RTT и пропускной способности |
||||
- Обеспечение прогресса доставки даже при длительных потерях |
||||
|
||||
Архитектура: |
||||
------------ |
||||
|
||||
На передающей стороне: |
||||
1. Очередь передачи (tx_queue) - данные, ожидающие отправки |
||||
2. Список отправленных пакетов (sent_list) - пакеты, ожидающие подтверждения |
||||
3. Таймеры для управления передачей и повторной отправкой |
||||
|
||||
На приемной стороне: |
||||
1. Сортированный список принятых пакетов (rx_list) - пакеты в порядке ID |
||||
2. Выходная очередь (output_queue) - собранные в правильном порядке данные |
||||
3. Буферы для накопления подтверждений и запросов на повторную передачу |
||||
|
||||
Структура данных: |
||||
----------------- |
||||
|
||||
struct epkt { |
||||
// Очереди |
||||
ll_queue_t* tx_queue; // Очередь данных для отправки |
||||
ll_queue_t* output_queue; // Выходная очередь (собранные данные) |
||||
|
||||
// Списки пакетов |
||||
struct rx_packet* rx_list; // Полученные пакеты (отсортированный список) |
||||
struct sent_packet* sent_list; // Отправленные пакеты (для повторной передачи) |
||||
|
||||
// Метрики |
||||
uint16_t rtt_last; // Последнее RTT (в единицах времени 0.1 мс) |
||||
uint16_t rtt_avg_10; // Среднее RTT за последние 10 пакетов |
||||
uint16_t rtt_avg_100; // Среднее RTT за последние 100 пакетов |
||||
uint16_t jitter; // Джиттер (усредненный) |
||||
uint16_t bandwidth; // Текущая пропускная способность (байты за единицу времени) |
||||
uint32_t bytes_sent_total; // Общее количество отправленных байт |
||||
uint16_t last_sent_timestamp; // Временная метка последнего отправленного пакета |
||||
uint32_t bytes_allowed; // Рассчитанное количество разрешенных к отправке байт |
||||
|
||||
// Состояние передачи |
||||
uint16_t next_tx_id; // Следующий ID для передачи |
||||
uint16_t last_sent_id; // Последний отправленный ID (для ретрансмиссии самого нового пакета) |
||||
uint16_t last_rx_id; // Последний полученный ID (для подтверждения) |
||||
uint16_t last_delivered_id; // Последний ID, переданный в output_queue |
||||
|
||||
// Таймеры |
||||
void* next_tx_timer; // Таймер для следующей передачи |
||||
void* retransmit_timer; // Таймер для повторных передач |
||||
|
||||
// Обратный вызов для отправки |
||||
etcp_tx_callback_t tx_callback; |
||||
void* tx_callback_arg; |
||||
|
||||
// История RTT для усреднения |
||||
uint16_t rtt_history[100]; |
||||
uint8_t rtt_history_idx; |
||||
uint8_t rtt_history_count; |
||||
|
||||
// Ожидающие подтверждения |
||||
uint16_t pending_ack_ids[32]; |
||||
uint16_t pending_ack_timestamps[32]; |
||||
uint8_t pending_ack_count; |
||||
|
||||
// Ожидающие запросы на повторную передачу |
||||
uint16_t pending_retransmit_ids[32]; |
||||
uint8_t pending_retransmit_count; |
||||
|
||||
// Управление окном |
||||
uint32_t unacked_bytes; // Количество байт, отправленных но еще не подтвержденных |
||||
uint32_t window_size; // Текущий размер окна в байтах (рассчитывается) |
||||
uint16_t last_acked_id; // Последний подтвержденный ID пакета |
||||
uint16_t last_rx_ack_id; // Последний полученный ID подтверждения от получателя |
||||
uint16_t retrans_timer_period; // Текущий период таймера повторной передачи (в единицах времени) |
||||
uint16_t next_retrans_time; // Время следующей проверки повторной передачи |
||||
uint8_t window_blocked; // Флаг: передача заблокирована из-за ограничения окна |
||||
|
||||
// Отслеживание прогресса доставки |
||||
uint16_t oldest_missing_id; // Самый старый отсутствующий ID пакета |
||||
uint16_t missing_since_time; // Время, когда самый старый отсутствующий пакет был впервые обнаружен |
||||
}; |
||||
|
||||
Формат пакета: |
||||
--------------- |
||||
|
||||
Пакет состоит из обязательного заголовка и опциональных секций: |
||||
|
||||
1. Обязательный заголовок (4 байта): |
||||
- ID пакета (2 байта): циклический порядковый номер (0 для пакетов только с метриками) |
||||
- Timestamp (2 байта): время отправки в единицах 0.1 мс (циклическое, 16 бит) |
||||
|
||||
2. Опциональные секции (одна или несколько, каждая начинается с байта-заголовка): |
||||
|
||||
а) Подтверждения (ACK) - заголовок 0x01: |
||||
[0x01] [count] [(id, timestamp) × count] [last_delivered_id] [last_rx_id] |
||||
- count: количество пар ID+timestamp (1 байт) |
||||
- Для каждого подтверждаемого пакета: ID (2 байта) + timestamp получения (2 байта) |
||||
- last_delivered_id (2 байта): последний ID, доставленный в выходную очередь |
||||
- last_rx_id (2 байта): последний полученный ID (новейший известный пакет) |
||||
|
||||
б) Запросы на повторную передачу - заголовок 0x10-0x2F: |
||||
[0x10 + (count-1)] [IDs × count] [last_delivered_id] [last_rx_id] |
||||
- count: (заголовок & 0x0F) + 1 (от 1 до 32) |
||||
- Для каждого запрашиваемого пакета: ID (2 байта) |
||||
- last_delivered_id (2 байта): последний ID, доставленный в выходную очередь |
||||
- last_rx_id (2 байта): последний полученный ID |
||||
|
||||
в) Полезные данные - заголовок 0x00: |
||||
[0x00] [данные...] |
||||
- Данные произвольной длины (до конца пакета) |
||||
|
||||
Пакет может содержать несколько секций (например, ACK + данные). Секция с полезными данными |
||||
обычно идет последней, если присутствует. |
||||
|
||||
Алгоритмы: |
||||
---------- |
||||
|
||||
1. Управление передачей: |
||||
- Передача происходит при наличии данных в tx_queue и доступной полосы пропускания |
||||
- Полоса пропускания контролируется через bytes_allowed, который накапливается со временем |
||||
- Размер окна рассчитывается как: window_size = RTT × bandwidth × 2 |
||||
- Передача блокируется, если unacked_bytes превышает window_size |
||||
|
||||
2. Подтверждение и RTT измерение: |
||||
- При получении пакета его ID и timestamp добавляются в pending_ack_ids |
||||
- При следующей отправке эти подтверждения включаются в пакет |
||||
- Получатель вычисляет RTT как разницу между текущим временем и полученным timestamp |
||||
- RTT усредняется за последние 10 и 100 пакетов |
||||
|
||||
3. Повторная передача: |
||||
- Отправленные пакеты хранятся в sent_list до подтверждения |
||||
- Таймер retransmit_timer периодически проверяет пакеты старше 1.5×RTT |
||||
- Если пакет не подтвержден, его ID добавляется в pending_retransmit_ids |
||||
- Новейший неподтвержденный пакет (last_sent_id) повторно передается после 2×RTT |
||||
- Запросы на повторную передачу от получателя также обрабатываются |
||||
|
||||
4. Сборка пакетов на приемной стороне: |
||||
- Полученные пакеты вставляются в отсортированный rx_list |
||||
- При обнаружении пропусков (gaps) отправляются запросы на повторную передачу |
||||
- Непрерывная последовательность пакетов перемещается в output_queue |
||||
- last_delivered_id отслеживает последний доставленный ID |
||||
|
||||
5. Обеспечение прогресса доставки (forward progress): |
||||
- Отслеживается самый старый отсутствующий пакет (oldest_missing_id) |
||||
- Если пакет отсутствует дольше 3×RTT (минимум 6 мс), last_delivered_id продвигается вперед |
||||
- Это предотвращает бесконечное ожидание потерянных пакетов |
||||
|
||||
6. Синхронизация состояния: |
||||
- Поля last_delivered_id и last_rx_id передаются в ACK и запросах на повторную передачу |
||||
- Получатель обновляет свой last_delivered_id, если полученное значение новее |
||||
- Это позволяет синхронизировать прогресс доставки между отправителем и получателем |
||||
|
||||
Таймеры: |
||||
-------- |
||||
|
||||
1. Таймер передачи (next_tx_timer): |
||||
- Срабатывает, когда передача невозможна (нет полосы или окно заполнено) |
||||
- Перезапускает процесс передачи |
||||
|
||||
2. Таймер повторной передачи (retransmit_timer): |
||||
- Период: max(RTT/2, 2 мс) |
||||
- Проверяет sent_list на наличие неподтвержденных пакетов старше 1.5×RTT |
||||
- Планирует повторную передачу |
||||
|
||||
API функции: |
||||
------------ |
||||
|
||||
epkt_t* etcp_init(void); |
||||
Инициализирует новый экземпляр ETCP |
||||
|
||||
void etcp_free(epkt_t* epkt); |
||||
Освобождает экземпляр ETCP и все связанные ресурсы |
||||
|
||||
void etcp_set_callback(epkt_t* epkt, etcp_tx_callback_t cb, void* arg); |
||||
Устанавливает обратный вызов для отправки пакетов через UDP |
||||
|
||||
int etcp_rx_input(epkt_t* epkt, uint8_t* pkt, uint16_t len); |
||||
Обрабатывает полученный UDP пакет |
||||
|
||||
int etcp_tx_put(epkt_t* epkt, uint8_t* data, uint16_t len); |
||||
Помещает данные в очередь передачи |
||||
|
||||
ll_queue_t* etcp_get_output_queue(epkt_t* epkt); |
||||
Возвращает выходную очередь для чтения полученных данных |
||||
|
||||
void etcp_set_bandwidth(epkt_t* epkt, uint16_t bandwidth); |
||||
Устанавливает ограничение пропускной способности |
||||
|
||||
int etcp_tx_queue_size(epkt_t* epkt); |
||||
Возвращает общее количество пакетов, ожидающих в очередях передачи |
||||
|
||||
void etcp_reset(epkt_t* epkt); |
||||
Сбрасывает состояние соединения (очищает очереди, метрики, таймеры) |
||||
|
||||
uint16_t etcp_get_rtt(epkt_t* epkt); |
||||
Возвращает текущее RTT |
||||
|
||||
uint16_t etcp_get_jitter(epkt_t* epkt); |
||||
Возвращает текущий джиттер |
||||
|
||||
Внутренние структуры: |
||||
--------------------- |
||||
|
||||
typedef struct rx_packet { |
||||
struct rx_packet* next; |
||||
uint16_t id; |
||||
uint16_t timestamp; |
||||
uint8_t* data; |
||||
uint16_t data_len; |
||||
uint8_t has_payload; |
||||
} rx_packet_t; |
||||
|
||||
typedef struct sent_packet { |
||||
struct sent_packet* next; |
||||
uint16_t id; |
||||
uint16_t timestamp; |
||||
uint8_t* data; |
||||
uint16_t data_len; // Общая длина пакета |
||||
uint16_t payload_len; // Длина полезных данных (для учета окна) |
||||
uint16_t send_time; // Время отправки |
||||
uint8_t need_ack; // Требуется подтверждение |
||||
uint8_t need_retransmit; // Требуется повторная передача |
||||
} sent_packet_t; |
||||
|
||||
Особенности реализации: |
||||
----------------------- |
||||
|
||||
1. Циклические счетчики: |
||||
- ID пакетов: 16-битные, циклические (0-65535, затем 0) |
||||
- Timestamp: 16-битные, циклические (0-65535 единиц по 0.1 мс ≈ 6.55 секунд) |
||||
- Сравнение с учетом цикличности через функцию id_compare() |
||||
|
||||
2. Единицы времени: |
||||
- Базовый интервал: 0.1 мс (100 микросекунд) |
||||
- Все таймеры и измерения RTT используют эту единицу |
||||
|
||||
3. Ограничения: |
||||
- Максимум 32 ожидающих подтверждения или запроса на повторную передачу |
||||
- История RTT хранит до 100 измерений |
||||
- Максимальный размер окна: 2^32-1 байт |
||||
|
||||
4. Обработка дубликатов: |
||||
- При получении пакета с уже существующим ID он игнорируется |
||||
- Повторная передача пакета имеет тот же ID, но новый timestamp |
||||
|
||||
Пример использования: |
||||
--------------------- |
||||
|
||||
1. Инициализация: |
||||
epkt_t* epkt = etcp_init(); |
||||
etcp_set_callback(epkt, udp_send_callback, udp_socket); |
||||
|
||||
2. Отправка данных: |
||||
etcp_tx_put(epkt, data, len); |
||||
|
||||
3. Обработка входящих пакетов: |
||||
etcp_rx_input(epkt, pkt, len); |
||||
|
||||
4. Чтение полученных данных: |
||||
ll_queue_t* output = etcp_get_output_queue(epkt); |
||||
ll_entry_t* entry = queue_entry_get(output); |
||||
// Обработка entry... |
||||
|
||||
5. Освобождение: |
||||
etcp_free(epkt); |
||||
|
||||
Примечания: |
||||
----------- |
||||
|
||||
- Протокол предназначен для работы в условиях умеренных потерь и реордеринга |
||||
- Механизм forward progress обеспечивает доставку даже при потере начальных пакетов |
||||
- Управление окном предотвращает перегрузку сети |
||||
- Поддержка двунаправленной связи (каждая сторона может быть одновременно отправителем и получателем) |
||||
- Минимальные накладные расходы: 4 байта на пакет + опциональные метаданные |
||||
Loading…
Reference in new issue