Browse Source
- Removed usleep(5000) from all test event loops - Changed to use single shared uasync for server and client instances - Removed uasync_destroy from utun_instance_destroy to prevent double-free - Added explicit uasync_destroy calls in all tests and main program - Fixed segfault in test_pkt_normalizer_etcp and test_etcp_100_packets - Added DEBUG_TRACE to all functions in etcp.c and etcp_connections.c Tests now run without artificial delays and complete successfully.nodeinfo-routing-update
Evgeny
2 months ago
20 changed files with 593 additions and 245 deletions
@ -0,0 +1,174 @@
|
||||
// ll_queue.h - Упрощенная архитектура: разделение создания элементов и работы с очередью |
||||
#ifndef LL_QUEUE_H |
||||
#define LL_QUEUE_H |
||||
|
||||
#include <stddef.h> // для size_t |
||||
#include <stdint.h> // для uint64_t |
||||
#include "memory_pool.h" // для struct memory_pool |
||||
|
||||
// Предварительные объявления |
||||
struct ll_queue; |
||||
struct ll_entry; |
||||
struct queue_waiter; |
||||
struct memory_pool; |
||||
struct UASYNC; |
||||
|
||||
// Автозабор элемента из очереди. вызывается когда в очереди что-то есть и коллбэк не занят обработкой. |
||||
// Параметры: указатель на очередь, указатель на структуру элемента (struct ll_entry*), пользовательский аргумент |
||||
// Когда коллбэк закончит обрабатывать элемент он должен вызвать queue_resume_callback - сообщить о готовности получить следующий элемент (приём очередным вызовом коллбэка в следующем цикле mainloop). |
||||
typedef void (*queue_callback_fn)(struct ll_queue* q, void* arg); |
||||
|
||||
// Структура элемента - переменный размер, данные расположены сразу после структуры |
||||
struct ll_entry { |
||||
struct ll_entry* next; // Указатель на следующий элемент в очереди |
||||
struct ll_entry* prev; // Указатель на предыдущий элемент в очереди |
||||
uint16_t size; // Размер доступной памяти после блока ll_entry - т.е. data[size]. используется для добавления доп. параметров |
||||
|
||||
uint16_t len; // размер пакета (dgram) |
||||
uint16_t memlen; // размер выделенной памяти (dgram) |
||||
uint16_t int_len; // размер (private, not use!) |
||||
uint8_t* dgram; // данные пакета |
||||
void (*dgram_free_fn)(uint8_t* data, void* arg); // функция освобождения блока |
||||
struct memory_pool* dgram_pool; // Пул, из которого выделен этот элемент (NULL, если выделен через malloc) |
||||
|
||||
uint32_t id; // Идентификатор для хеш-поиска |
||||
struct ll_entry* hash_next; // Следующий в хеш-цепочке |
||||
struct memory_pool* pool; // Пул, из которого выделен этот элемент (NULL, если выделен через malloc) |
||||
uint8_t data[0]; |
||||
}; |
||||
|
||||
// Ожидающий добавления элементов в очередь (пока очередь не освободться до нужного размера чтобы не забивать) |
||||
struct queue_waiter { |
||||
int max_packets; // Максимальное количество пакетов |
||||
size_t max_bytes; // Максимальное количество байт (0 = не проверять байты) |
||||
void (*callback)(struct ll_queue* q, void* arg); // Коллбэк для вызова |
||||
void* callback_arg; // Аргумент коллбэка |
||||
}; |
||||
|
||||
typedef void (*queue_threshold_callback_fn)(struct ll_queue* q, void* arg); |
||||
|
||||
// Структура очереди |
||||
struct ll_queue { |
||||
struct ll_entry* head; // Первый элемент (извлекается отсюда) |
||||
struct ll_entry* tail; // Последний элемент (добавляется сюда) |
||||
int count; // Текущее количество элементов |
||||
size_t total_bytes; // Общий размер данных всех элементов (байт) |
||||
int size_limit; // Максимальное количество (-1 = без ограничения) |
||||
|
||||
queue_callback_fn callback; // Функция коллбэка |
||||
void* callback_arg; // Пользовательский аргумент для коллбэка |
||||
int callback_suspended; |
||||
|
||||
void* resume_timeout_id; // ID таймаута uasync для отложенного возобновления |
||||
struct UASYNC* ua; // Экземпляр uasync для таймеров |
||||
|
||||
struct queue_waiter waiter; // Встроенный одиночный waiter |
||||
|
||||
// Хеш-таблица для быстрого поиска по ID |
||||
struct ll_entry** hash_table; |
||||
size_t hash_size; |
||||
}; |
||||
|
||||
// ==================== Управление очередью ==================== |
||||
|
||||
// Создать новую пустую очередь |
||||
// ua - экземпляр uasync для таймеров (обязательный параметр) |
||||
// hash_size - размер хеш-таблицы для быстрого поиска (0 = без хеш-таблицы) |
||||
// Возвращает: указатель на очередь или NULL при ошибке выделения памяти |
||||
struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size); |
||||
|
||||
// Освободить очередь и все её элементы |
||||
// ВАЖНО: освобождает только структуру очереди, элементы в очереди НЕ освобождаются! |
||||
// Элементы должны быть предварительно извлечены через queue_data_get() и освобождены через queue_data_free() |
||||
void queue_free(struct ll_queue* q); |
||||
|
||||
// Установить максимальное количество элементов в очереди |
||||
// При превышении лимита новый элемент автоматически освобождается |
||||
void queue_set_size_limit(struct ll_queue* q, int lim); |
||||
|
||||
// ==================== Асинхронное ожидание передачи ==================== |
||||
|
||||
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь |
||||
// не более max_packets пакетов и не более max_bytes байт (если max_bytes != 0). |
||||
// Если условие уже выполнено, коллбэк вызывается немедленно. |
||||
// Возвращает указатель на waiter для возможной отмены через queue_cancel_wait |
||||
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes, |
||||
queue_threshold_callback_fn callback, void* arg); |
||||
|
||||
// Отменить ожидание (удалить waiter из списка) |
||||
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter); |
||||
|
||||
// Получить общий размер данных в очереди (байт) |
||||
static inline size_t queue_total_bytes(struct ll_queue* q) { |
||||
if (!q) return 0; |
||||
return q->total_bytes; |
||||
} |
||||
|
||||
// ==================== Асинхронное ожидание приёма ==================== |
||||
|
||||
// Установить функцию и аргумент коллбэка для автозабора из очереди |
||||
// Коллбэк вызывается когда в очереди есть элемент и разрешен коллбэк |
||||
// обработчик должен обработать этот пакет (может использовать асинхронное ожидание). Когда будет готов к приёму следующего - должен вызвать resume_callback. обработка строго по одному пакету. |
||||
void queue_set_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg); |
||||
|
||||
// обрабатывается строго одина пакет за вызов |
||||
// Возобновить коллбэки после обработки элемента переданного в коллбэке (тянуть дополнительные элементы из очереди не предусмотернные api нельзя). |
||||
// эта функция должна вызываться всегда после того как cbk_fn обработала пакет (можно с ожиданием через async), иначе очередь застрянет. |
||||
// Если в очереди остались элементы, запланирует вызов коллбэка через uasync_set_timeout(0) |
||||
// Это предотвращает накопление рекурсии в стеке вызовов |
||||
void queue_resume_callback(struct ll_queue* q); |
||||
|
||||
// ==================== Управление элементами ==================== |
||||
|
||||
// выделить ll_entry и память под кодограмму |
||||
struct ll_entry* ll_alloc_lldgram(uint16_t len); |
||||
|
||||
// Создать новый элемент с областью данных указанного размера |
||||
// Память выделяется одним блоком: [struct ll_entry][область данных data_size байт] |
||||
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL при ошибке выделения памяти |
||||
struct ll_entry* queue_entry_new(size_t data_size); |
||||
|
||||
// Создать новый элемент из пула (размер был определен при создании пула) |
||||
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL при ошибке выделения памяти |
||||
struct ll_entry* queue_entry_new_from_pool(struct memory_pool* pool); |
||||
|
||||
// Освободить только entry (не влияет на очереди, dgram не освобождает) |
||||
void queue_entry_free(struct ll_entry* entry); |
||||
|
||||
// Освободить только entry->dgram (не влияет на очереди) |
||||
void queue_dgram_free(struct ll_entry* entry); |
||||
//void queue_data_free(void* data); |
||||
|
||||
// ==================== Операции с очередью ==================== |
||||
|
||||
// Добавить элемент в конец очереди (FIFO) |
||||
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк |
||||
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден) |
||||
int queue_data_put(struct ll_queue* q, struct ll_entry* entry, uint32_t id); |
||||
|
||||
// Добавить элемент в начало очереди (LIFO, высокий приоритет) |
||||
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк |
||||
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден) |
||||
int queue_data_put_first(struct ll_queue* q, struct ll_entry* entry, uint32_t id); |
||||
|
||||
// Извлечь элемент из начала очереди |
||||
// При извлечении приостанавливает коллбэки (callback_suspended = 1) чтобы предотвратить рекурсию |
||||
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL если очередь пуста |
||||
// ПРИМЕЧАНИЕ: не освобождает память элемента |
||||
struct ll_entry* queue_data_get(struct ll_queue* q); |
||||
|
||||
// Получить текущее количество элементов в очереди |
||||
int queue_entry_count(struct ll_queue* q); |
||||
|
||||
// ==================== Поиск и удаление по ID ==================== |
||||
|
||||
// Найти элемент по ID |
||||
// Возвращает: указатель на структуру элемента (struct ll_entry*) или NULL если не найден |
||||
struct ll_entry* queue_find_data_by_id(struct ll_queue* q, uint32_t id); |
||||
|
||||
// Удалить элемент из очереди по указателю на структуру элемента |
||||
// Возвращает: 0 при успехе, -1 если элемент не найден |
||||
// ПРИМЕЧАНИЕ: НЕ изменяет ref_count элемента, просто удаляет из очереди |
||||
int queue_remove_data(struct ll_queue* q, struct ll_entry* entry); |
||||
|
||||
#endif // LL_QUEUE_H |
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Loading…
Reference in new issue