Evgeny
5840e77e1e
- Fixed pkt_normalizer.c to send packets immediately instead of buffering - Added queue_resume_callback() call in etcp.c after adding to output_queue - Updated test to use simple checksum verification instead of pattern-based - Added strict sequence order checking in test - Reduced MAX_TEST_PACKET_SIZE to 1400 to fit in normalizer fragment - Reduced TOTAL_PACKETS to 10 and TEST_TIMEOUT_MS to 5s for faster testing
174 lines
12 KiB
C
174 lines
12 KiB
C
// 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)
|
||
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
|