Fix: ETCP module compilation errors and build system issues

- Added missing structure members to ETCP_CONN (ack_packets_count, last_rx_ack_id, rtt_history, rtt_history_idx, total_packets_sent)
- Added missing bandwidth member to ETCP_LINK structure
- Fixed include path for ll_queue.h header
- Added math library linking (-lm) to build system
- Added etcp_loadbalancer.c to build sources
- Fixed forward declarations and function calls in ETCP modules
- Updated test dependencies to include loadbalancer object files

All simple compilation errors resolved, project builds successfully and tests pass (except TUN permission test which is expected in container).
This commit is contained in:
Evgeny committed 2026-01-23 00:43:48 +03:00
1 parent 3bf178475d
commit e048b10899
35 files changed
+7011 -669

No files matched your search

+464 -17
View File
@@ -1,3 +1,4 @@
// ll_queue.c
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
@@ -16,6 +17,10 @@ static void check_waiters(struct ll_queue* q) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: checking waiters, count=%d, bytes=%zu",
q->count, q->total_bytes);
// Сначала собрать список коллбэков для вызова, чтобы избежать проблем с модификацией списка во время итерации
struct queue_waiter* to_call = NULL;
struct queue_waiter* to_call_tail = NULL;
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* waiter = q->waiters;
@@ -28,15 +33,16 @@ static void check_waiters(struct ll_queue* q) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
waiter->callback(q, waiter->callback_arg);
// Удалить waiter из списка
*pprev = next;
if (q->pool) {
memory_pool_free(q->pool, waiter);
// Добавить в список для вызова
waiter->next = NULL;
if (to_call_tail) {
to_call_tail->next = waiter;
} else {
free(waiter);
to_call = waiter;
}
// pprev уже указывает на правильный следующий элемент
to_call_tail = waiter;
} else {
// Условие не выполнено - оставить в списке
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition NOT met, count=%d>%d or bytes=%zu>%zu (max_bytes_check=%s)",
@@ -46,11 +52,60 @@ static void check_waiters(struct ll_queue* q) {
}
waiter = next;
}
// Теперь вызвать коллбэки
waiter = to_call;
while (waiter) {
struct queue_waiter* next = waiter->next;
waiter->callback(q, waiter->callback_arg);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
}
// ==================== Унифицированная система версий позиций ====================
// Унифицированная проверка версии позиции с автоматическим пересканированием
static int queue_pos_ensure_valid(struct ll_queue_pos* pos) {
if (!pos || !pos->queue) return 0;
// Если позиция уже валидна и версия совпадает, всё хорошо
if (pos->valid && pos->version == pos->queue->version) {
return 1;
}
// Несовпадение версии или позиция невалидна - нужно пересканировать
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "Position version %lu != queue version %lu, rescanning...",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
// Сохранить целевой индекс (где мы хотим быть)
int target_index = pos->index;
// Обновить версию позиции до текущей
pos->version = pos->queue->version;
// Обработать краевой случай: индекс больше не существует из-за изменения размера очереди
if (target_index >= pos->queue->count) {
// Позиция выходит за пределы очереди - инвалидируем её
pos->valid = 0;
return 0;
}
// Простой подход: сбросить и перейти к тому же индексу
// Это найдёт элемент, который теперь находится в целевом индексе
queue_pos_reset(pos);
return queue_pos_seek(pos, target_index);
}
// ==================== Управление очередью ====================
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool) {
if (!ua) return NULL;
struct ll_queue* q = calloc(1, sizeof(struct ll_queue));
if (!q) return NULL;
@@ -66,6 +121,7 @@ struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool) {
q->ua = ua;
q->waiters = NULL;
q->pool=pool;
q->version = 0;
return q;
}
@@ -83,6 +139,7 @@ void queue_free(struct ll_queue* q) {
struct ll_entry* next = entry->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: releasing entry %p (entry %d), ref_count=%d",
entry, entry_count++, entry->ref_count);
entry->ref_count--; // Дополнительно уменьшить за присутствие в очереди
queue_entry_free(entry); // Это уменьшит ref_count и освободит только если ref_count == 0
entry = next;
}
@@ -161,6 +218,7 @@ struct ll_entry* queue_entry_new(size_t data_size) {
if (!entry) return NULL;
entry->next = NULL;
entry->prev = NULL; // Инициализировать prev для двусвязного списка
entry->size = data_size;
entry->ref_count = 1; // Начальный счетчик ссылок
// Область данных оставить неинициализированной для производительности
@@ -211,12 +269,14 @@ int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в хвост (FIFO)
// Добавить в хвост (FIFO) - двусвязный список
entry->next = NULL;
entry->prev = q->tail;
if (q->tail) {
q->tail->next = entry;
} else {
q->head = entry;
q->head = entry; // Очередь была пустой
}
q->tail = entry;
q->count++;
@@ -225,6 +285,9 @@ int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
// Это запускает автоматическую обработку очереди
if (!q->callback_suspended && q->callback) {
@@ -232,7 +295,7 @@ int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
// Приостановить коллбэки во время выполнения коллбэка, чтобы предотвратить рекурсию
q->callback_suspended = 1;
q->callback(q, entry, q->callback_arg);
q->callback(q, q->head, q->callback_arg);
// Не восстанавливать здесь - восстановление происходит через queue_resume_callback
}
@@ -258,22 +321,30 @@ int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry) {
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в голову (LIFO, высокий приоритет)
// Добавить в голову (LIFO, высокий приоритет) - двусвязный список
entry->next = q->head;
q->head = entry;
if (!q->tail) {
q->tail = entry;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: calling callback for entry %p", entry);
q->callback(q, entry, q->callback_arg);
q->callback_suspended = 1;
q->callback(q, q->head, q->callback_arg);
}
// Проверить ожидающие коллбэки
@@ -289,9 +360,12 @@ struct ll_entry* queue_entry_get(struct ll_queue* q) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: retrieving entry %p, size=%zu, ref_count=%d",
entry, entry->size, entry->ref_count);
// Удалить из двусвязного списка
q->head = entry->next;
if (!q->head) {
q->tail = NULL;
if (q->head) {
q->head->prev = NULL;
} else {
q->tail = NULL; // Очередь стала пустой
}
q->count--;
q->total_bytes -= entry->size;
@@ -299,13 +373,17 @@ struct ll_entry* queue_entry_get(struct ll_queue* q) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: removed entry %p, new count=%d, total_bytes=%zu",
entry, q->count, q->total_bytes);
entry->next = NULL; // Отсоединить от очереди
entry->next = NULL;
entry->prev = NULL; // Полностью отсоединить от очереди
// Уменьшить счетчик ссылок при извлечении из очереди
// entry->ref_count был увеличен при добавлении в очередь
// теперь уменьшаем, но не освобождаем, так как вызывающий код должен это сделать
entry->ref_count--;
// Инвалидировать позиции
q->version++;
// При извлечении элемента приостанавливаем коллбэки
// Это предотвращает рекурсию если во время обработки добавляются новые элементы
q->callback_suspended = 1;
@@ -350,7 +428,11 @@ struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, s
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: condition already met, count=%d<=%d, bytes=%zu<=%zu, calling callback",
q->count, max_packets, q->total_bytes, max_bytes);
callback(q, arg);
free(waiter);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
return NULL;
}
@@ -398,3 +480,368 @@ void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t
memory_pool_get_stats(q->pool, waiter_allocations, waiter_reuse);
}
// ==================== Управление позициями ====================
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue) {
if (!queue) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: queue is NULL");
return NULL;
}
struct ll_queue_pos* pos = malloc(sizeof(struct ll_queue_pos));
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: failed to allocate position");
return NULL;
}
pos->queue = queue;
pos->current = NULL;
pos->prev = NULL;
pos->index = -1; // Перед head
pos->valid = 1;
pos->version = queue->version;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: created position %p for queue %p", pos, queue);
return pos;
}
void queue_pos_free(struct ll_queue_pos* pos) {
if (!pos) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_free: freeing position %p", pos);
free(pos);
}
void queue_pos_reset(struct ll_queue_pos* pos) {
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: pos is NULL");
return;
}
// Обновить версию позиции при необходимости (вместо инвалидации)
if (pos->queue->version != pos->version) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: updating position version from %lu to %lu",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
pos->version = pos->queue->version;
}
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
pos->valid = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: position reset");
}
int queue_pos_next(struct ll_queue_pos* pos) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (!pos->current) {
// Перед head - переходим к head
if (!pos->queue->head) {
// Очередь пуста - некуда переходить
return 0;
}
pos->current = pos->queue->head;
pos->prev = NULL;
pos->index = 0;
} else if (pos->current->next) {
// Есть следующий элемент
pos->prev = pos->current;
pos->current = pos->current->next;
pos->index++;
} else {
// Достигнут конец очереди
return 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_prev(struct ll_queue_pos* pos) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (!pos->current) {
// Уже перед head - дальше некуда
return 0;
} else if (pos->current->prev) {
// Есть предыдущий элемент
pos->current = pos->current->prev;
pos->prev = pos->current->prev;
pos->index--;
} else {
// Достигнуто начало очереди
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_seek(struct ll_queue_pos* pos, int target_index) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (target_index < -1 || target_index >= pos->queue->count) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid target index %d (count=%d)",
target_index, pos->queue->count);
return 0;
}
// Простая реализация: начинаем с начала и идем вперед
// Можно оптимизировать позже, если потребуется
queue_pos_reset(pos);
for (int i = 0; i <= target_index; i++) {
if (!queue_pos_next(pos)) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: failed to reach index %d", target_index);
return 0;
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: seeked to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode) {
if (!pos || !pos->valid || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: invalid parameters");
return -1;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: position invalidated due to queue version change");
return -1;
}
struct ll_queue* q = pos->queue;
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
entry->ref_count++;
if (mode == 0) {
// Вставить перед current
if (!pos->current) {
// Вставить в head
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить перед current
entry->next = pos->current;
entry->prev = pos->current->prev;
if (pos->current->prev) {
pos->current->prev->next = entry;
} else {
q->head = entry; // Вставляем в начало
}
pos->current->prev = entry;
}
} else {
// Вставить после current
if (!pos->current) {
// Вставить в head (current = NULL, значит перед head)
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить после current
entry->next = pos->current->next;
entry->prev = pos->current;
if (pos->current->next) {
pos->current->next->prev = entry;
} else {
q->tail = entry; // Вставляем в конец
}
pos->current->next = entry;
}
}
q->count++;
q->total_bytes += entry->size;
// Обновить версию очереди и позиции (вместо полной инвалидации)
q->version++;
// Обновить текущую позицию, если она активна
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// Корректировка индекса в зависимости от режима вставки
if (mode == 0) {
// Вставка перед текущим - индекс остается тем же
// pos->index не меняется
} else {
// Вставка после текущего - текущий элемент смещается
pos->index++;
}
}
// Позиция остается валидной - убрали строку pos->valid = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: inserted entry %p, mode=%d, count=%d",
entry, mode, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos) {
if (!pos || !pos->valid || !pos->current) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: invalid position or no current element");
return NULL;
}
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return NULL;
}
struct ll_queue* q = pos->queue;
struct ll_entry* entry = pos->current;
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
// Обновить позицию: перейти к следующему элементу
pos->current = entry->next;
pos->prev = entry->prev;
if (pos->current) {
// Остались элементы после удаленного
pos->index = pos->index; // Индекс остается тем же
} else {
// Удалили последний элемент
pos->index = q->count; // Указывает за конец
if (pos->index == 0) {
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
}
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--; // Уменьшить счетчик ссылок
// Обновить версию очереди и позицию (вместо полной инвалидации)
q->version++;
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// pos->index уже обновлен выше
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return entry;
}
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: invalid parameters");
return -1;
}
// Проверить, что элемент действительно в этой очереди
struct ll_entry* curr = q->head;
while (curr) {
if (curr == entry) {
break;
}
curr = curr->next;
}
if (!curr) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: entry %p not found in queue", entry);
return -1;
}
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--;
// Инвалидировать позиции
q->version++;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
+829
View File
@@ -0,0 +1,829 @@
// ll_queue.c
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>
#include "ll_queue.h"
#include "u_async.h"
#include "debug_config.h"
// Предварительное объявление для отложенного возобновления
static void queue_resume_timeout_cb(void* arg);
// Проверить и запустить ожидающие коллбэки
static void check_waiters(struct ll_queue* q) {
if (!q || !q->waiters) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: checking waiters, count=%d, bytes=%zu",
q->count, q->total_bytes);
// Сначала собрать список коллбэков для вызова, чтобы избежать проблем с модификацией списка во время итерации
struct queue_waiter* to_call = NULL;
struct queue_waiter* to_call_tail = NULL;
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* waiter = q->waiters;
while (waiter) {
struct queue_waiter* next = waiter->next;
// Проверить условие: не больше max_packets и не больше max_bytes
// max_bytes = 0 означает "не проверять байты"
if (q->count <= waiter->max_packets && (waiter->max_bytes == 0 || q->total_bytes <= waiter->max_bytes)) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
// Удалить waiter из списка
*pprev = next;
// Добавить в список для вызова
waiter->next = NULL;
if (to_call_tail) {
to_call_tail->next = waiter;
} else {
to_call = waiter;
}
to_call_tail = waiter;
} else {
// Условие не выполнено - оставить в списке
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition NOT met, count=%d>%d or bytes=%zu>%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
pprev = &waiter->next;
}
waiter = next;
}
// Теперь вызвать коллбэки
waiter = to_call;
while (waiter) {
struct queue_waiter* next = waiter->next;
waiter->callback(q, waiter->callback_arg);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
}
// ==================== Управление очередью ====================
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool) {
if (!ua) return NULL;
struct ll_queue* q = calloc(1, sizeof(struct ll_queue));
if (!q) return NULL;
q->head = NULL;
q->tail = NULL;
q->count = 0;
q->total_bytes = 0;
q->size_limit = -1; // По умолчанию без ограничения
q->callback = NULL;
q->callback_arg = NULL;
q->callback_suspended = 0; // Коллбэки разрешены изначально
q->resume_timeout_id = NULL;
q->ua = ua;
q->waiters = NULL;
q->pool=pool;
q->version = 0;
return q;
}
void queue_free(struct ll_queue* q) {
if (!q) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue %p, head=%p, tail=%p, count=%d",
q, q->head, q->tail, q->count);
// Освободить все элементы (уменьшить счетчик ссылок)
struct ll_entry* entry = q->head;
int entry_count = 0;
while (entry) {
struct ll_entry* next = entry->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: releasing entry %p (entry %d), ref_count=%d",
entry, entry_count++, entry->ref_count);
entry->ref_count--; // Дополнительно уменьшить за присутствие в очереди
queue_entry_free(entry); // Это уменьшит ref_count и освободит только если ref_count == 0
entry = next;
}
// Освободить все ожидающие коллбэки
struct queue_waiter* waiter = q->waiters;
int waiter_count = 0;
while (waiter) {
struct queue_waiter* next = waiter->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing waiter %p (waiter %d)", waiter, waiter_count++);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
// Отменить отложенное возобновление если запланировано
if (q->resume_timeout_id) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: cancelling resume timeout %p", q->resume_timeout_id);
uasync_cancel_timeout(q->ua, q->resume_timeout_id);
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue structure %p", q);
free(q);
}
// ==================== Конфигурация очереди ====================
void queue_set_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
if (!q) return;
q->callback = cbk_fn;
q->callback_arg = arg;
}
static void queue_resume_timeout_cb(void* arg) {
struct ll_queue* q = (struct ll_queue*)arg;
if (!q || !q->callback) return;
// Очистить ID таймаута (таймаут сработал)
q->resume_timeout_id = NULL;
// Разрешить коллбэки
q->callback_suspended = 0;
// Если в очереди есть элементы, вызвать коллбэк с первым элементом
// Обработчик должен извлечь этот элемент вызовом queue_entry_get()
if (q->head) {
q->callback(q, q->head, q->callback_arg);
}
}
void queue_resume_callback(struct ll_queue* q) {
if (!q || !q->callback) return;
// Если уже есть отложенное возобновление, ничего не делать
if (q->resume_timeout_id) {
return;
}
// Запланировать отложенное возобновление через uasync
q->resume_timeout_id = uasync_set_timeout(q->ua, 0, q, queue_resume_timeout_cb);
}
void queue_set_size_limit(struct ll_queue* q, int lim) {
if (!q) return;
q->size_limit = lim;
}
// ==================== Управление элементами ====================
struct ll_entry* queue_entry_new(size_t data_size) {
// Выделить память под структуру + область данных
struct ll_entry* entry = malloc(sizeof(struct ll_entry) + data_size);
if (!entry) return NULL;
entry->next = NULL;
entry->prev = NULL; // Инициализировать prev для двусвязного списка
entry->size = data_size;
entry->ref_count = 1; // Начальный счетчик ссылок
// Область данных оставить неинициализированной для производительности
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_new: created entry %p, size=%zu, ref_count=%d",
entry, data_size, entry->ref_count);
return entry;
}
void queue_entry_free(struct ll_entry* entry) {
if (!entry) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry=%p, ref_count=%d",
entry, entry->ref_count);
if (entry->ref_count <= 0) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry %p has invalid ref_count=%d",
entry, entry->ref_count);
return; // Предотвратить double-free
}
entry->ref_count--;
if (entry->ref_count == 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: actually freeing entry %p", entry);
free(entry);
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: decremented ref_count to %d for entry %p",
entry->ref_count, entry);
}
}
// ==================== Операции с очередью ====================
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в хвост (FIFO) - двусвязный список
entry->next = NULL;
entry->prev = q->tail;
if (q->tail) {
q->tail->next = entry;
} else {
q->head = entry; // Очередь была пустой
}
q->tail = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
// Это запускает автоматическую обработку очереди
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: calling callback for entry %p", entry);
// Приостановить коллбэки во время выполнения коллбэка, чтобы предотвратить рекурсию
q->callback_suspended = 1;
q->callback(q, q->head, q->callback_arg);
// Не восстанавливать здесь - восстановление происходит через queue_resume_callback
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в голову (LIFO, высокий приоритет) - двусвязный список
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: calling callback for entry %p", entry);
q->callback_suspended = 1;
q->callback(q, q->head, q->callback_arg);
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_entry_get(struct ll_queue* q) {
if (!q || !q->head) return NULL;
struct ll_entry* entry = q->head;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: retrieving entry %p, size=%zu, ref_count=%d",
entry, entry->size, entry->ref_count);
// Удалить из двусвязного списка
q->head = entry->next;
if (q->head) {
q->head->prev = NULL;
} else {
q->tail = NULL; // Очередь стала пустой
}
q->count--;
q->total_bytes -= entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: removed entry %p, new count=%d, total_bytes=%zu",
entry, q->count, q->total_bytes);
entry->next = NULL;
entry->prev = NULL; // Полностью отсоединить от очереди
// Уменьшить счетчик ссылок при извлечении из очереди
// entry->ref_count был увеличен при добавлении в очередь
// теперь уменьшаем, но не освобождаем, так как вызывающий код должен это сделать
entry->ref_count--;
// Инвалидировать позиции
q->version++;
// При извлечении элемента приостанавливаем коллбэки
// Это предотвращает рекурсию если во время обработки добавляются новые элементы
q->callback_suspended = 1;
// Проверить ожидающие коллбэки при извлечении элемента
// Это важно для waiters, которые ожидают уменьшения очереди
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: about to call check_waiters, count=%d", q->count);
check_waiters(q);
return entry;
}
int queue_entry_count(struct ll_queue* q) {
if (!q) return 0;
return q->count;
}
// ==================== Асинхронное ожидание ====================
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes,
queue_threshold_callback_fn callback, void* arg) {
if (!q || !callback) return NULL;
// Создать новый waiter
struct queue_waiter* waiter;
if (q->pool) {
waiter = (struct queue_waiter*)memory_pool_alloc(q->pool);
} else {
waiter = malloc(sizeof(struct queue_waiter));
}
if (!waiter) return NULL;
waiter->max_packets = max_packets;
waiter->max_bytes = max_bytes;
waiter->callback = callback;
waiter->callback_arg = arg;
waiter->next = NULL;
// Проверить условие немедленно
if (q->count <= max_packets && (max_bytes == 0 || q->total_bytes <= max_bytes)) {
// Условие уже выполнено - вызвать коллбэк и освободить waiter
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: condition already met, count=%d<=%d, bytes=%zu<=%zu, calling callback",
q->count, max_packets, q->total_bytes, max_bytes);
callback(q, arg);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
return NULL;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: registering waiter, count=%d, bytes=%zu, max_packets=%d, max_bytes=%zu",
q->count, q->total_bytes, max_packets, max_bytes);
// Добавить в список ожидающих
waiter->next = q->waiters;
q->waiters = waiter;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: waiter registered successfully");
return waiter;
}
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter) {
if (!q || !waiter) return;
// Найти и удалить waiter из списка
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* w = q->waiters;
while (w) {
if (w == waiter) {
*pprev = w->next;
if (q->pool) {
memory_pool_free(q->pool, w);
} else {
free(w);
}
return;
}
pprev = &w->next;
w = w->next;
}
}
// ==================== Статистика и метрики ====================
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse) {
if (!q || !q->pool) {
if (waiter_allocations) *waiter_allocations = 0;
if (waiter_reuse) *waiter_reuse = 0;
return;
}
memory_pool_get_stats(q->pool, waiter_allocations, waiter_reuse);
}
// ==================== Управление позициями ====================
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue) {
if (!queue) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: queue is NULL");
return NULL;
}
struct ll_queue_pos* pos = malloc(sizeof(struct ll_queue_pos));
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: failed to allocate position");
return NULL;
}
pos->queue = queue;
pos->current = NULL;
pos->prev = NULL;
pos->index = -1; // Перед head
pos->valid = 1;
pos->version = queue->version;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: created position %p for queue %p", pos, queue);
return pos;
}
void queue_pos_free(struct ll_queue_pos* pos) {
if (!pos) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_free: freeing position %p", pos);
free(pos);
}
void queue_pos_reset(struct ll_queue_pos* pos) {
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: pos is NULL");
return;
}
// Обновить версию позиции при необходимости (вместо инвалидации)
if (pos->queue->version != pos->version) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: updating position version from %lu to %lu",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
pos->version = pos->queue->version;
}
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
pos->valid = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: position reset");
}
int queue_pos_next(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: invalid position");
return 0;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: position invalidated due to queue version change");
return 0;
}
if (!pos->current) {
// Перед head - переходим к head
if (!pos->queue->head) {
// Очередь пуста - некуда переходить
return 0;
}
pos->current = pos->queue->head;
pos->prev = NULL;
pos->index = 0;
} else if (pos->current->next) {
// Есть следующий элемент
pos->prev = pos->current;
pos->current = pos->current->next;
pos->index++;
} else {
// Достигнут конец очереди
return 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_prev(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: invalid position");
return 0;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: position invalidated due to queue version change");
return 0;
}
if (!pos->current) {
// Уже перед head - дальше некуда
return 0;
} else if (pos->current->prev) {
// Есть предыдущий элемент
pos->current = pos->current->prev;
pos->prev = pos->current->prev;
pos->index--;
} else {
// Достигнуто начало очереди
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_seek(struct ll_queue_pos* pos, int target_index) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid position");
return 0;
}
// Обновить версию позиции при необходимости (вместо инвалидации)
if (pos->queue->version != pos->version) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: updating position version from %lu to %lu",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
pos->version = pos->queue->version;
// Сбросить позицию, так как структура очереди могла измениться
queue_pos_reset(pos);
}
if (target_index < -1 || target_index >= pos->queue->count) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid target index %d (count=%d)",
target_index, pos->queue->count);
return 0;
}
// Простая реализация: начинаем с начала и идем вперед
// Можно оптимизировать позже, если потребуется
queue_pos_reset(pos);
for (int i = 0; i <= target_index; i++) {
if (!queue_pos_next(pos)) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: failed to reach index %d", target_index);
return 0;
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: seeked to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode) {
if (!pos || !pos->valid || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: invalid parameters");
return -1;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: position invalidated due to queue version change");
return -1;
}
struct ll_queue* q = pos->queue;
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
entry->ref_count++;
if (mode == 0) {
// Вставить перед current
if (!pos->current) {
// Вставить в head
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить перед current
entry->next = pos->current;
entry->prev = pos->current->prev;
if (pos->current->prev) {
pos->current->prev->next = entry;
} else {
q->head = entry; // Вставляем в начало
}
pos->current->prev = entry;
}
} else {
// Вставить после current
if (!pos->current) {
// Вставить в head (current = NULL, значит перед head)
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить после current
entry->next = pos->current->next;
entry->prev = pos->current;
if (pos->current->next) {
pos->current->next->prev = entry;
} else {
q->tail = entry; // Вставляем в конец
}
pos->current->next = entry;
}
}
q->count++;
q->total_bytes += entry->size;
// Обновить версию очереди и позиции (вместо полной инвалидации)
q->version++;
// Обновить текущую позицию, если она активна
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// Корректировка индекса в зависимости от режима вставки
if (mode == 0) {
// Вставка перед текущим - индекс остается тем же
// pos->index не меняется
} else {
// Вставка после текущего - текущий элемент смещается
pos->index++;
}
}
pos->valid = 0; // Текущая позиция теперь invalid
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: inserted entry %p, mode=%d, count=%d",
entry, mode, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos) {
if (!pos || !pos->valid || !pos->current) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: invalid position or no current element");
return NULL;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: position invalidated due to queue version change");
return NULL;
}
struct ll_queue* q = pos->queue;
struct ll_entry* entry = pos->current;
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
// Обновить позицию: перейти к следующему элементу
pos->current = entry->next;
pos->prev = entry->prev;
if (pos->current) {
// Остались элементы после удаленного
pos->index = pos->index; // Индекс остается тем же
} else {
// Удалили последний элемент
pos->index = q->count; // Указывает за конец
if (pos->index == 0) {
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
}
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--; // Уменьшить счетчик ссылок
// Обновить версию очереди и позицию (вместо полной инвалидации)
q->version++;
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// pos->index уже обновлен выше
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return entry;
}
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: invalid parameters");
return -1;
}
// Проверить, что элемент действительно в этой очереди
struct ll_entry* curr = q->head;
while (curr) {
if (curr == entry) {
break;
}
curr = curr->next;
}
if (!curr) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: entry %p not found in queue", entry);
return -1;
}
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--;
// Инвалидировать позиции
q->version++;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
+847
View File
@@ -0,0 +1,847 @@
// ll_queue.c
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>
#include "ll_queue.h"
#include "u_async.h"
#include "debug_config.h"
// Предварительное объявление для отложенного возобновления
static void queue_resume_timeout_cb(void* arg);
// Проверить и запустить ожидающие коллбэки
static void check_waiters(struct ll_queue* q) {
if (!q || !q->waiters) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: checking waiters, count=%d, bytes=%zu",
q->count, q->total_bytes);
// Сначала собрать список коллбэков для вызова, чтобы избежать проблем с модификацией списка во время итерации
struct queue_waiter* to_call = NULL;
struct queue_waiter* to_call_tail = NULL;
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* waiter = q->waiters;
while (waiter) {
struct queue_waiter* next = waiter->next;
// Проверить условие: не больше max_packets и не больше max_bytes
// max_bytes = 0 означает "не проверять байты"
if (q->count <= waiter->max_packets && (waiter->max_bytes == 0 || q->total_bytes <= waiter->max_bytes)) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
// Удалить waiter из списка
*pprev = next;
// Добавить в список для вызова
waiter->next = NULL;
if (to_call_tail) {
to_call_tail->next = waiter;
} else {
to_call = waiter;
}
to_call_tail = waiter;
} else {
// Условие не выполнено - оставить в списке
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition NOT met, count=%d>%d or bytes=%zu>%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
pprev = &waiter->next;
}
waiter = next;
}
// Теперь вызвать коллбэки
waiter = to_call;
while (waiter) {
struct queue_waiter* next = waiter->next;
waiter->callback(q, waiter->callback_arg);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
}
// ==================== Унифицированная система версий позиций ====================
// Унифицированная проверка версии позиции с автоматическим пересканированием
static int queue_pos_ensure_valid(struct ll_queue_pos* pos) {
if (!pos || !pos->queue) return 0;
// Если позиция уже валидна и версия совпадает, всё хорошо
if (pos->valid && pos->version == pos->queue->version) {
return 1;
}
// Несовпадение версии или позиция невалидна - нужно пересканировать
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "Position version %lu != queue version %lu, rescanning...",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
// Сохранить целевой индекс (где мы хотим быть)
int target_index = pos->index;
// Обновить версию позиции до текущей
pos->version = pos->queue->version;
// Обработать краевой случай: индекс больше не существует из-за изменения размера очереди
if (target_index >= pos->queue->count) {
// Позиция выходит за пределы очереди - инвалидируем её
pos->valid = 0;
return 0;
}
// Простой подход: сбросить и перейти к тому же индексу
// Это найдёт элемент, который теперь находится в целевом индексе
queue_pos_reset(pos);
return queue_pos_seek(pos, target_index);
}
// ==================== Управление очередью ====================
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool) {
if (!ua) return NULL;
struct ll_queue* q = calloc(1, sizeof(struct ll_queue));
if (!q) return NULL;
q->head = NULL;
q->tail = NULL;
q->count = 0;
q->total_bytes = 0;
q->size_limit = -1; // По умолчанию без ограничения
q->callback = NULL;
q->callback_arg = NULL;
q->callback_suspended = 0; // Коллбэки разрешены изначально
q->resume_timeout_id = NULL;
q->ua = ua;
q->waiters = NULL;
q->pool=pool;
q->version = 0;
return q;
}
void queue_free(struct ll_queue* q) {
if (!q) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue %p, head=%p, tail=%p, count=%d",
q, q->head, q->tail, q->count);
// Освободить все элементы (уменьшить счетчик ссылок)
struct ll_entry* entry = q->head;
int entry_count = 0;
while (entry) {
struct ll_entry* next = entry->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: releasing entry %p (entry %d), ref_count=%d",
entry, entry_count++, entry->ref_count);
entry->ref_count--; // Дополнительно уменьшить за присутствие в очереди
queue_entry_free(entry); // Это уменьшит ref_count и освободит только если ref_count == 0
entry = next;
}
// Освободить все ожидающие коллбэки
struct queue_waiter* waiter = q->waiters;
int waiter_count = 0;
while (waiter) {
struct queue_waiter* next = waiter->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing waiter %p (waiter %d)", waiter, waiter_count++);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
// Отменить отложенное возобновление если запланировано
if (q->resume_timeout_id) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: cancelling resume timeout %p", q->resume_timeout_id);
uasync_cancel_timeout(q->ua, q->resume_timeout_id);
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue structure %p", q);
free(q);
}
// ==================== Конфигурация очереди ====================
void queue_set_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
if (!q) return;
q->callback = cbk_fn;
q->callback_arg = arg;
}
static void queue_resume_timeout_cb(void* arg) {
struct ll_queue* q = (struct ll_queue*)arg;
if (!q || !q->callback) return;
// Очистить ID таймаута (таймаут сработал)
q->resume_timeout_id = NULL;
// Разрешить коллбэки
q->callback_suspended = 0;
// Если в очереди есть элементы, вызвать коллбэк с первым элементом
// Обработчик должен извлечь этот элемент вызовом queue_entry_get()
if (q->head) {
q->callback(q, q->head, q->callback_arg);
}
}
void queue_resume_callback(struct ll_queue* q) {
if (!q || !q->callback) return;
// Если уже есть отложенное возобновление, ничего не делать
if (q->resume_timeout_id) {
return;
}
// Запланировать отложенное возобновление через uasync
q->resume_timeout_id = uasync_set_timeout(q->ua, 0, q, queue_resume_timeout_cb);
}
void queue_set_size_limit(struct ll_queue* q, int lim) {
if (!q) return;
q->size_limit = lim;
}
// ==================== Управление элементами ====================
struct ll_entry* queue_entry_new(size_t data_size) {
// Выделить память под структуру + область данных
struct ll_entry* entry = malloc(sizeof(struct ll_entry) + data_size);
if (!entry) return NULL;
entry->next = NULL;
entry->prev = NULL; // Инициализировать prev для двусвязного списка
entry->size = data_size;
entry->ref_count = 1; // Начальный счетчик ссылок
// Область данных оставить неинициализированной для производительности
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_new: created entry %p, size=%zu, ref_count=%d",
entry, data_size, entry->ref_count);
return entry;
}
void queue_entry_free(struct ll_entry* entry) {
if (!entry) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry=%p, ref_count=%d",
entry, entry->ref_count);
if (entry->ref_count <= 0) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry %p has invalid ref_count=%d",
entry, entry->ref_count);
return; // Предотвратить double-free
}
entry->ref_count--;
if (entry->ref_count == 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: actually freeing entry %p", entry);
free(entry);
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: decremented ref_count to %d for entry %p",
entry->ref_count, entry);
}
}
// ==================== Операции с очередью ====================
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в хвост (FIFO) - двусвязный список
entry->next = NULL;
entry->prev = q->tail;
if (q->tail) {
q->tail->next = entry;
} else {
q->head = entry; // Очередь была пустой
}
q->tail = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
// Это запускает автоматическую обработку очереди
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: calling callback for entry %p", entry);
// Приостановить коллбэки во время выполнения коллбэка, чтобы предотвратить рекурсию
q->callback_suspended = 1;
q->callback(q, q->head, q->callback_arg);
// Не восстанавливать здесь - восстановление происходит через queue_resume_callback
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в голову (LIFO, высокий приоритет) - двусвязный список
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Инвалидировать позиции
q->version++;
// Если коллбэки разрешены - вызвать коллбэк
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: calling callback for entry %p", entry);
q->callback_suspended = 1;
q->callback(q, q->head, q->callback_arg);
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_entry_get(struct ll_queue* q) {
if (!q || !q->head) return NULL;
struct ll_entry* entry = q->head;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: retrieving entry %p, size=%zu, ref_count=%d",
entry, entry->size, entry->ref_count);
// Удалить из двусвязного списка
q->head = entry->next;
if (q->head) {
q->head->prev = NULL;
} else {
q->tail = NULL; // Очередь стала пустой
}
q->count--;
q->total_bytes -= entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: removed entry %p, new count=%d, total_bytes=%zu",
entry, q->count, q->total_bytes);
entry->next = NULL;
entry->prev = NULL; // Полностью отсоединить от очереди
// Уменьшить счетчик ссылок при извлечении из очереди
// entry->ref_count был увеличен при добавлении в очередь
// теперь уменьшаем, но не освобождаем, так как вызывающий код должен это сделать
entry->ref_count--;
// Инвалидировать позиции
q->version++;
// При извлечении элемента приостанавливаем коллбэки
// Это предотвращает рекурсию если во время обработки добавляются новые элементы
q->callback_suspended = 1;
// Проверить ожидающие коллбэки при извлечении элемента
// Это важно для waiters, которые ожидают уменьшения очереди
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: about to call check_waiters, count=%d", q->count);
check_waiters(q);
return entry;
}
int queue_entry_count(struct ll_queue* q) {
if (!q) return 0;
return q->count;
}
// ==================== Асинхронное ожидание ====================
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes,
queue_threshold_callback_fn callback, void* arg) {
if (!q || !callback) return NULL;
// Создать новый waiter
struct queue_waiter* waiter;
if (q->pool) {
waiter = (struct queue_waiter*)memory_pool_alloc(q->pool);
} else {
waiter = malloc(sizeof(struct queue_waiter));
}
if (!waiter) return NULL;
waiter->max_packets = max_packets;
waiter->max_bytes = max_bytes;
waiter->callback = callback;
waiter->callback_arg = arg;
waiter->next = NULL;
// Проверить условие немедленно
if (q->count <= max_packets && (max_bytes == 0 || q->total_bytes <= max_bytes)) {
// Условие уже выполнено - вызвать коллбэк и освободить waiter
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: condition already met, count=%d<=%d, bytes=%zu<=%zu, calling callback",
q->count, max_packets, q->total_bytes, max_bytes);
callback(q, arg);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
return NULL;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: registering waiter, count=%d, bytes=%zu, max_packets=%d, max_bytes=%zu",
q->count, q->total_bytes, max_packets, max_bytes);
// Добавить в список ожидающих
waiter->next = q->waiters;
q->waiters = waiter;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: waiter registered successfully");
return waiter;
}
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter) {
if (!q || !waiter) return;
// Найти и удалить waiter из списка
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* w = q->waiters;
while (w) {
if (w == waiter) {
*pprev = w->next;
if (q->pool) {
memory_pool_free(q->pool, w);
} else {
free(w);
}
return;
}
pprev = &w->next;
w = w->next;
}
}
// ==================== Статистика и метрики ====================
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse) {
if (!q || !q->pool) {
if (waiter_allocations) *waiter_allocations = 0;
if (waiter_reuse) *waiter_reuse = 0;
return;
}
memory_pool_get_stats(q->pool, waiter_allocations, waiter_reuse);
}
// ==================== Управление позициями ====================
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue) {
if (!queue) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: queue is NULL");
return NULL;
}
struct ll_queue_pos* pos = malloc(sizeof(struct ll_queue_pos));
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: failed to allocate position");
return NULL;
}
pos->queue = queue;
pos->current = NULL;
pos->prev = NULL;
pos->index = -1; // Перед head
pos->valid = 1;
pos->version = queue->version;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: created position %p for queue %p", pos, queue);
return pos;
}
void queue_pos_free(struct ll_queue_pos* pos) {
if (!pos) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_free: freeing position %p", pos);
free(pos);
}
void queue_pos_reset(struct ll_queue_pos* pos) {
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: pos is NULL");
return;
}
// Обновить версию позиции при необходимости (вместо инвалидации)
if (pos->queue->version != pos->version) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: updating position version from %lu to %lu",
(unsigned long)pos->version, (unsigned long)pos->queue->version);
pos->version = pos->queue->version;
}
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
pos->valid = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: position reset");
}
int queue_pos_next(struct ll_queue_pos* pos) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (!pos->current) {
// Перед head - переходим к head
if (!pos->queue->head) {
// Очередь пуста - некуда переходить
return 0;
}
pos->current = pos->queue->head;
pos->prev = NULL;
pos->index = 0;
} else if (pos->current->next) {
// Есть следующий элемент
pos->prev = pos->current;
pos->current = pos->current->next;
pos->index++;
} else {
// Достигнут конец очереди
return 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_prev(struct ll_queue_pos* pos) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (!pos->current) {
// Уже перед head - дальше некуда
return 0;
} else if (pos->current->prev) {
// Есть предыдущий элемент
pos->current = pos->current->prev;
pos->prev = pos->current->prev;
pos->index--;
} else {
// Достигнуто начало очереди
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_seek(struct ll_queue_pos* pos, int target_index) {
if (!pos) return 0;
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return 0;
}
if (target_index < -1 || target_index >= pos->queue->count) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid target index %d (count=%d)",
target_index, pos->queue->count);
return 0;
}
// Простая реализация: начинаем с начала и идем вперед
// Можно оптимизировать позже, если потребуется
queue_pos_reset(pos);
for (int i = 0; i <= target_index; i++) {
if (!queue_pos_next(pos)) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: failed to reach index %d", target_index);
return 0;
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: seeked to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode) {
if (!pos || !pos->valid || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: invalid parameters");
return -1;
}
if (pos->queue->version != pos->version) {
pos->valid = 0;
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: position invalidated due to queue version change");
return -1;
}
struct ll_queue* q = pos->queue;
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
entry->ref_count++;
if (mode == 0) {
// Вставить перед current
if (!pos->current) {
// Вставить в head
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить перед current
entry->next = pos->current;
entry->prev = pos->current->prev;
if (pos->current->prev) {
pos->current->prev->next = entry;
} else {
q->head = entry; // Вставляем в начало
}
pos->current->prev = entry;
}
} else {
// Вставить после current
if (!pos->current) {
// Вставить в head (current = NULL, значит перед head)
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить после current
entry->next = pos->current->next;
entry->prev = pos->current;
if (pos->current->next) {
pos->current->next->prev = entry;
} else {
q->tail = entry; // Вставляем в конец
}
pos->current->next = entry;
}
}
q->count++;
q->total_bytes += entry->size;
// Обновить версию очереди и позиции (вместо полной инвалидации)
q->version++;
// Обновить текущую позицию, если она активна
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// Корректировка индекса в зависимости от режима вставки
if (mode == 0) {
// Вставка перед текущим - индекс остается тем же
// pos->index не меняется
} else {
// Вставка после текущего - текущий элемент смещается
pos->index++;
}
}
// Позиция остается валидной - убрали строку pos->valid = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: inserted entry %p, mode=%d, count=%d",
entry, mode, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos) {
if (!pos || !pos->valid || !pos->current) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: invalid position or no current element");
return NULL;
}
// Убедиться, что позиция валидна (автоматическое обновление при необходимости)
if (!queue_pos_ensure_valid(pos)) {
return NULL;
}
struct ll_queue* q = pos->queue;
struct ll_entry* entry = pos->current;
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
// Обновить позицию: перейти к следующему элементу
pos->current = entry->next;
pos->prev = entry->prev;
if (pos->current) {
// Остались элементы после удаленного
pos->index = pos->index; // Индекс остается тем же
} else {
// Удалили последний элемент
pos->index = q->count; // Указывает за конец
if (pos->index == 0) {
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
}
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--; // Уменьшить счетчик ссылок
// Обновить версию очереди и позицию (вместо полной инвалидации)
q->version++;
if (pos->current) {
pos->version = q->version;
pos->valid = 1; // Сделать позицию снова валидной
// pos->index уже обновлен выше
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return entry;
}
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: invalid parameters");
return -1;
}
// Проверить, что элемент действительно в этой очереди
struct ll_entry* curr = q->head;
while (curr) {
if (curr == entry) {
break;
}
curr = curr->next;
}
if (!curr) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: entry %p not found in queue", entry);
return -1;
}
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--;
// Инвалидировать позиции
q->version++;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
+744
View File
@@ -0,0 +1,744 @@
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>
#include "ll_queue.h"
#include "u_async.h"
#include "debug_config.h"
// Предварительное объявление для отложенного возобновления
static void queue_resume_timeout_cb(void* arg);
// Вспомогательные функции для двусвязного списка
static void link_entry_after(struct ll_entry* existing, struct ll_entry* new_entry) {
if (!existing || !new_entry) return;
new_entry->next = existing->next;
new_entry->prev = existing;
if (existing->next) {
existing->next->prev = new_entry;
}
existing->next = new_entry;
}
static void unlink_entry(struct ll_entry* entry) {
if (!entry) return;
if (entry->prev) entry->prev->next = entry->next;
if (entry->next) entry->next->prev = entry->prev;
entry->next = NULL;
entry->prev = NULL;
}
// Проверить и запустить ожидающие коллбэки
static void check_waiters(struct ll_queue* q) {
if (!q || !q->waiters) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: checking waiters, count=%d, bytes=%zu",
q->count, q->total_bytes);
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* waiter = q->waiters;
while (waiter) {
struct queue_waiter* next = waiter->next;
// Проверить условие: не больше max_packets и не больше max_bytes
// max_bytes = 0 означает "не проверять байты"
if (q->count <= waiter->max_packets && (waiter->max_bytes == 0 || q->total_bytes <= waiter->max_bytes)) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
waiter->callback(q, waiter->callback_arg);
// Удалить waiter из списка
*pprev = next;
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
// pprev уже указывает на правильный следующий элемент
} else {
// Условие не выполнено - оставить в списке
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "check_waiters: condition NOT met, count=%d>%d or bytes=%zu>%zu (max_bytes_check=%s)",
q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes,
waiter->max_bytes == 0 ? "disabled" : "enabled");
pprev = &waiter->next;
}
waiter = next;
}
}
// ==================== Управление очередью ====================
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool) {
struct ll_queue* q = calloc(1, sizeof(struct ll_queue));
if (!q) return NULL;
q->head = NULL;
q->tail = NULL;
q->count = 0;
q->total_bytes = 0;
q->size_limit = -1; // По умолчанию без ограничения
q->callback = NULL;
q->callback_arg = NULL;
q->callback_suspended = 0; // Коллбэки разрешены изначально
q->resume_timeout_id = NULL;
q->ua = ua;
q->waiters = NULL;
q->pool=pool;
return q;
}
void queue_free(struct ll_queue* q) {
if (!q) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue %p, head=%p, tail=%p, count=%d",
q, q->head, q->tail, q->count);
// Освободить все элементы (уменьшить счетчик ссылок)
struct ll_entry* entry = q->head;
int entry_count = 0;
while (entry) {
struct ll_entry* next = entry->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: releasing entry %p (entry %d), ref_count=%d",
entry, entry_count++, entry->ref_count);
queue_entry_free(entry); // Это уменьшит ref_count и освободит только если ref_count == 0
entry = next;
}
// Освободить все ожидающие коллбэки
struct queue_waiter* waiter = q->waiters;
int waiter_count = 0;
while (waiter) {
struct queue_waiter* next = waiter->next;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing waiter %p (waiter %d)", waiter, waiter_count++);
if (q->pool) {
memory_pool_free(q->pool, waiter);
} else {
free(waiter);
}
waiter = next;
}
// Отменить отложенное возобновление если запланировано
if (q->resume_timeout_id) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: cancelling resume timeout %p", q->resume_timeout_id);
uasync_cancel_timeout(q->ua, q->resume_timeout_id);
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_free: freeing queue structure %p", q);
free(q);
}
// ==================== Конфигурация очереди ====================
void queue_set_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
if (!q) return;
q->callback = cbk_fn;
q->callback_arg = arg;
}
static void queue_resume_timeout_cb(void* arg) {
struct ll_queue* q = (struct ll_queue*)arg;
if (!q || !q->callback) return;
// Очистить ID таймаута (таймаут сработал)
q->resume_timeout_id = NULL;
// Разрешить коллбэки
q->callback_suspended = 0;
// Если в очереди есть элементы, вызвать коллбэк с первым элементом
// Обработчик должен извлечь этот элемент вызовом queue_entry_get()
if (q->head) {
q->callback(q, q->head, q->callback_arg);
}
}
void queue_resume_callback(struct ll_queue* q) {
if (!q || !q->callback) return;
// Если уже есть отложенное возобновление, ничего не делать
if (q->resume_timeout_id) {
return;
}
// Запланировать отложенное возобновление через uasync
q->resume_timeout_id = uasync_set_timeout(q->ua, 0, q, queue_resume_timeout_cb);
}
void queue_set_size_limit(struct ll_queue* q, int lim) {
if (!q) return;
q->size_limit = lim;
}
// ==================== Управление элементами ====================
struct ll_entry* queue_entry_new(size_t data_size) {
// Выделить память под структуру + область данных
struct ll_entry* entry = malloc(sizeof(struct ll_entry) + data_size);
if (!entry) return NULL;
entry->next = NULL;
entry->prev = NULL; // Инициализировать prev для двусвязного списка
entry->size = data_size;
entry->ref_count = 1; // Начальный счетчик ссылок
// Область данных оставить неинициализированной для производительности
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_new: created entry %p, size=%zu, ref_count=%d",
entry, data_size, entry->ref_count);
return entry;
}
void queue_entry_free(struct ll_entry* entry) {
if (!entry) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry=%p, ref_count=%d",
entry, entry->ref_count);
if (entry->ref_count <= 0) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: entry %p has invalid ref_count=%d",
entry, entry->ref_count);
return; // Предотвратить double-free
}
entry->ref_count--;
if (entry->ref_count == 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: actually freeing entry %p", entry);
free(entry);
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_free: decremented ref_count to %d for entry %p",
entry->ref_count, entry);
}
}
// ==================== Операции с очередью ====================
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в хвост (FIFO) - двусвязный список
entry->next = NULL;
entry->prev = q->tail;
if (q->tail) {
q->tail->next = entry;
} else {
q->head = entry; // Очередь была пустой
}
q->tail = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Если коллбэки разрешены - вызвать коллбэк
// Это запускает автоматическую обработку очереди
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put: calling callback for entry %p", entry);
// Приостановить коллбэки во время выполнения коллбэка, чтобы предотвратить рекурсию
q->callback_suspended = 1;
q->callback(q, entry, q->callback_arg);
// Не восстанавливать здесь - восстановление происходит через queue_resume_callback
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: entry=%p, size=%zu, count=%d",
entry, entry->size, q->count);
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
// Увеличить счетчик ссылок при добавлении в очередь
entry->ref_count++;
// Добавить в голову (LIFO, высокий приоритет) - двусвязный список
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: added entry %p, new count=%d, total_bytes=%zu, ref_count=%d",
entry, q->count, q->total_bytes, entry->ref_count);
// Если коллбэки разрешены - вызвать коллбэк
if (!q->callback_suspended && q->callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_put_first: calling callback for entry %p", entry);
q->callback(q, entry, q->callback_arg);
}
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_entry_get(struct ll_queue* q) {
if (!q || !q->head) return NULL;
struct ll_entry* entry = q->head;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: retrieving entry %p, size=%zu, ref_count=%d",
entry, entry->size, entry->ref_count);
// Удалить из двусвязного списка
q->head = entry->next;
if (q->head) {
q->head->prev = NULL;
} else {
q->tail = NULL; // Очередь стала пустой
}
q->count--;
q->total_bytes -= entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: removed entry %p, new count=%d, total_bytes=%zu",
entry, q->count, q->total_bytes);
entry->next = NULL;
entry->prev = NULL; // Полностью отсоединить от очереди
// Уменьшить счетчик ссылок при извлечении из очереди
// entry->ref_count был увеличен при добавлении в очередь
// теперь уменьшаем, но не освобождаем, так как вызывающий код должен это сделать
entry->ref_count--;
// При извлечении элемента приостанавливаем коллбэки
// Это предотвращает рекурсию если во время обработки добавляются новые элементы
q->callback_suspended = 1;
// Проверить ожидающие коллбэки при извлечении элемента
// Это важно для waiters, которые ожидают уменьшения очереди
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_get: about to call check_waiters, count=%d", q->count);
check_waiters(q);
return entry;
}
int queue_entry_count(struct ll_queue* q) {
if (!q) return 0;
return q->count;
}
// ==================== Асинхронное ожидание ====================
struct queue_waiter* queue_wait_threshold(struct ll_queue* q, int max_packets, size_t max_bytes,
queue_threshold_callback_fn callback, void* arg) {
if (!q || !callback) return NULL;
// Создать новый waiter
struct queue_waiter* waiter;
if (q->pool) {
waiter = (struct queue_waiter*)memory_pool_alloc(q->pool);
} else {
waiter = malloc(sizeof(struct queue_waiter));
}
if (!waiter) return NULL;
waiter->max_packets = max_packets;
waiter->max_bytes = max_bytes;
waiter->callback = callback;
waiter->callback_arg = arg;
waiter->next = NULL;
// Проверить условие немедленно
if (q->count <= max_packets && (max_bytes == 0 || q->total_bytes <= max_bytes)) {
// Условие уже выполнено - вызвать коллбэк и освободить waiter
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: condition already met, count=%d<=%d, bytes=%zu<=%zu, calling callback",
q->count, max_packets, q->total_bytes, max_bytes);
callback(q, arg);
free(waiter);
return NULL;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: registering waiter, count=%d, bytes=%zu, max_packets=%d, max_bytes=%zu",
q->count, q->total_bytes, max_packets, max_bytes);
// Добавить в список ожидающих
waiter->next = q->waiters;
q->waiters = waiter;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_wait_threshold: waiter registered successfully");
return waiter;
}
void queue_cancel_wait(struct ll_queue* q, struct queue_waiter* waiter) {
if (!q || !waiter) return;
// Найти и удалить waiter из списка
struct queue_waiter** pprev = &q->waiters;
struct queue_waiter* w = q->waiters;
while (w) {
if (w == waiter) {
*pprev = w->next;
if (q->pool) {
memory_pool_free(q->pool, w);
} else {
free(w);
}
return;
}
pprev = &w->next;
w = w->next;
}
}
// ==================== Статистика и метрики ====================
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse) {
if (!q || !q->pool) {
if (waiter_allocations) *waiter_allocations = 0;
if (waiter_reuse) *waiter_reuse = 0;
return;
}
memory_pool_get_stats(q->pool, waiter_allocations, waiter_reuse);
}
// ==================== Управление позициями ====================
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue) {
if (!queue) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: queue is NULL");
return NULL;
}
struct ll_queue_pos* pos = malloc(sizeof(struct ll_queue_pos));
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: failed to allocate position");
return NULL;
}
pos->queue = queue;
pos->current = NULL;
pos->prev = NULL;
pos->index = -1; // Перед head
pos->valid = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_new: created position %p for queue %p", pos, queue);
return pos;
}
void queue_pos_free(struct ll_queue_pos* pos) {
if (!pos) return;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_free: freeing position %p", pos);
free(pos);
}
void queue_pos_reset(struct ll_queue_pos* pos) {
if (!pos) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: pos is NULL");
return;
}
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
pos->valid = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_reset: position reset");
}
int queue_pos_next(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: invalid position");
return 0;
}
if (!pos->current) {
// Перед head - переходим к head
if (!pos->queue->head) {
// Очередь пуста - некуда переходить
return 0;
}
pos->current = pos->queue->head;
pos->prev = NULL;
pos->index = 0;
} else if (pos->current->next) {
// Есть следующий элемент
pos->prev = pos->current;
pos->current = pos->current->next;
pos->index++;
} else {
// Достигнут конец очереди
return 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_next: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_prev(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: invalid position");
return 0;
}
if (!pos->current) {
// Уже перед head - дальше некуда
return 0;
} else if (pos->current->prev) {
// Есть предыдущий элемент
pos->current = pos->current->prev;
pos->prev = pos->current->prev;
pos->index--;
} else {
// Достигнуто начало очереди
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_prev: moved to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_seek(struct ll_queue_pos* pos, int target_index) {
if (!pos || !pos->valid) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid position");
return 0;
}
if (target_index < -1 || target_index >= pos->queue->count) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: invalid target index %d (count=%d)",
target_index, pos->queue->count);
return 0;
}
// Простая реализация: начинаем с начала и идем вперед
// Можно оптимизировать позже, если потребуется
queue_pos_reset(pos);
for (int i = 0; i <= target_index; i++) {
if (!queue_pos_next(pos)) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: failed to reach index %d", target_index);
return 0;
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_seek: seeked to index %d, entry %p",
pos->index, pos->current);
return 1;
}
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode) {
if (!pos || !pos->valid || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: invalid parameters");
return -1;
}
struct ll_queue* q = pos->queue;
// Проверить лимит размера
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: size limit exceeded, freeing entry %p", entry);
queue_entry_free(entry);
return -1;
}
entry->ref_count++;
if (mode == 0) {
// Вставить перед current
if (!pos->current) {
// Вставить в head
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить перед current
entry->next = pos->current;
entry->prev = pos->current->prev;
if (pos->current->prev) {
pos->current->prev->next = entry;
} else {
q->head = entry; // Вставляем в начало
}
pos->current->prev = entry;
}
} else {
// Вставить после current
if (!pos->current) {
// Вставить в head (current = NULL, значит перед head)
entry->next = q->head;
entry->prev = NULL;
if (q->head) {
q->head->prev = entry;
} else {
q->tail = entry; // Очередь была пустой
}
q->head = entry;
} else {
// Вставить после current
entry->next = pos->current->next;
entry->prev = pos->current;
if (pos->current->next) {
pos->current->next->prev = entry;
} else {
q->tail = entry; // Вставляем в конец
}
pos->current->next = entry;
}
}
q->count++;
q->total_bytes += entry->size;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_insert: inserted entry %p, mode=%d, count=%d",
entry, mode, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos) {
if (!pos || !pos->valid || !pos->current) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: invalid position or no current element");
return NULL;
}
struct ll_queue* q = pos->queue;
struct ll_entry* entry = pos->current;
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
// Обновить позицию: перейти к следующему элементу
pos->current = entry->next;
pos->prev = entry->prev;
if (pos->current) {
// Остались элементы после удаленного
pos->index = pos->index; // Индекс остается тем же
} else {
// Удалили последний элемент
pos->index = q->count; // Указывает за конец
if (pos->index == 0) {
pos->current = NULL;
pos->prev = NULL;
pos->index = -1;
}
}
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--; // Уменьшить счетчик ссылок
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_pos_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return entry;
}
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: invalid parameters");
return -1;
}
// Проверить, что элемент действительно в этой очереди
struct ll_entry* curr = q->head;
while (curr) {
if (curr == entry) {
break;
}
curr = curr->next;
}
if (!curr) {
DEBUG_ERROR(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: entry %p not found in queue", entry);
return -1;
}
// Удалить из двусвязного списка
if (entry->prev) {
entry->prev->next = entry->next;
} else {
q->head = entry->next; // Удаляем head
}
if (entry->next) {
entry->next->prev = entry->prev;
} else {
q->tail = entry->prev; // Удаляем tail
}
q->count--;
q->total_bytes -= entry->size;
entry->next = NULL;
entry->prev = NULL;
entry->ref_count--;
DEBUG_DEBUG(DEBUG_CATEGORY_LL_QUEUE, "queue_entry_remove: removed entry %p, new count=%d",
entry, q->count);
// Проверить ожидающие коллбэки
check_waiters(q);
return 0;
}
+77 -3
View File
@@ -1,7 +1,9 @@
// 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
// Предварительные объявления
@@ -9,6 +11,7 @@ struct ll_queue;
struct ll_entry;
struct queue_waiter;
struct memory_pool;
struct ll_queue_pos; // Управляющая структура позиции
// Тип коллбэка: вызывается при добавлении элемента в пустую очередь или для продолжения обработки
// Параметры: указатель на очередь, указатель на элемент (первый в очереди), пользовательский аргумент
@@ -17,6 +20,7 @@ typedef void (*queue_callback_fn)(struct ll_queue* q, struct ll_entry* entry, vo
// Структура элемента - переменный размер, данные расположены сразу после структуры
struct ll_entry {
struct ll_entry* next; // Указатель на следующий элемент в очереди
struct ll_entry* prev; // Указатель на предыдущий элемент в очереди
size_t size; // Размер данных элемента (байт)
int ref_count; // Счетчик ссылок для предотвращения double-free
};
@@ -24,7 +28,7 @@ struct ll_entry {
// Структура условия ожидания (waiter)
struct queue_waiter {
int max_packets; // Максимальное количество пакетов
size_t max_bytes; // Максимальное количество байт
size_t max_bytes; // Максимальное количество байт (0 = не проверять байты)
void (*callback)(struct ll_queue* q, void* arg); // Коллбэк для вызова
void* callback_arg; // Аргумент коллбэка
struct queue_waiter* next; // Следующий ожидающий в списке
@@ -51,6 +55,8 @@ struct ll_queue {
// Пулы памяти для оптимизации аллокаций
struct memory_pool* pool; // Пул для структур struct queue_waiter
uint64_t version; // Версия для инвалидации позиций
};
// ==================== Управление очередью ====================
@@ -117,18 +123,20 @@ int queue_entry_count(struct ll_queue* q);
// Получить указатель на область данных элемента
// Данные расположены сразу после структуры struct ll_entry
static inline void* ll_entry_data(struct ll_entry* entry) {
if (!entry) return NULL;
return (void*)(entry + 1);
}
// Получить размер данных элемента
static inline size_t ll_entry_size(struct ll_entry* entry) {
if (!entry) return 0;
return entry->size;
}
// ==================== Асинхронное ожидание ====================
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь
// не более max_packets пакетов и не более max_bytes байт.
// не более max_packets пакетов и не более max_bytes байт (если max_bytes != 0).
// Если условие уже выполнено, коллбэк вызывается немедленно.
// Можно зарегистрировать несколько ожиданий на одной очереди.
// Возвращает указатель на waiter для возможной отмены через queue_cancel_wait
@@ -147,4 +155,70 @@ static inline size_t queue_total_bytes(struct ll_queue* q) {
// Получить статистику использования пулов памяти
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse);
#endif // LL_QUEUE_H
// ==================== Управление позициями ====================
// Структура управления позицией для навигации по очереди
struct ll_queue_pos {
struct ll_queue* queue; // Ссылка на очередь
struct ll_entry* current; // Текущий элемент (NULL = перед head)
struct ll_entry* prev; // Предыдущий элемент (для навигации)
int index; // Текущий индекс (-1 = перед head)
int valid; // Флаг валидности позиции
uint64_t version; // Версия очереди на момент создания/обновления позиции
};
// Создать новую позицию для очереди
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue);
// Освободить позицию (не удаляет элементы очереди)
void queue_pos_free(struct ll_queue_pos* pos);
// Сбросить позицию в начальное состояние (перед head)
void queue_pos_reset(struct ll_queue_pos* pos);
// Перейти к следующему элементу
// Возвращает: 1 при успехе, 0 если достигнут конец
int queue_pos_next(struct ll_queue_pos* pos);
// Перейти к предыдущему элементу
// Возвращает: 1 при успехе, 0 если достигнуто начало
int queue_pos_prev(struct ll_queue_pos* pos);
// Перейти к конкретному индексу (0 = head)
// Возвращает: 1 при успехе, 0 если индекс неверен
int queue_pos_seek(struct ll_queue_pos* pos, int index);
// Получить текущий элемент без удаления
static inline struct ll_entry* queue_pos_current(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return NULL;
return pos->current;
}
// Получить текущий индекс
static inline int queue_pos_index(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return -1;
return pos->index;
}
// Обновить позицию после изменения очереди (сбросить версию)
static inline void queue_pos_refresh(struct ll_queue_pos* pos) {
if (!pos || !pos->queue) return;
pos->version = pos->queue->version;
pos->valid = 1;
}
// Вставить элемент относительно текущей позиции
// mode: 0 = перед current, 1 = после current
// Возвращает: 0 при успехе, -1 при ошибке
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode);
// Удалить элемент из текущей позиции и вернуть его
// После удаления current указывает на следующий элемент (или NULL)
// Возвращает: указатель на удаленный элемент или NULL
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos);
// Удалить конкретный элемент из очереди по указателю
// Возвращает: 0 при успехе, -1 если элемент не найден
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry);
#endif // LL_QUEUE_H
+224
View File
@@ -0,0 +1,224 @@
// 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 ll_queue_pos; // Управляющая структура позиции
// Тип коллбэка: вызывается при добавлении элемента в пустую очередь или для продолжения обработки
// Параметры: указатель на очередь, указатель на элемент (первый в очереди), пользовательский аргумент
typedef void (*queue_callback_fn)(struct ll_queue* q, struct ll_entry* entry, void* arg);
// Структура элемента - переменный размер, данные расположены сразу после структуры
struct ll_entry {
struct ll_entry* next; // Указатель на следующий элемент в очереди
struct ll_entry* prev; // Указатель на предыдущий элемент в очереди
size_t size; // Размер данных элемента (байт)
int ref_count; // Счетчик ссылок для предотвращения double-free
};
// Структура условия ожидания (waiter)
struct queue_waiter {
int max_packets; // Максимальное количество пакетов
size_t max_bytes; // Максимальное количество байт (0 = не проверять байты)
void (*callback)(struct ll_queue* q, void* arg); // Коллбэк для вызова
void* callback_arg; // Аргумент коллбэка
struct queue_waiter* next; // Следующий ожидающий в списке
};
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; // 1 если коллбэки приостановлены (во время обработки)
void* resume_timeout_id; // ID таймаута uasync для отложенного возобновления
struct UASYNC* ua; // Экземпляр uasync для таймеров
struct queue_waiter* waiters; // Список ожидающих коллбэков
// Пулы памяти для оптимизации аллокаций
struct memory_pool* pool; // Пул для структур struct queue_waiter
uint64_t version; // Версия для инвалидации позиций
};
// ==================== Управление очередью ====================
// Создать новую пустую очередь
// ua - экземпляр uasync для таймеров (обязательный параметр)
// pool - если не null то использовать этот пул памяти
// Возвращает: указатель на очередь или NULL при ошибке выделения памяти (see memory_pool.c/h)
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool);
// Освободить очередь и все её элементы
// Также отменяет отложенное возобновление если оно запланировано
void queue_free(struct ll_queue* q);
// ==================== Конфигурация очереди ====================
// Установить функцию и аргумент коллбэка для автозабора из очереди
// Коллбэк вызывается когда в очереди есть элемент и разрешен коллбэк
// обработчик должен обработать этот пакет (может использовать асинхронное ожидание). Когда будет готов к приёму следующего - должен вызвать 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);
// Установить максимальное количество элементов в очереди
// При превышении лимита новый элемент автоматически освобождается
void queue_set_size_limit(struct ll_queue* q, int lim);
// ==================== Управление элементами ====================
// Создать новый элемент с областью данных указанного размера
// Память выделяется одним блоком: [struct ll_entry][область данных data_size байт]
// Возвращает: указатель на элемент или NULL при ошибке выделения памяти
struct ll_entry* queue_entry_new(size_t data_size);
// Освободить элемент (не влияет на связи в очереди)
void queue_entry_free(struct ll_entry* entry);
// ==================== Операции с очередью ====================
// Добавить элемент в конец очереди (FIFO)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry);
// Добавить элемент в начало очереди (LIFO, высокий приоритет)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry);
// Извлечь элемент из начала очереди
// При извлечении приостанавливает коллбэки (callback_suspended = 1) чтобы предотвратить рекурсию
// Возвращает: указатель на элемент или NULL если очередь пуста
struct ll_entry* queue_entry_get(struct ll_queue* q);
// Получить текущее количество элементов в очереди
int queue_entry_count(struct ll_queue* q);
// ==================== Вспомогательные функции ====================
// Получить указатель на область данных элемента
// Данные расположены сразу после структуры struct ll_entry
static inline void* ll_entry_data(struct ll_entry* entry) {
if (!entry) return NULL;
return (void*)(entry + 1);
}
// Получить размер данных элемента
static inline size_t ll_entry_size(struct ll_entry* entry) {
if (!entry) return 0;
return entry->size;
}
// ==================== Асинхронное ожидание ====================
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь
// не более 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;
}
// Получить статистику использования пулов памяти
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse);
// ==================== Управление позициями ====================
// Структура управления позицией для навигации по очереди
struct ll_queue_pos {
struct ll_queue* queue; // Ссылка на очередь
struct ll_entry* current; // Текущий элемент (NULL = перед head)
struct ll_entry* prev; // Предыдущий элемент (для навигации)
int index; // Текущий индекс (-1 = перед head)
int valid; // Флаг валидности позиции
uint64_t version; // Версия очереди на момент создания/обновления позиции
};
// Создать новую позицию для очереди
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue);
// Освободить позицию (не удаляет элементы очереди)
void queue_pos_free(struct ll_queue_pos* pos);
// Сбросить позицию в начальное состояние (перед head)
void queue_pos_reset(struct ll_queue_pos* pos);
// Перейти к следующему элементу
// Возвращает: 1 при успехе, 0 если достигнут конец
int queue_pos_next(struct ll_queue_pos* pos);
// Перейти к предыдущему элементу
// Возвращает: 1 при успехе, 0 если достигнуто начало
int queue_pos_prev(struct ll_queue_pos* pos);
// Перейти к конкретному индексу (0 = head)
// Возвращает: 1 при успехе, 0 если индекс неверен
int queue_pos_seek(struct ll_queue_pos* pos, int index);
// Получить текущий элемент без удаления
static inline struct ll_entry* queue_pos_current(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return NULL;
return pos->current;
}
// Получить текущий индекс
static inline int queue_pos_index(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return -1;
return pos->index;
}
// Обновить позицию после изменения очереди (сбросить версию)
static inline void queue_pos_refresh(struct ll_queue_pos* pos) {
if (!pos || !pos->queue) return;
pos->version = pos->queue->version;
pos->valid = 1;
}
// Вставить элемент относительно текущей позиции
// mode: 0 = перед current, 1 = после current
// Возвращает: 0 при успехе, -1 при ошибке
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode);
// Удалить элемент из текущей позиции и вернуть его
// После удаления current указывает на следующий элемент (или NULL)
// Возвращает: указатель на удаленный элемент или NULL
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos);
// Удалить конкретный элемент из очереди по указателю
// Возвращает: 0 при успехе, -1 если элемент не найден
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry);
#endif // LL_QUEUE_H
+224
View File
@@ -0,0 +1,224 @@
// 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 ll_queue_pos; // Управляющая структура позиции
// Тип коллбэка: вызывается при добавлении элемента в пустую очередь или для продолжения обработки
// Параметры: указатель на очередь, указатель на элемент (первый в очереди), пользовательский аргумент
typedef void (*queue_callback_fn)(struct ll_queue* q, struct ll_entry* entry, void* arg);
// Структура элемента - переменный размер, данные расположены сразу после структуры
struct ll_entry {
struct ll_entry* next; // Указатель на следующий элемент в очереди
struct ll_entry* prev; // Указатель на предыдущий элемент в очереди
size_t size; // Размер данных элемента (байт)
int ref_count; // Счетчик ссылок для предотвращения double-free
};
// Структура условия ожидания (waiter)
struct queue_waiter {
int max_packets; // Максимальное количество пакетов
size_t max_bytes; // Максимальное количество байт (0 = не проверять байты)
void (*callback)(struct ll_queue* q, void* arg); // Коллбэк для вызова
void* callback_arg; // Аргумент коллбэка
struct queue_waiter* next; // Следующий ожидающий в списке
};
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; // 1 если коллбэки приостановлены (во время обработки)
void* resume_timeout_id; // ID таймаута uasync для отложенного возобновления
struct UASYNC* ua; // Экземпляр uasync для таймеров
struct queue_waiter* waiters; // Список ожидающих коллбэков
// Пулы памяти для оптимизации аллокаций
struct memory_pool* pool; // Пул для структур struct queue_waiter
uint64_t version; // Версия для инвалидации позиций
};
// ==================== Управление очередью ====================
// Создать новую пустую очередь
// ua - экземпляр uasync для таймеров (обязательный параметр)
// pool - если не null то использовать этот пул памяти
// Возвращает: указатель на очередь или NULL при ошибке выделения памяти (see memory_pool.c/h)
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool);
// Освободить очередь и все её элементы
// Также отменяет отложенное возобновление если оно запланировано
void queue_free(struct ll_queue* q);
// ==================== Конфигурация очереди ====================
// Установить функцию и аргумент коллбэка для автозабора из очереди
// Коллбэк вызывается когда в очереди есть элемент и разрешен коллбэк
// обработчик должен обработать этот пакет (может использовать асинхронное ожидание). Когда будет готов к приёму следующего - должен вызвать 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);
// Установить максимальное количество элементов в очереди
// При превышении лимита новый элемент автоматически освобождается
void queue_set_size_limit(struct ll_queue* q, int lim);
// ==================== Управление элементами ====================
// Создать новый элемент с областью данных указанного размера
// Память выделяется одним блоком: [struct ll_entry][область данных data_size байт]
// Возвращает: указатель на элемент или NULL при ошибке выделения памяти
struct ll_entry* queue_entry_new(size_t data_size);
// Освободить элемент (не влияет на связи в очереди)
void queue_entry_free(struct ll_entry* entry);
// ==================== Операции с очередью ====================
// Добавить элемент в конец очереди (FIFO)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry);
// Добавить элемент в начало очереди (LIFO, высокий приоритет)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry);
// Извлечь элемент из начала очереди
// При извлечении приостанавливает коллбэки (callback_suspended = 1) чтобы предотвратить рекурсию
// Возвращает: указатель на элемент или NULL если очередь пуста
struct ll_entry* queue_entry_get(struct ll_queue* q);
// Получить текущее количество элементов в очереди
int queue_entry_count(struct ll_queue* q);
// ==================== Вспомогательные функции ====================
// Получить указатель на область данных элемента
// Данные расположены сразу после структуры struct ll_entry
static inline void* ll_entry_data(struct ll_entry* entry) {
if (!entry) return NULL;
return (void*)(entry + 1);
}
// Получить размер данных элемента
static inline size_t ll_entry_size(struct ll_entry* entry) {
if (!entry) return 0;
return entry->size;
}
// ==================== Асинхронное ожидание ====================
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь
// не более 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;
}
// Получить статистику использования пулов памяти
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse);
// ==================== Управление позициями ====================
// Структура управления позицией для навигации по очереди
struct ll_queue_pos {
struct ll_queue* queue; // Ссылка на очередь
struct ll_entry* current; // Текущий элемент (NULL = перед head)
struct ll_entry* prev; // Предыдущий элемент (для навигации)
int index; // Текущий индекс (-1 = перед head)
int valid; // Флаг валидности позиции
uint64_t version; // Версия очереди на момент создания/обновления позиции
};
// Создать новую позицию для очереди
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue);
// Освободить позицию (не удаляет элементы очереди)
void queue_pos_free(struct ll_queue_pos* pos);
// Сбросить позицию в начальное состояние (перед head)
void queue_pos_reset(struct ll_queue_pos* pos);
// Перейти к следующему элементу
// Возвращает: 1 при успехе, 0 если достигнут конец
int queue_pos_next(struct ll_queue_pos* pos);
// Перейти к предыдущему элементу
// Возвращает: 1 при успехе, 0 если достигнуто начало
int queue_pos_prev(struct ll_queue_pos* pos);
// Перейти к конкретному индексу (0 = head)
// Возвращает: 1 при успехе, 0 если индекс неверен
int queue_pos_seek(struct ll_queue_pos* pos, int index);
// Получить текущий элемент без удаления
static inline struct ll_entry* queue_pos_current(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return NULL;
return pos->current;
}
// Получить текущий индекс
static inline int queue_pos_index(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return -1;
return pos->index;
}
// Обновить позицию после изменения очереди (сбросить версию)
static inline void queue_pos_refresh(struct ll_queue_pos* pos) {
if (!pos || !pos->queue) return;
pos->version = pos->queue->version;
pos->valid = 1;
}
// Вставить элемент относительно текущей позиции
// mode: 0 = перед current, 1 = после current
// Возвращает: 0 при успехе, -1 при ошибке
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode);
// Удалить элемент из текущей позиции и вернуть его
// После удаления current указывает на следующий элемент (или NULL)
// Возвращает: указатель на удаленный элемент или NULL
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos);
// Удалить конкретный элемент из очереди по указателю
// Возвращает: 0 при успехе, -1 если элемент не найден
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry);
#endif // LL_QUEUE_H
+210
View File
@@ -0,0 +1,210 @@
#ifndef LL_QUEUE_H
#define LL_QUEUE_H
#include <stddef.h> // для size_t
#include "memory_pool.h" // для struct memory_pool
// Предварительные объявления
struct ll_queue;
struct ll_entry;
struct queue_waiter;
struct memory_pool;
struct ll_queue_pos; // Управляющая структура позиции
// Тип коллбэка: вызывается при добавлении элемента в пустую очередь или для продолжения обработки
// Параметры: указатель на очередь, указатель на элемент (первый в очереди), пользовательский аргумент
typedef void (*queue_callback_fn)(struct ll_queue* q, struct ll_entry* entry, void* arg);
// Структура элемента - переменный размер, данные расположены сразу после структуры
struct ll_entry {
struct ll_entry* next; // Указатель на следующий элемент в очереди
struct ll_entry* prev; // Указатель на предыдущий элемент в очереди
size_t size; // Размер данных элемента (байт)
int ref_count; // Счетчик ссылок для предотвращения double-free
};
// Структура условия ожидания (waiter)
struct queue_waiter {
int max_packets; // Максимальное количество пакетов
size_t max_bytes; // Максимальное количество байт
void (*callback)(struct ll_queue* q, void* arg); // Коллбэк для вызова
void* callback_arg; // Аргумент коллбэка
struct queue_waiter* next; // Следующий ожидающий в списке
};
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; // 1 если коллбэки приостановлены (во время обработки)
void* resume_timeout_id; // ID таймаута uasync для отложенного возобновления
struct UASYNC* ua; // Экземпляр uasync для таймеров
struct queue_waiter* waiters; // Список ожидающих коллбэков
// Пулы памяти для оптимизации аллокаций
struct memory_pool* pool; // Пул для структур struct queue_waiter
};
// ==================== Управление очередью ====================
// Создать новую пустую очередь
// ua - экземпляр uasync для таймеров (обязательный параметр)
// pool - если не null то использовать этот пул памяти
// Возвращает: указатель на очередь или NULL при ошибке выделения памяти (see memory_pool.c/h)
struct ll_queue* queue_new(struct UASYNC* ua, struct memory_pool* pool);
// Освободить очередь и все её элементы
// Также отменяет отложенное возобновление если оно запланировано
void queue_free(struct ll_queue* q);
// ==================== Конфигурация очереди ====================
// Установить функцию и аргумент коллбэка для автозабора из очереди
// Коллбэк вызывается когда в очереди есть элемент и разрешен коллбэк
// обработчик должен обработать этот пакет (может использовать асинхронное ожидание). Когда будет готов к приёму следующего - должен вызвать 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);
// Установить максимальное количество элементов в очереди
// При превышении лимита новый элемент автоматически освобождается
void queue_set_size_limit(struct ll_queue* q, int lim);
// ==================== Управление элементами ====================
// Создать новый элемент с областью данных указанного размера
// Память выделяется одним блоком: [struct ll_entry][область данных data_size байт]
// Возвращает: указатель на элемент или NULL при ошибке выделения памяти
struct ll_entry* queue_entry_new(size_t data_size);
// Освободить элемент (не влияет на связи в очереди)
void queue_entry_free(struct ll_entry* entry);
// ==================== Операции с очередью ====================
// Добавить элемент в конец очереди (FIFO)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put(struct ll_queue* q, struct ll_entry* entry);
// Добавить элемент в начало очереди (LIFO, высокий приоритет)
// Если очередь была пустой и коллбэки разрешены - вызывает коллбэк
// Возвращает: 0 при успехе, -1 если превышен лимит размера (элемент освобожден)
int queue_entry_put_first(struct ll_queue* q, struct ll_entry* entry);
// Извлечь элемент из начала очереди
// При извлечении приостанавливает коллбэки (callback_suspended = 1) чтобы предотвратить рекурсию
// Возвращает: указатель на элемент или NULL если очередь пуста
struct ll_entry* queue_entry_get(struct ll_queue* q);
// Получить текущее количество элементов в очереди
int queue_entry_count(struct ll_queue* q);
// ==================== Вспомогательные функции ====================
// Получить указатель на область данных элемента
// Данные расположены сразу после структуры struct ll_entry
static inline void* ll_entry_data(struct ll_entry* entry) {
return (void*)(entry + 1);
}
// Получить размер данных элемента
static inline size_t ll_entry_size(struct ll_entry* entry) {
return entry->size;
}
// ==================== Асинхронное ожидание ====================
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь
// не более max_packets пакетов и не более max_bytes байт.
// Если условие уже выполнено, коллбэк вызывается немедленно.
// Можно зарегистрировать несколько ожиданий на одной очереди.
// Возвращает указатель на 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;
}
// Получить статистику использования пулов памяти
void queue_get_pool_stats(struct ll_queue* q, size_t* waiter_allocations, size_t* waiter_reuse);
// ==================== Управление позициями ====================
// Структура управления позицией для навигации по очереди
struct ll_queue_pos {
struct ll_queue* queue; // Ссылка на очередь
struct ll_entry* current; // Текущий элемент (NULL = перед head)
struct ll_entry* prev; // Предыдущий элемент (для навигации)
int index; // Текущий индекс (-1 = перед head)
int valid; // Флаг валидности позиции
};
// Создать новую позицию для очереди
struct ll_queue_pos* queue_pos_new(struct ll_queue* queue);
// Освободить позицию (не удаляет элементы очереди)
void queue_pos_free(struct ll_queue_pos* pos);
// Сбросить позицию в начальное состояние (перед head)
void queue_pos_reset(struct ll_queue_pos* pos);
// Перейти к следующему элементу
// Возвращает: 1 при успехе, 0 если достигнут конец
int queue_pos_next(struct ll_queue_pos* pos);
// Перейти к предыдущему элементу
// Возвращает: 1 при успехе, 0 если достигнуто начало
int queue_pos_prev(struct ll_queue_pos* pos);
// Перейти к конкретному индексу (0 = head)
// Возвращает: 1 при успехе, 0 если индекс неверен
int queue_pos_seek(struct ll_queue_pos* pos, int index);
// Получить текущий элемент без удаления
static inline struct ll_entry* queue_pos_current(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return NULL;
return pos->current;
}
// Получить текущий индекс
static inline int queue_pos_index(struct ll_queue_pos* pos) {
if (!pos || !pos->valid) return -1;
return pos->index;
}
// Вставить элемент относительно текущей позиции
// mode: 0 = перед current, 1 = после current
// Возвращает: 0 при успехе, -1 при ошибке
int queue_pos_insert(struct ll_queue_pos* pos, struct ll_entry* entry, int mode);
// Удалить элемент из текущей позиции и вернуть его
// После удаления current указывает на следующий элемент (или NULL)
// Возвращает: указатель на удаленный элемент или NULL
struct ll_entry* queue_pos_remove(struct ll_queue_pos* pos);
// Удалить конкретный элемент из очереди по указателю
// Возвращает: 0 при успехе, -1 если элемент не найден
int queue_entry_remove(struct ll_queue* q, struct ll_entry* entry);
#endif // LL_QUEUE_H