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

ll_queue: fix pointer arithmetic (xxx=0), callback now receives struct ll_entry* 

- Change xxx from 1 to 0, fixing pointer arithmetic in queue_resume_timeout_cb
- Update comments: callback receives struct ll_entry* not user data
- Rename payload field to data in struct ll_entry
- etcp: fix INFLIGHT_PACKET to ACK_PACKET type in etcp_conn_input
- debug: remove excessive DEBUG_ERROR/DEBUG_DEBUG messages
- tests: rewrite test_ll_queue.c for new architecture
nodeinfo-routing-update
Evgeny 8 months ago
parent
c80f26ff28
commit
edba76c4ed
9 changed files with 242 additions and 1097 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      lib/ll_queue.c
  2. 16
      lib/ll_queue.h
  3. 16
      lib/u_async.c
  4. 22
      src/etcp.c
  5. 15
      src/etcp_connections.c
  6. 4
      src/etcp_loadbalancer.c
  7. 6
      src/utun_instance.c
  8. 2
      tests/Makefile.am
  9. 1254
      tests/test_ll_queue.c

4
lib/ll_queue.c
Unescape View File

@ -13,7 +13,7 @@ static void check_waiters(struct ll_queue* q);
static void add_to_hash(struct ll_queue* q, struct ll_entry* entry);
static void remove_from_hash(struct ll_queue* q, struct ll_entry* entry);
#define xxx 1
#define xxx 0
// Вспомогательная функция для преобразования данных в запись
static inline struct ll_entry* data_to_entry(void* data) {
@ -86,7 +86,7 @@ static void queue_resume_timeout_cb(void* arg) {
// Вызвать коллбэк если есть элементы и коллбэки разрешены
if (q->head && !q->callback_suspended && q->callback) {
void* data = (void*)(q->head + 1);
void* data = (void*)(q->head + xxx);
q->callback(q, data, q->callback_arg);
}
}

16
lib/ll_queue.h
Unescape View File

@ -14,7 +14,7 @@ struct memory_pool;
struct UASYNC;
// Автозабор элемента из очереди. вызывается когда в очереди что-то есть и коллбэк не занят обработкой.
// Параметры: указатель на очередь, указатель на пользовательские данные, пользовательский аргумент
// Параметры: указатель на очередь, указатель на структуру элемента (struct ll_entry*), пользовательский аргумент
// Когда коллбэк закончит обрабатывать элемент он должен вызвать queue_resume_callback - сообщить о готовности получить следующий элемент (приём очередным вызовом коллбэка в следующем цикле mainloop).
typedef void (*queue_callback_fn)(struct ll_queue* q, void* data, void* arg);
@ -27,7 +27,7 @@ struct ll_entry {
uint32_t id; // Идентификатор для хеш-поиска
struct ll_entry* hash_next; // Следующий в хеш-цепочке
struct memory_pool* pool; // Пул, из которого выделен этот элемент (NULL, если выделен через malloc)
uint8_t payload[0];
uint8_t data[0];
};
// Ожидающий добавления элементов в очередь (пока очередь не освободться до нужного размера чтобы не забивать)
@ -96,12 +96,12 @@ void queue_set_size_limit(struct ll_queue* q, int lim);
// Создать новый элемент с областью данных указанного размера
// Память выделяется одним блоком: [struct ll_entry][область данных data_size байт]
// Возвращает: указатель на пользовательские данные или NULL при ошибке выделения памяти
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL при ошибке выделения памяти
// ПРИМЕЧАНИЕ: ref_count = 1, элемент не в очереди
void* queue_data_new(size_t data_size);
// Создать новый элемент из пула (размер был определен при создании пула)
// Возвращает: указатель на пользовательские данные или NULL при ошибке выделения памяти
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL при ошибке выделения памяти
// ПРИМЕЧАНИЕ: ref_count = 1, элемент не в очереди
void* queue_data_new_from_pool(struct memory_pool* pool);
@ -125,7 +125,7 @@ int queue_data_put_first(struct ll_queue* q, void* data, uint32_t id);
// Извлечь элемент из начала очереди
// При извлечении приостанавливает коллбэки (callback_suspended = 1) чтобы предотвратить рекурсию
// Возвращает: указатель на пользовательские данные или NULL если очередь пуста
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL если очередь пуста
// ПРИМЕЧАНИЕ: НЕ изменяет ref_count элемента, просто удаляет из очереди
void* queue_data_get(struct ll_queue* q);
@ -155,12 +155,12 @@ static inline size_t queue_total_bytes(struct ll_queue* q) {
// ==================== Поиск и удаление по ID ====================
// Найти элемент по ID
// Возвращает: указатель на пользовательские данные или NULL если не найден
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL если не найден
void* queue_find_data_by_id(struct ll_queue* q, uint32_t id);
// Удалить элемент из очереди по указателю на данные
// Удалить элемент из очереди по указателю на структуру элемента
// Возвращает: 0 при успехе, -1 если элемент не найден
// ПРИМЕЧАНИЕ: НЕ изменяет ref_count элемента, просто удаляет из очереди
int queue_remove_data(struct ll_queue* q, void* data);
#endif // LL_QUEUE_H
#endif // LL_QUEUE_H

16
lib/u_async.c
Unescape View File

@ -292,8 +292,7 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
if (!ua || timeout_tb < 0 || !callback) return NULL;
if (!ua->timeout_heap) return NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: ua=%p, timeout=%d tb, arg=%p, callback=%p",
ua, timeout_tb, arg, callback);
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: timeout=%d.%d ms, arg=%p, callback=%p", timeout_tb/10, timeout_tb%10, arg, callback);
struct timeout_node* node = malloc(sizeof(struct timeout_node));
if (!node) {
@ -302,9 +301,6 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
}
ua->timer_alloc_count++;
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: allocated node %p (alloc_count=%zu)",
node, ua->timer_alloc_count);
node->arg = arg;
node->callback = callback;
node->ua = ua;
@ -316,9 +312,6 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
timeval_add_tb(&now, timeout_tb);
node->expiration_ms = timeval_to_ms(&now);
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: node %p expires at %llu ms",
node, (unsigned long long)node->expiration_ms);
// Add to heap
if (timeout_heap_push(ua->timeout_heap, node->expiration_ms, node) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: failed to push to heap");
@ -327,7 +320,6 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
return NULL;
}
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: successfully created timer %p", node);
return node;
}
@ -343,9 +335,6 @@ err_t uasync_cancel_timeout(struct UASYNC* ua, void* t_id) {
struct timeout_node* node = (struct timeout_node*)t_id;
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_cancel_timeout: ua=%p, t_id=%p, node=%p, expires=%llu ms",
ua, t_id, node, (unsigned long long)node->expiration_ms);
// Try to cancel from heap first
if (timeout_heap_cancel(ua->timeout_heap, node->expiration_ms, node) == 0) {
// Successfully removed from heap - mark as cancelled and update counter
@ -357,6 +346,9 @@ err_t uasync_cancel_timeout(struct UASYNC* ua, void* t_id) {
return ERR_OK;
}
DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_cancel_timeout: not found in heap: ua=%p, t_id=%p, node=%p, expires=%llu ms",
ua, t_id, node, (unsigned long long)node->expiration_ms);
// If not found in heap, it may have already expired or been invalid
return ERR_FAIL;
}

