From 9e6892822250c45f5420be26cd2334fdf49ef1af Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 20:10:28 +0100 Subject: [PATCH] Test worker drain with completions in a detached batch --- tests/Makefile.am | 3 ++ tests/test_member_files.c | 108 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+) diff --git a/tests/Makefile.am b/tests/Makefile.am index 8022b889..7e6a2132 100644 --- a/tests/Makefile.am +++ b/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) diff --git a/tests/test_member_files.c b/tests/test_member_files.c index f55b0cfb..a3d68b4d 100644 --- a/tests/test_member_files.c +++ b/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 #include @@ -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; }