port the windows_thread_pool from libunifex

Author: ericniebler

CMakeLists.txt

@@ -524,6 +524,12 @@ if (CMAKE_SYSTEM_NAME STREQUAL "Darwin")
   target_compile_definitions(stdexec INTERFACE STDEXEC_ENABLE_LIBDISPATCH)
 endif()
 
+if (CMAKE_SYSTEM_NAME STREQUAL "Windows")
+  CHECK_INCLUDE_FILE_CXX("windows.h" STDEXEC_FOUND_WINDOWS_HEADER)
+  option(STDEXEC_ENABLE_WINDOWS_THREAD_POOL "Enable use of the Windows Thread Pool scheduler" ${STDEXEC_FOUND_WINDOWS_HEADER})
+  target_compile_definitions(stdexec INTERFACE STDEXEC_ENABLE_WINDOWS_THREAD_POOL)
+endif()
+
 option (STDEXEC_ENABLE_NUMA "Enable NUMA affinity for static_thread_pool" OFF)
 if (STDEXEC_ENABLE_NUMA)
   find_package(numa REQUIRED)

include/exec/__detail/__system_context_default_impl.hpp

@@ -22,6 +22,8 @@
 #  include "../libdispatch_queue.hpp" // IWYU pragma: keep
 #elif STDEXEC_ENABLE_IO_URING
 #  include "../linux/io_uring_context.hpp" // IWYU pragma: keep
+#elif STDEXEC_ENABLE_WINDOWS_THREAD_POOL
+#  include "../windows/windows_thread_pool.hpp" // IWYU pragma: keep
 #else
 #  include "../static_thread_pool.hpp" // IWYU pragma: keep
 #endif
@@ -201,7 +203,7 @@ namespace exec::__system_context_default_impl {
       }
 
       uint32_t __end(uint32_t __chunk_index) const noexcept {
-        return std::min(__begin(__chunk_index + 1), __max_size_);
+        return (std::min)(__begin(__chunk_index + 1), __max_size_);
       }
     };
 
