Skip to content

Commit 6942416

Browse files
kzhdevkzhdev
committed
Refactor websocket service: encapsulate internals, replace globals with accessors, and improve signal handling
1 parent b543d81 commit 6942416

3 files changed

Lines changed: 100 additions & 65 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@
33
## Changed
44
- Bump min cmake version to 3.21 required by PROJECT_IS_TOP_LEVEL
55
- Refactor CMake configuration: streamline dependency checks and remove redundant fetch logic for slick-queue and slick-stream-buffer
6+
- Refactor websocket service internals: move `ioc_`, `ctx_`, `run_`, `service_thread_`, and `init_service_thread_` from `extern` globals in `slick::net::detail` into anonymous-namespace locals in `websocket.cpp`; replace direct access with the accessor functions `websocket_ioc()`, `websocket_ssl_context()`, and `websocket_running()`, and the lifecycle functions `start_websocket_service()` / `stop_websocket_service()`. This hides internal state from the header and eliminates ODR-unsafe `extern` declarations.
7+
- Signal handler now chains to the previously installed `SIGINT`/`SIGTERM` handler (re-raises with `SIG_DFL` for default handlers) instead of unconditionally overwriting it; previous handler pointers are reset to `SIG_DFL` in `stop_websocket_service()`.
68

79
# [v3.0.0] - 2026-06-16
810

‎include/slick/net/detail/websocket_impl.hpp‎

Lines changed: 21 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -40,13 +40,11 @@ using tcp = boost::asio::ip::tcp;
4040

4141
namespace slick::net::detail {
4242

43-
extern asio::io_context ioc_;
44-
extern ssl::context ctx_;
45-
extern std::thread service_thread_;
46-
extern std::atomic_bool init_service_thread_;
47-
extern std::atomic_bool run_;
48-
49-
void install_signal_handlers();
43+
asio::io_context& websocket_ioc() noexcept;
44+
ssl::context& websocket_ssl_context() noexcept;
45+
bool websocket_running() noexcept;
46+
void start_websocket_service();
47+
void stop_websocket_service();
5048

5149
struct websocket_url_parts {
5250
std::string host;
@@ -255,10 +253,10 @@ Websocket<BufferT>::Impl::Impl(
255253

256254
if (use_ssl_) {
257255
wss_ = std::make_unique<websocket::stream<ssl::stream<beast::tcp_stream>>>(
258-
asio::make_strand(detail::ioc_), detail::ctx_);
256+
asio::make_strand(detail::websocket_ioc()), detail::websocket_ssl_context());
259257
} else {
260258
ws_ = std::make_unique<websocket::stream<beast::tcp_stream>>(
261-
asio::make_strand(detail::ioc_));
259+
asio::make_strand(detail::websocket_ioc()));
262260
}
263261
}
264262

