Cancel pending posts from active dispatch batches

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;
 }
 

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 уже публикует данные под этим мьютексом.

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),