From 79331d58656a6950f17f3162920c290725dd9c75 Mon Sep 17 00:00:00 2001 From: Daksh Mamodiya Date: Sat, 18 Jul 2026 18:29:45 +0200 Subject: [PATCH] CTest: Add Windows job server client Use a named semaphore polled from the libuv event loop to share GNU Make job slots on Windows. Cover token ordering, bounded draining, and teardown, and document the new support. Fixes: #27968 --- Help/manual/ctest.1.rst | 15 +- Help/release/dev/ctest-windows-jobserver.rst | 5 + Source/CTest/cmUVJobServerClient.cxx | 222 +++++++++++- Tests/CMakeLib/testUVJobServerClient.cxx | 347 ++++++++++++++++++- 4 files changed, 565 insertions(+), 24 deletions(-) create mode 100644 Help/release/dev/ctest-windows-jobserver.rst diff --git a/Help/manual/ctest.1.rst b/Help/manual/ctest.1.rst index 9151aec460..b72536f85f 100644 --- a/Help/manual/ctest.1.rst +++ b/Help/manual/ctest.1.rst @@ -2264,11 +2264,14 @@ Job Server Integration .. versionadded:: 3.29 -On POSIX systems, when running under the context of a `Job Server`_, -CTest shares its job slots. This is independent of the :prop_test:`PROCESSORS` -test property, which still counts against CTest's :ctest-option:`-j` parallel -level. CTest acquires exactly one token from the job server before running -each test, and returns it when the test finishes. +.. versionchanged:: 4.5 + Added support for job server integration on Windows. + +When running under the context of a `Job Server`_, CTest shares its job slots. +This is independent of the :prop_test:`PROCESSORS` test property, which still +counts against CTest's :ctest-option:`-j` parallel level. CTest acquires +exactly one token from the job server before running each test, and returns it +when the test finishes. For example, consider the ``Makefile``: @@ -2278,8 +2281,6 @@ For example, consider the ``Makefile``: When invoked via ``make -j 2 test``, CTest connects to the job server, acquires a token for each test, and runs at most 2 tests concurrently. -On Windows systems, job server integration is not yet implemented. - .. _`Job Server`: https://www.gnu.org/software/make/manual/html_node/Job-Slots.html See Also diff --git a/Help/release/dev/ctest-windows-jobserver.rst b/Help/release/dev/ctest-windows-jobserver.rst new file mode 100644 index 0000000000..d688047c1e --- /dev/null +++ b/Help/release/dev/ctest-windows-jobserver.rst @@ -0,0 +1,5 @@ +ctest-windows-jobserver +----------------------- + +* :manual:`ctest(1)` now supports :ref:`job server integration + ` on Windows. diff --git a/Source/CTest/cmUVJobServerClient.cxx b/Source/CTest/cmUVJobServerClient.cxx index 19fed2c499..8aa337487d 100644 --- a/Source/CTest/cmUVJobServerClient.cxx +++ b/Source/CTest/cmUVJobServerClient.cxx @@ -3,17 +3,18 @@ #include "cmUVJobServerClient.h" #include +#include +#include #include +#include #ifndef _WIN32 -# include -# include -# include - # include # include # include +#else +# include #endif #include @@ -468,6 +469,209 @@ void ImplPosix::StopReceivingTokens() } #endif +//--------------------------------------------------------------------------- +// Implementation on Windows platforms. +// https://www.gnu.org/software/make/manual/html_node/Windows-Jobserver.html + +#ifdef _WIN32 +namespace { +unsigned int const JobServerPollInterval = 10; +unsigned int const JobServerMaxTokensPerPoll = 32; + +class ImplWin32 : public cmUVJobServerClient::Impl +{ +public: + enum class Connection + { + None, + Semaphore, + }; + Connection Conn = Connection::None; + + HANDLE Semaphore = nullptr; + cm::uv_timer_ptr PollTimer; + bool ReceivingTokens = false; + + void Connect(); + void Disconnect(int status); + void PollTokens(); + + bool IsConnected() const; + + void SendToken() override; + void StartReceivingTokens() override; + void StopReceivingTokens() override; + + ImplWin32(uv_loop_t& loop); + ~ImplWin32() override; +}; + +ImplWin32::ImplWin32(uv_loop_t& loop) + : Impl(loop) +{ + if (this->PollTimer.init(this->Loop, this) == 0) { + this->Connect(); + } +} + +ImplWin32::~ImplWin32() +{ + this->Disconnect(0); +} + +void ImplWin32::Connect() +{ + static std::vector const prefixes = { + "--jobserver-auth=", "--jobserver-fds=", "-J" + }; + + cm::optional makeflags = cmSystemTools::GetEnvVar("MAKEFLAGS"); + if (!makeflags) { + return; + } + + cm::optional auth; + std::vector args; + cmSystemTools::ParseUnixCommandLine(makeflags->c_str(), args); + for (cm::string_view arg : cmReverseRange(args)) { + for (cm::string_view prefix : prefixes) { + if (cmHasPrefix(arg, prefix)) { + auth = cmTrimWhitespace(arg.substr(prefix.length())); + break; + } + } + if (auth) { + break; + } + } + + if (!auth || auth->empty()) { + return; + } + + int reader; + int writer; + char trailing; + if (cmHasLiteralPrefix(*auth, "fifo:") || + std::sscanf(auth->c_str(), "%d,%d%c", &reader, &writer, &trailing) == + 2) { + return; + } + + HANDLE semaphore = + OpenSemaphoreA(SYNCHRONIZE | SEMAPHORE_MODIFY_STATE, FALSE, auth->c_str()); + if (!semaphore) { + return; + } + + this->Semaphore = semaphore; + this->Conn = Connection::Semaphore; +} + +void ImplWin32::Disconnect(int status) +{ + if (this->Conn == Connection::None) { + return; + } + + this->StopReceivingTokens(); + this->Conn = Connection::None; + CloseHandle(this->Semaphore); + this->Semaphore = nullptr; + this->PollTimer.reset(); + + if (status != 0) { + this->Disconnected(status); + } +} + +void ImplWin32::PollTokens() +{ + unsigned int taken = 0; + while (this->Conn == Connection::Semaphore && this->NeedTokens > 0 && + !uv_is_active(this->ImplicitToken) && + taken < JobServerMaxTokensPerPoll) { + DWORD const result = WaitForSingleObject(this->Semaphore, 0); + if (result == WAIT_OBJECT_0) { + ++taken; + this->ReceivedToken(); + } else if (result == WAIT_TIMEOUT) { + return; + } else if (result == WAIT_FAILED) { + this->Disconnect(uv_translate_sys_error(GetLastError())); + return; + } else { + assert(result != WAIT_ABANDONED); + this->Disconnect(UV_EINVAL); + return; + } + } + + if (this->Conn == Connection::Semaphore && this->NeedTokens > 0 && + !uv_is_active(this->ImplicitToken) && + taken == JobServerMaxTokensPerPoll) { + int const status = uv_timer_start( + this->PollTimer, + [](uv_timer_t* handle) { + static_cast(handle->data)->PollTokens(); + }, + 0, JobServerPollInterval); + if (status != 0) { + this->Disconnect(status); + } + } +} + +bool ImplWin32::IsConnected() const +{ + return this->Conn != Connection::None; +} + +void ImplWin32::SendToken() +{ + if (this->Conn == Connection::None) { + return; + } + + if (!ReleaseSemaphore(this->Semaphore, 1, nullptr)) { + DWORD const error = GetLastError(); + assert(error != ERROR_TOO_MANY_POSTS); + this->Disconnect(uv_translate_sys_error(error)); + } +} + +void ImplWin32::StartReceivingTokens() +{ + if (this->Conn == Connection::None || this->ReceivingTokens) { + return; + } + + int const status = uv_timer_start( + this->PollTimer, + [](uv_timer_t* handle) { + static_cast(handle->data)->PollTokens(); + }, + 0, JobServerPollInterval); + if (status != 0) { + this->Disconnect(status); + return; + } + + this->ReceivingTokens = true; +} + +void ImplWin32::StopReceivingTokens() +{ + if (this->Conn == Connection::None || !this->ReceivingTokens) { + return; + } + + this->ReceivingTokens = false; + uv_timer_stop(this->PollTimer); +} +} +#endif + //--------------------------------------------------------------------------- // Implementation of public interface. @@ -508,10 +712,12 @@ cm::optional cmUVJobServerClient::Connect( std::function onDisconnect) { #if defined(_WIN32) - // FIXME: Windows job server client not yet implemented. - static_cast(loop); - static_cast(onToken); - static_cast(onDisconnect); + auto impl = cm::make_unique(loop); + if (impl && impl->IsConnected()) { + impl->OnToken = std::move(onToken); + impl->OnDisconnect = std::move(onDisconnect); + return cmUVJobServerClient(std::move(impl)); + } #else auto impl = cm::make_unique(loop); if (impl && impl->IsConnected()) { diff --git a/Tests/CMakeLib/testUVJobServerClient.cxx b/Tests/CMakeLib/testUVJobServerClient.cxx index 9fb2708fed..09e987b321 100644 --- a/Tests/CMakeLib/testUVJobServerClient.cxx +++ b/Tests/CMakeLib/testUVJobServerClient.cxx @@ -9,7 +9,9 @@ #include -#ifndef _WIN32 +#ifdef _WIN32 +# include +#else # include #endif @@ -24,6 +26,97 @@ namespace { std::size_t const kTOTAL_JOBS = 10; std::size_t const kTOTAL_TOKENS = 3; +#ifdef _WIN32 +std::size_t NextSemaphoreId = 0; + +struct JobServerSemaphore +{ + std::string Name; + HANDLE Handle = nullptr; + + JobServerSemaphore(LONG initialCount, LONG maximumCount) + : Name(cmStrCat("cmake_test_jobserver_", GetCurrentProcessId(), '_', + ++NextSemaphoreId)) + { + SetLastError(ERROR_SUCCESS); + this->Handle = CreateSemaphoreA(nullptr, initialCount, maximumCount, + this->Name.c_str()); + if (this->Handle && GetLastError() == ERROR_ALREADY_EXISTS) { + CloseHandle(this->Handle); + this->Handle = nullptr; + } + } + + ~JobServerSemaphore() + { + if (this->Handle) { + CloseHandle(this->Handle); + } + } + + JobServerSemaphore(JobServerSemaphore const&) = delete; + JobServerSemaphore& operator=(JobServerSemaphore const&) = delete; + + explicit operator bool() const { return this->Handle != nullptr; } + + bool CountAvailableTokens(std::size_t& count) const + { + count = 0; + for (;;) { + DWORD const result = WaitForSingleObject(this->Handle, 0); + if (result == WAIT_OBJECT_0) { + ++count; + } else if (result == WAIT_TIMEOUT) { + break; + } else { + return false; + } + } + + return count == 0 || + ReleaseSemaphore(this->Handle, static_cast(count), nullptr); + } +}; + +struct MakeFlagsGuard +{ + cm::optional Original = cmSystemTools::GetEnvVar("MAKEFLAGS"); + + ~MakeFlagsGuard() + { + if (this->Original) { + cmSystemTools::PutEnv(cmStrCat("MAKEFLAGS=", *this->Original)); + } else { + cmSystemTools::UnsetEnv("MAKEFLAGS"); + } + } +}; + +void SetJobServer(JobServerSemaphore const& jobServer) +{ + cmSystemTools::PutEnv(cmStrCat("MAKEFLAGS=--flags-before" + " --jobserver-auth=bogus" + " --flags-between" + " --jobserver-auth=", + jobServer.Name, " --flags-after")); +} + +bool CheckTokenCount(JobServerSemaphore const& jobServer, std::size_t expected) +{ + std::size_t count; + if (!jobServer.CountAvailableTokens(count)) { + std::cerr << "Failed to inspect job server semaphore\n"; + return false; + } + if (count != expected) { + std::cerr << "Expected " << expected << " job server tokens, got " << count + << '\n'; + return false; + } + return true; +} +#endif + struct Job { cm::uv_timer_ptr Timer; @@ -64,12 +157,7 @@ struct JobRunner std::cerr << "HeldTokens: " << this->JSC->GetHeldTokens() << '\n'; std::cerr << "NeedTokens: " << this->JSC->GetNeedTokens() << '\n'; } -#ifdef _WIN32 - // FIXME: Windows job server client not yet implemented. - return true; -#else return this->Okay; -#endif } void QueueNextJobs() @@ -140,7 +228,12 @@ struct JobRunner bool testJobServer() { #ifdef _WIN32 - // FIXME: Windows job server client not yet implemented. + JobServerSemaphore jobServer(kTOTAL_TOKENS - 1, kTOTAL_TOKENS - 1); + if (!jobServer) { + std::cerr << "Failed to create job server semaphore\n"; + return false; + } + SetJobServer(jobServer); #else // Create a job server pipe. int jobServerPipe[2]; @@ -167,14 +260,250 @@ bool testJobServer() " --flags-after")); #endif - JobRunner jobRunner; - return jobRunner.Run(); + bool passed; + { + JobRunner jobRunner; + passed = jobRunner.Run(); + } +#ifdef _WIN32 + passed = CheckTokenCount(jobServer, kTOTAL_TOKENS - 1) && passed; +#endif + return passed; } + +#ifdef _WIN32 +bool ConnectsWithMakeFlags(std::string const& makeFlags) +{ + cmSystemTools::PutEnv(cmStrCat("MAKEFLAGS=", makeFlags)); + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + cm::optional client = + cmUVJobServerClient::Connect(*loop, nullptr, nullptr); + return client.has_value(); +} + +bool testJobServerParsing() +{ + JobServerSemaphore jobServer(1, 1); + if (!jobServer) { + std::cerr << "Failed to create parsing test semaphore\n"; + return false; + } + + bool passed = true; + passed = + ConnectsWithMakeFlags(cmStrCat("--jobserver-auth=", jobServer.Name)) && + passed; + passed = ConnectsWithMakeFlags(cmStrCat("--jobserver-auth=bogus " + "--jobserver-fds=1,2 " + "--jobserver-auth=", + jobServer.Name)) && + passed; + passed = !ConnectsWithMakeFlags(cmStrCat("--jobserver-auth=", jobServer.Name, + " --jobserver-fds=1,2")) && + passed; + passed = !ConnectsWithMakeFlags("--jobserver-auth=fifo:somewhere") && passed; + passed = !ConnectsWithMakeFlags("--jobserver-auth=") && passed; + cmSystemTools::UnsetEnv("MAKEFLAGS"); + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + passed = !cmUVJobServerClient::Connect(*loop, nullptr, nullptr) && passed; + + if (!passed) { + std::cerr << "Job server MAKEFLAGS parsing test failed\n"; + } + return passed; +} + +bool testDeferredImplicitTokenOrdering() +{ + JobServerSemaphore jobServer(1, 1); + if (!jobServer) { + return false; + } + SetJobServer(jobServer); + + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + + std::size_t deliveries = 0; + cm::optional client; + client = cmUVJobServerClient::Connect( + *loop, + [&]() { + ++deliveries; + client->ReleaseToken(); + }, + nullptr); + if (!client) { + return false; + } + + client->RequestToken(); + client->RequestToken(); + bool passed = deliveries == 0; + uv_run(loop, UV_RUN_DEFAULT); + + passed = deliveries == 2 && client->GetHeldTokens() == 0 && + client->GetNeedTokens() == 0 && CheckTokenCount(jobServer, 1) && passed; + if (!passed) { + std::cerr << "Deferred implicit-token ordering test failed\n"; + } + return passed; +} + +bool testTimerRestartFromCallback() +{ + JobServerSemaphore jobServer(2, 2); + if (!jobServer) { + return false; + } + SetJobServer(jobServer); + + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + + std::size_t deliveries = 0; + cm::optional client; + client = cmUVJobServerClient::Connect( + *loop, + [&]() { + ++deliveries; + if (deliveries == 2) { + client->ReleaseToken(); + client->RequestToken(); + } else if (deliveries == 3) { + client->ReleaseToken(); + } + }, + nullptr); + if (!client) { + return false; + } + + client->RequestToken(); + uv_run(loop, UV_RUN_DEFAULT); + bool passed = deliveries == 1 && client->GetHeldTokens() == 1; + + client->RequestToken(); + uv_run(loop, UV_RUN_DEFAULT); + passed = deliveries == 3 && client->GetHeldTokens() == 1 && + client->GetNeedTokens() == 0 && passed; + client->ReleaseToken(); + passed = CheckTokenCount(jobServer, 2) && passed; + if (!passed) { + std::cerr << "Timer restart test failed\n"; + } + return passed; +} + +bool testBoundedTokenDrain() +{ + std::size_t const explicitTokens = 40; + JobServerSemaphore jobServer(explicitTokens, explicitTokens); + if (!jobServer) { + return false; + } + SetJobServer(jobServer); + + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + + std::size_t deliveries = 0; + cm::optional client = + cmUVJobServerClient::Connect(*loop, [&]() { ++deliveries; }, nullptr); + if (!client) { + return false; + } + + client->RequestToken(); + uv_run(loop, UV_RUN_DEFAULT); + for (std::size_t i = 0; i < explicitTokens; ++i) { + client->RequestToken(); + } + + uv_run(loop, UV_RUN_NOWAIT); + bool passed = deliveries == 33 && client->GetNeedTokens() == 8; + uv_run(loop, UV_RUN_DEFAULT); + passed = deliveries == explicitTokens + 1 && + client->GetHeldTokens() == static_cast(explicitTokens + 1) && + client->GetNeedTokens() == 0 && passed; + + for (std::size_t i = 0; i < explicitTokens + 1; ++i) { + client->ReleaseToken(); + } + passed = CheckTokenCount(jobServer, explicitTokens) && passed; + if (!passed) { + std::cerr << "Bounded token drain test failed\n"; + } + return passed; +} + +bool testPendingRequestTeardown() +{ + JobServerSemaphore jobServer(0, 1); + if (!jobServer) { + return false; + } + SetJobServer(jobServer); + + cm::uv_loop_ptr loop; + if (loop.init(nullptr) != 0) { + return false; + } + + std::size_t deliveries = 0; + { + cm::optional client = + cmUVJobServerClient::Connect(*loop, [&]() { ++deliveries; }, nullptr); + if (!client) { + return false; + } + + client->RequestToken(); + uv_run(loop, UV_RUN_NOWAIT); + client->RequestToken(); + uv_run(loop, UV_RUN_NOWAIT); + if (deliveries != 1 || client->GetHeldTokens() != 1 || + client->GetNeedTokens() != 1) { + std::cerr << "Pending request setup failed\n"; + return false; + } + } + + uv_run(loop, UV_RUN_DEFAULT); + bool const passed = CheckTokenCount(jobServer, 0); + if (!passed) { + std::cerr << "Pending request teardown test failed\n"; + } + return passed; +} +#endif } int testUVJobServerClient(int, char** const) { +#ifdef _WIN32 + MakeFlagsGuard makeFlagsGuard; +#endif bool passed = true; passed = testJobServer() && passed; +#ifdef _WIN32 + passed = testJobServerParsing() && passed; + passed = testDeferredImplicitTokenOrdering() && passed; + passed = testTimerRestartFromCallback() && passed; + passed = testBoundedTokenDrain() && passed; + passed = testPendingRequestTeardown() && passed; +#endif return passed ? 0 : -1; }