Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/concurrency/io_context_manager.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <atomic>
22 : : #include <condition_variable>
23 : : #include <mutex>
24 : : #include <stop_token>
25 : : #include <thread>
26 : :
27 : : #include "wirestead/concurrency/io_thread_hook.hpp"
28 : : #include "wirestead/diagnostics/logger.hpp"
29 : :
30 : : namespace wirestead {
31 : : namespace concurrency {
32 : :
33 : : struct IoContextManager::Impl {
34 : : bool owns_context_{true};
35 : : std::shared_ptr<IoContext> ioc_;
36 : : std::unique_ptr<WorkGuard> work_guard_;
37 : : std::jthread io_thread_;
38 : : std::atomic<bool> running_{false};
39 : : mutable std::mutex mutex_;
40 : : std::condition_variable cv_;
41 : : bool stopping_{false};
42 : :
43 : 140 : Impl() { diagnostics::Logger::instance(); }
44 : :
45 : 0 : explicit Impl(std::shared_ptr<IoContext> external_context) : owns_context_(false), ioc_(std::move(external_context)) {
46 : 0 : diagnostics::Logger::instance();
47 : 0 : }
48 : :
49 : 1 : explicit Impl(IoContext& external_context)
50 : 1 : : owns_context_(false), ioc_(std::shared_ptr<IoContext>(&external_context, [](IoContext*) {})) {
51 : 1 : diagnostics::Logger::instance();
52 : 1 : }
53 : :
54 : 4 : ~Impl() {
55 : : try {
56 : 4 : stop();
57 : 0 : } catch (...) {
58 : 0 : }
59 : 4 : }
60 : :
61 : 36 : void stop() {
62 : 36 : std::unique_lock<std::mutex> lock(mutex_);
63 [ + - ]: 78 : cv_.wait(lock, [this] { return !stopping_; });
64 [ + + + - : 36 : if (!owns_context_ && ioc_) return;
+ + ]
65 [ + + + + : 34 : if (!running_.load() && !io_thread_.joinable()) return;
+ + ]
66 : :
67 : 26 : stopping_ = true;
68 [ + + ]: 26 : if (work_guard_) work_guard_.reset();
69 [ + - + - : 26 : if (ioc_ && owns_context_) ioc_->stop();
+ - + - ]
70 : :
71 [ + - ]: 26 : if (io_thread_.joinable()) {
72 [ + + ]: 26 : if (io_thread_.get_id() == std::this_thread::get_id()) {
73 : 1 : WIRESTEAD_LOG_ERROR("io_context_manager", "stop", "Cannot join IoContext thread from within itself.");
74 : 1 : stopping_ = false;
75 : 1 : cv_.notify_all();
76 : 1 : return;
77 : : }
78 : 25 : lock.unlock();
79 : 25 : io_thread_.request_stop();
80 : 25 : io_thread_.join();
81 : 25 : lock.lock();
82 : : }
83 : :
84 : 25 : stopping_ = false;
85 : 25 : running_.store(false);
86 : 25 : cv_.notify_all();
87 : 36 : }
88 : : };
89 : :
90 : 140 : IoContextManager::IoContextManager() : impl_(std::make_unique<Impl>()) {}
91 : :
92 : 0 : IoContextManager::IoContextManager(std::shared_ptr<IoContext> external_context)
93 : 0 : : impl_(std::make_unique<Impl>(std::move(external_context))) {}
94 : :
95 : 1 : IoContextManager::IoContextManager(IoContext& external_context) : impl_(std::make_unique<Impl>(external_context)) {}
96 : :
97 : 4 : IoContextManager::~IoContextManager() = default;
98 : :
99 : 534 : IoContextManager& IoContextManager::instance() {
100 : 534 : static IoContextManager* instance = new IoContextManager();
101 : 534 : return *instance;
102 : : }
103 : :
104 : 13 : boost::asio::io_context& IoContextManager::get_context() {
105 : 13 : std::lock_guard<std::mutex> lock(impl_->mutex_);
106 [ + + ]: 13 : if (!impl_->ioc_) {
107 : 2 : impl_->ioc_ = std::make_shared<IoContext>();
108 : 2 : impl_->owns_context_ = true;
109 : : }
110 : 26 : return *impl_->ioc_;
111 : 13 : }
112 : :
113 : 161 : void IoContextManager::start() {
114 : 161 : std::shared_ptr<IoContext> context;
115 : : {
116 : 161 : std::unique_lock<std::mutex> lock(impl_->mutex_);
117 : :
118 [ + + + - : 161 : if (!impl_->owns_context_ && impl_->ioc_) {
+ + ]
119 [ + - - + ]: 1 : if (impl_->ioc_->stopped()) {
120 : 0 : WIRESTEAD_LOG_WARNING("io_context_manager", "start", "External io_context is stopped.");
121 : : }
122 : 1 : return;
123 : : }
124 : :
125 [ + - ]: 320 : impl_->cv_.wait(lock, [this] { return !impl_->stopping_; });
126 : :
127 [ + + ]: 160 : if (impl_->running_) return;
128 : :
129 [ - + - - : 155 : if (impl_->io_thread_.joinable() && impl_->io_thread_.get_id() == std::this_thread::get_id()) {
- + ]
130 : 0 : WIRESTEAD_LOG_ERROR("io_context_manager", "start", "Cannot restart from within its own thread.");
131 : 0 : return;
132 : : }
133 : :
134 [ + + ]: 155 : if (!impl_->ioc_) {
135 : 137 : impl_->ioc_ = std::make_shared<IoContext>();
136 : 137 : impl_->owns_context_ = true;
137 : : }
138 : :
139 [ + - + + ]: 155 : if (impl_->ioc_->stopped()) {
140 : 18 : impl_->ioc_->restart();
141 : : }
142 : 155 : impl_->work_guard_ = std::make_unique<WorkGuard>(impl_->ioc_->get_executor());
143 : 155 : context = impl_->ioc_;
144 : :
145 [ - + ]: 155 : if (impl_->io_thread_.joinable()) {
146 : 0 : impl_->io_thread_.join();
147 : : }
148 : :
149 : 310 : impl_->io_thread_ = std::jthread([this, context](std::stop_token st) {
150 : 155 : wirestead::concurrency::run_io_thread_init();
151 : : try {
152 : : // Register stop callback to gracefully stop io_context when jthread is stopped
153 : 176 : std::stop_callback cb(st, [context] { context->stop(); });
154 : 155 : context->run();
155 : 26 : } catch (const std::exception& e) {
156 : 1 : WIRESTEAD_LOG_ERROR("io_context_manager", "run", fmt::format("Thread error: {}", e.what()));
157 : 1 : } catch (...) {
158 : 0 : }
159 : 25 : impl_->running_.store(false);
160 : 180 : });
161 : 155 : impl_->running_.store(true);
162 : 161 : }
163 : 161 : }
164 : :
165 : 32 : void IoContextManager::stop() { impl_->stop(); }
166 : :
167 : 400 : bool IoContextManager::is_running() const { return get_impl()->running_.load(); }
168 : :
169 : 2 : std::unique_ptr<boost::asio::io_context> IoContextManager::create_independent_context() {
170 : 2 : return std::make_unique<IoContext>();
171 : : }
172 : :
173 : 0 : IoContextManager::IoContextManager(IoContextManager&& other) noexcept = default;
174 : 0 : IoContextManager& IoContextManager::operator=(IoContextManager&& other) noexcept = default;
175 : :
176 : : } // namespace concurrency
177 : : } // namespace wirestead
|
