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

@@ -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

Help/release/dev/ctest-windows-jobserver.rst

@@ -0,0 +1,5 @@
+ctest-windows-jobserver
+-----------------------
+
+* :manual:`ctest(1)` now supports :ref:`job server integration
+  <ctest-job-server-integration>` on Windows.

Source/CTest/cmUVJobServerClient.cxx

@@ -3,17 +3,18 @@
 #include "cmUVJobServerClient.h"
 
 #include <cassert>
+#include <cstdio>
+#include <string>
 #include <utility>
+#include <vector>
 
 #ifndef _WIN32
-#  include <cstdio>
-#  include <string>
-#  include <vector>
-
 #  include <fcntl.h>
 #  include <unistd.h>
 
 #  include <sys/types.h>
+#else
+#  include <windows.h>
 #endif
 
 #include <cm/memory>
@@ -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<cm::string_view> const prefixes = {
+    "--jobserver-auth=", "--jobserver-fds=", "-J"
+  };
+
+  cm::optional<std::string> makeflags = cmSystemTools::GetEnvVar("MAKEFLAGS");
+  if (!makeflags) {
+    return;
+  }
+
+  cm::optional<std::string> auth;
+  std::vector<std::string> 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<ImplWin32*>(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<ImplWin32*>(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> cmUVJobServerClient::Connect(
   std::function<void(int)> onDisconnect)
 {
 #if defined(_WIN32)
-  // FIXME: Windows job server client not yet implemented.
-  static_cast<void>(loop);
-  static_cast<void>(onToken);
-  static_cast<void>(onDisconnect);
+  auto impl = cm::make_unique<ImplWin32>(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<ImplPosix>(loop);
   if (impl && impl->IsConnected()) {

Tests/CMakeLib/testUVJobServerClient.cxx

@@ -9,7 +9,9 @@
 
 #include <cm3p/uv.h>
 
-#ifndef _WIN32
+#ifdef _WIN32
+#  include <windows.h>
+#else
 #  include <unistd.h>
 #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<LONG>(count), nullptr);
+  }
+};
+
+struct MakeFlagsGuard
+{
+  cm::optional<std::string> 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<cmUVJobServerClient> 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<cmUVJobServerClient> 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<cmUVJobServerClient> 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<cmUVJobServerClient> 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<int>(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<cmUVJobServerClient> 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;
 }