From 89b89376977f97a312049ada3ef981586b17612f Mon Sep 17 00:00:00 2001 From: csh Date: Mon, 13 Jul 2026 16:03:56 +0800 Subject: [PATCH] :bug: fix(async_executor): implement cooperative cancellation --- lsp-server/src/core/dispatcher.cppm | 13 +- lsp-server/src/core/server.cppm | 2 +- lsp-server/src/manager/bootstrap.cppm | 16 +- lsp-server/src/manager/symbol.cppm | 72 ++- .../src/provider/initialize/initialize.cppm | 15 +- .../did_change_workspace_folders.cppm | 52 +- .../provider/workspace/execute_command.cppm | 9 +- lsp-server/src/scheduler/async_executor.cppm | 492 ++++++++---------- .../test/test_provider/completion_test.cppm | 2 +- .../test/test_provider/definitions_test.cppm | 2 +- .../test/test_provider/interpreter_test.cppm | 2 +- .../test/test_provider/json_flow_test.cppm | 2 +- .../json_provider_coverage_test.cppm | 4 +- .../test_provider/provider_misc_test.cppm | 4 +- .../test_provider/provider_surface_test.cppm | 2 +- lsp-server/test/test_scheduler/main.cc | 2 +- .../test_scheduler/test_async_executor.cppm | 414 +++++++++++---- 17 files changed, 685 insertions(+), 420 deletions(-) diff --git a/lsp-server/src/core/dispatcher.cppm b/lsp-server/src/core/dispatcher.cppm index 37a1fd4..37ca790 100644 --- a/lsp-server/src/core/dispatcher.cppm +++ b/lsp-server/src/core/dispatcher.cppm @@ -1,6 +1,5 @@ module; - export module lsp.core.dispatcher; import spdlog; @@ -51,13 +50,13 @@ export namespace lsp::core { public: ExecutionContext(LifecycleCallback lifecycle_callback, - scheduler::AsyncExecutor& scheduler, + scheduler::async_executor::AsyncExecutor& scheduler, manager::ManagerHub& manager_hub) : lifecycle_callback_(lifecycle_callback), async_executor_(scheduler), manager_hub_(manager_hub) {} - scheduler::AsyncExecutor& GetScheduler() const { return async_executor_; } + scheduler::async_executor::AsyncExecutor& GetScheduler() const { return async_executor_; } manager::ManagerHub& GetManagerHub() const { return manager_hub_; } void TriggerLifecycleEvent(ServerLifecycleEvent event) const @@ -68,7 +67,7 @@ export namespace lsp::core private: LifecycleCallback lifecycle_callback_; - scheduler::AsyncExecutor& async_executor_; + scheduler::async_executor::AsyncExecutor& async_executor_; manager::ManagerHub& manager_hub_; }; @@ -80,7 +79,7 @@ export namespace lsp::core RequestDispatcher(); ~RequestDispatcher() = default; - void SetRequestScheduler(scheduler::AsyncExecutor* scheduler); + void SetRequestScheduler(scheduler::async_executor::AsyncExecutor* scheduler); void SetManagerHub(manager::ManagerHub* manager_hub); void RegisterRequestProvider(std::shared_ptr provider); @@ -114,7 +113,7 @@ export namespace lsp::core LifecycleCallback context_lifecycle_callback_; - scheduler::AsyncExecutor* async_executor_ = nullptr; + scheduler::async_executor::AsyncExecutor* async_executor_ = nullptr; manager::ManagerHub* manager_hub_ = nullptr; }; } @@ -128,7 +127,7 @@ namespace lsp::core }; } - void RequestDispatcher::SetRequestScheduler(scheduler::AsyncExecutor* scheduler) + void RequestDispatcher::SetRequestScheduler(scheduler::async_executor::AsyncExecutor* scheduler) { async_executor_ = scheduler; spdlog::debug("Request scheduler set"); diff --git a/lsp-server/src/core/server.cppm b/lsp-server/src/core/server.cppm index 0e76301..fd2c516 100644 --- a/lsp-server/src/core/server.cppm +++ b/lsp-server/src/core/server.cppm @@ -75,7 +75,7 @@ export namespace lsp::core private: RequestDispatcher dispatcher_; manager::ManagerHub manager_hub_; - scheduler::AsyncExecutor async_executor_; + scheduler::async_executor::AsyncExecutor async_executor_; std::string interpreter_path_; std::atomic is_initialized_ = false; diff --git a/lsp-server/src/manager/bootstrap.cppm b/lsp-server/src/manager/bootstrap.cppm index bc7b980..54340fe 100644 --- a/lsp-server/src/manager/bootstrap.cppm +++ b/lsp-server/src/manager/bootstrap.cppm @@ -12,7 +12,7 @@ export namespace lsp::manager::bootstrap { void InitializeManagerHub( ManagerHub& hub, - scheduler::AsyncExecutor& async_executor, + scheduler::async_executor::AsyncExecutor& async_executor, const std::vector& system_lib_paths); } @@ -20,7 +20,7 @@ namespace lsp::manager::bootstrap { void InitializeManagerHub( ManagerHub& hub, - scheduler::AsyncExecutor& async_executor, + scheduler::async_executor::AsyncExecutor& async_executor, const std::vector& system_lib_paths) { spdlog::info("Initializing manager hub..."); @@ -31,10 +31,10 @@ namespace lsp::manager::bootstrap async_executor.Submit( task_name, - [&hub, path]() -> std::optional { + [&hub, path](std::stop_token stop_token) -> std::optional { try { - hub.symbols().LoadSystemLibrary(path); + hub.symbols().LoadSystemLibrary(path, stop_token); return std::format("Loaded system library: {}", path); } catch (const std::exception& e) @@ -43,11 +43,11 @@ namespace lsp::manager::bootstrap throw; } }, - [path](const std::optional& result, bool cancelled) { - if (cancelled) + [path](const scheduler::async_executor::TaskResult& result) { + if (result.status == scheduler::async_executor::TaskStatus::kCancelled) spdlog::info("System library load task cancelled: {}", path); - else if (result) - spdlog::info("{}", *result); + else if (result.status == scheduler::async_executor::TaskStatus::kCompleted && result.value) + spdlog::info("{}", *result.value); }); } spdlog::info("Manager hub initialized, system library loading in background"); diff --git a/lsp-server/src/manager/symbol.cppm b/lsp-server/src/manager/symbol.cppm index 9192670..81c5a65 100644 --- a/lsp-server/src/manager/symbol.cppm +++ b/lsp-server/src/manager/symbol.cppm @@ -37,11 +37,11 @@ export namespace lsp::manager explicit Symbol(EventBus& event_bus); ~Symbol(); - void LoadSystemLibrary(const std::string& lib_path); - void LoadWorkspace(const protocol::DocumentUri& workspace_uri); + void LoadSystemLibrary(const std::string& lib_path, std::stop_token stop_token = {}); + void LoadWorkspace(const protocol::DocumentUri& workspace_uri, std::stop_token stop_token = {}); - void IndexWorkspaceFiles(const std::vector& uris); - void RemoveWorkspaceFiles(const std::vector& uris); + void IndexWorkspaceFiles(const std::vector& uris, std::stop_token stop_token = {}); + void RemoveWorkspaceFiles(const std::vector& uris, std::stop_token stop_token = {}); void RenameWorkspaceFiles(const std::vector>& files); const language::symbol::SymbolTable* GetSymbolTable(const protocol::DocumentUri& uri) const; @@ -306,11 +306,16 @@ namespace lsp::manager Symbol::~Symbol() = default; - void Symbol::LoadSystemLibrary(const std::string& lib_path) + void Symbol::LoadSystemLibrary(const std::string& lib_path, std::stop_token stop_token) { spdlog::info("Loading system library from: {}", lib_path); auto start = std::chrono::steady_clock::now(); + if (stop_token.stop_requested()) + { + return; + } + if (!std::filesystem::exists(lib_path)) { spdlog::warn("System library path does not exist: {}", lib_path); @@ -326,6 +331,11 @@ namespace lsp::manager for (const auto& entry : std::filesystem::recursive_directory_iterator(lib_path, options)) { + if (stop_token.stop_requested()) + { + return; + } + if (!entry.is_regular_file()) continue; @@ -338,6 +348,10 @@ namespace lsp::manager spdlog::trace("Indexing library file: {}", entry.path().string()); auto table = BuildSymbolTableFromFile(entry.path()); + if (stop_token.stop_requested()) + { + return; + } if (!table) { spdlog::trace("Failed to build symbol table for: {}", entry.path().string()); @@ -362,6 +376,11 @@ namespace lsp::manager ++loaded; } + if (stop_token.stop_requested()) + { + return; + } + { std::unique_lock lock(mutex_); system_symbols_ = std::move(new_symbols); @@ -378,12 +397,17 @@ namespace lsp::manager duration); } - void Symbol::LoadWorkspace(const protocol::DocumentUri& workspace_uri) + void Symbol::LoadWorkspace(const protocol::DocumentUri& workspace_uri, std::stop_token stop_token) { auto workspace_path = UriToPath(workspace_uri); spdlog::info("Loading workspace from: {}", workspace_path); auto start = std::chrono::steady_clock::now(); + if (stop_token.stop_requested()) + { + return; + } + if (!std::filesystem::exists(workspace_path)) { spdlog::warn("Workspace path does not exist: {}", workspace_path); @@ -399,6 +423,11 @@ namespace lsp::manager for (const auto& entry : std::filesystem::recursive_directory_iterator(workspace_path, options)) { + if (stop_token.stop_requested()) + { + return; + } + if (!entry.is_regular_file()) continue; @@ -407,6 +436,10 @@ namespace lsp::manager continue; auto table = BuildSymbolTableFromFile(entry.path()); + if (stop_token.stop_requested()) + { + return; + } if (!table) { ++failed; @@ -431,6 +464,11 @@ namespace lsp::manager ++loaded; } + if (stop_token.stop_requested()) + { + return; + } + { std::unique_lock lock(mutex_); workspace_symbols_ = std::move(new_symbols); @@ -447,7 +485,7 @@ namespace lsp::manager duration); } - void Symbol::IndexWorkspaceFiles(const std::vector& uris) + void Symbol::IndexWorkspaceFiles(const std::vector& uris, std::stop_token stop_token) { std::unordered_map updates; std::vector removals; @@ -457,6 +495,11 @@ namespace lsp::manager for (const auto& uri : uris) { + if (stop_token.stop_requested()) + { + return; + } + auto file_path = std::filesystem::path(UriToPath(uri)); auto kind = GetTslFileKind(file_path); if (kind == TslFileKind::kOther) @@ -473,6 +516,10 @@ namespace lsp::manager } auto table = BuildSymbolTableFromFile(file_path); + if (stop_token.stop_requested()) + { + return; + } if (!table) { removals.push_back(std::move(normalized_uri)); @@ -494,7 +541,7 @@ namespace lsp::manager updates[normalized_uri] = std::move(stored); } - if (updates.empty() && removals.empty()) + if (stop_token.stop_requested() || (updates.empty() && removals.empty())) { return; } @@ -513,7 +560,7 @@ namespace lsp::manager } } - void Symbol::RemoveWorkspaceFiles(const std::vector& uris) + void Symbol::RemoveWorkspaceFiles(const std::vector& uris, std::stop_token stop_token) { if (uris.empty()) { @@ -525,6 +572,11 @@ namespace lsp::manager for (const auto& uri : uris) { + if (stop_token.stop_requested()) + { + return; + } + auto file_path = std::filesystem::path(UriToPath(uri)); auto kind = GetTslFileKind(file_path); if (kind == TslFileKind::kOther) @@ -534,7 +586,7 @@ namespace lsp::manager removals.push_back(PathToUri(file_path)); } - if (removals.empty()) + if (stop_token.stop_requested() || removals.empty()) { return; } diff --git a/lsp-server/src/provider/initialize/initialize.cppm b/lsp-server/src/provider/initialize/initialize.cppm index cef3952..465ed7a 100644 --- a/lsp-server/src/provider/initialize/initialize.cppm +++ b/lsp-server/src/provider/initialize/initialize.cppm @@ -39,9 +39,6 @@ export namespace lsp::provider namespace lsp::provider { - - - std::string Initialize::ProvideResponse(const protocol::RequestMessage& request, ExecutionContext& context) { @@ -219,24 +216,24 @@ namespace lsp::provider for (const auto& workspace_folder : workspace_folders) { auto task_id = std::format("Load workspace symbols: {}", workspace_folder.uri); - scheduler.Submit(task_id, [&manager_hub, uri = workspace_folder.uri, folder_name = workspace_folder.name]() -> std::optional { + scheduler.Submit(task_id, [&manager_hub, uri = workspace_folder.uri, folder_name = workspace_folder.name](std::stop_token stop_token) -> std::optional { try { - manager_hub.symbols().LoadWorkspace(uri); + manager_hub.symbols().LoadWorkspace(uri, stop_token); return std::format("Loaded workspace {} symbols", uri); } catch (const std::exception& e) { spdlog::error("Failed to load workspace {} symbols: {}", folder_name, e.what()); throw; // Request会处理异常 - } }, [](const std::optional& result, bool cancelled) { - if (cancelled) + } }, [](const ::lsp::scheduler::async_executor::TaskResult& result) { + if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCancelled) { spdlog::debug("Workspace loading task cancelled"); } - else if (result) + else if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCompleted && result.value) { - spdlog::info("Workspace loading result: {}", *result); + spdlog::info("Workspace loading result: {}", *result.value); } }); } diff --git a/lsp-server/src/provider/workspace/did_change_workspace_folders.cppm b/lsp-server/src/provider/workspace/did_change_workspace_folders.cppm index dd10f2a..5bb0e06 100644 --- a/lsp-server/src/provider/workspace/did_change_workspace_folders.cppm +++ b/lsp-server/src/provider/workspace/did_change_workspace_folders.cppm @@ -1,6 +1,5 @@ module; - export module lsp.provider.workspace.did_change_workspace_folders; import spdlog; @@ -9,6 +8,7 @@ import std; import lsp.protocol; import lsp.codec.facade; import lsp.provider.base.interface; +import lsp.scheduler.async_executor; export namespace lsp::provider::workspace { @@ -123,10 +123,17 @@ namespace lsp::provider::workspace return folders; } - std::vector EnumerateWorkspaceFiles(const protocol::DocumentUri& workspace_uri) + std::vector EnumerateWorkspaceFiles( + const protocol::DocumentUri& workspace_uri, + std::stop_token stop_token) { std::vector uris; + if (stop_token.stop_requested()) + { + return uris; + } + std::filesystem::path workspace_path; try { @@ -147,6 +154,11 @@ namespace lsp::provider::workspace for (const auto& entry : std::filesystem::recursive_directory_iterator(workspace_path, options)) { + if (stop_token.stop_requested()) + { + return {}; + } + if (!entry.is_regular_file()) { continue; @@ -211,22 +223,26 @@ namespace lsp::provider::workspace auto task_id = std::format("Remove workspace folder: {}", folder.uri); scheduler.Submit( task_id, - [&hub, uri = folder.uri]() -> std::optional { - auto uris = EnumerateWorkspaceFiles(uri); + [&hub, uri = folder.uri](std::stop_token stop_token) -> std::optional { + auto uris = EnumerateWorkspaceFiles(uri, stop_token); + if (stop_token.stop_requested()) + { + return std::nullopt; + } if (!uris.empty()) { - hub.symbols().RemoveWorkspaceFiles(uris); + hub.symbols().RemoveWorkspaceFiles(uris, stop_token); } return std::format("Removed {} workspace file(s)", uris.size()); }, - [](const std::optional& result, bool cancelled) { - if (cancelled) + [](const ::lsp::scheduler::async_executor::TaskResult& result) { + if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCancelled) { spdlog::debug("Workspace folder removal task cancelled"); } - else if (result) + else if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCompleted && result.value) { - spdlog::info("{}", *result); + spdlog::info("{}", *result.value); } }); } @@ -236,22 +252,26 @@ namespace lsp::provider::workspace auto task_id = std::format("Index workspace folder: {}", folder.uri); scheduler.Submit( task_id, - [&hub, uri = folder.uri]() -> std::optional { - auto uris = EnumerateWorkspaceFiles(uri); + [&hub, uri = folder.uri](std::stop_token stop_token) -> std::optional { + auto uris = EnumerateWorkspaceFiles(uri, stop_token); + if (stop_token.stop_requested()) + { + return std::nullopt; + } if (!uris.empty()) { - hub.symbols().IndexWorkspaceFiles(uris); + hub.symbols().IndexWorkspaceFiles(uris, stop_token); } return std::format("Indexed {} workspace file(s)", uris.size()); }, - [](const std::optional& result, bool cancelled) { - if (cancelled) + [](const ::lsp::scheduler::async_executor::TaskResult& result) { + if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCancelled) { spdlog::debug("Workspace folder indexing task cancelled"); } - else if (result) + else if (result.status == ::lsp::scheduler::async_executor::TaskStatus::kCompleted && result.value) { - spdlog::info("{}", *result); + spdlog::info("{}", *result.value); } }); } diff --git a/lsp-server/src/provider/workspace/execute_command.cppm b/lsp-server/src/provider/workspace/execute_command.cppm index f165581..5366eda 100644 --- a/lsp-server/src/provider/workspace/execute_command.cppm +++ b/lsp-server/src/provider/workspace/execute_command.cppm @@ -1,6 +1,5 @@ module; - export module lsp.provider.workspace.execute_command; import spdlog; @@ -107,8 +106,8 @@ namespace lsp::provider::workspace auto& hub = context.GetManagerHub(); auto& scheduler = context.GetScheduler(); auto task_id = std::format("ExecuteCommand load workspace: {}", *uri); - scheduler.Submit(task_id, [&hub, uri = *uri]() -> std::optional { - hub.symbols().LoadWorkspace(uri); + scheduler.Submit(task_id, [&hub, uri = *uri](std::stop_token stop_token) -> std::optional { + hub.symbols().LoadWorkspace(uri, stop_token); return std::string("ok"); }); @@ -125,8 +124,8 @@ namespace lsp::provider::workspace auto& hub = context.GetManagerHub(); auto& scheduler = context.GetScheduler(); auto task_id = std::format("ExecuteCommand index files: {}", uris.size()); - scheduler.Submit(task_id, [&hub, uris = std::move(uris)]() -> std::optional { - hub.symbols().IndexWorkspaceFiles(uris); + scheduler.Submit(task_id, [&hub, uris = std::move(uris)](std::stop_token stop_token) -> std::optional { + hub.symbols().IndexWorkspaceFiles(uris, stop_token); return std::string("ok"); }); diff --git a/lsp-server/src/scheduler/async_executor.cppm b/lsp-server/src/scheduler/async_executor.cppm index 8a2ba14..2b0edfc 100644 --- a/lsp-server/src/scheduler/async_executor.cppm +++ b/lsp-server/src/scheduler/async_executor.cppm @@ -4,70 +4,61 @@ import spdlog; import taskflow; import std; -export namespace lsp::scheduler +export namespace lsp::scheduler::async_executor { - class AsyncExecutor; - - namespace detail + enum class TaskStatus { - struct ExecutionState - { - std::atomic cancelled{ false }; - std::mutex mutex; - std::condition_variable cv; - bool completed = false; - bool callback_completed = false; - std::optional result; - std::exception_ptr error; - std::chrono::steady_clock::time_point start_time{}; - }; + kCompleted, + kCancelled, + kFailed, + }; - struct ActiveEntry - { - std::shared_ptr state; - std::function& result, bool cancelled)> callback; - std::chrono::steady_clock::time_point start_time; - }; + struct TaskResult + { + TaskStatus status; + std::optional value; + std::exception_ptr error; + }; - struct ExecutorMetrics - { - std::size_t running = 0; - std::uint64_t submitted = 0; - std::uint64_t completed = 0; - std::uint64_t failed = 0; - std::uint64_t cancelled = 0; - }; - } + struct ExecutorMetrics + { + std::size_t running = 0; + std::uint64_t submitted = 0; + std::uint64_t completed = 0; + std::uint64_t failed = 0; + std::uint64_t cancelled = 0; + }; + + class AsyncExecutor; class TaskHandle { public: TaskHandle() = default; - TaskHandle(AsyncExecutor* executor, std::string task_id, std::shared_ptr state) : - executor_(executor), task_id_(std::move(task_id)), state_(std::move(state)) - { - } - - bool Valid() const - { - return executor_ != nullptr && !task_id_.empty() && !state_.expired(); - } + bool Valid() const; bool Cancel() const; - bool Wait() const; - std::optional GetResult() const; + std::optional Wait() const; + std::optional TryGetResult() const; private: - AsyncExecutor* executor_ = nullptr; - std::string task_id_; - std::weak_ptr state_; + struct State; + + explicit TaskHandle(std::shared_ptr state); + + std::shared_ptr state_; + + friend class AsyncExecutor; }; class AsyncExecutor { public: - using TaskClosure = std::function()>; - using TaskCallback = std::function&, bool)>; + using TaskClosure = std::function(std::stop_token)>; + + // A callback must not wait on its own handle or call WaitAll() on this + // executor because either operation would wait for the callback itself. + using TaskCallback = std::function; explicit AsyncExecutor(std::size_t concurrency = std::thread::hardware_concurrency()); ~AsyncExecutor(); @@ -77,89 +68,114 @@ export namespace lsp::scheduler TaskHandle Submit(const std::string& task_id, TaskClosure task, TaskCallback callback = nullptr); bool Cancel(const std::string& task_id); - bool WaitForTask(const std::string& task_id); void WaitAll(); std::size_t GetRunningTaskCount() const; void LogStatus() const; - detail::ExecutorMetrics GetStatistics() const; + ExecutorMetrics GetStatistics() const; private: - void ExecuteTask(const std::string& task_id, TaskClosure task, std::shared_ptr state, TaskCallback callback); - void CompleteTask(const std::string& task_id, - std::chrono::steady_clock::time_point start_time, - std::shared_ptr state, - const std::optional& result, - bool cancelled, - bool failed, - TaskCallback callback); - bool RegisterTask(const std::string& task_id, const detail::ActiveEntry& ctx); - void UnregisterTask(const std::string& task_id, const std::shared_ptr& state); - bool MarkTaskCancelled(const std::string& task_id); - std::chrono::milliseconds GetElapsedTime(std::chrono::steady_clock::time_point start_time) const; + using TaskState = TaskHandle::State; + + static bool RequestCancel(const std::shared_ptr& state); + void ExecuteTask(std::shared_ptr state, TaskClosure task, TaskCallback callback); + void CompleteTask(const std::shared_ptr& state, TaskResult result, const TaskCallback& callback); + void FinalizeTask(const std::shared_ptr& state); - private: tf::Executor executor_; mutable std::mutex mutex_; - std::unordered_map running_tasks_; + std::condition_variable active_tasks_cv_; + std::unordered_map> current_tasks_; + std::unordered_set> active_tasks_; std::atomic submitted_{ 0 }; std::atomic completed_{ 0 }; std::atomic failed_{ 0 }; std::atomic cancelled_{ 0 }; - static constexpr std::size_t kStatusLogInterval = 10; + friend class TaskHandle; }; } -namespace lsp::scheduler +namespace lsp::scheduler::async_executor { + struct TaskHandle::State + { + enum class Phase + { + kPending, + kRunning, + kCompleted, + }; + + explicit State(std::string id) : task_id(std::move(id)) {} + + std::string task_id; + std::chrono::steady_clock::time_point start_time = std::chrono::steady_clock::now(); + std::stop_source stop_source; + mutable std::mutex mutex; + std::condition_variable cv; + Phase phase = Phase::kPending; + bool cancel_requested = false; + std::optional result; + bool callback_completed = false; + }; + namespace { constexpr const char* kLogTag = "AsyncExecutor"; + std::size_t NormalizeConcurrency(std::size_t concurrency) + { + return std::max(1, concurrency); + } + template std::string FormatDuration(Duration duration) { - auto ms = std::chrono::duration_cast(duration).count(); - if (ms < 1000) - return std::to_string(ms) + "ms"; - else - return std::to_string(ms / 1000.0) + "s"; + auto milliseconds = std::chrono::duration_cast(duration).count(); + if (milliseconds < 1000) + return std::to_string(milliseconds) + "ms"; + return std::to_string(milliseconds / 1000.0) + "s"; } } + TaskHandle::TaskHandle(std::shared_ptr state) : state_(std::move(state)) {} + + bool TaskHandle::Valid() const + { + return state_ != nullptr; + } + bool TaskHandle::Cancel() const { - if (!Valid()) - return false; - return executor_->Cancel(task_id_); + return AsyncExecutor::RequestCancel(state_); } - bool TaskHandle::Wait() const + std::optional TaskHandle::Wait() const { - auto state = state_.lock(); - if (!state) - return false; - std::unique_lock lk(state->mutex); - state->cv.wait(lk, [state]() { return state->completed && state->callback_completed; }); - return true; - } - - std::optional TaskHandle::GetResult() const - { - auto state = state_.lock(); - if (!state) + if (!state_) return std::nullopt; - std::unique_lock lk(state->mutex); - if (!state->completed || !state->callback_completed || state->error) - return std::nullopt; - return state->result; + + std::unique_lock lock(state_->mutex); + state_->cv.wait(lock, [this] { return state_->callback_completed; }); + return state_->result; } - AsyncExecutor::AsyncExecutor(std::size_t concurrency) : executor_(concurrency) + std::optional TaskHandle::TryGetResult() const { - spdlog::info("{}: Initialized with {} threads", kLogTag, concurrency); + if (!state_) + return std::nullopt; + + std::lock_guard lock(state_->mutex); + if (!state_->callback_completed) + return std::nullopt; + return state_->result; + } + + AsyncExecutor::AsyncExecutor(std::size_t concurrency) : executor_(NormalizeConcurrency(concurrency)) + { + spdlog::info("{}: Initialized with {} threads", kLogTag, NormalizeConcurrency(concurrency)); } AsyncExecutor::~AsyncExecutor() @@ -171,91 +187,71 @@ namespace lsp::scheduler TaskHandle AsyncExecutor::Submit(const std::string& task_id, TaskClosure task, TaskCallback callback) { - detail::ActiveEntry entry; - entry.state = std::make_shared(); - entry.state->start_time = std::chrono::steady_clock::now(); - entry.callback = callback; - entry.start_time = entry.state->start_time; + auto state = std::make_shared(task_id); + std::shared_ptr replaced_state; + { + std::lock_guard lock(mutex_); + auto current = current_tasks_.find(task_id); + if (current != current_tasks_.end()) + replaced_state = current->second; + current_tasks_[task_id] = state; + active_tasks_.insert(state); + ++submitted_; + } - bool replaced_existing = RegisterTask(task_id, entry); - - if (replaced_existing) + if (replaced_state) + { + RequestCancel(replaced_state); spdlog::warn("[{}] Replaced existing task", task_id); + } try { - executor_.async([this, task_id, task = std::move(task), state = entry.state, callback]() mutable { - ExecuteTask(task_id, std::move(task), state, callback); + executor_.async([this, state, task = std::move(task), callback]() mutable { + ExecuteTask(std::move(state), std::move(task), std::move(callback)); }); } - catch (const std::exception& e) + catch (...) { - spdlog::error("[{}] Failed to submit task: {}", task_id, e.what()); - CompleteTask(task_id, entry.start_time, entry.state, std::nullopt, false, true, callback); + auto error = std::current_exception(); + spdlog::error("[{}] Failed to submit task", task_id); + CompleteTask(state, TaskResult{ TaskStatus::kFailed, std::nullopt, error }, callback); } - return TaskHandle(this, task_id, entry.state); + return TaskHandle(std::move(state)); } bool AsyncExecutor::Cancel(const std::string& task_id) { - return MarkTaskCancelled(task_id); - } - - bool AsyncExecutor::WaitForTask(const std::string& task_id) - { - std::shared_ptr state; + std::shared_ptr state; { - std::unique_lock lock(mutex_); - auto it = running_tasks_.find(task_id); - if (it == running_tasks_.end()) + std::lock_guard lock(mutex_); + auto current = current_tasks_.find(task_id); + if (current == current_tasks_.end()) return false; - state = it->second.state; + state = current->second; } - - std::unique_lock lk(state->mutex); - state->cv.wait(lk, [state]() { return state->completed && state->callback_completed; }); - return true; + return RequestCancel(state); } void AsyncExecutor::WaitAll() { - std::vector> tasks; { - std::unique_lock lock(mutex_); - for (const auto& [_, entry] : running_tasks_) - { - tasks.push_back(entry.state); - } + std::unique_lock lock(mutex_); + active_tasks_cv_.wait(lock, [this] { return active_tasks_.empty(); }); } - - for (const auto& state : tasks) - { - std::unique_lock lk(state->mutex); - state->cv.wait(lk, [state]() { return state->completed && state->callback_completed; }); - } - - // Ensure the underlying executor finishes any tasks that may have been - // cancelled or replaced but are still running. executor_.wait_for_all(); } std::size_t AsyncExecutor::GetRunningTaskCount() const { - std::unique_lock lock(mutex_); - return running_tasks_.size(); + std::lock_guard lock(mutex_); + return active_tasks_.size(); } void AsyncExecutor::LogStatus() const { - std::unique_lock lock(mutex_); - detail::ExecutorMetrics metrics; - metrics.running = running_tasks_.size(); - metrics.submitted = submitted_; - metrics.completed = completed_; - metrics.failed = failed_; - metrics.cancelled = cancelled_; - + const auto metrics = GetStatistics(); spdlog::info("[{}] Running: {}, Submitted: {}, Completed: {}, Failed: {}, Cancelled: {}", kLogTag, metrics.running, @@ -265,157 +261,131 @@ namespace lsp::scheduler metrics.cancelled); } - detail::ExecutorMetrics AsyncExecutor::GetStatistics() const + ExecutorMetrics AsyncExecutor::GetStatistics() const { - detail::ExecutorMetrics metrics; - std::unique_lock lock(mutex_); - metrics.running = running_tasks_.size(); - metrics.submitted = submitted_; - metrics.completed = completed_; - metrics.failed = failed_; - metrics.cancelled = cancelled_; + ExecutorMetrics metrics; + { + std::lock_guard lock(mutex_); + metrics.running = active_tasks_.size(); + } + metrics.submitted = submitted_.load(); + metrics.completed = completed_.load(); + metrics.failed = failed_.load(); + metrics.cancelled = cancelled_.load(); return metrics; } - void AsyncExecutor::ExecuteTask(const std::string& task_id, TaskClosure task, std::shared_ptr state, TaskCallback callback) + bool AsyncExecutor::RequestCancel(const std::shared_ptr& state) { + if (!state) + return false; + + { + std::lock_guard lock(state->mutex); + if (state->phase == TaskState::Phase::kCompleted || state->cancel_requested) + return false; + state->cancel_requested = true; + } + return state->stop_source.request_stop(); + } + + void AsyncExecutor::ExecuteTask(std::shared_ptr state, TaskClosure task, TaskCallback callback) + { + const auto stop_token = state->stop_source.get_token(); + bool cancelled_before_start = false; + { + std::lock_guard lock(state->mutex); + cancelled_before_start = state->cancel_requested; + if (!cancelled_before_start) + state->phase = TaskState::Phase::kRunning; + } + + if (cancelled_before_start) + { + CompleteTask(state, TaskResult{ TaskStatus::kCancelled, std::nullopt, nullptr }, callback); + return; + } + + TaskResult result; try { - auto result = task(); - CompleteTask(task_id, - state->start_time, - state, - result, - state->cancelled.load(std::memory_order_relaxed), - false, - callback); + auto value = task(stop_token); + if (stop_token.stop_requested()) + result = TaskResult{ TaskStatus::kCancelled, std::nullopt, nullptr }; + else + result = TaskResult{ TaskStatus::kCompleted, std::move(value), nullptr }; } catch (const std::exception& e) { - spdlog::error("[{}] Task threw exception: {}", task_id, e.what()); - CompleteTask(task_id, - state->start_time, - state, - std::nullopt, - state->cancelled.load(std::memory_order_relaxed), - true, - callback); + spdlog::error("[{}] Task threw exception: {}", state->task_id, e.what()); + result = TaskResult{ TaskStatus::kFailed, std::nullopt, std::current_exception() }; } catch (...) { - spdlog::error("[{}] Task threw unknown exception", task_id); - CompleteTask(task_id, - state->start_time, - state, - std::nullopt, - state->cancelled.load(std::memory_order_relaxed), - true, - callback); + spdlog::error("[{}] Task threw unknown exception", state->task_id); + result = TaskResult{ TaskStatus::kFailed, std::nullopt, std::current_exception() }; } + + CompleteTask(state, std::move(result), callback); } - void AsyncExecutor::CompleteTask(const std::string& task_id, - std::chrono::steady_clock::time_point start_time, - std::shared_ptr state, - const std::optional& result, - bool cancelled, - bool failed, - TaskCallback callback) + void AsyncExecutor::CompleteTask(const std::shared_ptr& state, TaskResult result, const TaskCallback& callback) { - if (!state) - return; - - const bool is_cancelled = cancelled || state->cancelled.load(std::memory_order_relaxed); { - std::unique_lock lk(state->mutex); - state->completed = true; + std::lock_guard lock(state->mutex); + if (result.status == TaskStatus::kCompleted && state->cancel_requested) + result = TaskResult{ TaskStatus::kCancelled, std::nullopt, nullptr }; + state->phase = TaskState::Phase::kCompleted; state->result = result; - state->cancelled = is_cancelled; - if (failed) - state->error = std::make_exception_ptr(std::runtime_error("Task failed")); } - UnregisterTask(task_id, state); - - auto elapsed = GetElapsedTime(start_time); - bool callback_failed = false; if (callback) { try { - callback(result, is_cancelled); + callback(result); } catch (...) { - callback_failed = true; - std::unique_lock lk(state->mutex); - state->error = std::current_exception(); + spdlog::error("[{}] Task callback threw exception", state->task_id); } } - const bool has_failed = failed || callback_failed; - if (has_failed) - ++failed_; - else if (is_cancelled) - ++cancelled_; - else - ++completed_; - + switch (result.status) { - std::unique_lock lk(state->mutex); + case TaskStatus::kCompleted: + ++completed_; + break; + case TaskStatus::kCancelled: + ++cancelled_; + break; + case TaskStatus::kFailed: + ++failed_; + break; + } + + FinalizeTask(state); + + const auto elapsed = std::chrono::steady_clock::now() - state->start_time; + spdlog::info("[{}] Task completed. status={}, elapsed={}", + state->task_id, + static_cast(result.status), + FormatDuration(elapsed)); + } + + void AsyncExecutor::FinalizeTask(const std::shared_ptr& state) + { + { + std::lock_guard lock(mutex_); + active_tasks_.erase(state); + auto current = current_tasks_.find(state->task_id); + if (current != current_tasks_.end() && current->second == state) + current_tasks_.erase(current); + + std::lock_guard state_lock(state->mutex); state->callback_completed = true; } state->cv.notify_all(); - - if (callback_failed) - spdlog::error("[{}] Task callback threw exception", task_id); - - spdlog::info("[{}] Task completed. cancelled={}, failed={}, elapsed={}", task_id, is_cancelled, has_failed, FormatDuration(elapsed)); + active_tasks_cv_.notify_all(); } - - bool AsyncExecutor::RegisterTask(const std::string& task_id, const detail::ActiveEntry& ctx) - { - std::unique_lock lock(mutex_); - auto it = running_tasks_.find(task_id); - const bool replaced = it != running_tasks_.end(); - if (it != running_tasks_.end()) - { - // Mark prior task as cancelled; completion will observe this flag. - it->second.state->cancelled.store(true, std::memory_order_relaxed); - } - running_tasks_[task_id] = ctx; - ++submitted_; - return replaced; - } - - void AsyncExecutor::UnregisterTask(const std::string& task_id, const std::shared_ptr& state) - { - std::unique_lock lock(mutex_); - auto it = running_tasks_.find(task_id); - if (it != running_tasks_.end() && it->second.state == state) - { - running_tasks_.erase(it); - } - } - - bool AsyncExecutor::MarkTaskCancelled(const std::string& task_id) - { - std::unique_lock lock(mutex_); - auto it = running_tasks_.find(task_id); - if (it == running_tasks_.end()) - return false; - - auto& state = it->second.state; - { - std::unique_lock lk(state->mutex); - state->cancelled = true; - } - return true; - } - - std::chrono::milliseconds AsyncExecutor::GetElapsedTime(std::chrono::steady_clock::time_point start_time) const - { - return std::chrono::duration_cast(std::chrono::steady_clock::now() - start_time); - } - } diff --git a/lsp-server/test/test_provider/completion_test.cppm b/lsp-server/test/test_provider/completion_test.cppm index 3d44ae4..2a6b1b1 100644 --- a/lsp-server/test/test_provider/completion_test.cppm +++ b/lsp-server/test/test_provider/completion_test.cppm @@ -64,7 +64,7 @@ namespace lsp::test::provider { struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::ExecutionContext context; diff --git a/lsp-server/test/test_provider/definitions_test.cppm b/lsp-server/test/test_provider/definitions_test.cppm index 0ae1eb0..a9fc977 100644 --- a/lsp-server/test/test_provider/definitions_test.cppm +++ b/lsp-server/test/test_provider/definitions_test.cppm @@ -33,7 +33,7 @@ namespace lsp::test::provider { struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::ExecutionContext context; diff --git a/lsp-server/test/test_provider/interpreter_test.cppm b/lsp-server/test/test_provider/interpreter_test.cppm index 1e04fae..abc70a7 100644 --- a/lsp-server/test/test_provider/interpreter_test.cppm +++ b/lsp-server/test/test_provider/interpreter_test.cppm @@ -34,7 +34,7 @@ namespace lsp::test::provider struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 4 }; + scheduler::async_executor::AsyncExecutor scheduler{ 4 }; manager::ManagerHub hub{}; core::ExecutionContext context; diff --git a/lsp-server/test/test_provider/json_flow_test.cppm b/lsp-server/test/test_provider/json_flow_test.cppm index 2f88ed3..f578231 100644 --- a/lsp-server/test/test_provider/json_flow_test.cppm +++ b/lsp-server/test/test_provider/json_flow_test.cppm @@ -35,7 +35,7 @@ namespace lsp::test::provider { struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::ExecutionContext context; diff --git a/lsp-server/test/test_provider/json_provider_coverage_test.cppm b/lsp-server/test/test_provider/json_provider_coverage_test.cppm index 2e16fba..d8a3f8e 100644 --- a/lsp-server/test/test_provider/json_provider_coverage_test.cppm +++ b/lsp-server/test/test_provider/json_provider_coverage_test.cppm @@ -49,7 +49,7 @@ namespace lsp::test::provider struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::RequestDispatcher dispatcher{}; @@ -1225,7 +1225,7 @@ namespace lsp::test::provider auto code_action_tree = env.hub.parser().GetTree(code_action_uri); auto code_action_diagnostics = BuildDiagnosticsFromSyntaxErrors(code_action_tree, code_action_content); - env.scheduler.Submit("json_cancel_me", []() -> std::optional { + env.scheduler.Submit("json_cancel_me", [](std::stop_token) -> std::optional { std::this_thread::sleep_for(std::chrono::milliseconds(200)); return std::string("done"); }); diff --git a/lsp-server/test/test_provider/provider_misc_test.cppm b/lsp-server/test/test_provider/provider_misc_test.cppm index 1049c44..64200e4 100644 --- a/lsp-server/test/test_provider/provider_misc_test.cppm +++ b/lsp-server/test/test_provider/provider_misc_test.cppm @@ -176,7 +176,7 @@ namespace lsp::test::provider struct ProviderEnv { std::vector events; - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::ExecutionContext context; @@ -3426,7 +3426,7 @@ namespace lsp::test::provider ProviderEnv env; std::atomic started{ false }; - env.scheduler.Submit("cancel_me", [&started]() -> std::optional { + env.scheduler.Submit("cancel_me", [&started](std::stop_token) -> std::optional { started.store(true); std::this_thread::sleep_for(std::chrono::milliseconds(200)); return std::string("done"); diff --git a/lsp-server/test/test_provider/provider_surface_test.cppm b/lsp-server/test/test_provider/provider_surface_test.cppm index ee9b604..6ffe179 100644 --- a/lsp-server/test/test_provider/provider_surface_test.cppm +++ b/lsp-server/test/test_provider/provider_surface_test.cppm @@ -113,7 +113,7 @@ namespace lsp::test::provider struct ProviderEnv { - scheduler::AsyncExecutor scheduler{ 1 }; + scheduler::async_executor::AsyncExecutor scheduler{ 1 }; manager::ManagerHub hub{}; core::ExecutionContext context; diff --git a/lsp-server/test/test_scheduler/main.cc b/lsp-server/test/test_scheduler/main.cc index cda47a5..2e6c56e 100644 --- a/lsp-server/test/test_scheduler/main.cc +++ b/lsp-server/test/test_scheduler/main.cc @@ -2,5 +2,5 @@ import lsp.test.scheduler.async_executor; int main() { - return Run(); + return lsp::test::scheduler::async_executor::Run(); } diff --git a/lsp-server/test/test_scheduler/test_async_executor.cppm b/lsp-server/test/test_scheduler/test_async_executor.cppm index da61ba1..dc97daa 100644 --- a/lsp-server/test/test_scheduler/test_async_executor.cppm +++ b/lsp-server/test/test_scheduler/test_async_executor.cppm @@ -10,6 +10,9 @@ using namespace std::chrono_literals; namespace { + using lsp::scheduler::async_executor::AsyncExecutor; + using lsp::scheduler::async_executor::TaskStatus; + void Expect(bool condition, const std::string& message) { if (!condition) @@ -42,7 +45,7 @@ namespace } catch (const std::exception& e) { - failures++; + ++failures; std::cout << "[FAIL] " << entry.name << " -> " << e.what() << std::endl; } } @@ -53,112 +56,337 @@ namespace private: std::vector entries_; }; -} - -export int Run() -{ - SchedulerTestSuite suite; - - suite.Add("Completes basic task", [] { - lsp::scheduler::AsyncExecutor executor(2); - std::mutex callback_mutex; - std::optional callback_result; - bool callback_cancelled = false; - - auto handle = executor.Submit("task.simple", []() -> std::optional { - std::this_thread::sleep_for(5ms); - return std::string("done"); }, [&](const std::optional& result, bool cancelled) { - std::lock_guard lk(callback_mutex); - callback_result = result; - callback_cancelled = cancelled; }); - - Expect(handle.Valid(), "Task handle should be valid"); - Expect(handle.Wait(), "Handle wait should succeed"); + class BlockingCallbackCopy + { + public: + BlockingCallbackCopy(std::latch& copy_started, std::latch& release_copy) : + copy_started_(copy_started), release_copy_(release_copy) { - std::lock_guard lk(callback_mutex); - Expect(callback_result.has_value(), "Task result should be reported by callback"); - Expect(!callback_cancelled, "Callback should not be marked cancelled"); - Expect(callback_result.value() == "done", "Task result should match"); } - executor.WaitAll(); - auto stats = executor.GetStatistics(); - Expect(stats.completed == 1, "Completed count should be 1"); - Expect(stats.cancelled == 0, "Cancelled count should be 0"); - Expect(stats.failed == 0, "Failed count should be 0"); - }); + BlockingCallbackCopy(const BlockingCallbackCopy& other) : + copy_started_(other.copy_started_), release_copy_(other.release_copy_) + { + copy_started_.count_down(); + release_copy_.wait(); + } - suite.Add("Cancels running task via handle", [] { - lsp::scheduler::AsyncExecutor executor(1); - std::atomic callback_cancelled{ false }; + BlockingCallbackCopy(BlockingCallbackCopy&&) = default; - auto handle = executor.Submit("task.cancel", []() -> std::optional { - std::this_thread::sleep_for(50ms); - return std::string("late"); }, [&](const std::optional&, bool cancelled) { callback_cancelled.store(cancelled, std::memory_order_relaxed); }); + void operator()(const lsp::scheduler::async_executor::TaskResult&) const {} - std::this_thread::sleep_for(10ms); - Expect(handle.Cancel(), "Handle cancel should report success"); + private: + std::latch& copy_started_; + std::latch& release_copy_; + }; +} - Expect(handle.Wait(), "Wait after cancellation should still succeed"); - executor.WaitAll(); - Expect(callback_cancelled.load(std::memory_order_relaxed), "Callback should observe cancellation"); +export namespace lsp::test::scheduler::async_executor +{ + int Run() + { + SchedulerTestSuite suite; - auto stats = executor.GetStatistics(); - Expect(stats.cancelled == 1, "Cancelled count should be 1"); - }); - - suite.Add("Replaces existing task with same id", [] { - lsp::scheduler::AsyncExecutor executor(2); - std::atomic first_cancelled{ false }; - - auto first = executor.Submit("task.duplicate", []() -> std::optional { - std::this_thread::sleep_for(30ms); - return std::string("first"); }, [&](const std::optional&, bool cancelled) { - if (cancelled) - first_cancelled.store(true, std::memory_order_relaxed); }); - - std::this_thread::sleep_for(5ms); - std::atomic second_completed{ false }; - auto second = executor.Submit("task.duplicate", [&]() -> std::optional { - std::this_thread::sleep_for(20ms); - second_completed.store(true, std::memory_order_relaxed); - return std::string("second"); + suite.Add("Rejects operations on a default handle", [] { + lsp::scheduler::async_executor::TaskHandle handle; + Expect(!handle.Valid(), "default handle should be invalid"); + Expect(!handle.Cancel(), "invalid handle cancellation should fail"); + Expect(!handle.Wait().has_value(), "invalid handle wait should be empty"); + Expect(!handle.TryGetResult().has_value(), "invalid handle result should be empty"); }); - Expect(second.Wait(), "Second handle should finish"); - Expect(first.Wait(), "First handle should finish"); - executor.WaitAll(); - Expect(first_cancelled.load(std::memory_order_relaxed), "First callback should report cancellation"); + suite.Add("Retains completed task result through handle", [] { + AsyncExecutor executor{ 1 }; + auto handle = executor.Submit("fast", [](std::stop_token) { + return std::optional{ "done" }; + }); - auto stats = executor.GetStatistics(); - Expect(stats.completed == 1, "One task should complete successfully"); - Expect(stats.cancelled >= 1, "At least one task should be cancelled"); - Expect(second_completed.load(std::memory_order_relaxed), "Second task body should run"); - }); + auto result = handle.Wait(); + Expect(handle.Valid(), "completed handle should remain valid"); + Expect(result && result->status == TaskStatus::kCompleted, "task should complete"); + Expect(result->value == "done", "task value should remain available"); + Expect(handle.TryGetResult().has_value(), "completed result should be non-blocking"); + Expect(!handle.Cancel(), "completed task cancellation should fail"); + }); - suite.Add("Handles many concurrent submissions", [] { - constexpr int kTaskCount = 32; - lsp::scheduler::AsyncExecutor executor(8); - std::atomic callback_count{ 0 }; + suite.Add("Waits for callback completion", [] { + AsyncExecutor executor{ 1 }; + std::latch callback_started{ 1 }; + std::latch release_callback{ 1 }; + auto handle = executor.Submit( + "callback.wait", + [](std::stop_token) { return std::optional{ "done" }; }, + [&](const lsp::scheduler::async_executor::TaskResult&) { + callback_started.count_down(); + release_callback.wait(); + }); - for (int i = 0; i < kTaskCount; ++i) - { - executor.Submit("task." + std::to_string(i), [i]() -> std::optional { - std::this_thread::sleep_for(2ms); - return std::to_string(i * i); }, [&](const std::optional& result, bool cancelled) { - Expect(!cancelled, "Concurrent tasks should not be cancelled"); - Expect(result.has_value(), "Concurrent task should produce a result"); - callback_count.fetch_add(1, std::memory_order_relaxed); }); - } + callback_started.wait(); + auto waiter = std::async(std::launch::async, [&handle] { return handle.Wait(); }); + const bool blocked = waiter.wait_for(20ms) == std::future_status::timeout; + const bool unavailable = !handle.TryGetResult().has_value(); + release_callback.count_down(); + auto result = waiter.get(); - executor.WaitAll(); - auto stats = executor.GetStatistics(); - Expect(stats.completed == kTaskCount, "All concurrent tasks should complete"); - Expect(stats.cancelled == 0 && stats.failed == 0, "No concurrent tasks should fail or cancel"); - Expect(callback_count.load(std::memory_order_relaxed) == kTaskCount, "Callbacks should run for every task"); - }); + Expect(blocked, "wait should block until callback returns"); + Expect(unavailable, "result should remain unavailable during callback"); + Expect(result && result->status == TaskStatus::kCompleted, "wait should return task result"); + }); - const int failures = suite.RunAll(); - return failures == 0 ? 0 : 1; + suite.Add("Skips a queued task cancelled through its handle", [] { + AsyncExecutor executor{ 1 }; + std::latch blocker_started{ 1 }; + std::latch release_blocker{ 1 }; + auto blocker = executor.Submit("blocker", [&](std::stop_token) { + blocker_started.count_down(); + release_blocker.wait(); + return std::optional{}; + }); + + blocker_started.wait(); + std::atomic ran{ false }; + auto queued = executor.Submit("queued", [&](std::stop_token) { + ran = true; + return std::optional{}; + }); + const bool first_cancel = queued.Cancel(); + const bool second_cancel = queued.Cancel(); + release_blocker.count_down(); + + auto queued_result = queued.Wait(); + auto blocker_result = blocker.Wait(); + Expect(first_cancel, "first queued cancellation should succeed"); + Expect(!second_cancel, "repeated cancellation should fail"); + Expect(queued_result && queued_result->status == TaskStatus::kCancelled, "queued task should cancel"); + Expect(!ran, "cancelled queued closure must not execute"); + Expect(blocker_result && blocker_result->status == TaskStatus::kCompleted, "blocker should finish"); + }); + + suite.Add("Running task observes stop token", [] { + AsyncExecutor executor{ 1 }; + std::latch started{ 1 }; + std::atomic observed_stop{ false }; + std::atomic release_task{ false }; + auto handle = executor.Submit("running", [&](std::stop_token stop_token) { + started.count_down(); + while (!stop_token.stop_requested() && !release_task.load()) + std::this_thread::yield(); + observed_stop = stop_token.stop_requested(); + return std::optional{ "ignored" }; + }); + + started.wait(); + const bool cancelled = executor.Cancel("running"); + release_task = true; + auto result = handle.Wait(); + Expect(cancelled, "ID cancellation should find current task"); + Expect(observed_stop, "running task should observe stop token"); + Expect(result && result->status == TaskStatus::kCancelled, "running task should finish cancelled"); + Expect(!result->value.has_value(), "cancelled task should not expose a value"); + }); + + suite.Add("Does not hold state lock while requesting stop", [] { + AsyncExecutor executor{ 1 }; + lsp::scheduler::async_executor::TaskHandle handle; + std::latch allow_callback_registration{ 1 }; + std::latch callback_registered{ 1 }; + std::atomic reentrant_cancel_blocked{ false }; + std::thread reentrant_cancel; + + handle = executor.Submit("stop.callback", [&](std::stop_token stop_token) { + allow_callback_registration.wait(); + std::stop_callback callback(stop_token, [&] { + auto reentrant_done = std::make_shared>(); + auto reentrant_future = reentrant_done->get_future(); + reentrant_cancel = std::thread([&, reentrant_done] { + handle.Cancel(); + reentrant_done->set_value(); + }); + reentrant_cancel_blocked = reentrant_future.wait_for(20ms) == std::future_status::timeout; + }); + callback_registered.count_down(); + while (!stop_token.stop_requested()) + std::this_thread::yield(); + return std::optional{}; + }); + + allow_callback_registration.count_down(); + callback_registered.wait(); + const bool cancelled = handle.Cancel(); + reentrant_cancel.join(); + auto result = handle.Wait(); + + Expect(cancelled, "initial cancellation should succeed"); + Expect(!reentrant_cancel_blocked, "stop callback should reenter cancellation without blocking on state lock"); + Expect(result && result->status == TaskStatus::kCancelled, "task should finish cancelled"); + }); + + suite.Add("Old same-ID handle cannot cancel replacement", [] { + AsyncExecutor executor{ 2 }; + std::latch first_started{ 1 }; + std::atomic release_first{ false }; + auto first = executor.Submit("duplicate", [&](std::stop_token stop_token) { + first_started.count_down(); + while (!stop_token.stop_requested() && !release_first.load()) + std::this_thread::yield(); + return std::optional{ "first" }; + }); + first_started.wait(); + + std::latch second_started{ 1 }; + std::latch release_second{ 1 }; + std::atomic second_saw_stop{ false }; + auto second = executor.Submit("duplicate", [&](std::stop_token stop_token) { + second_started.count_down(); + release_second.wait(); + second_saw_stop = stop_token.stop_requested(); + return std::optional{ "second" }; + }); + second_started.wait(); + + const bool old_cancel = first.Cancel(); + release_first = true; + release_second.count_down(); + auto first_result = first.Wait(); + auto second_result = second.Wait(); + + Expect(!old_cancel, "already replaced task should reject repeated cancellation"); + Expect(first_result && first_result->status == TaskStatus::kCancelled, "replaced task should cancel"); + Expect(second_result && second_result->status == TaskStatus::kCompleted, "replacement should complete"); + Expect(!second_saw_stop, "old handle must not stop replacement"); + }); + + suite.Add("Counts every active same-ID instance", [] { + AsyncExecutor executor{ 2 }; + std::latch first_started{ 1 }; + std::latch second_started{ 1 }; + std::latch release_tasks{ 1 }; + auto first = executor.Submit("same", [&](std::stop_token) { + first_started.count_down(); + release_tasks.wait(); + return std::optional{}; + }); + first_started.wait(); + auto second = executor.Submit("same", [&](std::stop_token) { + second_started.count_down(); + release_tasks.wait(); + return std::optional{}; + }); + second_started.wait(); + + const auto active_count = executor.GetRunningTaskCount(); + release_tasks.count_down(); + executor.WaitAll(); + + Expect(active_count == 2, "both same-ID instances should be active"); + Expect(executor.GetRunningTaskCount() == 0, "all instances should unregister"); + Expect(first.Wait().has_value() && second.Wait().has_value(), "both handles should retain results"); + }); + + suite.Add("WaitAll observes a task before Taskflow submission", [] { + AsyncExecutor executor{ 1 }; + std::latch callback_copy_started{ 1 }; + std::latch release_callback_copy{ 1 }; + AsyncExecutor::TaskCallback callback{ + BlockingCallbackCopy(callback_copy_started, release_callback_copy) + }; + + auto submitter = std::async(std::launch::async, [&] { + return executor.Submit( + "submission.window", + [](std::stop_token) { return std::optional{}; }, + std::move(callback)); + }); + callback_copy_started.wait(); + auto waiter = std::async(std::launch::async, [&] { executor.WaitAll(); }); + const bool waited_for_registered_task = waiter.wait_for(20ms) == std::future_status::timeout; + release_callback_copy.count_down(); + + auto handle = submitter.get(); + waiter.get(); + auto result = handle.Wait(); + + Expect(waited_for_registered_task, "WaitAll should not return while a registered task is being submitted"); + Expect(result && result->status == TaskStatus::kCompleted, "submitted task should complete"); + }); + + suite.Add("Normalizes zero concurrency", [] { + AsyncExecutor executor{ 0 }; + auto result = executor.Submit("zero", [](std::stop_token) { + return std::optional{ "done" }; + }) + .Wait(); + Expect(result && result->status == TaskStatus::kCompleted, "zero concurrency should use one worker"); + }); + + suite.Add("Preserves task exception", [] { + AsyncExecutor executor{ 1 }; + auto result = executor.Submit("failure", [](std::stop_token) -> std::optional { + throw std::logic_error("original failure"); + }) + .Wait(); + + Expect(result && result->status == TaskStatus::kFailed, "throwing task should fail"); + Expect(result->error != nullptr, "failed task should retain exception"); + try + { + std::rethrow_exception(result->error); + } + catch (const std::logic_error& e) + { + Expect(std::string_view(e.what()) == "original failure", "original exception should be preserved"); + } + }); + + suite.Add("Ignores callback exception in task result", [] { + AsyncExecutor executor{ 1 }; + auto handle = executor.Submit( + "callback.failure", + [](std::stop_token) { return std::optional{ "done" }; }, + [](const lsp::scheduler::async_executor::TaskResult&) { + throw std::runtime_error("callback failure"); + }); + + auto result = handle.Wait(); + Expect(result && result->status == TaskStatus::kCompleted, "callback failure should not alter task status"); + Expect(result->value == "done", "callback failure should not alter task value"); + Expect(result->error == nullptr, "callback failure should not become task error"); + }); + + suite.Add("Reports terminal metrics", [] { + AsyncExecutor executor{ 1 }; + std::latch blocker_started{ 1 }; + std::latch release_blocker{ 1 }; + auto completed = executor.Submit("complete", [&](std::stop_token) { + blocker_started.count_down(); + release_blocker.wait(); + return std::optional{ "done" }; + }); + blocker_started.wait(); + auto cancelled = executor.Submit("cancel", [](std::stop_token) { + return std::optional{}; + }); + auto failed = executor.Submit("fail", [](std::stop_token) -> std::optional { + throw std::runtime_error("failed"); + }); + const bool cancel_requested = cancelled.Cancel(); + release_blocker.count_down(); + executor.WaitAll(); + + const auto metrics = executor.GetStatistics(); + Expect(cancel_requested, "queued task cancellation should succeed"); + Expect(completed.Wait()->status == TaskStatus::kCompleted, "completed status should match metrics"); + Expect(cancelled.Wait()->status == TaskStatus::kCancelled, "cancelled status should match metrics"); + Expect(failed.Wait()->status == TaskStatus::kFailed, "failed status should match metrics"); + Expect(metrics.running == 0, "no task should remain active"); + Expect(metrics.submitted == 3, "submitted count should be 3"); + Expect(metrics.completed == 1, "completed count should be 1"); + Expect(metrics.cancelled == 1, "cancelled count should be 1"); + Expect(metrics.failed == 1, "failed count should be 1"); + }); + + const int failures = suite.RunAll(); + return failures == 0 ? 0 : 1; + } }