22
src/etcp.c
Unescape View File

@ -63,10 +63,12 @@ static uint16_t timestamp_diff(uint16_t t1, uint16_t t2) {
struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance) {
if (!instance) return NULL;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create: creating connection for instance %p", instance);
struct ETCP_CONN* etcp = calloc(1, sizeof(struct ETCP_CONN));
if (!etcp) return NULL;
if (!etcp) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create: creating connection failed for instance %p", instance);
return NULL;
}
etcp->instance = instance;
etcp->input_queue = queue_new(instance->ua, 0); // No hash for input_queue
@ -78,11 +80,10 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance) {
etcp->inflight_pool = memory_pool_init(sizeof(struct INFLIGHT_PACKET));
etcp->rx_pool = memory_pool_init(sizeof(struct ETCP_FRAGMENT));
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create: queues created - input:%p output:%p send:%p wait:%x pool:%p",
etcp->input_queue, etcp->output_queue, etcp->input_send_q, etcp->input_wait_ack, etcp->inflight_pool);
if (!etcp->input_queue || !etcp->output_queue || !etcp->input_send_q || !etcp->recv_q || !etcp->ack_q ||
!etcp->input_wait_ack || !etcp->inflight_pool || !etcp->rx_pool) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create: error - closing - input:%p output:%p send:%p wait:%x pool:%p",
etcp->input_queue, etcp->output_queue, etcp->input_send_q, etcp->input_wait_ack, etcp->inflight_pool);
etcp_connection_close(etcp);
return NULL;
}
@ -100,8 +101,8 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance) {
queue_set_callback(etcp->input_send_q, input_send_q_cb, etcp);
etcp->link_ready_for_send_fn = etcp_link_ready_callback;
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp_connection_create: queues created, mtu=%d, window_size=%u, next_tx_id=%u",
etcp->mtu, etcp->window_size, etcp->next_tx_id);
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp_connection_create: queues created, ETCP=%p mtu=%d, window_size=%u, next_tx_id=%u",
etcp, etcp->mtu, etcp->window_size, etcp->next_tx_id);
return etcp;
}
@ -265,7 +266,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) {
// Create INFLIGHT_PACKET
struct INFLIGHT_PACKET* p = memory_pool_alloc(etcp->inflight_pool);
if (!p) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "input_queue_cb: cannot allocate INFLIGHT_PACKET");
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "input_queue_cb: cannot allocate INFLIGHT_PACKET (pool=%p etcp=%p)", etcp->inflight_pool, etcp);
queue_data_free(in_pkt); // Free the ETCP_FRAGMENT
queue_resume_callback(q);
return;
@ -555,10 +556,11 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
if (len>=5) {
// формируем ACK
struct INFLIGHT_PACKET* p = memory_pool_alloc(etcp->instance->ack_pool);
struct ACK_PACKET* p = memory_pool_alloc(etcp->instance->ack_pool);
uint32_t seq=data[1] | (data[2]<<8) | (data[3]<<16) | (data[4]<<24);
p->seq=seq;
p->last_timestamp=pkt->timestamp;
p->pkt_timestamp=pkt->timestamp;
p->recv_timestamp=get_current_timestamp();
queue_data_put(etcp->ack_q, p, p->seq);
if ((int32_t)(etcp->last_delivered_id-seq)<0) if (queue_find_data_by_id(etcp->recv_q, seq)==NULL) {// проверяем есть ли пакет с этим seq

15
src/etcp_connections.c
Unescape View File

@ -756,21 +756,16 @@ int init_connections(struct UTUN_INSTANCE* instance) {
// Initialize clients - create outgoing connections
struct CFG_CLIENT* client = config->clients;
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "init_connections: START - client list = %p, first client name = %s",
client, client ? client->name : "NULL");
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "init_connections called, instance=%p, config=%p, clients=%p, connections_count=%d",
instance, config, config ? config->clients : NULL, instance ? instance->connections_count : -1);
while (client) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Processing client: %s", client->name);
// Check if client has required configuration
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Client %s - keepalive=%d, links=%p, peer_key_len=%zu",
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "Client %s - keepalive=%d, links=%p, peer_key_len=%zu",
client->name, client->keepalive, client->links,
strlen(client->peer_public_key_hex));
// Create ETCP connection for this client
struct ETCP_CONN* etcp_conn = etcp_connection_create(instance);
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create returned: %p", etcp_conn);
if (!etcp_conn) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Failed to create ETCP connection for client %s", client->name);
@ -778,8 +773,6 @@ int init_connections(struct UTUN_INSTANCE* instance) {
continue;
}
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Successfully created connection %p for client %s", etcp_conn, client->name);
// Initialize crypto context for this connection
if (sc_init_ctx(&etcp_conn->crypto_ctx, &instance->my_keys) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "init_connections: failed to initialize crypto context for client %s", client->name);
@ -920,15 +913,11 @@ int init_connections(struct UTUN_INSTANCE* instance) {
client_link = client_link->next;
}
// Add connection to instance's connection list only after all validations pass
// This prevents double-free when cleanup tries to free connections that were freed during failed initialization
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Adding connection %p to instance list", etcp_conn);
etcp_conn->next = instance->connections;
instance->connections = etcp_conn;
instance->connections_count++;
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Added connection %p to instance, total count: %d", etcp_conn, instance->connections_count);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "Added connection %p to instance, total count: %d", etcp_conn, instance->connections_count);
client = client->next;
}

