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

Сохрани накопленные исправления аудио, async и диагностики

Browse Source
This commit is contained in:
evgeny committed 2026-10-02 01:49:32 +03:00
1 parent b1e3193899
commit bdcf8b5634
59 files changed
+4611 -1441
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lib/ll_queue.c
+103 -33
View File
@@ -52,15 +52,23 @@ static void on_get_trampoline(void* arg);
static void empty_trampoline(void* arg) {
struct ll_queue* q = (struct ll_queue*)arg;
q->empty_call_soon_id = NULL;
if (q->count != 0) return;
queue_callback_fn cb = q->empty_callback;
void* cb_arg = q->empty_callback_arg;
q->empty_callback = NULL;
q->empty_callback_arg = NULL;
if (q->count == 0 && cb) cb(q, cb_arg);
if (cb) cb(q, cb_arg);
}
static void on_get_trampoline(void* arg) {
struct ll_queue* q = (struct ll_queue*)arg;
q->on_get_call_soon_id = NULL;
if (!q->on_get_pending || !q->on_get) return;
q->on_get_pending--;
if (q->on_get_pending) {
q->on_get_call_soon_id = uasync_call_soon(q->ua, q, on_get_trampoline);
if (!q->on_get_call_soon_id) DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] on_get scheduling failed: pending=%zu", q->name, q->on_get_pending);
}
if (q->on_get) q->on_get(q, q->on_get_arg);
}
@@ -130,6 +138,7 @@ void queue_free(struct ll_queue* q) {
// ВАЖНО: Не освобождаем элементы в очереди - они должны быть извлечены отдельно
// Это упрощает архитектуру и предотвращает double-u_free
while (q->deferred_waiters) queue_waiter_cancel(q, q->deferred_waiters);
// Освободить список waiters (сбросить handle->internal, отменить отложенные вызовы)
struct queue_waiter* w = q->waiter_head;
while (w) {
@@ -159,6 +168,10 @@ void queue_free(struct ll_queue* q) {
q->empty_callback_arg = NULL;
q->on_get = NULL;
q->on_get_arg = NULL;
if (q->on_get_call_soon_id) uasync_call_soon_cancel(q->ua, q->on_get_call_soon_id);
q->on_get_call_soon_id = NULL; q->on_get_pending = 0;
if (q->head) DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] freeing nonempty queue: count=%d; entries remain caller-owned", q->name, q->count);
for (struct ll_entry* e = q->head; e; e = e->next) e->owner = NULL;
u_free(q);
}
@@ -316,22 +329,42 @@ static uint32_t make_hash(const void* data, uint16_t len) {// алгоритм F
return hash;
}
static void waiter_defer_cb(void* arg) {
struct queue_waiter_handle* h = (struct queue_waiter_handle*)arg;
h->call_soon_id = NULL;
if (h->defer_cb) h->defer_cb(h->defer_q, h->defer_arg);
else DEBUG_ERROR(DEBUG_CATEGORY_SYS, "waiter_defer_cb: NULL callback (stale?) q=%p", (void*)h->defer_q);
static void unlink_deferred_waiter(struct ll_queue* q, struct queue_waiter_handle* h) {
struct queue_waiter_handle** p = &q->deferred_waiters;
while (*p && *p != h) p = &(*p)->defer_next;
if (*p) *p = h->defer_next;
else DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] deferred waiter not found: handle=%p", q->name, h);
h->defer_next = NULL; h->defer_q = NULL;
}
static inline void queue_waiter_call(struct ll_queue* q, queue_threshold_callback_fn cb, void* arg,
static void waiter_defer_cb(void* arg) {
struct queue_waiter_handle* h = (struct queue_waiter_handle*)arg;
struct ll_queue* q = h->defer_q;
queue_threshold_callback_fn cb = h->defer_cb;
void* cb_arg = h->defer_arg;
h->call_soon_id = NULL;
unlink_deferred_waiter(q, h);
h->defer_cb = NULL; h->defer_arg = NULL;
if (cb) cb(q, cb_arg);
else DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] deferred waiter has no callback", q->name);
}
static inline int queue_waiter_call(struct ll_queue* q, queue_threshold_callback_fn cb, void* arg,
struct queue_waiter_handle* h) {
if (q->waiter_defer) {
if (h->call_soon_id) { DEBUG_WARN(DEBUG_CATEGORY_SYS, "queue_waiter_call: overwriting existing call_soon_id=%p", h->call_soon_id); uasync_call_soon_cancel(q->ua, h->call_soon_id); }
if (h->call_soon_id) queue_waiter_cancel(h->defer_q, h);
h->defer_q = q; h->defer_cb = cb; h->defer_arg = arg;
h->call_soon_id = uasync_call_soon(q->ua, h, waiter_defer_cb);
if (h->call_soon_id) { h->defer_next = q->deferred_waiters; q->deferred_waiters = h; }
else {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] deferred waiter scheduling failed: handle=%p", q->name, h);
h->defer_q = NULL; h->defer_cb = NULL; h->defer_arg = NULL;
return -1;
}
} else {
cb(q, arg);
}
return 0;
}
// Проверить и запустить первый ожидающий коллбэк (FIFO / round-robin)
@@ -351,13 +384,18 @@ static void check_waiters(struct ll_queue* q) {
queue_threshold_callback_fn cb = waiter->callback;
void* arg = waiter->callback_arg;
u_free(waiter);
if (cb) {
DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "check_waiters: waking head waiter, count=%d<=%d, bytes=%zu<=%zu",
q->count, q->threshold_max_packets, q->total_bytes, q->threshold_max_bytes);
queue_waiter_call(q, cb, arg, h);
if (queue_waiter_call(q, cb, arg, h) < 0) {
// Отказ планирования не теряет ожидание: следующий get/remove повторит попытку.
waiter->next = q->waiter_head; q->waiter_head = waiter;
if (!q->waiter_tail) q->waiter_tail = waiter;
h->internal = waiter;
return;
}
}
u_free(waiter);
}
int queue_data_put(struct ll_queue* q, struct ll_entry* entry) {
@@ -366,7 +404,7 @@ int queue_data_put(struct ll_queue* q, struct ll_entry* entry) {
queue_check_thread(q);
#endif
if (entry->prev || entry->next || q->head == entry || q->tail == entry) {
if (entry->owner || entry->prev || entry->next || q->head == entry || q->tail == entry) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] double put detected! entry=%p prev=%p next=%p len=%u",
q->name, (void*)entry, (void*)entry->prev, (void*)entry->next, entry->len);
abort();
@@ -380,6 +418,7 @@ int queue_data_put(struct ll_queue* q, struct ll_entry* entry) {
}
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] queue full: count=%d limit=%d", q->name, q->count, q->size_limit);
queue_dgram_free(entry);
queue_entry_free(entry);
return -1;
@@ -391,17 +430,18 @@ int queue_data_put(struct ll_queue* q, struct ll_entry* entry) {
if (q->tail) q->tail->next = entry; else q->head = entry;
q->tail = entry;
entry->owner = q;
q->count++;
entry->int_len = entry->len;
q->total_bytes += entry->int_len;
// НЕ вызываем add_to_hash
if (q->count == 1) queue_trigger_callback(q);
#ifdef QUEUE_DEBUG
queue_check_consistency(q);
#endif
// Пользовательский callback может освободить очередь.
if (q->count == 1) queue_trigger_callback(q);
return 0;
}
@@ -411,7 +451,7 @@ int queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry) {
queue_check_thread(q);
#endif
if (entry->prev || entry->next || q->head == entry || q->tail == entry) {
if (entry->owner || entry->prev || entry->next || q->head == entry || q->tail == entry) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] double put_with_index detected! entry=%p prev=%p next=%p",
q->name, (void*)entry, (void*)entry->prev, (void*)entry->next);
abort();
@@ -427,6 +467,7 @@ int queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry) {
entry->index_hash = make_hash(entry->data + q->index_offset, q->index_size);
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] queue full: count=%d limit=%d", q->name, q->count, q->size_limit);
queue_dgram_free(entry);
queue_entry_free(entry);
return -1;
@@ -438,17 +479,18 @@ int queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry) {
if (q->tail) q->tail->next = entry; else q->head = entry;
q->tail = entry;
entry->owner = q;
q->count++;
entry->int_len = entry->len;
q->total_bytes += entry->int_len;
add_to_hash(q, entry);
if (q->count == 1) queue_trigger_callback(q);
#ifdef QUEUE_DEBUG
queue_check_consistency(q);
#endif
// Пользовательский callback может освободить очередь.
if (q->count == 1) queue_trigger_callback(q);
return 0;
}
@@ -458,7 +500,7 @@ int queue_data_put_first(struct ll_queue* q, struct ll_entry* entry) {
queue_check_thread(q);
#endif
if (entry->prev || entry->next || q->head == entry || q->tail == entry) {
if (entry->owner || entry->prev || entry->next || q->head == entry || q->tail == entry) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] double put_first detected! entry=%p prev=%p next=%p",
q->name, (void*)entry, (void*)entry->prev, (void*)entry->next);
abort();
@@ -471,20 +513,25 @@ int queue_data_put_first(struct ll_queue* q, struct ll_entry* entry) {
return -1;
}
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] queue full: count=%d limit=%d", q->name, q->count, q->size_limit);
queue_dgram_free(entry); queue_entry_free(entry); return -1;
}
entry->next = q->head;
entry->prev = NULL;
if (q->head) q->head->prev = entry; else q->tail = entry;
q->head = entry;
entry->owner = q;
q->count++;
entry->int_len = entry->len;
q->total_bytes += entry->int_len;
if (q->count == 1) queue_trigger_callback(q);
#ifdef QUEUE_DEBUG
queue_check_consistency(q);
#endif
// Пользовательский callback может освободить очередь.
if (q->count == 1) queue_trigger_callback(q);
return 0;
}
@@ -495,7 +542,7 @@ int queue_data_put_first_with_index(struct ll_queue* q, struct ll_entry* entry)
queue_check_thread(q);
#endif
if (entry->prev || entry->next || q->head == entry || q->tail == entry) {
if (entry->owner || entry->prev || entry->next || q->head == entry || q->tail == entry) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] double put_first_with_index detected! entry=%p prev=%p next=%p",
q->name, (void*)entry, (void*)entry->prev, (void*)entry->next);
abort();
@@ -511,6 +558,7 @@ int queue_data_put_first_with_index(struct ll_queue* q, struct ll_entry* entry)
entry->index_hash = make_hash(entry->data + q->index_offset, q->index_size);
if (q->size_limit >= 0 && q->count >= q->size_limit) {
DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] queue full: count=%d limit=%d", q->name, q->count, q->size_limit);
queue_dgram_free(entry);
queue_entry_free(entry);
return -1;
@@ -521,17 +569,18 @@ int queue_data_put_first_with_index(struct ll_queue* q, struct ll_entry* entry)
if (q->head) q->head->prev = entry; else q->tail = entry;
q->head = entry;
entry->owner = q;
q->count++;
entry->int_len = entry->len;
q->total_bytes += entry->int_len;
add_to_hash(q, entry);
if (q->count == 1) queue_trigger_callback(q);
#ifdef QUEUE_DEBUG
queue_check_consistency(q);
#endif
// Пользовательский callback может освободить очередь.
if (q->count == 1) queue_trigger_callback(q);
return 0;
}
@@ -554,6 +603,7 @@ struct ll_entry* queue_data_get(struct ll_queue* q) {
entry->next = NULL;
entry->prev = NULL;
entry->owner = NULL;
remove_from_hash(q, entry);
@@ -562,18 +612,21 @@ struct ll_entry* queue_data_get(struct ll_queue* q) {
// Приостановить коллбэки для предотвращения рекурсии
q->callback_suspended = 1;
// Проверить ожидающие коллбэки
check_waiters(q);
if (q->count == 0 && q->empty_callback && !q->empty_call_soon_id)
q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline);
if (q->on_get)
uasync_call_soon(q->ua, q, (timeout_callback_t)on_get_trampoline);
if (q->on_get) {
q->on_get_pending++;
if (!q->on_get_call_soon_id) {
q->on_get_call_soon_id = uasync_call_soon(q->ua, q, on_get_trampoline);
if (!q->on_get_call_soon_id) DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[%s] on_get scheduling failed: pending=%zu", q->name, q->on_get_pending);
}
}
#ifdef QUEUE_DEBUG
queue_check_consistency(q);// !!!! for debug
#endif
check_waiters(q);
return entry;
}
@@ -672,18 +725,24 @@ void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
if (!q) return;
q->on_get = cbk_fn;
q->on_get_arg = arg;
if (!cbk_fn) {
if (q->on_get_call_soon_id) uasync_call_soon_cancel(q->ua, q->on_get_call_soon_id);
q->on_get_call_soon_id = NULL; q->on_get_pending = 0;
}
}
int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h,
queue_threshold_callback_fn callback, void* arg) {
if (!q || !h) return -1;
if (!q || !h || !callback) {
DEBUG_ERROR(DEBUG_CATEGORY_SYS, "queue_waiter_wait: invalid arguments q=%p handle=%p callback=%p", q, h, (void*)callback);
return -1;
}
if (h->internal) queue_waiter_cancel(q, h);
if (h->internal || h->call_soon_id) queue_waiter_cancel(h->call_soon_id ? h->defer_q : q, h);
if (q->count <= q->threshold_max_packets
&& (q->threshold_max_bytes == 0 || q->total_bytes <= q->threshold_max_bytes)) {
queue_waiter_call(q, callback, arg, h);
return 1;
return queue_waiter_call(q, callback, arg, h) < 0 ? -1 : 1;
}
struct queue_waiter* waiter = u_malloc(sizeof(struct queue_waiter));
@@ -709,7 +768,11 @@ void queue_waiter_cancel(struct ll_queue* q, struct queue_waiter_handle* h) {
* queue_waiter_wait по ветке «очередь свободна» планирует call_soon_id,
* не регистрируя waiter. Ранний выход по !internal оставлял бы висячий
* callback с arg=handle, который после teardown разыменовывал бы freed-память. */
if (h->call_soon_id) { uasync_call_soon_cancel(q->ua, h->call_soon_id); h->call_soon_id = NULL; }
if (h->call_soon_id) {
struct ll_queue* deferred_q = h->defer_q;
uasync_call_soon_cancel(deferred_q->ua, h->call_soon_id); h->call_soon_id = NULL;
unlink_deferred_waiter(deferred_q, h); h->defer_cb = NULL; h->defer_arg = NULL;
}
if (!h->internal) return;
struct queue_waiter* waiter = h->internal;
@@ -769,6 +832,10 @@ struct ll_entry* queue_find_next_by_index(struct ll_queue* q, const void* index_
int queue_remove_data(struct ll_queue* q, struct ll_entry* entry) {
if (!q || !entry) return -1;
if (entry->owner != q) {
DEBUG_WARN(DEBUG_CATEGORY_SYS, "[%s] cannot remove foreign entry=%p owner=%p", q->name, entry, entry->owner);
return -1;
}
#ifdef QUEUE_THREAD_CHECK
queue_check_thread(q);
@@ -792,6 +859,7 @@ int queue_remove_data(struct ll_queue* q, struct ll_entry* entry) {
entry->next = NULL;
entry->prev = NULL;
entry->owner = NULL;
remove_from_hash(q, entry);
@@ -799,6 +867,8 @@ int queue_remove_data(struct ll_queue* q, struct ll_entry* entry) {
#ifdef QUEUE_DEBUG
queue_check_consistency(q);// !!!! for debug
#endif
if (q->count == 0 && q->empty_callback && !q->empty_call_soon_id)
q->empty_call_soon_id = uasync_call_soon(q->ua, q, empty_trampoline);
check_waiters(q);
return 0;
}