@@ -370,6 +372,8 @@ namespace exec::__system_context_default_impl {
   using __parallel_scheduler_backend_impl = __generic_impl<exec::libdispatch_queue>;
 #elif STDEXEC_ENABLE_IO_URING
   using __parallel_scheduler_backend_impl = __generic_impl<exec::io_uring_context>;
+#elif STDEXEC_ENABLE_WINDOWS_THREAD_POOL
+  using __parallel_scheduler_backend_impl = __generic_impl<exec::windows_thread_pool>;
 #else
   using __parallel_scheduler_backend_impl = __generic_impl<exec::static_thread_pool>;
 #endif

include/exec/repeat_effect_until.hpp

@@ -24,12 +24,11 @@
 #include "trampoline_scheduler.hpp"
 
 #include <atomic>
-#include <concepts>
 #include <exception>
 #include <type_traits>
 
 namespace exec {
-  namespace __repeat_effect_until {
+  namespace __repeat_effect {
     using namespace stdexec;
 
     template <class _Sender, class _Receiver>
@@ -66,7 +65,6 @@ namespace exec {
     };
 
     STDEXEC_PRAGMA_PUSH()
-
     STDEXEC_PRAGMA_IGNORE_GNU("-Wtsan")
 
     template <class _Sender, class _Receiver>
@@ -173,6 +171,7 @@ namespace exec {
       __error_t
     >;
 
+    struct __repeat_effect_tag { };
     struct __repeat_effect_until_tag { };
 
     struct __repeat_effect_until_impl : __sexpr_defaults {
@@ -214,22 +213,55 @@ namespace exec {
           });
       }
     };
-  } // namespace __repeat_effect_until
 
-  using __repeat_effect_until::repeat_effect_until_t;
+    struct repeat_effect_t {
+      struct _never {
+        template <class... _Args>
+        STDEXEC_ATTRIBUTE(host, device, always_inline)
+        constexpr auto operator()(_Args &&...) const noexcept -> bool{
+          return false;
+        }
+      };
+
+      template <sender _Sender>
+      auto operator()(_Sender &&__sndr) const {
+        auto __domain = __get_early_domain(__sndr);
+        return stdexec::transform_sender(
+          __domain, __make_sexpr<repeat_effect_t>({}, static_cast<_Sender &&>(__sndr)));
+      }
+
+      STDEXEC_ATTRIBUTE(always_inline)
+      constexpr auto operator()() const -> __binder_back<repeat_effect_t> {
+        return {{}, {}, {}};
+      }
+
+      template <class _Sender>
+      auto transform_sender(_Sender &&__sndr, __ignore) {
+        return __sexpr_apply(
+          static_cast<_Sender &&>(__sndr), []<class _Child>(__ignore, __ignore, _Child __child) {
+            return repeat_effect_until_t{}(stdexec::then(std::move(__child)), _never{});
+          });
+      }
+    };
+  } // namespace __repeat_effect
+
+  using __repeat_effect::repeat_effect_until_t;
   inline constexpr repeat_effect_until_t repeat_effect_until{};
+
+  using __repeat_effect::repeat_effect_t;
+  inline constexpr repeat_effect_t repeat_effect{};
 } // namespace exec
 
 namespace stdexec {
   template <>
-  struct __sexpr_impl<exec::__repeat_effect_until::__repeat_effect_until_tag>
-    : exec::__repeat_effect_until::__repeat_effect_until_impl { }; // namespace stdexec
+  struct __sexpr_impl<exec::__repeat_effect::__repeat_effect_until_tag>
+    : exec::__repeat_effect::__repeat_effect_until_impl { }; // namespace stdexec
 
   template <>
   struct __sexpr_impl<exec::repeat_effect_until_t> : __sexpr_defaults {
     static constexpr auto get_completion_signatures =
       []<class _Sender>(
-        _Sender &&) noexcept -> exec::__repeat_effect_until::__completions_t<__data_of<_Sender>> {
+        _Sender &&) noexcept -> exec::__repeat_effect::__completions_t<__data_of<_Sender>> {
       return {};
     };
   };

include/exec/system_context.hpp

@@ -378,7 +378,7 @@ namespace exec {
           // If we are not parallelizing, we need to run all the iterations sequentially.
           uint32_t __increments = 1;
           if constexpr (!_BulkState::__parallelize) {
-            __increments = __state->__size_;
+            __increments = static_cast<uint32_t>(__state->__size_);
           }
           for (uint32_t __i = __begin; __i < __begin + __increments; __i++) {
             std::apply(
@@ -389,7 +389,7 @@ namespace exec {
           // If we are not parallelizing, we need to pass the entire range to the functor.
           if constexpr (!_BulkState::__parallelize) {
             __begin = 0;
-            __end = __state->__size_;
+            __end = static_cast<uint32_t>(__state->__size_);
           }
           std::apply(
             [&](auto&&... __args) { __state->__fun_(__begin, __end, __args...); },

include/exec/windows/filetime_clock.hpp

@@ -0,0 +1,144 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+
+#include <windows.h>
+
+namespace exec::__win32 {
+
+class filetime_clock {
+public:
+  using rep = std::int64_t;
+  using ratio = std::ratio<1, 10'000'000>; // 100ns
+  using duration = std::chrono::duration<rep, ratio>;
+
+  static constexpr bool is_steady = false;
+
+  class time_point {
+  public:
+    using duration = filetime_clock::duration;
+
+    constexpr time_point() noexcept = default;
+
+    constexpr time_point(const time_point &) noexcept = default;
+
+    auto operator=(const time_point &) noexcept -> time_point & = default;
+
+    [[nodiscard]] auto get_ticks() const noexcept -> std::uint64_t {
+      return ticks_;
+    }
+
+    static constexpr auto(max)() noexcept -> time_point {
+      time_point tp;
+      tp.ticks_ = (std::numeric_limits<std::int64_t>::max)();
+      return tp;
+    }
+
+    static constexpr auto(min)() noexcept -> time_point { return time_point{}; }
+
+    static auto from_ticks(std::uint64_t ticks) noexcept -> time_point {
+      time_point tp;
+      tp.ticks_ = ticks;
+      return tp;
+    }
+
+    template <typename Rep, typename Ratio>
+    auto operator+=(const std::chrono::duration<Rep, Ratio> &d) noexcept
+        -> time_point & {
+      ticks_ += std::chrono::duration_cast<duration>(d).count();
+      return *this;
+    }
+
+    template <typename Rep, typename Ratio>
+    auto operator-=(const std::chrono::duration<Rep, Ratio> &d) noexcept
+        -> time_point & {
+      ticks_ -= std::chrono::duration_cast<duration>(d).count();
+      return *this;
+    }
+
+    friend auto operator-(time_point a, time_point b) noexcept -> duration {
+      return duration{a.ticks_} - duration{b.ticks_};
+    }
+
+    template <typename Rep, typename Ratio>
+    friend auto operator-(time_point t,
+                          std::chrono::duration<Rep, Ratio> d) noexcept
+        -> time_point {
+      time_point tp = t;
+      tp -= d;
+      return tp;
+    }
+
+    template <typename Rep, typename Ratio>
+    friend auto operator+(time_point t,
+                          std::chrono::duration<Rep, Ratio> d) noexcept
+        -> time_point {
+      time_point tp = t;
+      tp += d;
+      return tp;
+    }
+
+    template <typename Rep, typename Ratio>
+    friend auto operator+(std::chrono::duration<Rep, Ratio> d, time_point t) noexcept
+        -> time_point {
+      return t + d;
+    }
+
+    friend auto operator==(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ == b.ticks_;
+    }
+
+    friend auto operator!=(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ != b.ticks_;
+    }
+
+    friend auto operator<(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ < b.ticks_;
+    }
+
+    friend auto operator>(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ > b.ticks_;
+    }
+
+    friend auto operator<=(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ <= b.ticks_;
+    }
+
+    friend auto operator>=(time_point a, time_point b) noexcept -> bool {
+      return a.ticks_ >= b.ticks_;
+    }
+
+  private:
+    // Ticks since Jan 1, 1601 (UTC)
+    std::uint64_t ticks_{};
+  };
+
+  static auto now() noexcept -> time_point {
+    FILETIME filetime;
+    ::GetSystemTimeAsFileTime(&filetime);
+
+    ULARGE_INTEGER ticks;
+    ticks.HighPart = filetime.dwHighDateTime;
+    ticks.LowPart = filetime.dwLowDateTime;
+
+    return time_point::from_ticks(ticks.QuadPart);
+  }
+};
+
+} // namespace exec::__win32