4
src/etcp_loadbalancer.c
Unescape View File

@ -25,8 +25,6 @@ static void shaper_timer_cb(void* arg); // Shaper wait callback
// Select link for transmission
struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp) {
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp_loadbalancer_select_link: etcp=%p", etcp);
if (!etcp || !etcp->links) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_loadbalancer_select_link: invalid parameters (etcp=%p, links=%p)",
etcp, etcp ? etcp->links : NULL);
@ -37,8 +35,6 @@ struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp) {
uint64_t min_load_tb = UINT64_MAX;
uint64_t now_tb=get_current_time_units();
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp_loadbalancer_select_link: current tb=%llu, scanning links", (unsigned long long)now_tb);
struct ETCP_LINK* link = etcp->links;
int link_index = 0;
while (link) {

6
src/utun_instance.c
Unescape View File

@ -36,7 +36,6 @@ void utun_instance_set_tun_init_enabled(int enabled) {
// Create and initialize root instance
struct UTUN_INSTANCE* utun_instance_create(struct UASYNC* ua, const char *config_file) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMORY, "blablabla 11");
struct UTUN_INSTANCE *instance = calloc(1, sizeof(struct UTUN_INSTANCE));
if (!instance) return NULL;
@ -47,7 +46,6 @@ struct UTUN_INSTANCE* utun_instance_create(struct UASYNC* ua, const char *config
// Note: Global debug system is initialized from command line arguments if provided
// If not initialized via command line, instance-specific logging can be set up here
// The first initialization wins - either global (from main) or instance-specific
DEBUG_ERROR(DEBUG_CATEGORY_MEMORY, "blablabla 11");
// Ensure keys and node_id exist in config
if (config_ensure_keys_and_node_id(config_file) < 0) {
@ -56,7 +54,6 @@ struct UTUN_INSTANCE* utun_instance_create(struct UASYNC* ua, const char *config
return NULL;
}
DEBUG_ERROR(DEBUG_CATEGORY_MEMORY, "blablabla 12");
// Load configuration
instance->config = parse_config(config_file);
@ -78,9 +75,6 @@ struct UTUN_INSTANCE* utun_instance_create(struct UASYNC* ua, const char *config
DEBUG_ERROR(DEBUG_CATEGORY_MEMORY, "Failed to initialize local keys");
}
DEBUG_ERROR(DEBUG_CATEGORY_MEMORY, "blablabla 2");
instance->ack_pool=memory_pool_init(sizeof(struct ACK_PACKET));
instance->data_pool=memory_pool_init(PACKET_DATA_SIZE);
instance->pkt_pool=memory_pool_init(sizeof(struct ETCP_DGRAM) + PACKET_DATA_SIZE);

2
tests/Makefile.am
Unescape View File

@ -41,7 +41,7 @@ test_crypto_SOURCES = test_crypto.c
test_crypto_CFLAGS = -I$(top_srcdir)/tinycrypt/lib/include -I$(top_srcdir)/tinycrypt/lib/source -I$(top_srcdir)/lib
test_crypto_LDADD = $(top_builddir)/tinycrypt/lib/source/utun-aes_encrypt.o $(top_builddir)/tinycrypt/lib/source/utun-aes_decrypt.o $(top_builddir)/tinycrypt/lib/source/utun-ccm_mode.o $(top_builddir)/tinycrypt/lib/source/utun-cmac_mode.o $(top_builddir)/tinycrypt/lib/source/utun-ctr_mode.o $(top_builddir)/tinycrypt/lib/source/utun-sha256.o $(top_builddir)/tinycrypt/lib/source/utun-utils.o $(top_builddir)/lib/libuasync.a -lpthread -lcrypto
# New comprehensive ll_queue test with current API
# Working ll_queue test for xxx=0 architecture
test_ll_queue_SOURCES = test_ll_queue.c
test_ll_queue_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_ll_queue_LDADD = $(top_builddir)/lib/libuasync.a -lpthread -lcrypto

1254
tests/test_ll_queue.c
View File

File diff suppressed because it is too large Load Diff
Write Preview
Loading…
Cancel
Save