diff --git a/lib/ll_queue.h b/lib/ll_queue.h index fad9fa71..a98647a1 100644 --- a/lib/ll_queue.h +++ b/lib/ll_queue.h @@ -41,8 +41,8 @@ extern "C" { * @note Важные правила использования автозабора (queue_set_callback): * 1. Коллбэк получает управление только если в очереди есть элементы И !callback_suspended * 2. Внутри коллбэка ОБЯЗАТЕЛЬНО: queue_data_get() → обработать элемент → queue_resume_callback() - * 3. Без вызова queue_resume_callback очередь навсегда застрянет - * 4. Не извлекайте элементы вручную вне коллбэка — это нарушит логику + * 3. Без queue_resume_callback автозабор остаётся приостановленным + * 4. Ручное извлечение допустимо; для продолжения автозабора после него тоже нужен resume * 5. queue_data_get() автоматически suspend'ит коллбэки для предотвращения рекурсии * * @note Память: @@ -61,7 +61,7 @@ struct queue_waiter_handle; * @brief Элемент очереди (переменного размера). * * Память выделяется одним блоком: [struct ll_entry + data[size]]. - * Поле dgram — отдельный блок (malloc / пул / внешняя память). + * Поле dgram — отдельный блок (u_malloc / пул / внешняя память со своим dgram_free_fn). */ struct ll_entry { char* name; @@ -77,7 +77,7 @@ struct ll_entry { struct memory_pool* dgram_pool; // Пул для dgram (если используется) uint32_t index_hash; // FNV-1a (32 бита) по всем index_size байтам data[] от index_offset struct ll_entry* hash_next; // Следующий в хеш-цепочке - struct memory_pool* pool; // Пул, из которого выделен сам entry (NULL = malloc) + struct memory_pool* pool; // Пул, из которого выделен сам entry (NULL = u_malloc) uint8_t data[0]; // Гибкий массив пользовательских данных (размер = size) }; @@ -189,8 +189,8 @@ struct ll_queue { * @param hash_size размер хеш-таблицы (0 = поиск по id отключён) * @param index_offset смещение ключа индекса в data[] каждого entry (0 если hash_size=0) * @param index_size размер ключа индекса (0 если hash_size=0) - * @param name имя очереди для отладки - * @return указатель на очередь или NULL при ошибке выделения памяти + * @param name заимствованное имя для отладки, должно жить до queue_free + * @return указатель на очередь или NULL при NULL ua / ошибке выделения памяти */ struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size, uint16_t index_offset, uint16_t index_size, char* name); @@ -200,6 +200,8 @@ struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size, uint16_t index_o * * @warning НЕ ОСВОБОЖДАЕТ элементы в очереди! * Их нужно извлечь через queue_data_get() и освободить через queue_entry_free(). + * Перед дренажем снять callback и отменить ожидания владельцев, затем для каждого entry освободить dgram и entry. + * queue_free отменяет свои отложенные уведомления и waiters; user_arg не освобождает. NULL допустим. */ void queue_free(struct ll_queue* q); @@ -208,7 +210,7 @@ void queue_free(struct ll_queue* q); /** * @brief Устанавливает максимальное количество элементов в очереди. * @param q очередь - * @param lim лимит (-1 = без ограничения) + * @param lim лимит (любое отрицательное значение = без ограничения) * * При превышении лимита новые элементы автоматически освобождаются. */ @@ -249,11 +251,15 @@ void queue_resume_callback(struct ll_queue* q); * @param max_bytes максимальный объём данных (0 = не проверять) * * Порог — общий на всю очередь. Все waiters в связном списке используют этот порог. - * При достижении порога вызывается следующий waiter из списка (FIFO / round-robin). + * После успешного get/remove при выполнении порога вызывается один следующий waiter из FIFO-списка. + * Сам setter старые ожидания не проверяет. Байт-порог использует int_len, зафиксированный при put. * По умолчанию: max_packets=0, max_bytes=0 (пустая очередь). */ void queue_set_threshold(struct ll_queue* q, int max_packets, size_t max_bytes); +// enable=0 — синхронный waiter (по умолчанию), !=0 — через uasync_call_soon. +// Deferred-уведомление не резервирует место и повторно порог не проверяет: очередь к исполнению может заполниться. +// Повторную регистрацию producer из его callback при свободной очереди откладывать, чтобы избежать рекурсии. void queue_set_waiter_defer(struct ll_queue* q, int enable); /** @@ -282,10 +288,13 @@ void queue_set_callback_defer(struct ll_queue* q, int enable); void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg); /** - * @brief Устанавливает коллбэк вызываемый после каждого queue_data_get (deferred через call_soon). + * @brief Устанавливает коллбэк после каждого успешного queue_data_get (deferred через call_soon). * @param q очередь * @param cbk_fn коллбэк (NULL для отмены) * @param arg пользовательский аргумент + * + * Уведомления подсчитываются, не сливаются в одно; remove их не создаёт. + * NULL отменяет отложенный вызов и обнуляет накопленные уведомления. Замена использует новые функцию/arg. */ void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg); @@ -391,29 +400,29 @@ int queue_entry_count(struct ll_queue* q); int queue_check_consistency(struct ll_queue* q); /** - * @brief Выделяет элемент с отдельным буфером dgram (malloc). + * @brief Выделяет элемент с отдельным буфером dgram через u_malloc (entry->len=0, memlen=len). * @param len требуемый размер dgram * @return элемент или NULL при ошибке */ struct ll_entry* ll_alloc_lldgram(uint16_t len); /** - * @brief Создаёт элемент с пользовательским буфером data[] (malloc одним блоком). - * @param data_size размер data[] + * @brief Создаёт обнулённый элемент с пользовательским буфером data[] (u_malloc одним блоком). + * @param data_size размер data[], должен помещаться в uint16_t * @return элемент или NULL при ошибке */ struct ll_entry* queue_entry_new(size_t data_size); /** * @brief Создаёт элемент из пула памяти. - * @param pool пул + * @param pool пул с object_size >= sizeof(struct ll_entry); разность размеров должна помещаться в uint16_t * @return элемент или NULL при ошибке выделения; пустой кеш пула пополняется автоматически */ struct ll_entry* queue_entry_new_from_pool(struct memory_pool* pool); /** - * @brief Освобождает структуру элемента (возвращает в пул или free). - * @param entry элемент + * @brief Освобождает структуру элемента (возвращает в пул или u_free). + * @param entry извлечённый из очереди элемент; NULL допустим * * @note НЕ освобождает dgram! Используйте queue_dgram_free() отдельно. */ @@ -464,7 +473,7 @@ int queue_remove_data(struct ll_queue* q, struct ll_entry* entry); /** * @brief Возвращает суммарный объём данных в очереди. * @param q очередь - * @return количество байт + * @return сумма int_len (снимки len при put); изменение len связанного entry счётчик не пересчитывает */ static inline size_t queue_total_bytes(struct ll_queue* q) { return q->total_bytes; diff --git a/lib/ll_queue_doc.md b/lib/ll_queue_doc.md deleted file mode 100644 index ea922368..00000000 --- a/lib/ll_queue_doc.md +++ /dev/null @@ -1,235 +0,0 @@ -# ll_queue — Двусвязная FIFO-очередь с автозабором и backpressure - -## 1. Назначение - -Двусвязная очередь элементов (FIFO) с автоматическим вызовом callback при появлении данных, пороговым ожиданием освобождения (backpressure), хеш-индексом для поиска и поддержкой пулов памяти. Используется как основной механизм передачи данных между компонентами внутри одного потока uasyc: TUN ↔ ETCP ↔ routing ↔ normalizer и т.д. - -Работает **строго в одном потоке** (один uasyc). Для многопоточного доступа не предназначена. - -## 2. Как пользоваться - -### 2.1. Создание очереди - -```c -// Без хеша (простая очередь) -struct ll_queue* q = queue_new(ua, 0, 0, 0, "my_queue"); - -// С хешем — для быстрого поиска по ключу внутри data[] -// hash_size=256, ключ лежит в data[] по смещению 0, размер ключа 4 байта -struct ll_queue* q = queue_new(ua, 256, 0, 4, "my_hash_queue"); -``` - -### 2.2. Создание элемента - -Три способа, по убыванию производительности: - -```c -// 1. Из memory pool (самый быстрый) -struct ll_entry* e = queue_entry_new_from_pool(instance->pkt_pool); - -// 2. Через malloc (data[] фиксированного размера одним блоком) -struct ll_entry* e = queue_entry_new(sizeof(struct my_data)); - -// 3. Элемент + отдельный буфер dgram (для переменного размера данных) -struct ll_entry* e = ll_alloc_lldgram(dgram_len); -memcpy(e->dgram, src_data, dgram_len); -e->len = dgram_len; -``` - -### 2.3. Запись в очередь (backpressure) - -**Главное правило: не забиваем очередь!** Добавляем следующий элемент только когда очередь опустела до заданного порога. Для этого используется Пороговое ожидание: - -```c -// 1. Настроить порог (например: ждать пока count<=0, т.е. очередь совсем пуста) -queue_set_threshold(q, 0, 0); - -// 2. В структуре-производителе завести handle: -struct queue_waiter_handle my_waiter = {0}; - -// 3. При готовности отправить — зарегистрировать ожидание: -static void my_send_cb(struct ll_queue* q, void* arg) { - struct my_producer* p = (struct my_producer*)arg; - struct ll_entry* e = ... // создать элемент - e->len = data_len; - queue_data_put_with_index(q, e); - // Снова зарегистрироваться на следующую порцию - queue_waiter_wait(q, &p->waiter, my_send_cb, p); -} - -// Первичная регистрация: -queue_waiter_wait(q, &my_waiter, my_send_cb, producer); - -// 4. При деинициализации — отменить ожидание: -queue_waiter_cancel(q, &my_waiter); -``` - -### 2.4. Чтение из очереди (callback) - -```c -// Установить callback — будет вызываться при появлении элементов -queue_set_callback(q, my_callback, my_data); - -static void my_callback(struct ll_queue* q, void* arg) { - // 1. ИЗВЛЕЧЬ элемент - struct ll_entry* e = queue_data_get(q); - if (!e) { queue_resume_callback(q); return; } - - // 2. ОБРАБОТАТЬ - if (e->dgram && e->len > 0) { - do_something(e->dgram, e->len); - queue_dgram_free(e); // сначала освободить dgram - } - queue_entry_free(e); // потом освободить сам entry - - // 3. ОБЯЗАТЕЛЬНО возобновить callback - queue_resume_callback(q); -} -``` - -**Критически важно:** `queue_data_get()` автоматически приостанавливает callback-и (`callback_suspended = 1`). Без вызова `queue_resume_callback()` очередь **навсегда застрянет** — callback больше никогда не вызовется. - -Во **всех ветках** (включая ранние return по ошибке) должен быть `queue_resume_callback(q)`: - -```c -// ПРАВИЛЬНО: -if (!e) { queue_resume_callback(q); return; } -if (!e->dgram) { queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; } - -// НЕПРАВИЛЬНО — очередь зависнет: -if (!e) return; -``` - -### 2.5. Поиск по индексу - -```c -// Создать очередь с хешем (ключ 4 байта по смещению 0): -struct ll_queue* q = queue_new(ua, 256, 0, 4, "indexed_q"); - -// Добавлять элементы через _with_index (хеш вычисляется автоматически): -queue_data_put_with_index(q, entry); - -// Искать: -uint32_t key = 42; -struct ll_entry* found = queue_find_data_by_index(q, &key); - -// Если несколько элементов с одним ключом — продолжать поиск: -struct ll_entry* next = queue_find_next_by_index(q, &key, found); -``` - -### 2.6. Освобождение ресурсов - -```c -// 1. Снять callback -queue_set_callback(q, NULL, NULL); - -// 2. Отменить всех waiter-ов -queue_waiter_cancel(q, &my_waiter); - -// 3. Извлечь и освободить ВСЕ оставшиеся элементы -struct ll_entry* e; -while ((e = queue_data_get(q)) != NULL) { - queue_dgram_free(e); - queue_entry_free(e); -} -// После извлечения всех элементов resume не нужен — callback уже снят. - -// 4. Освободить очередь -queue_free(q); -``` - -**Важно:** `queue_free()` **не освобождает элементы**. Их нужно извлечь через `queue_data_get()` и освободить вручную. - -### 2.7. Приоритетная вставка (LIFO) - -```c -// Вставить в начало очереди (высокий приоритет): -queue_data_put_first(q, entry); -queue_data_put_first_with_index(q, entry); -``` - -### 2.8. Удаление из середины - -```c -queue_remove_data(q, entry); // удаляет из очереди, НЕ освобождает память -queue_dgram_free(entry); // освободить dgram отдельно -queue_entry_free(entry); // освободить entry отдельно -``` - -После `queue_remove_data()` элемент больше не принадлежит очереди — вызывающий сам отвечает за его освобождение. - -### 2.9. Callback при опустошении - -```c -static void on_empty(struct ll_queue* q, void* arg) { - // Очередь стала пустой (count==0) - // Вызывается однократно, после вызова сбрасывается -} - -queue_set_empty_callback(q, on_empty, my_data); -``` - -## 3. API - -### Создание / уничтожение - -| Функция | Назначение | -|---------|-----------| -| `queue_new(ua, hash_size, index_offset, index_size, name)` | Создать очередь. `hash_size=0` — без хеша. `ua` обязателен (для uasync_call_soon). | -| `queue_free(q)` | Освободить очередь (НЕ элементы — их надо извлечь и освободить отдельно). | - -### Создание / освобождение элементов - -| Функция | Назначение | -|---------|-----------| -| `queue_entry_new(data_size)` | Выделить entry с data[] через malloc. | -| `queue_entry_new_from_pool(pool)` | Выделить entry из memory pool. | -| `ll_alloc_lldgram(len)` | Выделить entry + отдельный буфер dgram (malloc). | -| `queue_entry_free(entry)` | Освободить entry (возврат в pool или free). НЕ освобождает dgram! | -| `queue_dgram_free(entry)` | Освободить dgram (через dgram_free_fn, pool или free). | - -### Чтение (callback) - -| Функция | Назначение | -|---------|-----------| -| `queue_set_callback(q, cbk_fn, arg)` | Установить callback автозабора. Вызывается при наличии элементов. | -| `queue_data_get(q)` | Извлечь элемент из головы. Приостанавливает callback до `queue_resume_callback()`. | -| `queue_resume_callback(q)` | **Обязательно** вызвать после обработки элемента. Планирует отложенный вызов callback через uasync_call_soon. | -| `queue_set_callback_defer(q, enable)` | `enable=1` — первый запуск callback из `put` откладывается через uasync_call_soon (очистка стека). По умолчанию `0` (синхронно). Повторные запуски всегда deferred. | - -### Запись - -| Функция | Назначение | -|---------|-----------| -| `queue_data_put(q, entry)` | Добавить в конец (FIFO). Для очередей БЕЗ хеша. | -| `queue_data_put_with_index(q, entry)` | Добавить в конец с хеш-индексом. | -| `queue_data_put_first(q, entry)` | Добавить в начало (LIFO, высокий приоритет). | -| `queue_data_put_first_with_index(q, entry)` | Добавить в начало с хеш-индексом. | - -### Backpressure - -| Функция | Назначение | -|---------|-----------| -| `queue_set_threshold(q, max_packets, max_bytes)` | Задать общий порог. Когда `count <= max_packets` И `total_bytes <= max_bytes`, waiter-ы пробуждаются (FIFO). | -| `queue_waiter_wait(q, h, callback, arg)` | Зарегистрировать ожидание. Возвращает 1 (уже выполнено), 0 (в очереди), -1 (ошибка). | -| `queue_waiter_cancel(q, h)` | Отменить ожидание. Безопасно вызывать в любом состоянии. | -| `queue_set_waiter_defer(q, enable)` | Отложенный вызов waiter-ов (через uasync_call_soon). | -| `queue_set_empty_callback(q, cbk_fn, arg)` | Одноразовый callback при count==0. | -| `queue_set_on_get(q, cbk_fn, arg)` | Callback после каждого queue_data_get (deferred). | - -### Поиск / удаление - -| Функция | Назначение | -|---------|-----------| -| `queue_find_data_by_index(q, index_key)` | Найти по ключу (требуется hash_size>0). Размер ключа = q->index_size. | -| `queue_find_next_by_index(q, index_key, prev_entry)` | Следующий элемент с тем же ключом. | -| `queue_remove_data(q, entry)` | Удалить из очереди (память НЕ освобождает). | - -### Утилиты - -| Функция | Назначение | -|---------|-----------| -| `queue_entry_count(q)` | Текущее количество элементов. | -| `queue_total_bytes(q)` | Суммарный объём данных (сумма len). | -| `queue_set_size_limit(q, lim)` | Ограничение на количество элементов (при превышении новые удаляются). | -| `queue_check_consistency(q)` | Проверка целостности (счётчики, циклы, prev/next). Только для отладки. |