Test worker drain with completions in a detached batch

tests/Makefile.am

@@ -1247,6 +1247,9 @@ test_tun_route_LDFLAGS = -Wl,--wrap=sendto -Wl,--wrap=recvfrom -Wl,--wrap=__recv
 test_member_files_SOURCES = test_member_files.c
 test_member_files_CFLAGS = $(test_services_CFLAGS)
 test_member_files_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+if OS_LINUX
+test_member_files_LDFLAGS = -Wl,--wrap=uasync_post_reserved
+endif
 test_member_files_network_SOURCES = test_member_files_network.c $(test_clock_sources)
 test_member_files_network_CFLAGS = $(test_services_CFLAGS)
 test_member_files_network_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)

tests/test_member_files.c

@@ -9,6 +9,7 @@
 #include "../src/member_files/worker.h"
 #include "../lib/u_async.h"
 #include "../lib/debug_config.h"
+#include "../lib/mem.h"
 #include "test_utils.h"
 #include <openssl/sha.h>
 #include <string.h>
@@ -414,6 +415,110 @@ static void worker_test(void) {
     uasync_destroy(t.ua, 1);
 }
 
+#ifdef __linux__
+static pthread_mutex_t post_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t post_condition = PTHREAD_COND_INITIALIZER;
+static struct UASYNC* observed_ua;
+static pthread_t observed_owner;
+static unsigned worker_posts;
+
+/* Барьер только наблюдает настоящую публикацию: completion не подменяется и не исполняется вручную. */
+extern void __real_uasync_post_reserved(struct UASYNC*, struct posted_task*);
+void __wrap_uasync_post_reserved(struct UASYNC* ua, struct posted_task* task) {
+    __real_uasync_post_reserved(ua, task);
+    pthread_mutex_lock(&post_lock);
+    if (ua == observed_ua && !pthread_equal(pthread_self(), observed_owner)) {
+        worker_posts++; pthread_cond_signal(&post_condition);
+    }
+    pthread_mutex_unlock(&post_lock);
+}
+
+static void wait_worker_posts(unsigned count) {
+    pthread_mutex_lock(&post_lock);
+    while (worker_posts < count) pthread_cond_wait(&post_condition, &post_lock);
+    pthread_mutex_unlock(&post_lock);
+}
+
+struct drain_case;
+struct drain_job { struct drain_case* test; unsigned index, worked, done; int disk_thread, owner_thread, error; };
+struct drain_case {
+    struct UASYNC* ua;
+    struct mf_worker* worker;
+    struct mf_work* cleanup;
+    pthread_t owner;
+    struct drain_job jobs[3];
+    int destroy, stops;
+};
+
+static void drain_work(void* arg) {
+    struct drain_job* j = arg;
+    j->worked++; j->disk_thread = !pthread_equal(pthread_self(), j->test->owner);
+}
+
+/* Первый done добавляет обязательную cleanup continuation, в том числе во время drain/destroy. */
+static void drain_done(void* arg, int error) {
+    struct drain_job* j = arg;
+    struct drain_case* t = j->test;
+    j->done++; j->error = error; j->owner_thread = pthread_equal(pthread_self(), t->owner);
+    if (!j->index) {
+        struct mf_work* cleanup = t->cleanup; t->cleanup = NULL;
+        mf_worker_submit_reserved(t->worker, cleanup, drain_work, &t->jobs[2], drain_done, &t->jobs[2]);
+    }
+}
+
+static void drain_in_batch(void* arg) {
+    struct drain_case* t = arg; t->stops++;
+    if (t->destroy) { mf_worker_destroy(t->worker); t->worker = NULL; }
+    else mf_worker_drain(t->worker);
+    for (unsigned i = 0; i < 3; i++) EXPECT(t->jobs[i].done == 1, "stop delivers each done before returning");
+}
+
+/* normal, queued destroy, detached drain, detached destroy; два completion и cleanup в каждом случае. */
+static void worker_completion_test(void) {
+    for (unsigned mode = 0; mode < 4; mode++) {
+        size_t before = u_get_allocated_count();
+        struct drain_case t = {0}; t.owner = pthread_self(); t.destroy = mode == 3;
+        t.ua = uasync_create(); EXPECT(t.ua != NULL, "drain fixture uasync"); if (!t.ua) return;
+        t.worker = mf_worker_create(t.ua); t.cleanup = mf_worker_reserve();
+        EXPECT(t.worker && t.cleanup, "drain fixture worker and cleanup reserve");
+        if (!t.worker || !t.cleanup) {
+            mf_worker_destroy(t.worker); mf_worker_unreserve(t.cleanup); uasync_destroy(t.ua, 1); return;
+        }
+        for (unsigned i = 0; i < 3; i++) { t.jobs[i].test = &t; t.jobs[i].index = i; }
+        pthread_mutex_lock(&post_lock);
+        observed_ua = t.ua; observed_owner = t.owner; worker_posts = 0;
+        pthread_mutex_unlock(&post_lock);
+        if (mode >= 2) {
+            /* Stop раньше completion, а poll начнётся только после обеих публикаций. */
+            struct posted_task* stop = u_calloc(1, sizeof(*stop));
+            EXPECT(stop != NULL, "stop task reserve");
+            if (!stop) {
+                mf_worker_destroy(t.worker); mf_worker_unreserve(t.cleanup); uasync_destroy(t.ua, 1);
+                pthread_mutex_lock(&post_lock); observed_ua = NULL; pthread_mutex_unlock(&post_lock); return;
+            }
+            stop->callback = drain_in_batch; stop->arg = &t; uasync_post_reserved(t.ua, stop);
+        }
+        for (unsigned i = 0; i < 2; i++) mf_worker_submit(t.worker, drain_work, &t.jobs[i], drain_done, &t.jobs[i]);
+        wait_worker_posts(2);
+        EXPECT(!t.jobs[0].done && !t.jobs[1].done, "publication does not deliver done before poll/drain");
+        if (mode == 1) { mf_worker_destroy(t.worker); t.worker = NULL; }
+        uasync_poll(t.ua, 0);
+        wait_worker_posts(3); uasync_poll(t.ua, 0);
+        EXPECT(t.stops == (mode >= 2), "batch stop exactly once");
+        for (unsigned i = 0; i < 3; i++) {
+            struct drain_job* j = &t.jobs[i];
+            EXPECT(j->worked == 1 && j->done == 1 && !j->error && j->disk_thread && j->owner_thread,
+                   "work and done once, correct threads, cleanup completed");
+        }
+        EXPECT(t.cleanup == NULL, "cleanup reserve transferred exactly once");
+        mf_worker_destroy(t.worker); uasync_poll(t.ua, 0); uasync_destroy(t.ua, 1);
+        pthread_mutex_lock(&post_lock); observed_ua = NULL; pthread_mutex_unlock(&post_lock);
+        EXPECT(u_get_allocated_count() == before, "worker completion tasks and jobs all freed");
+        DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "worker completion regression mode=%u: work/done=3/3, cleanup and allocations verified", mode);
+    }
+}
+#endif
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG);
     char directory[1024] = "/tmp/utun_member_files_XXXXXX";
@@ -424,6 +529,9 @@ int main(void) {
     local_fifo_test(directory); large_offset_test(directory);
 #endif
     worker_test();
+#ifdef __linux__
+    worker_completion_test();
+#endif
     printf("member_files: %s failures=%d directory=%s\n", failures ? "FAIL" : "PASS", failures, directory);
     return failures ? 1 : 0;
 }