From 06fd37bd9abba63c8acd461b78a16915f2bd4ce6 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 21:20:25 +0100 Subject: [PATCH 1/9] Refused pushes to the processing thread from stop() until the next start() - The flag is set and checked under taskMutex_, so a push that passed the check no longer lands after the drain - stop() runs the drained tasks outside the lock, pushes meanwhile are refused instead of waiting for the drain --- ssh/include/ssh/async/processing_thread.hpp | 5 +- ssh/source/ssh/async/processing_thread.cpp | 33 ++++---- ssh/test/ssh/test_processing_thread.hpp | 92 ++++++++++++++++++++- 3 files changed, 108 insertions(+), 22 deletions(-) diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp index 0df4f7a1..38490bcb 100644 --- a/ssh/include/ssh/async/processing_thread.hpp +++ b/ssh/include/ssh/async/processing_thread.hpp @@ -92,7 +92,7 @@ namespace SecureShell std::expected watchSocket(std::function watched); /** - * @brief Stops the processing thread. Executes all pending tasks. + * @brief Stops the processing thread. Executes all pending tasks. Refuses pushes from now until the next start. */ void stop(); @@ -109,7 +109,7 @@ namespace SecureShell * * @param task The task to push. * @return true If the task was pushed. - * @return false If the task was empty or the processing thread is shutting down. + * @return false If the processing thread was stopped and not started again. */ bool pushTask(std::function task); @@ -259,6 +259,7 @@ namespace SecureShell std::thread thread_{}; mutable std::recursive_mutex taskMutex_{}; std::atomic running_ = false; + /** @brief Refuses pushes from stop() until the next start(). Written and checked by pushes under taskMutex_. */ std::atomic shuttingDown_ = false; std::atomic_bool permanentTasksAvailable_ = false; std::atomic permanentTaskIdCounter_ = 0; diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp index 49f1a1b3..8538fff3 100644 --- a/ssh/source/ssh/async/processing_thread.cpp +++ b/ssh/source/ssh/async/processing_thread.cpp @@ -26,8 +26,8 @@ namespace SecureShell { std::lock_guard lock{taskMutex_}; running_ = true; + shuttingDown_ = false; } - shuttingDown_ = false; std::promise awaitThreadStart{}; thread_ = std::thread( [this, &awaitThreadStart, pollInterval] @@ -41,10 +41,10 @@ namespace SecureShell } void ProcessingThread::stop() { - shuttingDown_ = true; bool interruptSocketWait = false; { std::lock_guard lock{taskMutex_}; + shuttingDown_ = true; running_ = false; wakeRequested_ = true; interruptSocketWait = waitingOnSocket_; @@ -53,14 +53,17 @@ namespace SecureShell if (thread_.joinable()) thread_.join(); - // execute all pending tasks in FIFO order, matching the run loop: - std::lock_guard lock{taskMutex_}; + // execute all pending tasks in FIFO order, matching the run loop, and outside the lock so pushes meanwhile are + // refused instead of waiting for the drain: + std::unique_lock lock{taskMutex_}; while (!tasks_.empty()) { - tasks_.front()(); + auto task = std::move(tasks_.front()); tasks_.pop_front(); + lock.unlock(); + task(); + lock.lock(); } - shuttingDown_ = false; } bool ProcessingThread::pushTask(std::function task) { @@ -69,14 +72,13 @@ namespace SecureShell throw std::invalid_argument("Task must not be empty."); } - if (shuttingDown_) - { - return false; - } - bool interruptSocketWait = false; { std::lock_guard lock{taskMutex_}; + if (shuttingDown_) + { + return false; + } tasks_.push_back(std::move(task)); wakeRequested_ = true; interruptSocketWait = waitingOnSocket_; @@ -147,15 +149,14 @@ namespace SecureShell throw std::invalid_argument("Task must not be empty."); } - if (shuttingDown_) - { - return {false, PermanentTaskId{-1}}; - } - PermanentTaskId id{-1}; bool interruptSocketWait = false; { std::lock_guard lock{taskMutex_}; + if (shuttingDown_) + { + return {false, PermanentTaskId{-1}}; + } ++permanentTaskIdCounter_; id = PermanentTaskId{permanentTaskIdCounter_}; permanentTasks_.insert({id, PermanentTask{.run = std::move(task), .pollInterval = pollInterval}}); diff --git a/ssh/test/ssh/test_processing_thread.hpp b/ssh/test/ssh/test_processing_thread.hpp index 373e66e2..16f51dd7 100644 --- a/ssh/test/ssh/test_processing_thread.hpp +++ b/ssh/test/ssh/test_processing_thread.hpp @@ -125,12 +125,90 @@ namespace SecureShell::Test ASSERT_EQ(ProcessingThread::maximumTasksProcessableAtOnce * 3, counter.load()); } - TEST_F(ProcessingThreadTest, CanPushTaskAfterStop) + TEST_F(ProcessingThreadTest, CannotPushAfterStopUntilStartedAgain) { + std::atomic_int ran = 0; + const auto countRun = [&ran] + { + ++ran; + }; ProcessingThread processingThread; processingThread.start(std::chrono::milliseconds{1}); processingThread.stop(); - EXPECT_TRUE(processingThread.pushTask([] {})); + + EXPECT_FALSE(processingThread.pushTask(countRun)); + EXPECT_FALSE(processingThread + .pushPermanentTask( + [](auto) + { + return true; + } + ) + .first); + processingThread.stop(); + EXPECT_EQ(ran.load(), 0); + + processingThread.start(std::chrono::milliseconds{1}); + Awaiter awaiter{}; + EXPECT_TRUE(processingThread.pushTask( + [&awaiter] + { + awaiter.arrive(); + } + )); + EXPECT_TRUE(awaiter.waitFor()); + } + + TEST_F(ProcessingThreadTest, CannotPushWhileStopWaitsForTheLastTask) + { + Awaiter running{}; + std::promise release{}; + const auto released = release.get_future().share(); + std::atomic_int lateRan = 0; + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + processingThread.pushTask( + [&running, released] + { + running.arrive(); + released.wait_for(5s); + } + ); + ASSERT_TRUE(running.waitFor()); + std::thread stopper{[&processingThread] + { + processingThread.stop(); + }}; + int accepted = 0; + const auto refused = waitUntil( + [&processingThread, &lateRan, &accepted] + { + const auto pushed = processingThread.pushTask( + [&lateRan] + { + ++lateRan; + } + ); + if (pushed) + ++accepted; + return !pushed; + } + ); + const auto permanentPushed = processingThread + .pushPermanentTask( + [](auto) + { + return true; + } + ) + .first; + release.set_value(); + stopper.join(); + + EXPECT_TRUE(refused); + EXPECT_FALSE(permanentPushed); + EXPECT_FALSE(processingThread.pushTask([] {})); + EXPECT_EQ(lateRan.load(), accepted); } TEST_F(ProcessingThreadTest, CannotPushTaskWhileStopping) @@ -1117,13 +1195,15 @@ namespace SecureShell::Test EXPECT_EQ(ran.load(), 1); } - TEST_F(ProcessingThreadTest, EveryAcceptedTaskRunsWhenStopRacesWithPushes) + TEST_F(ProcessingThreadTest, EveryAcceptedTaskRunsBeforeStopReturnsWhenPushesRace) { constexpr int pushers = 4; std::atomic_int accepted = 0; std::atomic_int ran = 0; std::atomic_bool keepPushing = true; bool pushing = false; + int ranWhenStopped = 0; + bool refusedAfterStop = false; { ProcessingThread processingThread; processingThread.start(1ms); @@ -1156,12 +1236,16 @@ namespace SecureShell::Test } ); processingThread.stop(); + ranWhenStopped = ran.load(); + refusedAfterStop = !processingThread.pushTask([] {}); keepPushing = false; for (auto& thread : threads) thread.join(); - // Tasks accepted after stop wait for the next start or for the destructor. } EXPECT_TRUE(pushing); + EXPECT_TRUE(refusedAfterStop); + // A push that passed the check before stop() but queued after its drain would run only in the destructor. + EXPECT_EQ(ranWhenStopped, accepted.load()); EXPECT_EQ(ran.load(), accepted.load()); } From e633fec463d4ff240b47a3da915954f8b7cc7b79 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 21:30:35 +0100 Subject: [PATCH 2/9] Completed refused promise tasks with an error value instead of a broken promise - UnfinishedTaskResult makes it: nothing for void, false for bool, the error of a std::expected (SftpError and OperationError define one); other results pass their own - A finalized strand completes its promise tasks the same way instead of with a runtime_error - awaitCycle reads the value, workQueue fails the first strand operation when its batch did not run --- backend/include/backend/sftp/operation.hpp | 12 ++ .../source/backend/sftp/operation_queue.cpp | 47 +++++-- ssh/include/ssh/async/processing_strand.hpp | 40 +++--- ssh/include/ssh/async/processing_thread.hpp | 113 +++++++++++++--- .../ssh/async/unfinished_task_result.hpp | 39 ++++++ ssh/include/ssh/sftp_error.hpp | 14 ++ ssh/include/ssh/sftp_session.hpp | 9 +- ssh/source/ssh/async/processing_thread.cpp | 13 +- ssh/test/ssh/test_processing_thread.hpp | 123 ++++++++++++++++++ ssh/test/ssh/test_sftp.hpp | 5 + 10 files changed, 358 insertions(+), 57 deletions(-) create mode 100644 ssh/include/ssh/async/unfinished_task_result.hpp diff --git a/backend/include/backend/sftp/operation.hpp b/backend/include/backend/sftp/operation.hpp index bfd03ea6..80e739eb 100644 --- a/backend/include/backend/sftp/operation.hpp +++ b/backend/include/backend/sftp/operation.hpp @@ -22,6 +22,18 @@ #include #include +template <> +struct SecureShell::UnfinishedTaskResult +{ + SharedData::OperationError operator()() const + { + return SharedData::OperationError{ + .type = SharedData::OperationErrorType::ChannelClosed, + .extraInfo = "The task did not finish, the SFTP session is closed or its thread stopped", + }; + } +}; + class Operation { public: diff --git a/backend/source/backend/sftp/operation_queue.cpp b/backend/source/backend/sftp/operation_queue.cpp index 1e2a25ef..02413bc3 100644 --- a/backend/source/backend/sftp/operation_queue.cpp +++ b/backend/source/backend/sftp/operation_queue.cpp @@ -1049,6 +1049,7 @@ bool OperationQueue::workQueue(std::dequehasWake()) operation->setWake(wake_); if (operation->usesStrand() && !strand) + { strand = operation->strand(); + strandIndex = item.index; + } } struct ItemResult @@ -1103,17 +1107,42 @@ bool OperationQueue::workQueue(std::deque results; if (strand) { - auto fut = strand->pushPromiseTask( - [&runBatch]() + using Batch = std::optional>; + auto batch = strand + ->pushPromiseTask( + [&runBatch]() -> Batch + { + return runBatch(true); + }, + []() -> Batch + { + return std::nullopt; + } + ) + .get(); + // The processing thread no longer runs tasks, the operations that need it fail one per cycle. + if (!batch) + { + results.push_back( + {strandIndex, + std::unexpected( + Operation::Error{ + .type = Operation::ErrorType::ChannelClosed, + .extraInfo = "The SFTP processing thread no longer runs tasks", + } + )} + ); + } + else + { + results = std::move(batch).value(); + if (std::ranges::none_of(results, isTerminal)) { - return runBatch(true); + auto local = runBatch(false); + results.insert( + results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end()) + ); } - ); - results = fut.get(); - if (std::ranges::none_of(results, isTerminal)) - { - auto local = runBatch(false); - results.insert(results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end())); } } else diff --git a/ssh/include/ssh/async/processing_strand.hpp b/ssh/include/ssh/async/processing_strand.hpp index caf95de7..92247bf9 100644 --- a/ssh/include/ssh/async/processing_strand.hpp +++ b/ssh/include/ssh/async/processing_strand.hpp @@ -123,21 +123,20 @@ namespace SecureShell /** * @brief Works like pushFinalTask but returns a future for the return value of the task. * - * @param task The task to run. + * @param func The task to run. + * @param unfinished Makes the future's value when the task does not finish, also when the strand was finalized + * before. See UnfinishedTaskResult. */ - template - auto pushFinalPromiseTask(FunctionT&& func) -> std::future>> + template < + typename FunctionT, + typename UnfinishedFunc = UnfinishedTaskResult>>> + auto pushFinalPromiseTask(FunctionT&& func, UnfinishedFunc unfinished = {}) + -> std::future>> { auto* processingThread = finalize(); if (processingThread == nullptr) - { - std::promise>> promise{}; - promise.set_exception( - std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand.")) - ); - return promise.get_future(); - } - return processingThread->pushPromiseTask(std::forward(func)); + return Detail::unfinishedFuture>>(std::move(unfinished)); + return processingThread->pushPromiseTask(std::forward(func), std::move(unfinished)); } /** @@ -164,21 +163,20 @@ namespace SecureShell * * @tparam Func The type of the task. * @param func The task to push. + * @param unfinished Makes the future's value when the task does not finish, also when the strand was + * finalized. See UnfinishedTaskResult. * @return std::future>> The future of the return value of the task. */ - template - auto pushPromiseTask(Func&& func) -> std::future>> + template < + typename Func, + typename UnfinishedFunc = UnfinishedTaskResult>>> + auto pushPromiseTask(Func&& func, UnfinishedFunc unfinished = {}) + -> std::future>> { std::scoped_lock lock(mutex_); if (finalized_) - { - std::promise>> promise{}; - promise.set_exception( - std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand.")) - ); - return promise.get_future(); - } - return processingThread_->pushPromiseTask(std::forward(func)); + return Detail::unfinishedFuture>>(std::move(unfinished)); + return processingThread_->pushPromiseTask(std::forward(func), std::move(unfinished)); } /** diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp index 38490bcb..5739dc51 100644 --- a/ssh/include/ssh/async/processing_thread.hpp +++ b/ssh/include/ssh/async/processing_thread.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -12,12 +13,96 @@ #include #include #include +#include +#include #include namespace SecureShell { class ProcessingStrand; + namespace Detail + { + /** + * @brief The promise of a promise task, shared by the copies of the task. When the last copy is gone and the + * task did not finish, the promise gets the unfinished result. + */ + template + class PromiseTaskState + { + public: + explicit PromiseTaskState(UnfinishedT unfinished) + : unfinished_{std::move(unfinished)} + {} + ~PromiseTaskState() + { + if (!finished_) + complete(promise_, unfinished_); + } + PromiseTaskState(PromiseTaskState const&) = delete; + PromiseTaskState& operator=(PromiseTaskState const&) = delete; + PromiseTaskState(PromiseTaskState&&) = delete; + PromiseTaskState& operator=(PromiseTaskState&&) = delete; + + std::future future() + { + return promise_.get_future(); + } + + template + void run(FunctionT& function) + { + complete(promise_, function); + finished_ = true; + } + + /** + * @brief Sets @p promise to what @p produce returns. + */ + template + static void complete(std::promise& promise, FunctionT& produce) + { + if constexpr (std::is_void_v) + { + produce(); + promise.set_value(); + } + else + { + promise.set_value(produce()); + } + } + + private: + std::promise promise_{}; + UnfinishedT unfinished_; + bool finished_{false}; + }; + + /** + * @brief Makes a task of @p func that completes the promise of @p state. + */ + template + std::function makePromiseTask(std::shared_ptr state, FunctionT&& func) + { + return [state = std::move(state), func = std::forward(func)]() mutable + { + state->run(func); + }; + } + + /** + * @brief A future that already holds the unfinished result, for a task refused before it reached a thread. + */ + template + std::future unfinishedFuture(UnfinishedT unfinished) + { + std::promise promise{}; + PromiseTaskState::complete(promise, unfinished); + return promise.get_future(); + } + } + /** * @brief A processing thread that can be used to execute tasks sequentially in a separate thread. * This allows for complex simultaneous operations to be executed in a single thread. @@ -118,28 +203,20 @@ namespace SecureShell * The return value is then accessible through the returned future. * * @param func The function to execute. + * @param unfinished Makes the future's value when the task does not finish, see UnfinishedTaskResult. * @return std::future>> The future that will contain the return value. */ - template - auto pushPromiseTask(Func&& func) -> std::future>> + template < + typename Func, + typename UnfinishedFunc = UnfinishedTaskResult>>> + auto pushPromiseTask(Func&& func, UnfinishedFunc unfinished = {}) + -> std::future>> { using ReturnType = std::invoke_result_t>; - auto promise = std::make_shared>(); - pushTask( - [promise, func = std::forward(func)]() mutable - { - if constexpr (std::is_void_v) - { - func(); - promise->set_value(); - } - else - { - promise->set_value(func()); - } - } - ); - return promise->get_future(); + auto state = std::make_shared>(std::move(unfinished)); + auto future = state->future(); + pushTask(Detail::makePromiseTask(std::move(state), std::forward(func))); + return future; } /** diff --git a/ssh/include/ssh/async/unfinished_task_result.hpp b/ssh/include/ssh/async/unfinished_task_result.hpp new file mode 100644 index 00000000..ca5aa817 --- /dev/null +++ b/ssh/include/ssh/async/unfinished_task_result.hpp @@ -0,0 +1,39 @@ +#pragma once + +#include + +namespace SecureShell +{ + /** + * @brief Makes what the future of a promise task holds when the task does not finish: its push was refused, the + * thread dropped it, or it threw. Defined for void (nothing), bool (false) and std::expected of an error + * that defines it (that error), other results pass their own. + */ + template + struct UnfinishedTaskResult; + + template <> + struct UnfinishedTaskResult + { + void operator()() const noexcept + {} + }; + + template <> + struct UnfinishedTaskResult + { + bool operator()() const noexcept + { + return false; + } + }; + + template + struct UnfinishedTaskResult> + { + std::expected operator()() const + { + return std::unexpected(UnfinishedTaskResult{}()); + } + }; +} diff --git a/ssh/include/ssh/sftp_error.hpp b/ssh/include/ssh/sftp_error.hpp index af03b01b..cda81270 100644 --- a/ssh/include/ssh/sftp_error.hpp +++ b/ssh/include/ssh/sftp_error.hpp @@ -1,5 +1,7 @@ #pragma once +#include + #include #include @@ -71,4 +73,16 @@ namespace SecureShell ); } }; + + template <> + struct UnfinishedTaskResult + { + SftpError operator()() const + { + return SftpError{ + .message = "The task did not finish, the SFTP session is closed or its thread stopped", + .wrapperError = WrapperErrors::TaskPushFailed, + }; + } + }; } \ No newline at end of file diff --git a/ssh/include/ssh/sftp_session.hpp b/ssh/include/ssh/sftp_session.hpp index 6c6b27c5..9a20c44d 100644 --- a/ssh/include/ssh/sftp_session.hpp +++ b/ssh/include/ssh/sftp_session.hpp @@ -57,10 +57,13 @@ namespace SecureShell strand_->pushTask(std::forward(func)); } - template - auto performPromise(FunctionT&& func) -> std::future>> + template < + typename FunctionT, + typename UnfinishedFunc = UnfinishedTaskResult>>> + auto performPromise(FunctionT&& func, UnfinishedFunc unfinished = {}) + -> std::future>> { - return strand_->pushPromiseTask(std::forward(func)); + return strand_->pushPromiseTask(std::forward(func), std::move(unfinished)); } /** diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp index 8538fff3..e804213b 100644 --- a/ssh/source/ssh/async/processing_thread.cpp +++ b/ssh/source/ssh/async/processing_thread.cpp @@ -432,12 +432,13 @@ namespace SecureShell } if (!withinProcessingThread() && running) { - return pushPromiseTask( - []() - { - return true; - } - ).wait_for(maxWait) == std::future_status::ready; + auto cycled = pushPromiseTask( + []() + { + return true; + } + ); + return cycled.wait_for(maxWait) == std::future_status::ready && cycled.get(); } return false; } diff --git a/ssh/test/ssh/test_processing_thread.hpp b/ssh/test/ssh/test_processing_thread.hpp index 16f51dd7..7d1bc4c0 100644 --- a/ssh/test/ssh/test_processing_thread.hpp +++ b/ssh/test/ssh/test_processing_thread.hpp @@ -15,10 +15,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -211,6 +213,115 @@ namespace SecureShell::Test EXPECT_EQ(lateRan.load(), accepted); } + TEST_F(ProcessingThreadTest, RefusedPromiseTaskCompletesWithTheUnfinishedResult) + { + std::atomic_int ran = 0; + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + processingThread.stop(); + + auto refusedBool = processingThread.pushPromiseTask( + [&ran] + { + ++ran; + return true; + } + ); + auto refusedVoid = processingThread.pushPromiseTask( + [&ran] + { + ++ran; + } + ); + auto refusedExpected = processingThread.pushPromiseTask( + [&ran]() -> std::expected + { + ++ran; + return 1; + }, + []() -> std::expected + { + return std::unexpected("refused"s); + } + ); + + ASSERT_EQ(refusedBool.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(refusedBool.get()); + ASSERT_EQ(refusedVoid.wait_for(0s), std::future_status::ready); + EXPECT_NO_THROW(refusedVoid.get()); + ASSERT_EQ(refusedExpected.wait_for(0s), std::future_status::ready); + EXPECT_EQ(refusedExpected.get(), std::unexpected("refused"s)); + EXPECT_FALSE(processingThread.awaitCycle(0ms)); + EXPECT_EQ(ran.load(), 0); + } + + TEST_F(ProcessingThreadTest, PromiseTaskOfAFinalizedStrandCompletesWithTheUnfinishedResult) + { + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + auto strand = processingThread.createStrand(); + auto finalTask = strand->pushFinalPromiseTask( + [] + { + return true; + } + ); + ASSERT_EQ(finalTask.wait_for(5s), std::future_status::ready); + EXPECT_TRUE(finalTask.get()); + + auto afterFinal = strand->pushPromiseTask( + [] + { + return true; + } + ); + auto secondFinal = strand->pushFinalPromiseTask( + []() -> std::expected + { + return 1; + }, + []() -> std::expected + { + return std::unexpected("finalized"s); + } + ); + ASSERT_EQ(afterFinal.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(afterFinal.get()); + ASSERT_EQ(secondFinal.wait_for(0s), std::future_status::ready); + EXPECT_EQ(secondFinal.get(), std::unexpected("finalized"s)); + } + + TEST_F(ProcessingThreadTest, PromiseTaskOfAStrandOnAStoppedThreadCompletesWithTheUnfinishedResult) + { + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + auto strand = processingThread.createStrand(); + processingThread.stop(); + + auto refused = strand->pushPromiseTask( + [] + { + return true; + } + ); + ASSERT_EQ(refused.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(refused.get()); + } + + TEST_F(ProcessingThreadTest, ThrowingPromiseTaskCompletesWithTheUnfinishedResult) + { + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + auto threw = processingThread.pushPromiseTask( + []() -> bool + { + throw std::runtime_error("boom"); + } + ); + ASSERT_EQ(threw.wait_for(5s), std::future_status::ready); + EXPECT_FALSE(threw.get()); + } + TEST_F(ProcessingThreadTest, CannotPushTaskWhileStopping) { Awaiter awaiter{}; @@ -1104,6 +1215,10 @@ namespace SecureShell::Test [&slowTurns, &fastTurns] { return std::pair{slowTurns.load(), fastTurns.load()}; + }, + [] + { + return std::pair{-1, -2}; } ) .get(); @@ -1146,6 +1261,10 @@ namespace SecureShell::Test [] { return 42; + }, + [] + { + return 0; } ); ASSERT_EQ(done.wait_for(5s), std::future_status::ready); @@ -1444,6 +1563,10 @@ namespace SecureShell::Test [] { return 42; + }, + [] + { + return 0; } ); ASSERT_EQ(done.wait_for(5s), std::future_status::ready); diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp index 7db830ec..2abdb4dc 100644 --- a/ssh/test/ssh/test_sftp.hpp +++ b/ssh/test/ssh/test_sftp.hpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include @@ -1139,6 +1140,10 @@ namespace SecureShell::Test for (auto* entry = static_cast(sftp)->queue; entry != nullptr; entry = entry->next) ++count; return count; + }, + [] + { + return std::numeric_limits::max(); } ) .get(); From 8c764a0fc7da1148320e5a2f33d90da7be171451 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 21:40:31 +0100 Subject: [PATCH 3/9] Refused pushes once the processing loop threw, failing the tasks still queued - Queued tasks are dropped there, so their promise tasks complete as unfinished instead of waiting for stop() - Strand final tasks are cleanup tasks: they still run when the loop fails, and after stop() on the calling thread, so SFTP sessions and channels close and leave their owner - stop() drains as the processing thread; cleanup tasks outside the loop run one at a time under cleanupMutex_ --- ssh/include/ssh/async/processing_strand.hpp | 7 +- ssh/include/ssh/async/processing_thread.hpp | 63 ++++++- ssh/source/ssh/async/processing_thread.cpp | 109 ++++++++++-- ssh/test/ssh/test_processing_thread.hpp | 179 ++++++++++++++++++++ ssh/test/ssh/test_sftp.hpp | 46 +++++ 5 files changed, 384 insertions(+), 20 deletions(-) diff --git a/ssh/include/ssh/async/processing_strand.hpp b/ssh/include/ssh/async/processing_strand.hpp index 92247bf9..2c4906d0 100644 --- a/ssh/include/ssh/async/processing_strand.hpp +++ b/ssh/include/ssh/async/processing_strand.hpp @@ -109,7 +109,8 @@ namespace SecureShell } /** - * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks. + * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks. The task + * runs even when the thread no longer runs tasks, see ProcessingThread::pushCleanupTask. * * @param task The task to push. */ @@ -117,7 +118,7 @@ namespace SecureShell { auto* processingThread = finalize(); if (processingThread != nullptr) - processingThread->pushTask(std::move(task)); + processingThread->pushCleanupTask(std::move(task)); } /** @@ -136,7 +137,7 @@ namespace SecureShell auto* processingThread = finalize(); if (processingThread == nullptr) return Detail::unfinishedFuture>>(std::move(unfinished)); - return processingThread->pushPromiseTask(std::forward(func), std::move(unfinished)); + return processingThread->pushCleanupPromiseTask(std::forward(func), std::move(unfinished)); } /** diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp index 5739dc51..f2d3609d 100644 --- a/ssh/include/ssh/async/processing_thread.hpp +++ b/ssh/include/ssh/async/processing_thread.hpp @@ -177,7 +177,8 @@ namespace SecureShell std::expected watchSocket(std::function watched); /** - * @brief Stops the processing thread. Executes all pending tasks. Refuses pushes from now until the next start. + * @brief Stops the processing thread. Executes all pending tasks, on this thread acting as the processing + * thread. Refuses pushes from now until the next start. */ void stop(); @@ -194,10 +195,19 @@ namespace SecureShell * * @param task The task to push. * @return true If the task was pushed. - * @return false If the processing thread was stopped and not started again. + * @return false If the processing thread was stopped and not started again, or its loop ended on an exception. */ bool pushTask(std::function task); + /** + * @brief Pushes a task that must run, a strand's final task. It is queued like pushTask while the loop or the + * drain of stop() will run it, and still runs when the loop ends on an exception. Once neither will, it + * runs on the calling thread at once, which acts as the processing thread meanwhile. + * + * @param task The task to push. + */ + void pushCleanupTask(std::function task); + /** * @brief Pushes a task that has a return value to the processing thread. * The return value is then accessible through the returned future. @@ -219,6 +229,25 @@ namespace SecureShell return future; } + /** + * @brief Works like pushCleanupTask but returns a future for the return value of the task. + * + * @param func The function to execute. + * @param unfinished Makes the future's value when the task throws, see UnfinishedTaskResult. + */ + template < + typename Func, + typename UnfinishedFunc = UnfinishedTaskResult>>> + auto pushCleanupPromiseTask(Func&& func, UnfinishedFunc unfinished = {}) + -> std::future>> + { + using ReturnType = std::invoke_result_t>; + auto state = std::make_shared>(std::move(unfinished)); + auto future = state->future(); + pushCleanupTask(Detail::makePromiseTask(std::move(state), std::forward(func))); + return future; + } + /** * @brief Pushes a task that is not removed upon execution. Will be executed every cycle. * @@ -332,12 +361,36 @@ namespace SecureShell std::optional pollInterval; }; + /** + * @brief A queued task. Only cleanup tasks still run when the loop ends on an exception. + */ + struct QueuedTask + { + std::function run; + bool cleanup{false}; + }; + + /** + * @brief What happens to pushed tasks. + */ + enum class Intake + { + /** @brief Tasks are queued: the thread was constructed or started. */ + Open, + /** @brief stop() is running: tasks are refused, cleanup tasks are queued for its drain. */ + Draining, + /** @brief Stopped, or the loop ended on an exception: tasks are refused, cleanup tasks run on the caller. */ + Closed, + }; + private: std::thread thread_{}; + /** @brief Held while cleanup tasks run outside the loop and while intake_ changes, taken before taskMutex_. */ + std::recursive_mutex cleanupMutex_{}; mutable std::recursive_mutex taskMutex_{}; std::atomic running_ = false; - /** @brief Refuses pushes from stop() until the next start(). Written and checked by pushes under taskMutex_. */ - std::atomic shuttingDown_ = false; + /** @brief Under taskMutex_. */ + Intake intake_{Intake::Open}; std::atomic_bool permanentTasksAvailable_ = false; std::atomic permanentTaskIdCounter_ = 0; std::atomic processingThreadId_{}; @@ -356,7 +409,7 @@ namespace SecureShell std::function watchedSocket_{}; std::unique_ptr socketWait_{}; - std::deque> tasks_{}; + std::deque tasks_{}; std::map permanentTasks_{}; }; } \ No newline at end of file diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp index e804213b..7793f3d9 100644 --- a/ssh/source/ssh/async/processing_thread.cpp +++ b/ssh/source/ssh/async/processing_thread.cpp @@ -7,6 +7,33 @@ namespace SecureShell { + namespace + { + /** + * @brief Makes the calling thread act as the processing thread until destroyed, for tasks run outside the loop. + */ + class ActingProcessingThread + { + public: + explicit ActingProcessingThread(std::atomic& processingThreadId) + : processingThreadId_{processingThreadId} + , previous_{processingThreadId.exchange(std::this_thread::get_id())} + {} + ~ActingProcessingThread() + { + processingThreadId_.store(previous_); + } + ActingProcessingThread(ActingProcessingThread const&) = delete; + ActingProcessingThread& operator=(ActingProcessingThread const&) = delete; + ActingProcessingThread(ActingProcessingThread&&) = delete; + ActingProcessingThread& operator=(ActingProcessingThread&&) = delete; + + private: + std::atomic& processingThreadId_; + std::thread::id previous_; + }; + } + ProcessingThread::ProcessingThread() {} ProcessingThread::~ProcessingThread() @@ -24,9 +51,10 @@ namespace SecureShell "ProcessingThread::start called while a previous thread is still joinable; call stop() first." ); { + std::lock_guard cleanupLock{cleanupMutex_}; std::lock_guard lock{taskMutex_}; running_ = true; - shuttingDown_ = false; + intake_ = Intake::Open; } std::promise awaitThreadStart{}; thread_ = std::thread( @@ -43,8 +71,10 @@ namespace SecureShell { bool interruptSocketWait = false; { + // Waits for cleanup tasks that run on a caller, the drain must not run alongside them. + std::lock_guard cleanupLock{cleanupMutex_}; std::lock_guard lock{taskMutex_}; - shuttingDown_ = true; + intake_ = Intake::Draining; running_ = false; wakeRequested_ = true; interruptSocketWait = waitingOnSocket_; @@ -52,18 +82,22 @@ namespace SecureShell notifyWake(interruptSocketWait); if (thread_.joinable()) thread_.join(); + // The id of a joined thread may be reused. + processingThreadId_.store({}); // execute all pending tasks in FIFO order, matching the run loop, and outside the lock so pushes meanwhile are // refused instead of waiting for the drain: + const ActingProcessingThread acting{processingThreadId_}; std::unique_lock lock{taskMutex_}; while (!tasks_.empty()) { - auto task = std::move(tasks_.front()); + auto task = std::move(tasks_.front().run); tasks_.pop_front(); lock.unlock(); task(); lock.lock(); } + intake_ = Intake::Closed; } bool ProcessingThread::pushTask(std::function task) { @@ -75,17 +109,46 @@ namespace SecureShell bool interruptSocketWait = false; { std::lock_guard lock{taskMutex_}; - if (shuttingDown_) + if (intake_ != Intake::Open) { return false; } - tasks_.push_back(std::move(task)); + tasks_.push_back({.run = std::move(task)}); wakeRequested_ = true; interruptSocketWait = waitingOnSocket_; } notifyWake(interruptSocketWait); return true; } + void ProcessingThread::pushCleanupTask(std::function task) + { + if (!task) + { + throw std::invalid_argument("Task must not be empty."); + } + + // Held while the task runs here, so cleanup tasks run one at a time and never alongside a restarted thread. + std::lock_guard cleanupLock{cleanupMutex_}; + bool queued = false; + bool interruptSocketWait = false; + { + std::lock_guard lock{taskMutex_}; + if (intake_ != Intake::Closed) + { + tasks_.push_back({.run = std::move(task), .cleanup = true}); + wakeRequested_ = true; + interruptSocketWait = waitingOnSocket_; + queued = true; + } + } + if (queued) + { + notifyWake(interruptSocketWait); + return; + } + const ActingProcessingThread acting{processingThreadId_}; + task(); + } void ProcessingThread::wake() { bool interruptSocketWait = false; @@ -153,7 +216,7 @@ namespace SecureShell bool interruptSocketWait = false; { std::lock_guard lock{taskMutex_}; - if (shuttingDown_) + if (intake_ != Intake::Open) { return {false, PermanentTaskId{-1}}; } @@ -288,7 +351,7 @@ namespace SecureShell needs.awaitsInput = true; needs.inputFallback = std::min(needs.inputFallback, lookAgain); } - return running_ && !shuttingDown_; + return running_.load(); }; bool keepGoing = true; for (auto task = rotatedBegin; keepGoing && task != permaTasksMoved.end(); ++task) @@ -329,7 +392,7 @@ namespace SecureShell } { - std::vector> tasks{}; + std::vector tasks{}; { std::lock_guard lock(taskMutex_); tasks.reserve(std::min(tasks_.size(), maximumTasksProcessableAtOnce)); @@ -347,14 +410,14 @@ namespace SecureShell { // Task is checked before adding, shouldnt possibly be empty: #ifndef NDEBUG - if (!task) + if (!task.run) { throw std::runtime_error("Task must not be empty."); } #endif try { - task(); + task.run(); } catch (...) { @@ -372,8 +435,30 @@ namespace SecureShell } catch (...) { - std::lock_guard lock{taskMutex_}; - running_ = false; + // Cleanup tasks pushed from now on run on their caller, after the queued ones ran here. + std::lock_guard cleanupLock{cleanupMutex_}; + std::deque pending{}; + { + std::lock_guard lock{taskMutex_}; + running_ = false; + intake_ = Intake::Closed; + pending = std::exchange(tasks_, {}); + } + // Queued tasks would run only when stop() drains, which the threads waiting for them may never reach. + // Dropped promise tasks complete with their unfinished result, cleanup tasks still run. + for (auto& task : pending) + { + if (!task.cleanup) + continue; + try + { + task.run(); + } + catch (...) + { + // The loop already ended on an exception. + } + } } } void ProcessingThread::waitForWork(CycleNeeds const& needs) diff --git a/ssh/test/ssh/test_processing_thread.hpp b/ssh/test/ssh/test_processing_thread.hpp index 7d1bc4c0..931966bf 100644 --- a/ssh/test/ssh/test_processing_thread.hpp +++ b/ssh/test/ssh/test_processing_thread.hpp @@ -1394,6 +1394,185 @@ namespace SecureShell::Test )); } + TEST_F(ProcessingThreadTest, PushesAfterATaskThrewAreRefusedAndQueuedTasksFail) + { + std::atomic_int ran = 0; + std::future queuedBehind{}; + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + // Queued from within, the task waits for the next cycle the loop does not reach. + processingThread.pushTask( + [&processingThread, &queuedBehind, &ran] + { + queuedBehind = processingThread.pushPromiseTask( + [&ran] + { + ++ran; + return true; + } + ); + throw std::runtime_error("boom"); + } + ); + ASSERT_TRUE(waitUntil( + [&processingThread] + { + return !processingThread.isRunning(); + } + )); + ASSERT_EQ(queuedBehind.wait_for(5s), std::future_status::ready); + EXPECT_FALSE(queuedBehind.get()); + + EXPECT_FALSE(processingThread.pushTask( + [&ran] + { + ++ran; + } + )); + EXPECT_FALSE(processingThread + .pushPermanentTask( + [](auto) + { + return true; + } + ) + .first); + auto refused = processingThread.pushPromiseTask( + [&ran] + { + ++ran; + return true; + } + ); + ASSERT_EQ(refused.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(refused.get()); + processingThread.stop(); + EXPECT_EQ(ran.load(), 0); + } + + TEST_F(ProcessingThreadTest, CleanupTasksStillRunAfterATaskThrew) + { + std::atomic_bool queuedRanOnTheThread = false; + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + processingThread.pushTask( + [&processingThread, &queuedRanOnTheThread] + { + processingThread.pushCleanupTask( + [&processingThread, &queuedRanOnTheThread] + { + queuedRanOnTheThread = processingThread.withinProcessingThread(); + } + ); + throw std::runtime_error("boom"); + } + ); + ASSERT_TRUE(waitUntil( + [&processingThread] + { + return !processingThread.isRunning(); + } + )); + + // Nothing else runs it now, so it runs here before the push returns, acting as the processing thread. + auto later = processingThread.pushCleanupPromiseTask( + [&processingThread] + { + return processingThread.withinProcessingThread(); + } + ); + ASSERT_EQ(later.wait_for(0s), std::future_status::ready); + EXPECT_TRUE(later.get()); + EXPECT_FALSE(processingThread.withinProcessingThread()); + EXPECT_TRUE(queuedRanOnTheThread.load()); + } + + TEST_F(ProcessingThreadTest, FinalTaskAfterStopRunsOnTheCallerAndMayDestroyItsStrand) + { + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + auto strand = processingThread.createStrand(); + auto* const finalizing = strand.get(); + processingThread.stop(); + + auto closed = finalizing->pushFinalPromiseTask( + [&strand, &processingThread] + { + const auto within = processingThread.withinProcessingThread(); + strand.reset(); + return within; + } + ); + ASSERT_EQ(closed.wait_for(0s), std::future_status::ready); + EXPECT_TRUE(closed.get()); + EXPECT_EQ(strand, nullptr); + EXPECT_FALSE(processingThread.withinProcessingThread()); + } + + TEST_F(ProcessingThreadTest, NoTaskRunsWhileItsOwnerIsDestroyed) + { + // Members after the thread are destroyed before it, like the libssh session and channels of a Session. + struct Owner + { + ProcessingThread processingThread{}; + std::vector touched{}; + std::function afterStop{}; + + ~Owner() + { + processingThread.stop(); + afterStop(); + } + }; + + std::atomic_bool stopped = false; + std::atomic_bool keepPushing = true; + std::atomic_int pushedAfterStop = 0; + std::atomic_int acceptedAfterStop = 0; + std::atomic_int ranAfterStop = 0; + auto owner = std::make_unique(); + auto* const raw = owner.get(); + raw->processingThread.start(std::chrono::milliseconds{1}); + std::thread pusher{[&] + { + while (keepPushing) + { + const auto afterStop = stopped.load(); + const auto accepted = raw->processingThread.pushTask( + [raw, &stopped, &ranAfterStop] + { + if (stopped) + ++ranAfterStop; + raw->touched.push_back(1); + } + ); + if (afterStop) + { + ++pushedAfterStop; + if (accepted) + ++acceptedAfterStop; + } + } + }}; + raw->afterStop = [&] + { + stopped = true; + waitUntil( + [&pushedAfterStop] + { + return pushedAfterStop.load() >= 100; + } + ); + keepPushing = false; + pusher.join(); + }; + + owner.reset(); + EXPECT_GE(pushedAfterStop.load(), 100); + EXPECT_EQ(acceptedAfterStop.load(), 0); + EXPECT_EQ(ranAfterStop.load(), 0); + } + TEST_F(ProcessingThreadTest, StrandWakeRunsAnIdleTaskAgain) { std::atomic_int turns = 0; diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp index 2abdb4dc..b8190fa9 100644 --- a/ssh/test/ssh/test_sftp.hpp +++ b/ssh/test/ssh/test_sftp.hpp @@ -12,8 +12,11 @@ #include #include #include +#include +#include #include #include +#include #include using namespace std::chrono_literals; @@ -83,6 +86,49 @@ namespace SecureShell::Test EXPECT_TRUE(sftp.has_value()); } + TEST_F(SftpTests, SftpSessionAndSessionCloseAfterATaskOnTheirThreadThrew) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [session, sftp] = createSftpSession(serverStartResult->port); + auto opened = + sftp->openFile(remote("/home/test/large.txt"), SftpSession::OpenType::Read, std::filesystem::perms::owner_read); + ASSERT_EQ(opened.wait_for(replyTimeout), std::future_status::ready); + ASSERT_TRUE(opened.get().has_value()); + + sftp->perform( + [] + { + throw std::runtime_error("boom"); + } + ); + // Once the loop ended, promise tasks complete with an error at once. + bool refused = false; + const auto deadline = std::chrono::steady_clock::now() + replyTimeout; + while (!refused && std::chrono::steady_clock::now() < deadline) + { + auto probe = sftp->performPromise( + []() -> std::expected + { + return {}; + } + ); + refused = probe.wait_for(replyTimeout) == std::future_status::ready && !probe.get().has_value(); + } + ASSERT_TRUE(refused); + + auto closing = std::async( + std::launch::async, + [&session, &sftp] + { + const auto closed = sftp->close(); + session.reset(); + return closed; + } + ); + ASSERT_EQ(closing.wait_for(10s), std::future_status::ready); + EXPECT_TRUE(closing.get()); + } + TEST_F(SftpTests, CanListDirectory) { CREATE_SERVER_AND_JOINER(Sftp); From a682c3c288a5293e8e5bbc8bd740c69889377f9f Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 22:13:51 +0100 Subject: [PATCH 4/9] Returned whether a deferred removal of a permanent task removed it - From another thread during a permanent cycle, removePermanentTask returned whether the deferred removal ran, so a second removal reported success (CanSafelyDoubleDeletePermanentTask flaked, 9 in 50000 TSan runs) --- ssh/source/ssh/async/processing_thread.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp index 7793f3d9..daae729f 100644 --- a/ssh/source/ssh/async/processing_thread.cpp +++ b/ssh/source/ssh/async/processing_thread.cpp @@ -279,7 +279,8 @@ namespace SecureShell } ); lock.unlock(); - return promise->get_future().wait_for(std::chrono::seconds{5}) == std::future_status::ready; + auto removed = promise->get_future(); + return removed.wait_for(std::chrono::seconds{5}) == std::future_status::ready && removed.get(); } } From 80dfbe644f9421621080159930b7ed29a70e95c9 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 22:17:43 +0100 Subject: [PATCH 5/9] Created SSH channels and SFTP sessions through promise tasks - A refused or dropped push completes the future with an error instead of a broken promise --- ssh/source/ssh/session.cpp | 43 +++++++++++++++++--------------------- ssh/test/ssh/test_sftp.hpp | 3 +++ 2 files changed, 22 insertions(+), 24 deletions(-) diff --git a/ssh/source/ssh/session.cpp b/ssh/source/ssh/session.cpp index faa9e0cb..9c85e0ab 100644 --- a/ssh/source/ssh/session.cpp +++ b/ssh/source/ssh/session.cpp @@ -173,10 +173,9 @@ namespace SecureShell std::future, int>> Session::createPtyChannel(PtyCreationOptions options) { - auto promise = std::make_shared, int>>>(); - - processingThread_.pushTask( - [this, options = std::move(options), promise]() mutable + using Created = std::expected, int>; + return processingThread_.pushPromiseTask( + [this, options = std::move(options)]() mutable -> Created { auto ptyChannel = std::make_unique(session_); auto& channel = *ptyChannel; @@ -227,7 +226,7 @@ namespace SecureShell if (result.result != SSH_OK) { - return promise->set_value(std::unexpected(session_.getErrorCode())); + return std::unexpected(session_.getErrorCode()); } auto sharedChannel = @@ -236,31 +235,29 @@ namespace SecureShell hiddenChannel_ = sharedChannel; else channels_.push_back(sharedChannel); - return promise->set_value(sharedChannel); + return sharedChannel; + }, + []() -> Created + { + return std::unexpected(SSH_ERROR); } ); - - return promise->get_future(); } std::future, SftpError>> Session::createSftpSession() { - auto promise = std::make_shared, SftpError>>>(); - - processingThread_.pushTask( - [this, promise]() -> void + return processingThread_.pushPromiseTask( + [this]() -> std::expected, SftpError> { auto sftp = sftp_new(session_.getCSession()); if (sftp == nullptr) { - return promise->set_value( - std::unexpected( - SftpError{ - .message = session_.getError(), - .sshError = session_.getErrorCode(), - .sftpError = 0, - } - ) + return std::unexpected( + SftpError{ + .message = session_.getError(), + .sshError = session_.getErrorCode(), + .sftpError = 0, + } ); } @@ -273,17 +270,15 @@ namespace SecureShell .sftpError = sftp_get_error(sftp), }; sftp_free(sftp); - return promise->set_value(std::unexpected(error)); + return std::unexpected(error); } // Kept before the caller hears of it, who may shut the session down and walk the list right away. auto sftpSession = std::make_shared(this, processingThread_.createStrand(), sftp); sftpSessions_.push_back(sftpSession); - promise->set_value(sftpSession); + return sftpSession; } ); - - return promise->get_future(); } std::expected, std::string> makeSession( diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp index b8190fa9..8faa5024 100644 --- a/ssh/test/ssh/test_sftp.hpp +++ b/ssh/test/ssh/test_sftp.hpp @@ -115,6 +115,9 @@ namespace SecureShell::Test refused = probe.wait_for(replyTimeout) == std::future_status::ready && !probe.get().has_value(); } ASSERT_TRUE(refused); + auto another = session->createSftpSession(); + ASSERT_EQ(another.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(another.get().has_value()); auto closing = std::async( std::launch::async, From 79a9651179af538a4f72e9e8a5a549204bc53082 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 22:19:24 +0100 Subject: [PATCH 6/9] Formatted the processing thread changes with clang-format --- .../source/backend/sftp/operation_queue.cpp | 12 ++++---- ssh/include/ssh/async/processing_thread.hpp | 2 +- ssh/test/ssh/test_processing_thread.hpp | 28 +++++++++---------- ssh/test/ssh/test_sftp.hpp | 5 ++-- 4 files changed, 24 insertions(+), 23 deletions(-) diff --git a/backend/source/backend/sftp/operation_queue.cpp b/backend/source/backend/sftp/operation_queue.cpp index 02413bc3..ddc3591f 100644 --- a/backend/source/backend/sftp/operation_queue.cpp +++ b/backend/source/backend/sftp/operation_queue.cpp @@ -1125,12 +1125,12 @@ bool OperationQueue::workQueue(std::dequeport); - auto opened = - sftp->openFile(remote("/home/test/large.txt"), SftpSession::OpenType::Read, std::filesystem::perms::owner_read); + auto opened = sftp->openFile( + remote("/home/test/large.txt"), SftpSession::OpenType::Read, std::filesystem::perms::owner_read + ); ASSERT_EQ(opened.wait_for(replyTimeout), std::future_status::ready); ASSERT_TRUE(opened.get().has_value()); From cbb86f2fb2deb2d49aaef5958f58a55127c8914c Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 22:40:40 +0100 Subject: [PATCH 7/9] Completed a chunked FileStream::write with an error when a chunk cannot run - A refused, unpushed or dropped chunk completes the future with TaskPushFailed, a closed file with FileNull --- ssh/include/ssh/async/processing_thread.hpp | 4 +- ssh/source/ssh/file_stream.cpp | 24 +++- ssh/test/ssh/test_sftp.hpp | 129 +++++++++++++++++--- 3 files changed, 129 insertions(+), 28 deletions(-) diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp index 81cca1c4..7e7b5721 100644 --- a/ssh/include/ssh/async/processing_thread.hpp +++ b/ssh/include/ssh/async/processing_thread.hpp @@ -24,8 +24,8 @@ namespace SecureShell namespace Detail { /** - * @brief The promise of a promise task, shared by the copies of the task. When the last copy is gone and the - * task did not finish, the promise gets the unfinished result. + * @brief The promise of a promise task, shared by the copies of the task or of the callback that completes it. + * When the last copy is gone and the task did not finish, the promise gets the unfinished result. */ template class PromiseTaskState diff --git a/ssh/source/ssh/file_stream.cpp b/ssh/source/ssh/file_stream.cpp index 050db1e8..274f4414 100644 --- a/ssh/source/ssh/file_stream.cpp +++ b/ssh/source/ssh/file_stream.cpp @@ -681,7 +681,11 @@ namespace SecureShell [this, toWrite, onWriteComplete = std::move(onWriteComplete)]() { if (!file_) - return; + { + return onWriteComplete( + std::unexpected(SftpError{.message = "File is null", .wrapperError = WrapperErrors::FileNull}) + ); + } const auto written = sftp_write(file_.get(), toWrite.data(), std::min(toWrite.size(), writeLengthLimit())); @@ -753,16 +757,24 @@ namespace SecureShell ); } - // Write in chunks for large data: - auto promise = std::make_shared>>(); + // Write in chunks for large data. A chunk that is refused, never pushed or dropped leaves the unfinished error: + using WriteResult = std::expected; + auto state = std::make_shared>>( + UnfinishedTaskResult{} + ); + auto future = state->future(); writePart( data, - [promise](std::expected&& result) + [state](WriteResult&& result) { - promise->set_value(std::move(result)); + auto finished = [&result]() + { + return std::move(result); + }; + state->run(finished); } ); - return promise->get_future(); + return future; } sftp_file FileStream::release() { diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp index dc078c00..0ddfb0a4 100644 --- a/ssh/test/ssh/test_sftp.hpp +++ b/ssh/test/ssh/test_sftp.hpp @@ -59,6 +59,49 @@ namespace SecureShell::Test return std::make_pair(std::move(session).value(), sftpFuture.get().value().lock()); } + /** + * @brief Ends the processing loop of @p sftp with a throwing task, then waits until promise tasks come back + * with an error at once. + */ + ::testing::AssertionResult endProcessingLoop(SftpSession& sftp) + { + sftp.perform( + [] + { + throw std::runtime_error("boom"); + } + ); + const auto deadline = std::chrono::steady_clock::now() + replyTimeout; + while (std::chrono::steady_clock::now() < deadline) + { + auto probe = sftp.performPromise( + []() -> std::expected + { + return {}; + } + ); + if (probe.wait_for(replyTimeout) == std::future_status::ready && !probe.get().has_value()) + return ::testing::AssertionSuccess(); + } + return ::testing::AssertionFailure() << "the processing loop kept running tasks"; + } + + /** + * @brief Opens a new remote file for writing. + */ + std::shared_ptr openForWriting(SftpSession& sftp, std::string_view path) + { + auto opened = sftp.openFile( + remote(path), + SftpSession::OpenType::Write | SftpSession::OpenType::Create | SftpSession::OpenType::Truncate, + std::filesystem::perms{0644u} + ); + if (opened.wait_for(replyTimeout) != std::future_status::ready) + return nullptr; + auto result = opened.get(); + return result.has_value() ? result.value().lock() : nullptr; + } + std::string createAlphabetString(std::size_t length) { std::string result; @@ -96,26 +139,7 @@ namespace SecureShell::Test ASSERT_EQ(opened.wait_for(replyTimeout), std::future_status::ready); ASSERT_TRUE(opened.get().has_value()); - sftp->perform( - [] - { - throw std::runtime_error("boom"); - } - ); - // Once the loop ended, promise tasks complete with an error at once. - bool refused = false; - const auto deadline = std::chrono::steady_clock::now() + replyTimeout; - while (!refused && std::chrono::steady_clock::now() < deadline) - { - auto probe = sftp->performPromise( - []() -> std::expected - { - return {}; - } - ); - refused = probe.wait_for(replyTimeout) == std::future_status::ready && !probe.get().has_value(); - } - ASSERT_TRUE(refused); + ASSERT_TRUE(endProcessingLoop(*sftp)); auto another = session->createSftpSession(); ASSERT_EQ(another.wait_for(0s), std::future_status::ready); EXPECT_FALSE(another.get().has_value()); @@ -133,6 +157,71 @@ namespace SecureShell::Test EXPECT_TRUE(closing.get()); } + TEST_F(SftpTests, LargeWriteIsWrittenInChunks) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForWriting(*sftp, "/home/test/uploaded.bin"); + ASSERT_TRUE(file); + const std::string content(file->writeLengthLimit() * 3u + 17u, 'w'); + + auto written = file->write(content); + ASSERT_EQ(written.wait_for(replyTimeout), std::future_status::ready); + EXPECT_TRUE(written.get().has_value()); + file->close(); + + auto information = sftp->stat(remote("/home/test/uploaded.bin")); + ASSERT_EQ(information.wait_for(replyTimeout), std::future_status::ready); + auto stated = information.get(); + ASSERT_TRUE(stated.has_value()); + EXPECT_EQ(stated->size, content.size()); + } + + TEST_F(SftpTests, LargeWriteToAClosedFileFailsWithFileNull) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForWriting(*sftp, "/home/test/uploaded.bin"); + ASSERT_TRUE(file); + file->close(); + + auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w')); + ASSERT_EQ(written.wait_for(replyTimeout), std::future_status::ready); + const auto result = written.get(); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().wrapperError, WrapperErrors::FileNull); + } + + TEST_F(SftpTests, LargeWriteAfterTheSftpSessionClosedFailsAtOnce) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForWriting(*sftp, "/home/test/uploaded.bin"); + ASSERT_TRUE(file); + ASSERT_TRUE(sftp->close()); + + auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w')); + ASSERT_EQ(written.wait_for(0s), std::future_status::ready); + const auto result = written.get(); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed); + } + + TEST_F(SftpTests, LargeWriteAfterATaskOnTheThreadThrewFailsAtOnce) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForWriting(*sftp, "/home/test/uploaded.bin"); + ASSERT_TRUE(file); + ASSERT_TRUE(endProcessingLoop(*sftp)); + + auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w')); + ASSERT_EQ(written.wait_for(0s), std::future_status::ready); + const auto result = written.get(); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed); + } + TEST_F(SftpTests, CanListDirectory) { CREATE_SERVER_AND_JOINER(Sftp); From 8d873f7f5486063d395ad387a4414ca329072675 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 23:30:16 +0100 Subject: [PATCH 8/9] Kept stop() and its drain safe from cleanup tasks, throwing tasks and barriers - A loop that throws while stop() drains leaves the queue and intake to the drain, which resets the thread id and closes intake together, so no cleanup task runs on its caller beside stop() - The drain and cleanup tasks run on a caller contain task exceptions like the loop does - A promise task pushed while stop() drains is queued refused and completes when the drain reaches it, so Operation::waitForStrandTasks still waits for the tasks queued before it --- ssh/include/ssh/async/processing_thread.hpp | 33 +++- ssh/source/ssh/async/processing_thread.cpp | 71 +++++--- ssh/test/ssh/test_processing_thread.hpp | 169 ++++++++++++++++++++ 3 files changed, 248 insertions(+), 25 deletions(-) diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp index 7e7b5721..66e0f4d1 100644 --- a/ssh/include/ssh/async/processing_thread.hpp +++ b/ssh/include/ssh/async/processing_thread.hpp @@ -212,6 +212,9 @@ namespace SecureShell * @brief Pushes a task that has a return value to the processing thread. * The return value is then accessible through the returned future. * + * A push refused while stop() drains completes only once the drain reached it, so the future is still a + * barrier for the tasks queued before it. + * * @param func The function to execute. * @param unfinished Makes the future's value when the task does not finish, see UnfinishedTaskResult. * @return std::future>> The future that will contain the return value. @@ -225,7 +228,7 @@ namespace SecureShell using ReturnType = std::invoke_result_t>; auto state = std::make_shared>(std::move(unfinished)); auto future = state->future(); - pushTask(Detail::makePromiseTask(std::move(state), std::forward(func))); + queuePromiseTask(Detail::makePromiseTask(std::move(state), std::forward(func))); return future; } @@ -327,6 +330,17 @@ namespace SecureShell private: void run(std::chrono::milliseconds const& pollInterval); + /** + * @brief Queues a promise task like pushTask. While stop() drains it is queued refused instead: the drain + * drops it in order, so its future completes after the tasks queued before it. + */ + void queuePromiseTask(std::function task); + + /** + * @brief Runs a task outside the loop, where a throwing task must not keep the others from running. + */ + static void runContained(std::function const& task); + /** * @brief What the permanent tasks of one cycle need from the wait after it. */ @@ -362,12 +376,21 @@ namespace SecureShell }; /** - * @brief A queued task. Only cleanup tasks still run when the loop ends on an exception. + * @brief What a queued task is. */ + enum class TaskKind + { + Regular, + /** @brief Still runs when the loop ends on an exception. */ + Cleanup, + /** @brief A promise task pushed while stop() drains: dropped, not run, when its turn comes. */ + Refused, + }; + struct QueuedTask { std::function run; - bool cleanup{false}; + TaskKind kind{TaskKind::Regular}; }; /** @@ -377,7 +400,7 @@ namespace SecureShell { /** @brief Tasks are queued: the thread was constructed or started. */ Open, - /** @brief stop() is running: tasks are refused, cleanup tasks are queued for its drain. */ + /** @brief stop() is running: tasks are refused, cleanup tasks are queued for its drain to run. */ Draining, /** @brief Stopped, or the loop threw: tasks are refused, cleanup tasks run on the caller. */ Closed, @@ -385,7 +408,7 @@ namespace SecureShell private: std::thread thread_{}; - /** @brief Held while cleanup tasks run outside the loop and while intake_ changes, taken before taskMutex_. */ + /** @brief Held while cleanup tasks run outside the loop and while intake_ leaves Closed. Taken first. */ std::recursive_mutex cleanupMutex_{}; mutable std::recursive_mutex taskMutex_{}; std::atomic running_ = false; diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp index daae729f..c923224a 100644 --- a/ssh/source/ssh/async/processing_thread.cpp +++ b/ssh/source/ssh/async/processing_thread.cpp @@ -82,23 +82,38 @@ namespace SecureShell notifyWake(interruptSocketWait); if (thread_.joinable()) thread_.join(); - // The id of a joined thread may be reused. - processingThreadId_.store({}); // execute all pending tasks in FIFO order, matching the run loop, and outside the lock so pushes meanwhile are - // refused instead of waiting for the drain: - const ActingProcessingThread acting{processingThreadId_}; + // refused instead of waiting for the drain. Intake stays Draining until the end, so no cleanup task runs on a + // caller meanwhile. std::unique_lock lock{taskMutex_}; + processingThreadId_.store(std::this_thread::get_id()); while (!tasks_.empty()) { - auto task = std::move(tasks_.front().run); - tasks_.pop_front(); - lock.unlock(); - task(); + { + auto task = std::move(tasks_.front()); + tasks_.pop_front(); + lock.unlock(); + if (task.kind != TaskKind::Refused) + runContained(task.run); + } lock.lock(); } + // The id of the joined thread may be reused. + processingThreadId_.store({}); intake_ = Intake::Closed; } + void ProcessingThread::runContained(std::function const& task) + { + try + { + task(); + } + catch (...) + { + // Outside the loop no task ends anything, a promise task completes with its unfinished result. + } + } bool ProcessingThread::pushTask(std::function task) { if (!task) @@ -120,6 +135,24 @@ namespace SecureShell notifyWake(interruptSocketWait); return true; } + void ProcessingThread::queuePromiseTask(std::function task) + { + bool interruptSocketWait = false; + { + std::lock_guard lock{taskMutex_}; + if (intake_ == Intake::Closed) + return; + if (intake_ == Intake::Draining) + { + tasks_.push_back({.run = std::move(task), .kind = TaskKind::Refused}); + return; + } + tasks_.push_back({.run = std::move(task)}); + wakeRequested_ = true; + interruptSocketWait = waitingOnSocket_; + } + notifyWake(interruptSocketWait); + } void ProcessingThread::pushCleanupTask(std::function task) { if (!task) @@ -135,7 +168,7 @@ namespace SecureShell std::lock_guard lock{taskMutex_}; if (intake_ != Intake::Closed) { - tasks_.push_back({.run = std::move(task), .cleanup = true}); + tasks_.push_back({.run = std::move(task), .kind = TaskKind::Cleanup}); wakeRequested_ = true; interruptSocketWait = waitingOnSocket_; queued = true; @@ -147,7 +180,7 @@ namespace SecureShell return; } const ActingProcessingThread acting{processingThreadId_}; - task(); + runContained(task); } void ProcessingThread::wake() { @@ -416,6 +449,9 @@ namespace SecureShell throw std::runtime_error("Task must not be empty."); } #endif + // Pushed while stop() drains, completed when the batch is gone. + if (task.kind == TaskKind::Refused) + continue; try { task.run(); @@ -442,6 +478,9 @@ namespace SecureShell { std::lock_guard lock{taskMutex_}; running_ = false; + // stop() drains what is queued once it joined this thread, and closes intake afterwards. + if (intake_ == Intake::Draining) + return; intake_ = Intake::Closed; pending = std::exchange(tasks_, {}); } @@ -449,16 +488,8 @@ namespace SecureShell // Dropped promise tasks complete with their unfinished result, cleanup tasks still run. for (auto& task : pending) { - if (!task.cleanup) - continue; - try - { - task.run(); - } - catch (...) - { - // The loop already ended on an exception. - } + if (task.kind == TaskKind::Cleanup) + runContained(task.run); } } } diff --git a/ssh/test/ssh/test_processing_thread.hpp b/ssh/test/ssh/test_processing_thread.hpp index 89c2ba70..95765383 100644 --- a/ssh/test/ssh/test_processing_thread.hpp +++ b/ssh/test/ssh/test_processing_thread.hpp @@ -1487,6 +1487,175 @@ namespace SecureShell::Test EXPECT_TRUE(queuedRanOnTheThread.load()); } + TEST_F(ProcessingThreadTest, CleanupTaskPushedWhileStopDrainsRunsInTheDrain) + { + Awaiter lastBatchRunning{}; + std::promise releaseLastBatch{}; + const auto lastBatchReleased = releaseLastBatch.get_future().share(); + Awaiter firstCleanupRunning{}; + std::promise releaseFirstCleanup{}; + const auto firstCleanupReleased = releaseFirstCleanup.get_future().share(); + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + // The last batch throws after stop() began, the cleanup task it queued holds up the rest of stop(). + processingThread.pushTask( + [&processingThread, &lastBatchRunning, &firstCleanupRunning, lastBatchReleased, firstCleanupReleased] + { + lastBatchRunning.arrive(); + lastBatchReleased.wait_for(5s); + processingThread.pushCleanupTask( + [&firstCleanupRunning, firstCleanupReleased] + { + firstCleanupRunning.arrive(); + firstCleanupReleased.wait_for(5s); + } + ); + throw std::runtime_error("boom"); + } + ); + ASSERT_TRUE(lastBatchRunning.waitFor()); + std::thread stopper{[&processingThread] + { + processingThread.stop(); + }}; + const auto stopperId = stopper.get_id(); + ASSERT_TRUE(waitUntil( + [&processingThread] + { + return !processingThread.pushTask([] {}); + } + )); + releaseLastBatch.set_value(); + ASSERT_TRUE(firstCleanupRunning.waitFor()); + + // Pushed while stop() is still running, it must not run on its caller beside the rest of stop(). + std::thread::id ranOn{}; + bool within = false; + std::atomic_bool pushed = false; + std::thread pusher{[&] + { + processingThread.pushCleanupTask( + [&] + { + ranOn = std::this_thread::get_id(); + within = processingThread.withinProcessingThread(); + } + ); + pushed = true; + }}; + const auto pushedAtOnce = waitUntil( + [&pushed] + { + return pushed.load(); + }, + 1s + ); + releaseFirstCleanup.set_value(); + pusher.join(); + stopper.join(); + + EXPECT_TRUE(pushedAtOnce); + EXPECT_EQ(ranOn, stopperId); + EXPECT_TRUE(within); + EXPECT_FALSE(processingThread.withinProcessingThread()); + } + + TEST_F(ProcessingThreadTest, ThrowingTaskInTheDrainOfStopKeepsTheOthersRunning) + { + std::atomic_int ran = 0; + // Never started, so stop() drains what was pushed. + ProcessingThread processingThread; + auto threw = processingThread.pushPromiseTask( + []() -> bool + { + throw std::runtime_error("boom"); + } + ); + processingThread.pushTask( + [&ran] + { + ++ran; + } + ); + + EXPECT_NO_THROW(processingThread.stop()); + EXPECT_EQ(ran.load(), 1); + ASSERT_EQ(threw.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(threw.get()); + // The drain finished, so a final task runs here at once. + auto strand = processingThread.createStrand(); + auto finalTask = strand->pushFinalPromiseTask( + [] + { + return true; + } + ); + ASSERT_EQ(finalTask.wait_for(0s), std::future_status::ready); + EXPECT_TRUE(finalTask.get()); + } + + TEST_F(ProcessingThreadTest, ThrowingCleanupTaskAfterStopDoesNotThrowAtItsCaller) + { + ProcessingThread processingThread; + processingThread.start(std::chrono::milliseconds{1}); + processingThread.stop(); + + std::future threw{}; + EXPECT_NO_THROW( + threw = processingThread.pushCleanupPromiseTask( + []() -> bool + { + throw std::runtime_error("boom"); + } + ) + ); + ASSERT_TRUE(threw.valid()); + ASSERT_EQ(threw.wait_for(0s), std::future_status::ready); + EXPECT_FALSE(threw.get()); + EXPECT_FALSE(processingThread.withinProcessingThread()); + } + + TEST_F(ProcessingThreadTest, PromiseTaskPushedWhileStopDrainsCompletesAfterTheTasksBeforeIt) + { + Awaiter drainRunning{}; + std::promise release{}; + const auto released = release.get_future().share(); + std::atomic_bool laterRan = false; + // Never started, so stop() drains what was pushed. + ProcessingThread processingThread; + auto strand = processingThread.createStrand(); + strand->pushTask( + [&drainRunning, released] + { + drainRunning.arrive(); + released.wait_for(5s); + } + ); + strand->pushTask( + [&laterRan] + { + laterRan = true; + } + ); + std::thread stopper{[&processingThread] + { + processingThread.stop(); + }}; + ASSERT_TRUE(drainRunning.waitFor()); + + // A barrier like Operation::waitForStrandTasks: refused, but complete only once the drain got to it. + auto barrier = strand->pushPromiseTask([] {}); + const auto readyEarly = barrier.wait_for(0s) == std::future_status::ready; + release.set_value(); + const auto readyLater = barrier.wait_for(5s) == std::future_status::ready; + const auto laterRanFirst = laterRan.load(); + stopper.join(); + + EXPECT_FALSE(readyEarly); + EXPECT_TRUE(readyLater); + EXPECT_TRUE(laterRanFirst); + } + TEST_F(ProcessingThreadTest, FinalTaskAfterStopRunsOnTheCallerAndMayDestroyItsStrand) { ProcessingThread processingThread; From 3d7232809f3539fc6b45087ede097fbc74b44e03 Mon Sep 17 00:00:00 2001 From: 5cript Date: Wed, 28 Oct 2026 23:30:16 +0100 Subject: [PATCH 9/9] Completed FileStream::readAll with an error when a read cannot run - A refused, unpushed or dropped read completes the future with TaskPushFailed instead of a broken promise --- ssh/source/ssh/file_stream.cpp | 26 ++++++++--- ssh/test/ssh/test_sftp.hpp | 80 +++++++++++++++++++++++++++++----- 2 files changed, 88 insertions(+), 18 deletions(-) diff --git a/ssh/source/ssh/file_stream.cpp b/ssh/source/ssh/file_stream.cpp index 274f4414..50002723 100644 --- a/ssh/source/ssh/file_stream.cpp +++ b/ssh/source/ssh/file_stream.cpp @@ -562,22 +562,33 @@ namespace SecureShell std::future> FileStream::readAll(std::function onReadFunc) { + using ReadResult = std::expected; struct ReadState : public std::enable_shared_from_this { FileStream& stream; BufferLease buffer; std::function callback; - std::promise> promise; + // A read that is refused, never pushed or dropped leaves the unfinished error when the state is gone. + Detail::PromiseTaskState> completion; std::size_t totalRead; ReadState(FileStream& stream, BufferLease buffer, std::function callback) : stream{stream} , buffer{std::move(buffer)} , callback{std::move(callback)} - , promise{} + , completion{UnfinishedTaskResult{}} , totalRead{0} {} + void finish(ReadResult result) + { + auto finished = [&result]() + { + return std::move(result); + }; + completion.run(finished); + } + void onRead(std::make_signed_t amount) { // The callback runs before the promise is set: the waiting caller may destroy what it captured. @@ -585,14 +596,14 @@ namespace SecureShell { const auto error = stream.lastError(); callback({}); - promise.set_value(std::unexpected(error)); + finish(std::unexpected(error)); return; } if (amount == 0) { callback({}); - promise.set_value(totalRead); + finish(totalRead); return; } @@ -600,7 +611,7 @@ namespace SecureShell if (!callback({buffer.data(), static_cast(amount)})) { - promise.set_value(totalRead); + finish(totalRead); return; } @@ -614,7 +625,7 @@ namespace SecureShell { if (!state->stream.file_) { - state->promise.set_value( + state->finish( std::unexpected( SftpError{ .message = "File is null", @@ -661,8 +672,9 @@ namespace SecureShell } auto state = std::make_shared(*this, std::move(lease), std::move(onReadFunc)); + auto future = state->completion.future(); state->doRead(); - return state->promise.get_future(); + return future; } std::size_t FileStream::writeLengthLimit() const { diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp index 0ddfb0a4..df39abb6 100644 --- a/ssh/test/ssh/test_sftp.hpp +++ b/ssh/test/ssh/test_sftp.hpp @@ -86,20 +86,42 @@ namespace SecureShell::Test return ::testing::AssertionFailure() << "the processing loop kept running tasks"; } + /** + * @brief Opens a remote file, null when that failed. + */ + std::shared_ptr openStream( + SftpSession& sftp, + std::string_view path, + SftpSession::OpenType openType, + std::filesystem::perms permissions + ) + { + auto opened = sftp.openFile(remote(path), openType, permissions); + if (opened.wait_for(replyTimeout) != std::future_status::ready) + return nullptr; + auto result = opened.get(); + return result.has_value() ? result.value().lock() : nullptr; + } + /** * @brief Opens a new remote file for writing. */ std::shared_ptr openForWriting(SftpSession& sftp, std::string_view path) { - auto opened = sftp.openFile( - remote(path), + return openStream( + sftp, + path, SftpSession::OpenType::Write | SftpSession::OpenType::Create | SftpSession::OpenType::Truncate, std::filesystem::perms{0644u} ); - if (opened.wait_for(replyTimeout) != std::future_status::ready) - return nullptr; - auto result = opened.get(); - return result.has_value() ? result.value().lock() : nullptr; + } + + /** + * @brief Opens a remote file for reading. + */ + std::shared_ptr openForReading(SftpSession& sftp, std::string_view path) + { + return openStream(sftp, path, SftpSession::OpenType::Read, std::filesystem::perms::owner_read); } std::string createAlphabetString(std::size_t length) @@ -133,11 +155,7 @@ namespace SecureShell::Test { CREATE_SERVER_AND_JOINER(Sftp); auto [session, sftp] = createSftpSession(serverStartResult->port); - auto opened = sftp->openFile( - remote("/home/test/large.txt"), SftpSession::OpenType::Read, std::filesystem::perms::owner_read - ); - ASSERT_EQ(opened.wait_for(replyTimeout), std::future_status::ready); - ASSERT_TRUE(opened.get().has_value()); + ASSERT_TRUE(openForReading(*sftp, "/home/test/large.txt")); ASSERT_TRUE(endProcessingLoop(*sftp)); auto another = session->createSftpSession(); @@ -157,6 +175,46 @@ namespace SecureShell::Test EXPECT_TRUE(closing.get()); } + TEST_F(SftpTests, ReadAllAfterTheSftpSessionClosedFailsAtOnce) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForReading(*sftp, "/home/test/large.txt"); + ASSERT_TRUE(file); + ASSERT_TRUE(sftp->close()); + + auto read = file->readAll( + [](std::string_view) + { + return true; + } + ); + ASSERT_EQ(read.wait_for(0s), std::future_status::ready); + const auto result = read.get(); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed); + } + + TEST_F(SftpTests, ReadAllAfterATaskOnTheThreadThrewFailsAtOnce) + { + CREATE_SERVER_AND_JOINER(Sftp); + auto [_, sftp] = createSftpSession(serverStartResult->port); + auto file = openForReading(*sftp, "/home/test/large.txt"); + ASSERT_TRUE(file); + ASSERT_TRUE(endProcessingLoop(*sftp)); + + auto read = file->readAll( + [](std::string_view) + { + return true; + } + ); + ASSERT_EQ(read.wait_for(0s), std::future_status::ready); + const auto result = read.get(); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed); + } + TEST_F(SftpTests, LargeWriteIsWrittenInChunks) { CREATE_SERVER_AND_JOINER(Sftp);