diff --git a/doc/etcp_arch.md b/doc/etcp_arch.md new file mode 100755 index 00000000..5788e6fe --- /dev/null +++ b/doc/etcp_arch.md @@ -0,0 +1,17 @@ +Архитектура etcp: + +## etcp_connections: +- обслуживает encrypt/decrypt и установку защищенного подключения. +- для одного etcp соединения может испольоваться несколько подключений одновременно (load balancing / filover) + +## etcp_loadbalancer: +- выбирает через какой маршрут отправить пакет +- ограничивает rate отправки чтобы не забивать очереди каналов + +## etcp: +- обеспечивает ретрансмиссии при передаче и сборку в правильной последовательности при приёме + +## etcp_metric: +- обеспечивает обновление метрик каналов etcp_connections +- вызывается при получении ack + diff --git a/doc/etcp_protocol.txt b/doc/etcp_protocol.txt index c96f7eb9..fcd84a68 100644 --- a/doc/etcp_protocol.txt +++ b/doc/etcp_protocol.txt @@ -105,12 +105,14 @@ bandwidth по каждому линку адаптивно подстраива **** Формат кодограмм для etcp_connections.c/h **** Кодограммы с этими секциями обрабатываются в etcp_connections (в этих кодограммах всегда только одна секция). в обязательном заголовке ID не используется, при передаче для порядка =0: [СПОРНО: заголовок в etcp — только 2 байта TS; здесь подразумевается ID? Уточнить единый формат] 1) Init запрос - заголовок 0x02 (со сбросом etcp сессии) или 0x04 (без сброса): - [0x02/0x04] [my_node_id 64bit] [my mtu high] [my mtu low] [keepalive high] [keepalive low] [my public key (64 байта, не шифруется)] + [0x02/0x04] [my_node_id 64bit] [my mtu high] [my mtu low] [keepalive high] [keepalive low] [my link_id 1 байт] [my public key (64 байта, не шифруется)] - Инициирует новый connection для tcp instance. если tcp instance нет (первое подключение) - создаёт. Между нодами только одно подключение, но можно добавлять каналы. + - link_id: локальный идентификатор канала (0-255), назначается отправителем для идентификации канала - Публичный ключ отправляется в конце пакета без шифрования, чтобы получатель мог установить его и расшифровать остальную часть пакета (т.к. инициатор соединения всегда имеет оригинальный peer public key в конфиге - по нему исключаем MITM) 2) Init подтверждение - заголовок 0x03/0x05 (ответы соответственно на коды 0x02 и 0x04): - [0x03/0x05] [my_node_id 64bit] [my mtu high] [my mtu low] + [0x03/0x05] [my_node_id 64bit] [my mtu high] [my mtu low] [my link_id 1 байт] - Подтверждение инициализации (канал успешно создан, можно начинать обмен) + - link_id: локальный идентификатор канала (0-255) отправителя ответа При получении init получатель пакета должен: - reset ETCP_LINK с этим ip_port если он есть - создать новый ETCP_CONN с этим node_id. если уже существует подключение с этим node_id - вызвать etcp_reset (функция сброса окон неподтвержденных данных и нумерации) diff --git a/src/etcp.c b/src/etcp.c index b47ba8cd..101a086f 100644 --- a/src/etcp.c +++ b/src/etcp.c @@ -627,8 +627,6 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) { delivered_count++; DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "etcp_output_try_assembly: moved packet id=%u to output_queue", next_expected_id); - // Resume callback to notify listeners (e.g., pkt_normalizer) - queue_resume_callback(etcp->output_queue); } else { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_output_try_assembly: failed to add packet id=%u to output_queue", next_expected_id); diff --git a/src/etcp_connections.c b/src/etcp_connections.c index 45cf9406..23e1d4ed 100644 --- a/src/etcp_connections.c +++ b/src/etcp_connections.c @@ -66,7 +66,9 @@ static void etcp_link_send_init(struct ETCP_LINK* link) { dgram->data[offset++] = (link->keepalive_interval >> 8) & 0xFF; dgram->data[offset++] = link->keepalive_interval & 0xFF; - + + dgram->data[offset++] = link->local_link_id; + memcpy(dgram->data + offset, link->etcp->instance->my_keys.public_key, SC_PUBKEY_SIZE); dgram->data_len = offset + SC_PUBKEY_SIZE; @@ -201,13 +203,44 @@ static void remove_link(struct ETCP_SOCKET* e_sock, uint32_t hash) { // надо править, используй sockaddr_hash struct ETCP_LINK* etcp_link_find_by_addr(struct ETCP_SOCKET* e_sock, struct sockaddr_storage* addr) { if (!e_sock || !addr) return NULL; - + int idx = find_link_index(e_sock, sockaddr_hash(addr)); if (idx < 0) return NULL; - + return e_sock->links[idx]; } +int etcp_find_free_local_link_id(struct ETCP_CONN* etcp) { + if (!etcp) return -1; + + // Битовый массив для 256 id (32 байта * 8 бит = 256) + uint8_t used_ids[32] = {0}; + + // Помечаем занятые id + struct ETCP_LINK* link = etcp->links; + while (link) { + if (link->local_link_id < 256) { + used_ids[link->local_link_id >> 3] |= (1 << (link->local_link_id & 7)); + } + link = link->next; + } + + // Ищем первый свободный id + for (int i = 0; i < 32; i++) { + if (used_ids[i] != 0xFF) { + // Есть свободные биты в этом байте + for (int bit = 0; bit < 8; bit++) { + if (!(used_ids[i] & (1 << bit))) { + return (i << 3) + bit; + } + } + } + } + + // Все id заняты + return -1; +} + // =============================== @@ -344,10 +377,10 @@ void etcp_socket_remove(struct ETCP_SOCKET* conn) { struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn, struct sockaddr_storage* remote_addr, uint8_t is_server) { if (!remote_addr) return NULL; - + struct ETCP_LINK* link = calloc(1, sizeof(struct ETCP_LINK)); if (!link) return NULL; - + link->conn = conn; link->etcp = etcp; link->is_server = is_server; @@ -358,17 +391,26 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn link->init_timeout = 0; link->init_retry_count = 0; + // Выделяем свободный local_link_id + int free_id = etcp_find_free_local_link_id(etcp); + if (free_id < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_link_new: no free local_link_id available"); + free(link); + return NULL; + } + link->local_link_id = (uint8_t)free_id; + memcpy(&link->remote_addr, remote_addr, sizeof(struct sockaddr_storage)); - link->last_activity = time(NULL); - +// link->last_activity = time(NULL); + link->ip_port_hash = sockaddr_hash(remote_addr); insert_link(conn, link); - + struct ETCP_LINK* l=etcp->links; while (l && l->next) l=l->next; if (l) l->next = link; else etcp->links = link; - + if (is_server == 0) { DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp_link_new: client link, calling etcp_link_send_init"); etcp_link_send_init(link); @@ -524,6 +566,7 @@ static void etcp_connections_read_callback(int fd, void* arg) { uint8_t id[8]; uint8_t mtu[2]; uint8_t keepalive[2]; + uint8_t link_id; uint8_t pubkey[SC_PUBKEY_SIZE]; } *ack_hdr=(void*)&pkt->data[0]; uint64_t peer_id; @@ -557,22 +600,25 @@ static void etcp_connections_read_callback(int fd, void* arg) { } struct ETCP_LINK* link = etcp_link_new(conn, e_sock, &addr, 1); if (!link) { if (new_conn) etcp_connection_close(conn); errorcode=66; DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_connections_read_callback: failed to create link for connection"); goto ec_fr; }// облом + link->remote_link_id = ack_hdr->link_id; if (ack_hdr->code==0x02) etcp_conn_reset(conn); struct { uint8_t code; uint8_t id[8]; uint8_t mtu[2]; + uint8_t link_id; } *ack_repl_hdr=(void*)&pkt->data[0]; ack_repl_hdr->code+=1; memcpy(ack_repl_hdr->id, &e_sock->instance->node_id, 8); int mtu=e_sock->instance->config->global.mtu; ack_repl_hdr->mtu[0]=mtu>>8; ack_repl_hdr->mtu[1]=mtu; + ack_repl_hdr->link_id = link->local_link_id; pkt->data_len=sizeof(*ack_repl_hdr); pkt->noencrypt_len=0; pkt->link=link; - DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "Sending INIT RESPONSE, link=%p", link); + DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "Sending INIT RESPONSE, link=%p, local_link_id=%d, remote_link_id=%d", link, link->local_link_id, link->remote_link_id); DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[ETCP DEBUG] Send INIT RESPONSE"); etcp_encrypt_send(pkt); // printf("[ETCP DEBUG] Send INIT RESPONSE ok\n"); @@ -598,6 +644,7 @@ static void etcp_connections_read_callback(int fd, void* arg) { link->last_recv_local_time=get_current_time_units(); link->last_recv_timestamp=pkt->timestamp; + link->last_recv_updated=1; size_t offset = 0; uint8_t code = pkt->data[offset++]; @@ -610,17 +657,18 @@ static void etcp_connections_read_callback(int fd, void* arg) { server_node_id = (server_node_id << 8) | pkt->data[offset++]; } link->mtu = (pkt->data[offset++] << 8) | pkt->data[offset++]; + link->remote_link_id = pkt->data[offset++]; if (offset > pkt_len) { errorcode=13; DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_connections_read_callback: packet parsing overflow, offset=%zu, pkt_len=%zu", offset, pkt_len); goto ec_fr; } - + // DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "Received INIT_RESPONSE from server_node_id=%llu, mtu=%d", (unsigned long long)server_node_id, link->mtu); - + link->etcp->peer_node_id = server_node_id; // If not set - + // Mark link as initialized // DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "Setting link->initialized=1, link=%p, is_server=%d", link, link->is_server); link->initialized = 1;// получен init response (client) loadbalancer_link_ready(link); - DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp client: Link initialized successfully! Server node_id=%llu, mtu=%d", (unsigned long long)server_node_id, link->mtu); + DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp client: Link initialized successfully! Server node_id=%llu, mtu=%d, local_link_id=%d, remote_link_id=%d", (unsigned long long)server_node_id, link->mtu, link->local_link_id, link->remote_link_id); // Cancel init timer if exists if (link->init_timer) { diff --git a/src/etcp_connections.h b/src/etcp_connections.h index f1d07ca9..e13cd66e 100644 --- a/src/etcp_connections.h +++ b/src/etcp_connections.h @@ -21,7 +21,7 @@ struct ETCP_DGRAM {// пакет (незашифрованный) struct ETCP_LINK* link;// откуда получена или куда отправялем uint16_t data_len;// общий размер пакета не включая timestamp uint16_t noencrypt_len;// число байт (с конца) которые не надо шифровать. для передачи pubkey - uint16_t timestamp;// timestamp принятого или для отправки + uint16_t timestamp;// timestamp отправляющего узла при отправке пакета uint8_t data[0];// данные пакета (без timestamp) }; @@ -58,15 +58,18 @@ struct ETCP_LINK { uint16_t keepalive_interval; // Keepalive интервал uint8_t is_server; // инициирует подключение клиент uint8_t initialized; // Флаг инициализации (1=подтверждено или получен request) + uint8_t local_link_id; // id моего линка + uint8_t remote_link_id; // id этого линка на peer (устанавливается в момент initialized) // Состояние установки соединения (только для клиентов) void* init_timer; // Таймер для повторов INIT (NULL=не подключается) uint16_t init_timeout; // Текущий таймаут в мс uint16_t init_retry_count; // Счетчик попыток - uint64_t last_activity; // Время последней активности +// uint64_t last_activity; // Время последней активности uint64_t last_recv_local_time; uint16_t last_recv_timestamp; + uint8_t last_recv_updated; // =1 при обновлении timestamp, =0 при отправке (чтобы не дублировать отправки при отсутствии обновлений) uint64_t shaper_load_time_tb; // основной в 0.1 мс units uint64_t shaper_sub_nanotime; // sub 0-999999 для 0.1 нс - 0.1 мс @@ -101,4 +104,9 @@ int etcp_encrypt_send(struct ETCP_DGRAM* dgram);// зашифровывает и // find link by address struct ETCP_LINK* etcp_link_find_by_addr(struct ETCP_SOCKET* e_sock, struct sockaddr_storage* addr); +// find free local_link_id for connection +// scans all links in connection, marks used ids in bit array +// returns first free id (0-255) or -1 if all occupied +int etcp_find_free_local_link_id(struct ETCP_CONN* etcp); + #endif // ETCP_CONNECTIONS_H diff --git a/tests/test_etcp_link_id b/tests/test_etcp_link_id new file mode 100755 index 00000000..502611e6 Binary files /dev/null and b/tests/test_etcp_link_id differ diff --git a/tests/test_etcp_link_id.c b/tests/test_etcp_link_id.c new file mode 100644 index 00000000..fcc226b3 --- /dev/null +++ b/tests/test_etcp_link_id.c @@ -0,0 +1,279 @@ +// test_etcp_link_id.c - Unit test for etcp_find_free_local_link_id function +// Tests: empty connection, full allocation, random deletion/addition cycles + +#include "../src/etcp_connections.h" +#include "../src/etcp.h" +#include +#include +#include +#include +#include + +// Mock ETCP_LINK structure for testing (minimal version) +struct TEST_LINK { + struct TEST_LINK* next; + uint8_t local_link_id; +}; + +// Helper function to create a mock ETCP_CONN with empty links list +static struct ETCP_CONN* create_test_conn(void) { + struct ETCP_CONN* conn = calloc(1, sizeof(struct ETCP_CONN)); + if (!conn) { + fprintf(stderr, "Failed to allocate test connection\n"); + exit(1); + } + conn->links = NULL; + return conn; +} + +// Helper function to add a link with specific id to connection +static void add_link_with_id(struct ETCP_CONN* conn, uint8_t id) { + struct ETCP_LINK* link = calloc(1, sizeof(struct ETCP_LINK)); + if (!link) { + fprintf(stderr, "Failed to allocate test link\n"); + exit(1); + } + link->local_link_id = id; + link->next = conn->links; + conn->links = link; +} + +// Helper function to remove link with specific id +static int remove_link_with_id(struct ETCP_CONN* conn, uint8_t id) { + struct ETCP_LINK** pp = &conn->links; + while (*pp) { + if ((*pp)->local_link_id == id) { + struct ETCP_LINK* to_remove = *pp; + *pp = (*pp)->next; + free(to_remove); + return 1; // Success + } + pp = &(*pp)->next; + } + return 0; // Not found +} + +// Helper function to count links +static int count_links(struct ETCP_CONN* conn) { + int count = 0; + struct ETCP_LINK* link = conn->links; + while (link) { + count++; + link = link->next; + } + return count; +} + +// Helper function to free all links +static void free_all_links(struct ETCP_CONN* conn) { + struct ETCP_LINK* link = conn->links; + while (link) { + struct ETCP_LINK* next = link->next; + free(link); + link = next; + } + conn->links = NULL; +} + +// Test 1: NULL argument should return -1 +static void test_null_argument(void) { + printf("Test 1: NULL argument... "); + int result = etcp_find_free_local_link_id(NULL); + assert(result == -1); + printf("PASSED\n"); +} + +// Test 2: Empty connection should return 0 +static void test_empty_connection(void) { + printf("Test 2: Empty connection... "); + struct ETCP_CONN* conn = create_test_conn(); + int result = etcp_find_free_local_link_id(conn); + assert(result == 0); + free(conn); + printf("PASSED\n"); +} + +// Test 3: Fill all 256, randomly delete and add back (30 cycles) +static void test_random_deletion_addition(void) { + printf("Test 3: Random deletion/addition cycles (30 iterations)...\n"); + + for (int cycle = 0; cycle < 30; cycle++) { + struct ETCP_CONN* conn = create_test_conn(); + uint8_t used[256] = {0}; + + // Fill all 256 ids + for (int i = 0; i < 256; i++) { + add_link_with_id(conn, i); + used[i] = 1; + } + + assert(count_links(conn) == 256); + + // Random number of links to delete (30-100) + int num_to_delete = 30 + (rand() % 71); + int deleted_ids[100]; + int deleted_count = 0; + + // Randomly delete links + while (deleted_count < num_to_delete) { + int id = rand() % 256; + if (used[id] && remove_link_with_id(conn, id)) { + used[id] = 0; + deleted_ids[deleted_count++] = id; + } + } + + assert(count_links(conn) == 256 - num_to_delete); + + // Add back the same number of links + int added_ids[100]; + int added_count = 0; + int prev_id = -1; + + while (added_count < num_to_delete) { + int new_id = etcp_find_free_local_link_id(conn); + + // Verify the id is actually free + assert(new_id >= 0 && new_id < 256); + assert(used[new_id] == 0); + + // Verify ids are returned in ascending order (filling gaps from smallest) + assert(new_id > prev_id); + prev_id = new_id; + + // Add the link + add_link_with_id(conn, new_id); + used[new_id] = 1; + added_ids[added_count++] = new_id; + } + + assert(count_links(conn) == 256); + + // Verify all ids are marked as used + for (int i = 0; i < 256; i++) { + assert(used[i] == 1); + } + + // Verify no duplicates + uint8_t check[256] = {0}; + struct ETCP_LINK* link = conn->links; + while (link) { + assert(check[link->local_link_id] == 0); // No duplicate + check[link->local_link_id] = 1; + link = link->next; + } + + // Cleanup + free_all_links(conn); + free(conn); + + if ((cycle + 1) % 10 == 0) { + printf(" Completed %d cycles...\n", cycle + 1); + } + } + + printf("Test 3: PASSED (all 30 cycles)\n"); +} + +// Test 4: All 256 occupied should return -1 +static void test_all_occupied(void) { + printf("Test 4: All 256 occupied... "); + struct ETCP_CONN* conn = create_test_conn(); + + // Fill all 256 + for (int i = 0; i < 256; i++) { + add_link_with_id(conn, i); + } + + int result = etcp_find_free_local_link_id(conn); + assert(result == -1); + + free_all_links(conn); + free(conn); + printf("PASSED\n"); +} + +// Test 5: Delete specific ids (5, 10, 100), should return 5 +static void test_specific_deletion(void) { + printf("Test 5: Delete ids 5, 10, 100... "); + struct ETCP_CONN* conn = create_test_conn(); + + // Fill all 256 + for (int i = 0; i < 256; i++) { + add_link_with_id(conn, i); + } + + // Delete specific ids + assert(remove_link_with_id(conn, 5)); + assert(remove_link_with_id(conn, 10)); + assert(remove_link_with_id(conn, 100)); + + // Should return 5 (smallest free) + int result = etcp_find_free_local_link_id(conn); + assert(result == 5); + + free_all_links(conn); + free(conn); + printf("PASSED\n"); +} + +// Test 6: Delete id 0, should return 0 +static void test_delete_zero(void) { + printf("Test 6: Delete id 0... "); + struct ETCP_CONN* conn = create_test_conn(); + + // Fill all 256 + for (int i = 0; i < 256; i++) { + add_link_with_id(conn, i); + } + + // Delete id 0 + assert(remove_link_with_id(conn, 0)); + + // Should return 0 + int result = etcp_find_free_local_link_id(conn); + assert(result == 0); + + free_all_links(conn); + free(conn); + printf("PASSED\n"); +} + +// Test 7: Delete all, should return 0 +static void test_delete_all(void) { + printf("Test 7: Delete all... "); + struct ETCP_CONN* conn = create_test_conn(); + + // Fill all 256 + for (int i = 0; i < 256; i++) { + add_link_with_id(conn, i); + } + + // Delete all + free_all_links(conn); + + // Should return 0 + int result = etcp_find_free_local_link_id(conn); + assert(result == 0); + + free(conn); + printf("PASSED\n"); +} + +int main(void) { + printf("=== ETCP Link ID Unit Tests ===\n\n"); + + // Initialize random seed + srand(time(NULL)); + + test_null_argument(); + test_empty_connection(); + test_random_deletion_addition(); + test_all_occupied(); + test_specific_deletion(); + test_delete_zero(); + test_delete_all(); + + printf("\n=== All tests PASSED ===\n"); + return 0; +} diff --git a/tests/test_pkt_normalizer_etcp b/tests/test_pkt_normalizer_etcp index 5943afac..4c911a29 100755 Binary files a/tests/test_pkt_normalizer_etcp and b/tests/test_pkt_normalizer_etcp differ