Bound iterative-deepening AI worker threads (#8764)

src/main/cpp/net/eagle0/shardok/ai/AICommandEvaluator.cpp

@@ -23,7 +23,6 @@
 
 namespace shardok {
 
-constexpr bool kMultithread = true;
 constexpr bool kLogging = false;
 
 [[nodiscard]] static auto AverageGenerator() -> std::shared_ptr<SequenceRandomGenerator> {
@@ -120,10 +119,12 @@ static auto CommandSorter(
 AICommandEvaluator::AICommandEvaluator(
         const AIScoreCalculator& scorer,
         const APDCache& apdCache,
-        BattalionTypeGetter battalionTypeGetter)
+        BattalionTypeGetter battalionTypeGetter,
+        AIThreadPool& threadPool)
     : scorer_(scorer),
       apdCache_(apdCache),
-      battalionTypeGetter_(std::move(battalionTypeGetter)) {}
+      battalionTypeGetter_(std::move(battalionTypeGetter)),
+      threadPool_(threadPool) {}
 
 auto AICommandEvaluator::PerformLookahead(
         const PlayerId pid,
@@ -295,7 +296,7 @@ auto AICommandEvaluator::EvaluateWithRandomness(
         returnValue.lookaheadScore = p.get_future();
         p.set_value(EvaluationResult{.score = innerUtility, .completed = true});
     } else {
-        auto lookaheadLambda = [this,
+        auto lookaheadLambda = [evaluator = *this,
                                 pid,
                                 isDefender,
                                 remainingLookahead,
@@ -307,8 +308,11 @@ auto AICommandEvaluator::EvaluateWithRandomness(
                                 guessedCommands,
                                 allCastleCoords,
                                 deadline]() -> EvaluationResult {
+            if (std::chrono::steady_clock::now() > deadline) {
+                return EvaluationResult{.score = 0.0, .completed = false};
+            }
             try {
-                auto lookaheadFuture = PerformLookahead(
+                auto lookaheadFuture = evaluator.PerformLookahead(
                         pid,
                         isDefender,
                         remainingLookahead,
@@ -329,15 +333,14 @@ auto AICommandEvaluator::EvaluateWithRandomness(
             }
         };
 
-        if constexpr (kMultithread) {
-            auto launchPolicy =
-                    remainingLookahead == 1 ? std::launch::async : std::launch::deferred;
-            returnValue.lookaheadScore = std::async(launchPolicy, lookaheadLambda);
+        // Only the depth-one frontier is parallel. A pool worker therefore reaches depth zero,
+        // never another queued frontier. AIThreadPool also executes same-pool reentrant
+        // submissions inline as a hard backstop against worker-waits-on-worker deadlocks.
+        if (remainingLookahead == 1) {
+            returnValue.lookaheadScore = threadPool_.Submit(std::move(lookaheadLambda));
         } else {
-            std::promise<EvaluationResult> p;
-            returnValue.lookaheadScore = p.get_future();
-            auto lambdaResult = lookaheadLambda();
-            p.set_value(lambdaResult);
+            returnValue.lookaheadScore =
+                    std::async(std::launch::deferred, std::move(lookaheadLambda));
         }
     }
 

src/main/cpp/net/eagle0/shardok/ai/AICommandEvaluator.hpp

@@ -10,6 +10,7 @@
 #include <future>
 
 #include "src/main/cpp/net/eagle0/shardok/ai/AIStrategy.hpp"
+#include "src/main/cpp/net/eagle0/shardok/ai/AIThreadPool.hpp"
 #include "src/main/cpp/net/eagle0/shardok/library/GameStateW.hpp"
 #include "src/main/cpp/net/eagle0/shardok/library/ShardokCTypes.h"
 #include "src/main/cpp/net/eagle0/shardok/library/action_point_distances/ActionPointDistancesCache.hpp"
@@ -35,7 +36,8 @@ public:
     AICommandEvaluator(
             const AIScoreCalculator& scorer,
             const APDCache& apdCache,
-            BattalionTypeGetter battalionTypeGetter);  // Pass by value
+            BattalionTypeGetter battalionTypeGetter,
+            AIThreadPool& threadPool = SharedAIThreadPool());  // Pass getter by value
 
     /// Evaluates the score for a particular command index with lookahead.
     struct EvaluationResult {
@@ -79,6 +81,7 @@ private:
     const AIScoreCalculator& scorer_;
     const APDCache& apdCache_;
     BattalionTypeGetter battalionTypeGetter_;
+    AIThreadPool& threadPool_;
 
     struct ImmediateAndLookaheadScore {
         ScoreValue immediateScore{};

src/main/cpp/net/eagle0/shardok/ai/AIThreadPool.cpp

@@ -0,0 +1,70 @@
+#include "AIThreadPool.hpp"
+
+#include <algorithm>
+
+namespace shardok {
+
+namespace {
+
+// Keep the process responsive to gRPC and command handling while still giving lookahead enough
+// parallelism to saturate typical production hosts.
+constexpr size_t kMaximumAIWorkerCount = 8;
+
+}  // namespace
+
+thread_local const AIThreadPool* AIThreadPool::currentPool_ = nullptr;
+
+AIThreadPool::AIThreadPool(const size_t workerCount) {
+    const size_t normalizedWorkerCount = std::max<size_t>(1, workerCount);
+    workers_.reserve(normalizedWorkerCount);
+    try {
+        for (size_t index = 0; index < normalizedWorkerCount; ++index) {
+            workers_.emplace_back([this] { RunWorker(); });
+        }
+    } catch (...) {
+        StopAndJoin();
+        throw;
+    }
+}
+
+AIThreadPool::~AIThreadPool() { StopAndJoin(); }
+
+void AIThreadPool::RunWorker() {
+    currentPool_ = this;
+    while (true) {
+        std::function<void()> task;
+        {
+            std::unique_lock lock(mutex_);
+            condition_.wait(lock, [this] { return stopping_ || !tasks_.empty(); });
+            if (stopping_ && tasks_.empty()) { break; }
+            task = std::move(tasks_.front());
+            tasks_.pop();
+        }
+        task();
+    }
+    currentPool_ = nullptr;
+}
+
+void AIThreadPool::StopAndJoin() {
+    {
+        std::lock_guard lock(mutex_);
+        stopping_ = true;
+    }
+    condition_.notify_all();
+    for (auto& worker : workers_) {
+        if (worker.joinable()) { worker.join(); }
+    }
+}
+
+auto AIWorkerCountForHardwareConcurrency(const unsigned int hardwareConcurrency) -> size_t {
+    if (hardwareConcurrency <= 1) { return 1; }
+    return std::min(kMaximumAIWorkerCount, static_cast<size_t>(hardwareConcurrency - 1));
+}
+
+auto SharedAIThreadPool() -> AIThreadPool& {
+    static AIThreadPool pool(
+            AIWorkerCountForHardwareConcurrency(std::thread::hardware_concurrency()));
+    return pool;
+}
+
+}  // namespace shardok

src/main/cpp/net/eagle0/shardok/ai/AIThreadPool.hpp

@@ -0,0 +1,79 @@
+//
+// Fixed-size worker pool for iterative-deepening lookahead.
+//
+
+#ifndef EAGLE0_SHARDOK_AI_AI_THREAD_POOL_HPP
+#define EAGLE0_SHARDOK_AI_AI_THREAD_POOL_HPP
+
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <stdexcept>
+#include <thread>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+namespace shardok {
+
+class AIThreadPool {
+public:
+    explicit AIThreadPool(size_t workerCount);
+    ~AIThreadPool();
+
+    AIThreadPool(const AIThreadPool&) = delete;
+    auto operator=(const AIThreadPool&) -> AIThreadPool& = delete;
+    AIThreadPool(AIThreadPool&&) = delete;
+    auto operator=(AIThreadPool&&) -> AIThreadPool& = delete;
+
+    template<typename Function>
+    [[nodiscard]] auto Submit(Function&& function)
+            -> std::future<std::invoke_result_t<std::decay_t<Function>&>> {
+        using Result = std::invoke_result_t<std::decay_t<Function>&>;
+
+        auto task =
+                std::make_shared<std::packaged_task<Result()>>(std::forward<Function>(function));
+        auto result = task->get_future();
+
+        // A worker must never enqueue work and then wait for the same bounded pool. Running
+        // reentrant submissions inline makes that safe even if recursive search scheduling
+        // changes in the future.
+        if (currentPool_ == this) {
+            (*task)();
+            return result;
+        }
+
+        {
+            std::lock_guard lock(mutex_);
+            if (stopping_) { throw std::runtime_error("Submit on stopped AIThreadPool"); }
+            tasks_.emplace([task] { (*task)(); });
+        }
+        condition_.notify_one();
+        return result;
+    }
+
+    [[nodiscard]] auto WorkerCount() const -> size_t { return workers_.size(); }
+
+private:
+    void RunWorker();
+    void StopAndJoin();
+
+    static thread_local const AIThreadPool* currentPool_;
+
+    std::vector<std::thread> workers_;
+    std::queue<std::function<void()>> tasks_;
+    std::mutex mutex_;
+    std::condition_variable condition_;
+    bool stopping_ = false;
+};
+
+[[nodiscard]] auto AIWorkerCountForHardwareConcurrency(unsigned int hardwareConcurrency) -> size_t;
+[[nodiscard]] auto SharedAIThreadPool() -> AIThreadPool&;
+
+}  // namespace shardok
+
+#endif  // EAGLE0_SHARDOK_AI_AI_THREAD_POOL_HPP

src/main/cpp/net/eagle0/shardok/ai/BUILD.bazel

@@ -194,6 +194,16 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "ai_thread_pool",
+    srcs = ["AIThreadPool.cpp"],
+    hdrs = ["AIThreadPool.hpp"],
+    copts = COPTS,
+    visibility = [
+        "//src/test/cpp/net/eagle0/shardok/ai:__pkg__",
+    ],
+)
+
 cc_library(
     name = "ai_command_evaluator",
     srcs = ["AICommandEvaluator.cpp"],
@@ -207,6 +217,7 @@ cc_library(
         ":ai_command_filter",
         ":ai_experiment_config",
         ":ai_strategy",
+        ":ai_thread_pool",
         ":transposition_table",
         "//src/main/cpp/net/eagle0/common:sequence_random_generator",
         "//src/main/cpp/net/eagle0/shardok/ai/score:ai_score_calculator_interface",

src/test/cpp/net/eagle0/shardok/ai/AICommandEvaluator_test.cpp

@@ -173,7 +173,8 @@ TEST_F(AICommandEvaluatorTest, LookaheadFutureOwnsCastleCoords) {
     const auto battalionTypeGetter = [this](BattalionTypeId typeId) {
         return settings->GetGetter().GetBattalionType(typeId);
     };
-    const AICommandEvaluator evaluator(scorer, apdCache, battalionTypeGetter);
+    AIThreadPool threadPool(1);
+    const AICommandEvaluator evaluator(scorer, apdCache, battalionTypeGetter, threadPool);
 
     std::future<AICommandEvaluator::EvaluationResult> result;
     {

src/test/cpp/net/eagle0/shardok/ai/AIThreadPool_test.cpp

@@ -0,0 +1,46 @@
+#include "src/main/cpp/net/eagle0/shardok/ai/AIThreadPool.hpp"
+
+#include <gtest/gtest.h>
+
+#include <thread>
+
+namespace shardok {
+namespace {
+
+TEST(AIThreadPoolTest, CalculatesBoundedWorkerCountAndLeavesOneHardwareThreadFree) {
+    EXPECT_EQ(1, AIWorkerCountForHardwareConcurrency(0));
+    EXPECT_EQ(1, AIWorkerCountForHardwareConcurrency(1));
+    EXPECT_EQ(3, AIWorkerCountForHardwareConcurrency(4));
+    EXPECT_EQ(8, AIWorkerCountForHardwareConcurrency(16));
+    EXPECT_EQ(8, AIWorkerCountForHardwareConcurrency(128));
+}
+
+TEST(AIThreadPoolTest, NormalizesZeroWorkersToOne) {
+    const AIThreadPool pool(0);
+    EXPECT_EQ(1, pool.WorkerCount());
+}
+
+TEST(AIThreadPoolTest, ReusesWorkerBetweenSubmissions) {
+    AIThreadPool pool(1);
+
+    const auto firstThread = pool.Submit([] { return std::this_thread::get_id(); }).get();
+    const auto secondThread = pool.Submit([] { return std::this_thread::get_id(); }).get();
+
+    EXPECT_EQ(firstThread, secondThread);
+}
+
+TEST(AIThreadPoolTest, RunsNestedSubmissionInlineSoAWorkerCanWaitForIt) {
+    AIThreadPool pool(1);
+
+    auto outer = pool.Submit([&pool] {
+        const auto outerThread = std::this_thread::get_id();
+        auto inner =
+                pool.Submit([outerThread] { return std::this_thread::get_id() == outerThread; });
+        return inner.get();
+    });
+
+    EXPECT_TRUE(outer.get());
+}
+
+}  // namespace
+}  // namespace shardok

src/test/cpp/net/eagle0/shardok/ai/BUILD.bazel

@@ -37,6 +37,18 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "ai_thread_pool_test",
+    srcs = ["AIThreadPool_test.cpp"],
+    copts = TEST_COPTS,
+    linkstatic = True,
+    deps = [
+        "//src/main/cpp/net/eagle0/shardok/ai:ai_thread_pool",
+        "@googletest//:gtest",
+        "@googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "ai_flee_decision_calculator_test",
     srcs = ["AIFleeDecisionCalculator_test.cpp"],