Skip to content

Commit 11ca3d1

Browse files
MorganCaronMorganCaron
committed
Thread/ThreadLoop: onError callback
1 parent 4702462 commit 11ca3d1

3 files changed

Lines changed: 153 additions & 48 deletions

File tree

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@
8383
### 🚦 Multithreading & Synchronization
8484
- [`Event`](modules/Thread/Event.mpp) - An event for thread synchronization
8585
- [`SharedLocker`](modules/Thread/SharedLocker.mpp) - Scoped lock wrapper with shared or exclusive mode
86-
- [`ThreadLoop`](modules/Thread/ThreadLoop.mpp) - Thread loop with exception propagation
86+
- [`ThreadLoop`](modules/Thread/ThreadLoop.mpp) - Thread loop with exception handling
8787
- [`ThreadPool`](modules/Thread/ThreadPool.mpp) - Fixed-size thread pool for parallel task execution
8888
- [`TryAsync`](modules/Thread/TryAsync.mpp) - Launches a function asynchronously, forwards exception to caller
8989
- [`UniqueLocker`](modules/Thread/UniqueLocker.mpp) - Scoped exclusive lock wrapper

‎modules/Thread/ThreadLoop.mpp‎

Lines changed: 119 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -7,90 +7,162 @@ export namespace CppUtils::Thread
77
{
88
class ThreadLoop final
99
{
10+
using Logger = Logger<"CppUtils">;
11+
1012
public:
1113
inline ThreadLoop() = delete;
1214

13-
inline explicit ThreadLoop(std::function<void()> function, std::function<void()> interruptFunction = nullptr, auto&&... args):
14-
m_data{std::make_unique<Data>(std::bind(function, std::forward<decltype(args)>(args)...), std::move(interruptFunction))}
15+
inline explicit ThreadLoop(std::function<void()> function,
16+
std::function<void()> interruptFunction = nullptr,
17+
std::function<void(std::exception_ptr)> onError = nullptr) noexcept:
18+
m_function{std::move(function)},
19+
m_interruptFunction{std::move(interruptFunction)},
20+
m_onError{std::move(onError)}
1521
{}
1622

1723
inline ThreadLoop(const ThreadLoop&) = delete;
18-
inline ThreadLoop(ThreadLoop&& src) noexcept:
19-
m_data{std::move(src.m_data)}
20-
{}
2124
inline auto operator=(const ThreadLoop&) -> ThreadLoop& = delete;
22-
inline auto operator=(ThreadLoop&& rhs) noexcept -> ThreadLoop&
25+
26+
inline ThreadLoop(ThreadLoop&& other) noexcept
27+
{
28+
auto lockGuard = std::unique_lock{other.m_mutex};
29+
m_function = std::move(other.m_function);
30+
m_interruptFunction = std::move(other.m_interruptFunction);
31+
m_onError = std::move(other.m_onError);
32+
m_thread = std::move(other.m_thread);
33+
}
34+
35+
inline auto operator=(ThreadLoop&& other) noexcept -> ThreadLoop&
2336
{
24-
m_data = std::move(rhs.m_data);
37+
if (this == std::addressof(other))
38+
return *this;
39+
auto lockGuard = std::scoped_lock{m_mutex, other.m_mutex};
40+
m_function = std::move(other.m_function);
41+
m_interruptFunction = std::move(other.m_interruptFunction);
42+
m_onError = std::move(other.m_onError);
43+
m_thread = std::move(other.m_thread);
2544
return *this;
2645
}
2746

28-
inline ~ThreadLoop() noexcept(false)
47+
inline ~ThreadLoop() noexcept
2948
{
30-
if (m_data)
31-
stop();
49+
stop();
3250
}
3351

3452
[[nodiscard]] inline auto isRunning() const noexcept -> bool
3553
{
36-
return m_data->thread.joinable() and not m_data->exceptionOccurred.load();
54+
auto lockGuard = std::unique_lock{m_mutex};
55+
return m_thread.joinable();
56+
}
57+
58+
inline auto setInterruptFunction(std::function<void()> interruptFunction) noexcept -> void
59+
{
60+
auto lockGuard = std::unique_lock{m_mutex};
61+
m_interruptFunction = std::move(interruptFunction);
62+
}
63+
64+
inline auto setOnError(std::function<void(std::exception_ptr)> onError) noexcept -> void
65+
{
66+
auto lockGuard = std::unique_lock{m_mutex};
67+
m_onError = std::move(onError);
3768
}
3869

3970
inline auto start() -> void
4071
{
41-
m_data->exceptionOccurred = false;
42-
m_data->thread = std::jthread{[data = m_data.get()](std::stop_token stopToken) -> void {
43-
try
44-
{
45-
while (not stopToken.stop_requested())
46-
if (data->function)
47-
data->function();
48-
return;
49-
}
50-
catch (const std::exception& exception)
51-
{
52-
logException(std::runtime_error{std::format("Exception occurred during thread execution:\n{}\nThread execution is interrupted", exception.what())});
53-
}
54-
catch (...)
72+
auto lockGuard = std::unique_lock{m_mutex};
73+
if (m_thread.joinable())
74+
return;
75+
76+
m_thread = std::jthread{[function = m_function, interruptFunction = m_interruptFunction, onError = m_onError](std::stop_token stopToken) mutable -> void {
77+
while (not stopToken.stop_requested())
5578
{
56-
Logger<"CppUtils">::print<"error">("Exception occurred during thread execution.\nThread execution is interrupted");
79+
try
80+
{
81+
if (function)
82+
function();
83+
}
84+
catch (...)
85+
{
86+
if (onError)
87+
{
88+
try
89+
{
90+
onError(std::current_exception());
91+
}
92+
catch (const std::exception& exception)
93+
{
94+
Logger::print<"error">("ThreadLoop: onError threw an exception: {}", exception.what());
95+
}
96+
catch (...)
97+
{
98+
Logger::print<"error">("ThreadLoop: onError threw a non-std exception");
99+
}
100+
}
101+
else
102+
{
103+
Logger::print<"error">("ThreadLoop: Unhandled exception");
104+
}
105+
}
57106
}
58-
data->exceptionOccurred = true;
59107
}};
60108
}
61109

62110
inline auto isStopRequested() const -> bool
63111
{
64-
return m_data->thread.get_stop_token().stop_requested();
112+
auto lockGuard = std::unique_lock{m_mutex};
113+
if (not m_thread.joinable())
114+
return false;
115+
return m_thread.get_stop_token().stop_requested();
65116
}
66117

67118
inline auto requestStop() -> void
68119
{
69-
if (not isRunning() or isStopRequested())
120+
auto lockGuard = std::unique_lock{m_mutex};
121+
if (not m_thread.joinable() or m_thread.get_stop_token().stop_requested())
70122
return;
71-
m_data->thread.request_stop();
123+
m_thread.request_stop();
72124
}
73125

74-
inline auto stop() -> void
126+
inline auto stop() noexcept -> void
75127
{
76-
if (m_data->exceptionOccurred.load())
77-
throw std::runtime_error{"Exception occurred during thread execution"};
78-
if (not isRunning())
79-
return;
80-
requestStop();
81-
if (m_data->interruptFunction)
82-
m_data->interruptFunction();
83-
m_data->thread.join();
128+
try
129+
{
130+
auto lockGuard = std::unique_lock{m_mutex};
131+
if (not m_thread.joinable() or m_thread.get_stop_token().stop_requested())
132+
return;
133+
m_thread.request_stop();
134+
if (m_interruptFunction)
135+
{
136+
auto interruptFunction = m_interruptFunction;
137+
lockGuard.unlock();
138+
try
139+
{
140+
interruptFunction();
141+
}
142+
catch (...)
143+
{
144+
Logger::print<"error">("ThreadLoop: interruptFunction threw during stop");
145+
}
146+
lockGuard.lock();
147+
}
148+
if (m_thread.joinable())
149+
m_thread.join();
150+
}
151+
catch (const std::exception& exception)
152+
{
153+
Logger::print<"error">("Exception in ThreadLoop::stop(): {}", exception.what());
154+
}
155+
catch (...)
156+
{
157+
Logger::print<"error">("Unknown exception in ThreadLoop::stop()");
158+
}
84159
}
85160

86161
private:
87-
struct Data final
88-
{
89-
std::function<void()> function;
90-
std::function<void()> interruptFunction;
91-
std::atomic_bool exceptionOccurred = false;
92-
std::jthread thread;
93-
};
94-
std::unique_ptr<Data> m_data;
162+
mutable std::mutex m_mutex;
163+
std::function<void()> m_function;
164+
std::function<void()> m_interruptFunction;
165+
std::function<void(std::exception_ptr)> m_onError;
166+
std::jthread m_thread;
95167
};
96168
}

‎tests/Thread/ThreadLoop.mpp‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ export namespace CppUtils::UnitTest::Thread::ThreadLoop
77
{
88
inline auto _ = CppUtils::UnitTest::TestSuite{"Thread/ThreadLoop", {"UnitTest"}, [](auto& suite) {
99
using namespace std::chrono_literals;
10+
using Logger = CppUtils::Logger<"CppUtils">;
1011

1112
suite.addTest("isRunning", [&] {
1213
auto threadLoop = CppUtils::Thread::ThreadLoop{[] {}};
@@ -35,5 +36,37 @@ export namespace CppUtils::UnitTest::Thread::ThreadLoop
3536
auto nb = accessor.value();
3637
suite.expectEqual(nb, 10uz);
3738
});
39+
40+
suite.addTest("onError callback", [&] {
41+
auto onErrorCalled = std::atomic_bool{false};
42+
43+
auto threadLoop = CppUtils::Thread::ThreadLoop{[] {
44+
static auto count = 0;
45+
if (++count == 1)
46+
throw std::runtime_error{"Exception"};
47+
}, nullptr, [&](std::exception_ptr exceptionPointer) {
48+
onErrorCalled = true;
49+
try
50+
{
51+
if (exceptionPointer)
52+
std::rethrow_exception(exceptionPointer);
53+
}
54+
catch (const std::exception& exception)
55+
{
56+
suite.expectEqual(std::string{exception.what()}, "Exception");
57+
throw std::runtime_error{"onError fail"};
58+
}
59+
}};
60+
61+
Logger::print<"detail">("Normal error message:");
62+
threadLoop.start();
63+
std::this_thread::sleep_for(50ms);
64+
65+
suite.expect(onErrorCalled);
66+
67+
suite.expect(threadLoop.isRunning());
68+
threadLoop.stop();
69+
suite.expect(not threadLoop.isRunning());
70+
});
3871
}};
3972
}

0 commit comments

Comments
 (0)