diff --git a/lib/u_async.c b/lib/u_async.c index 86b097d1..8e4e7b51 100644 --- a/lib/u_async.c +++ b/lib/u_async.c @@ -403,19 +403,23 @@ static void drain_wakeup_pipe(struct UASYNC* ua) { while (read(ua->wakeup_pipe[0], &val, sizeof(val)) > 0) {} } -// Process posted tasks (lock-u_free during execution) +/* Стек пачек сохраняет доступ к ожидающим задачам при отмене из вложенного poll. */ +struct uasync_post_batch { + struct uasync_post_batch* previous; + struct posted_task* head; +}; + +// Исполняем снимок очереди без блокировки; новые post остаются для следующей пачки. static void process_posted_tasks(struct UASYNC* ua) { if (!ua) return; - struct posted_task* list = NULL; - - + struct uasync_post_batch batch = { .previous = ua->posted_batches }; #ifdef _WIN32 EnterCriticalSection(&ua->posted_lock); #else pthread_mutex_lock(&ua->posted_lock); #endif - list = ua->posted_tasks_head; + batch.head = ua->posted_tasks_head; ua->posted_tasks_head = ua->posted_tasks_tail = NULL; #ifdef _WIN32 LeaveCriticalSection(&ua->posted_lock); @@ -423,16 +427,18 @@ static void process_posted_tasks(struct UASYNC* ua) { pthread_mutex_unlock(&ua->posted_lock); #endif - while (list) { - DEBUG_DEBUG(DEBUG_CATEGORY_TUN, "POSTed task get"); - struct posted_task* t = list; - list = list->next; + ua->posted_batches = &batch; + while (batch.head) { + struct posted_task* t = batch.head; + batch.head = t->next; // Исполняемый callback уже не подлежит отмене. + DEBUG_TRACE(DEBUG_CATEGORY_SYS, "uasync: executing post task=%p arg=%p", (void*)t, t->arg); if (t->callback) { t->callback(t->arg); } u_free(t); } + ua->posted_batches = batch.previous; } #ifdef _WIN32 @@ -1846,7 +1852,7 @@ void uasync_post_reserved(struct UASYNC* ua, struct posted_task* task) { uasync_wakeup(ua); // будим mainloop } -/* Владелец отменяет pending completion после join производителя вне callbacks. */ +/* В потоке loop удаляем ожидающую задачу из общей очереди или ещё не исполненной части любой пачки. */ err_t uasync_cancel_post(struct UASYNC* ua, struct posted_task* task) { #ifdef _WIN32 EnterCriticalSection(&ua->posted_lock); @@ -1866,8 +1872,20 @@ err_t uasync_cancel_post(struct UASYNC* ua, struct posted_task* task) { #else pthread_mutex_unlock(&ua->posted_lock); #endif - if (found) u_free(task); - else DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "uasync_cancel_post: completion already taken or absent"); + const char* source = "queue"; + for (struct uasync_post_batch* batch = ua->posted_batches; !found && batch; batch = batch->previous) { + p = &batch->head; + while (*p && *p != task) p = &(*p)->next; + if (*p) { + *p = task->next; + found = 1; + source = "batch"; + } + } + if (found) { + DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "uasync_cancel_post: cancelled task=%p source=%s", (void*)task, source); + u_free(task); + } else DEBUG_DEBUG(DEBUG_CATEGORY_SYS, "uasync_cancel_post: task=%p running or absent", (void*)task); return found ? ERR_OK : ERR_FAIL; } diff --git a/lib/u_async.h b/lib/u_async.h index 0e949b9c..683c180f 100644 --- a/lib/u_async.h +++ b/lib/u_async.h @@ -39,6 +39,7 @@ typedef void (*uasync_post_callback_t)(void* user_arg); #include "memory_pool.h" struct timeout_node; // Forward declaration +struct uasync_post_batch; /* CPU одного потока: 100% = одно ядро. При работе цикла/запросе снимок обновляется раз в 100 мс, без пробуждений. * До первого замера valid=0; window_us растёт до секунды. При задержке замера @@ -87,6 +88,7 @@ struct UASYNC { int use_epoll; // 1 = epoll; 0 = poll на POSIX или select на Windows struct posted_task* posted_tasks_head; struct posted_task* posted_tasks_tail; + struct uasync_post_batch* posted_batches; // Ожидающие задачи вложенных dispatch; только поток loop #ifdef _WIN32 CRITICAL_SECTION posted_lock; @@ -216,8 +218,9 @@ void uasync_post(struct UASYNC* ua, uasync_post_callback_t callback, void* user_ * uasync освобождает task после callback, успешной отмены или при destroy без callback; arg не освобождает. * Публикация не проверяет running/stop. Ошибка wakeup не удаляет уже поставленную задачу и не возвращается. */ void uasync_post_reserved(struct UASYNC* ua, struct posted_task* task); -// Отмена в потоке владельца вне callbacks после завершения производителя; ua/task не NULL. -// ERR_OK — task найден в очереди и освобождён; ERR_FAIL — не найден (в том числе уже взят на исполнение). +// Отмена в потоке loop после завершения производителя; допустима из callback, ua/task не NULL. +// ERR_OK — ожидающий task удалён и освобождён, в том числе из исполняемой пачки; arg остаётся владельцу. +// ERR_FAIL — task отсутствует или его callback уже начат; исполняющийся task отменить нельзя. err_t uasync_cancel_post(struct UASYNC* ua, struct posted_task* task); // Lock/unlock posted_lock у живого ua; не ждёт исполнения задач. post уже публикует данные под этим мьютексом. diff --git a/tests/test_uasync_regressions.c b/tests/test_uasync_regressions.c index c1516100..46ab26d3 100644 --- a/tests/test_uasync_regressions.c +++ b/tests/test_uasync_regressions.c @@ -344,6 +344,59 @@ static int test_reserved_post_cancel(int param) { uasync_destroy(ua, 0); return 0; } +struct post_batch_test { + struct UASYNC* ua; + struct posted_task* running; + struct posted_task* pending[3]; + int target, nested, calls, cancellations, followups, callback_error; +}; + +/* Ожидающий completion отменяется до освобождения его arg, даже из вложенного dispatch. */ +static void cancel_batch_pending(void* arg) { + struct post_batch_test* t = arg; + if (uasync_cancel_post(t->ua, t->running) != ERR_FAIL) t->callback_error = 1; + if (uasync_cancel_post(t->ua, t->pending[t->target]) != ERR_OK) t->callback_error = 1; + t->cancellations++; + + // Публикация и отмена новой задачи не должны затронуть уже извлечённую пачку. + struct posted_task* queued = u_calloc(1, sizeof(*queued)); + if (!queued) { t->callback_error = 1; return; } + queued->callback = count_cb; queued->arg = &t->calls; + uasync_post_reserved(t->ua, queued); + if (uasync_cancel_post(t->ua, queued) != ERR_OK) t->callback_error = 1; + uasync_post(t->ua, count_cb, &t->followups); +} + +static void cancel_batch_running(void* arg) { + struct post_batch_test* t = arg; + if (t->nested) { + uasync_post(t->ua, cancel_batch_pending, t); + uasync_poll(t->ua, 0); + // После возврата из вложенного poll внешняя пачка снова доступна для отмены. + if (uasync_cancel_post(t->ua, t->pending[(t->target + 1) % 3]) != ERR_OK) t->callback_error = 1; + } else cancel_batch_pending(t); + if (t->calls) t->callback_error = 1; +} + +static int test_post_batch_cancel(int param) { + if (param & 1) regression_epoll_create_fail(); + struct post_batch_test t = { .target = (param / 2) % 3, .nested = param >= 6 }; + t.ua = uasync_create(); CHECK(t.ua); + t.running = u_calloc(1, sizeof(*t.running)); CHECK(t.running); + t.running->callback = cancel_batch_running; t.running->arg = &t; + uasync_post_reserved(t.ua, t.running); + for (int i = 0; i < 3; i++) { + t.pending[i] = u_calloc(1, sizeof(*t.pending[i])); CHECK(t.pending[i]); + t.pending[i]->callback = count_cb; t.pending[i]->arg = &t.calls; + uasync_post_reserved(t.ua, t.pending[i]); + } + uasync_poll(t.ua, 0); uasync_poll(t.ua, 0); + CHECK_EQ(t.callback_error, 0); CHECK_EQ(t.cancellations, 1); + CHECK_EQ(t.calls, t.nested ? 1 : 2); CHECK_EQ(t.followups, 1); + CHECK(!t.ua->posted_batches); CHECK(!t.ua->posted_tasks_head); CHECK(!t.ua->posted_tasks_tail); + uasync_destroy(t.ua, 0); return 0; +} + static int test_socket_growth_oom(int param) { struct socket_test t; CHECK_EQ(socket_setup(&t, 0, 1, 0, 0), 0); int high = fcntl(t.fd[0], F_DUPFD, 128); CHECK(high >= 128); @@ -550,6 +603,18 @@ int main(int argc, char** argv) { CASE("post/worker-wakeup/epoll", test_worker_wakeup, 0), CASE("post/worker-wakeup/poll", test_worker_wakeup, 1), CASE("post/cancel-head", test_reserved_post_cancel, 0), CASE("post/cancel-middle", test_reserved_post_cancel, 1), CASE("post/cancel-tail", test_reserved_post_cancel, 2), + CASE("post/cancel-batch-head/epoll", test_post_batch_cancel, 0), + CASE("post/cancel-batch-head/poll", test_post_batch_cancel, 1), + CASE("post/cancel-batch-middle/epoll", test_post_batch_cancel, 2), + CASE("post/cancel-batch-middle/poll", test_post_batch_cancel, 3), + CASE("post/cancel-batch-tail/epoll", test_post_batch_cancel, 4), + CASE("post/cancel-batch-tail/poll", test_post_batch_cancel, 5), + CASE("post/cancel-outer-batch-head/epoll", test_post_batch_cancel, 6), + CASE("post/cancel-outer-batch-head/poll", test_post_batch_cancel, 7), + CASE("post/cancel-outer-batch-middle/epoll", test_post_batch_cancel, 8), + CASE("post/cancel-outer-batch-middle/poll", test_post_batch_cancel, 9), + CASE("post/cancel-outer-batch-tail/epoll", test_post_batch_cancel, 10), + CASE("post/cancel-outer-batch-tail/poll", test_post_batch_cancel, 11), CASE("poll-reserve/create-oom-once", test_poll_create_oom, 1), CASE("poll-reserve/create-oom-persistent", test_poll_create_oom, 0), CASE("poll-reserve/add-oom-fd-recovery", test_poll_growth_oom, 0),