@@ -292,7 +290,7 @@ void Websocket<BufferT>::Impl::open() {
292290
session_started_.store(true, std::memory_order_release);
293291
}
294292
LOG_INFO("Opening WebSocket {}", url_);
295-
asio::co_spawn(detail::ioc_, do_ws_session(),
293+
asio::co_spawn(detail::websocket_ioc(), do_ws_session(),
296294
[self = this->shared_from_this()](std::exception_ptr eptr) {
297295
std::string err;
298296
bool failed = false;
@@ -308,15 +306,15 @@ void Websocket<BufferT>::Impl::open() {
308306
if (failed) {
309307
self->close();
310308
self->status_.store(Status::DISCONNECTED, std::memory_order_release);
311-
if (detail::run_.load(std::memory_order_relaxed) && !self->is_detached()) {
309+
if (detail::websocket_running() && !self->is_detached()) {
312310
self->on_error_(std::move(err));
313311
self->on_diconnected_();
314312
}
315313
}
316314
return;
317315
}
318316
const auto prev = self->status_.exchange(Status::DISCONNECTED, std::memory_order_acq_rel);
319-
if (detail::run_.load(std::memory_order_relaxed) && !self->is_detached()) {
317+
if (detail::websocket_running() && !self->is_detached()) {
320318
if (failed && prev != Status::DISCONNECTING) {
321319
self->on_error_(std::move(err));
322320
}
@@ -327,34 +325,7 @@ void Websocket<BufferT>::Impl::open() {
327325
self->release_buffer();
328326
});
329327

330-
auto init_service = detail::init_service_thread_.load(std::memory_order_relaxed);
331-
if (detail::init_service_thread_.compare_exchange_strong(init_service, true,
332-
std::memory_order_acq_rel) && !init_service) {
333-
detail::install_signal_handlers();
334-
detail::run_.store(true, std::memory_order_release);
335-
detail::service_thread_ = std::thread([]() {
336-
LOG_INFO("Websocket service thread started.");
337-
while (detail::run_.load(std::memory_order_relaxed)) {
338-
try {
339-
if (detail::ioc_.stopped()) {
340-
detail::ioc_.restart();
341-
}
342-
detail::ioc_.run();
343-
}
344-
catch(const std::exception& e) {
345-
detail::ioc_.restart();
346-
LOG_ERROR("{}", e.what());
347-
}
348-
}
349-
350-
if (!detail::ioc_.stopped()) [[unlikely]] {
351-
LOG_TRACE("call ioc_.stop at the end of run");
352-
detail::ioc_.stop();
353-
}
354-
LOG_INFO("Websocket service thread exit");
355-
detail::init_service_thread_.store(false, std::memory_order_release);
356-
});
357-
}
328+
detail::start_websocket_service();
358329
}
359330

360331
template<typename BufferT>
@@ -378,7 +349,7 @@ asio::awaitable<void> Websocket<BufferT>::Impl::do_ws_session_ssl() {
378349
std::memory_order_acq_rel, std::memory_order_relaxed);
379350
co_return;
380351
}
381-
tcp::resolver resolver(asio::make_strand(detail::ioc_));
352+
tcp::resolver resolver(asio::make_strand(detail::websocket_ioc()));
382353
auto result = co_await resolver.async_resolve(host_, std::to_string(port_), asio::use_awaitable);
383354

384355
if (!SSL_set_tlsext_host_name(wss_->next_layer().native_handle(), host_.c_str())) {
@@ -405,7 +376,7 @@ asio::awaitable<void> Websocket<BufferT>::Impl::do_ws_session_ssl() {
405376

406377
co_await wss_->async_handshake(host_header, path_, asio::use_awaitable);
407378

408-
if (!detail::run_.load(std::memory_order_relaxed) ||
379+
if (!detail::websocket_running() ||
409380
open_cancelled_.load(std::memory_order_acquire) || is_detached()) [[unlikely]] {
410381
Status expected = Status::CONNECTING;
411382
status_.compare_exchange_strong(expected, Status::DISCONNECTED,
@@ -452,7 +423,7 @@ asio::awaitable<void> Websocket<BufferT>::Impl::do_ws_session_plain() {
452423
std::memory_order_acq_rel, std::memory_order_relaxed);
453424
co_return;
454425
}
455-
tcp::resolver resolver(asio::make_strand(detail::ioc_));
426+
tcp::resolver resolver(asio::make_strand(detail::websocket_ioc()));
456427
auto result = co_await resolver.async_resolve(host_, std::to_string(port_), asio::use_awaitable);
457428

458429
beast::get_lowest_layer(*ws_).expires_after(std::chrono::seconds(30));
@@ -469,7 +440,7 @@ asio::awaitable<void> Websocket<BufferT>::Impl::do_ws_session_plain() {
469440

470441
co_await ws_->async_handshake(host_header, path_, asio::use_awaitable);
471442

472-
if (!detail::run_.load(std::memory_order_relaxed) ||
443+
if (!detail::websocket_running() ||
473444
open_cancelled_.load(std::memory_order_acquire) || is_detached()) [[unlikely]] {
474445
Status expected = Status::CONNECTING;
475446
status_.compare_exchange_strong(expected, Status::DISCONNECTED,
@@ -553,7 +524,7 @@ template<typename BufferT>
553524
void Websocket<BufferT>::Impl::on_write(beast::error_code ec, std::size_t bytes_transferred) {
554525
boost::ignore_unused(bytes_transferred);
555526
if(ec) {
556-
if (detail::run_.load(std::memory_order_relaxed) &&
527+
if (detail::websocket_running() &&
557528
status_.load(std::memory_order_relaxed) == Status::CONNECTED &&
558529
ec != beast::websocket::error::closed &&
559530
ec != asio::error::eof &&
@@ -574,7 +545,7 @@ void Websocket<BufferT>::Impl::on_write(beast::error_code ec, std::size_t bytes_
574545
template<typename BufferT>
575546
void Websocket<BufferT>::Impl::on_read(beast::error_code ec, std::size_t bytes_transferred) {
576547
if(ec) {
577-
if (detail::run_.load(std::memory_order_relaxed) &&
548+
if (detail::websocket_running() &&
578549
status_.load(std::memory_order_relaxed) == Status::CONNECTED &&
579550
ec != beast::websocket::error::closed &&
580551
ec != asio::error::eof &&
@@ -597,7 +568,7 @@ void Websocket<BufferT>::Impl::on_read(beast::error_code ec, std::size_t bytes_t
597568
return;
598569
}
599570

600-
if (detail::run_.load(std::memory_order_relaxed) &&
571+
if (detail::websocket_running() &&
601572
status_.load(std::memory_order_relaxed) == Status::CONNECTED &&
602573
!is_detached()) {
603574
LOG_TRACE("<-- {}", std::string((const char*)r_buffer_->data().data(), bytes_transferred));
@@ -626,7 +597,7 @@ void Websocket<BufferT>::Impl::on_read(beast::error_code ec, std::size_t bytes_t
626597

627598
template<typename BufferT>
628599
void Websocket<BufferT>::Impl::on_close(beast::error_code ec) {
629-
if (ec && detail::run_.load(std::memory_order_relaxed) &&
600+
if (ec && detail::websocket_running() &&
630601
ec != beast::websocket::error::closed &&
631602
ec != asio::error::eof &&
632603
ec != asio::error::operation_aborted &&
@@ -838,19 +809,12 @@ typename Websocket<BufferT>::Status Websocket<BufferT>::status() const noexcept
838809

839810
template<typename BufferT>
840811
bool Websocket<BufferT>::is_running() noexcept {
841-
return detail::run_.load(std::memory_order_relaxed);
812+
return detail::websocket_running();
842813
}
843814

844815
template<typename BufferT>
845816
void Websocket<BufferT>::shutdown() {
846-
if (detail::run_.load(std::memory_order_relaxed)) {
847-
LOG_DEBUG("Shutting down WebSocket service thread.");
848-
detail::run_.store(false, std::memory_order_release);
849-
detail::ioc_.stop();
850-
if (detail::service_thread_.joinable()) {
851-
detail::service_thread_.join();
852-
}
853-
}
817+
detail::stop_websocket_service();
854818
}
855819

856820
template<typename BufferT>

‎src/websocket.cpp‎

Lines changed: 77 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,96 @@
11
#include <slick/net/detail/websocket_impl.hpp>
22

3+
namespace {
4+
asio::io_context ioc_;
5+
ssl::context ctx_{ssl::context::tlsv12_client};
6+
std::thread service_thread_;
7+
std::atomic_bool init_service_thread_{false};
8+
std::atomic_bool run_{false};
9+
10+
using signal_handler_t = void (*)(int);
11+
signal_handler_t previous_sigint = SIG_DFL;
12+
signal_handler_t previous_sigterm = SIG_DFL;
13+
}
14+
315
namespace slick::net::detail {
416

5-
asio::io_context ioc_;
6-
ssl::context ctx_{ssl::context::tlsv12_client};
7-
std::thread service_thread_;
8-
std::atomic_bool init_service_thread_{ false };
9-
std::atomic_bool run_;
17+
asio::io_context& websocket_ioc() noexcept {
18+
return ioc_;
19+
}
20+
ssl::context& websocket_ssl_context() noexcept {
21+
return ctx_;
22+
}
23+
24+
bool websocket_running() noexcept {
25+
return run_.load(std::memory_order_relaxed);
26+
}
1027

1128
extern "C" inline void __signal_handler(int signal) {
1229
if (signal == SIGINT || signal == SIGTERM) {
1330
Websocket<>::shutdown();
1431
}
32+
33+
auto previous = signal == SIGINT ? previous_sigint : previous_sigterm;
34+
if (previous && previous != SIG_DFL && previous != SIG_IGN) {
35+
previous(signal);
36+
} else if (previous == SIG_DFL) {
37+
std::signal(signal, SIG_DFL);
38+
std::raise(signal);
39+
}
1540
}
1641

1742
void install_signal_handlers() {
18-
std::signal(SIGINT, __signal_handler);
19-
std::signal(SIGTERM, __signal_handler);
43+
previous_sigint = std::signal(SIGINT, __signal_handler);
44+
previous_sigterm = std::signal(SIGTERM, __signal_handler);
45+
}
46+
47+
void start_websocket_service() {
48+
auto init_service = init_service_thread_.load(std::memory_order_relaxed);
49+
if (init_service_thread_.compare_exchange_strong(init_service, true,
50+
std::memory_order_acq_rel) && !init_service) {
51+
install_signal_handlers();
52+
run_.store(true, std::memory_order_release);
53+
service_thread_ = std::thread([]() {
54+
LOG_INFO("Websocket service thread started.");
55+
while (run_.load(std::memory_order_relaxed)) {
56+
try {
57+
if (ioc_.stopped()) {
58+
ioc_.restart();
59+
}
60+
ioc_.run();
61+
}
62+
catch(const std::exception& e) {
63+
ioc_.restart();
64+
LOG_ERROR("{}", e.what());
65+
}
66+
}
67+
68+
if (!ioc_.stopped()) [[unlikely]] {
69+
LOG_TRACE("call ioc_.stop at the end of run");
70+
ioc_.stop();
71+
}
72+
LOG_INFO("Websocket service thread exit");
73+
init_service_thread_.store(false, std::memory_order_release);
74+
});
75+
}
76+
}
77+
78+
void stop_websocket_service() {
79+
if (run_.load(std::memory_order_relaxed)) {
80+
LOG_DEBUG("Shutting down WebSocket service thread.");
81+
run_.store(false, std::memory_order_release);
82+
ioc_.stop();
83+
if (service_thread_.joinable()) {
84+
service_thread_.join();
85+
}
86+
}
87+
previous_sigint = SIG_DFL;
88+
previous_sigterm = SIG_DFL;
2089
}
2190

2291
struct WebsocketServiceTerminater {
2392
~WebsocketServiceTerminater() {
24-
Websocket<>::shutdown();
93+
stop_websocket_service();
2594
}
2695
};
2796

0 commit comments

Comments
 (0)