Line data Source code
1 : #include "workers.hpp"
2 : #include <algorithm>
3 : #include <stdexcept>
4 : #include <pthread.h>
5 : #include <boost/asio/error.hpp>
6 : #include <boost/asio/executor_work_guard.hpp>
7 : #include <boost/asio/post.hpp>
8 : #include <boost/asio/signal_set.hpp>
9 : #include "../util/logger.hpp"
10 :
11 : namespace thinger::asio{
12 :
13 : namespace {
14 :
15 : // The workers that started the caller thread, and its io_context: set by the thread itself,
16 : // before any other work, so it is read without a lock
17 : struct thread_mark {
18 : const workers* owner = nullptr;
19 : boost::asio::io_context* context = nullptr;
20 : };
21 :
22 21323 : thread_mark& this_thread_mark() {
23 : static thread_local thread_mark mark;
24 21323 : return mark;
25 : }
26 :
27 : // The name of the caller thread, as the system shows it (truncated to its 15 characters)
28 6959 : void name_this_thread(const std::string& name) {
29 6959 : auto truncated = name.substr(0, 15);
30 : #if defined(__APPLE__)
31 : pthread_setname_np(truncated.c_str());
32 : #elif defined(__linux__)
33 6959 : pthread_setname_np(pthread_self(), truncated.c_str());
34 : #endif
35 6959 : }
36 :
37 : }
38 :
39 1237 : workers::workers() = default;
40 :
41 1237 : workers::~workers()
42 : {
43 1237 : if(running_) {
44 3 : LOG_WARNING("workers destructor called while still running - forcing stop");
45 3 : stop();
46 : }
47 : // A stop() called from one of its threads may still be joining the others
48 : {
49 1237 : std::unique_lock<std::mutex> lock(mutex_);
50 2477 : stopped_.wait(lock, [this] { return stopping_ == 0; });
51 1237 : }
52 : // Those that called stop() themselves, then the io_contexts with what they left pending
53 8196 : for(auto const& thread : stopped_threads_){
54 6959 : thread->stop();
55 : }
56 1237 : stopped_threads_.clear();
57 1237 : }
58 :
59 4897 : boost::system::error_code workers::start(size_t threads)
60 : {
61 4897 : std::scoped_lock<std::mutex> lock(mutex_);
62 4897 : if(running_) return boost::asio::error::already_started;
63 4894 : if(threads == 0) return make_error_code(boost::system::errc::invalid_argument);
64 :
65 4891 : LOG_DEBUG("starting {} working threads in the shared pool", threads);
66 4891 : auto table = std::make_unique<pool_table>();
67 4891 : worker_threads_.reserve(threads);
68 11796 : for(size_t thread_number=1; thread_number<=threads; ++thread_number){
69 6905 : auto worker = start_thread("worker thread " + std::to_string(thread_number));
70 6905 : table->contexts.push_back(&worker->get_io_context());
71 6905 : worker_threads_.emplace_back(std::move(worker));
72 6905 : }
73 :
74 4891 : next_io_context_.store(0, std::memory_order_relaxed);
75 4891 : table_.store(table.get());
76 4891 : tables_.emplace_back(std::move(table));
77 4891 : running_ = true;
78 4891 : return {};
79 4897 : }
80 :
81 6959 : std::unique_ptr<worker_thread> workers::start_thread(std::string name)
82 : {
83 6959 : auto worker = std::make_unique<worker_thread>(name);
84 6959 : auto& context = worker->get_io_context();
85 : // The first work of the thread: it marks itself as one of these workers
86 6959 : boost::asio::post(context, [this, &context, name = std::move(name)] {
87 6959 : this_thread_mark() = {this, &context};
88 6959 : name_this_thread(name);
89 6959 : });
90 6959 : worker->start();
91 6959 : return worker;
92 0 : }
93 :
94 60 : boost::asio::any_io_executor workers::isolated_executor(std::string name)
95 : {
96 60 : std::scoped_lock<std::mutex> lock(mutex_);
97 60 : if(!running_) return {};
98 54 : LOG_DEBUG("starting '{}' worker thread", name);
99 54 : auto worker = start_thread(std::move(name));
100 54 : auto executor = worker->get_io_context().get_executor();
101 54 : isolated_threads_.emplace_back(std::move(worker));
102 54 : return executor;
103 60 : }
104 :
105 4906 : void workers::stop()
106 : {
107 4906 : std::vector<std::unique_ptr<worker_thread>> threads;
108 : {
109 4906 : std::scoped_lock<std::mutex> lock(mutex_);
110 4906 : if(!running_) {
111 15 : LOG_WARNING("workers already stopped");
112 15 : return;
113 : }
114 4891 : running_ = false;
115 4891 : ++stopping_;
116 4891 : table_.store(nullptr);
117 4909 : for(auto* wait : waits_) wait->stop();
118 4891 : threads = std::move(isolated_threads_);
119 4891 : isolated_threads_.clear();
120 4891 : std::move(worker_threads_.begin(), worker_threads_.end(), std::back_inserter(threads));
121 4891 : worker_threads_.clear();
122 4906 : }
123 :
124 : // Out of the lock, as a handler finishing meanwhile may call back into the workers:
125 : // every io_context stops at once, then each thread is joined, but the caller's own
126 4891 : LOG_DEBUG("stopping {} worker threads", threads.size());
127 11850 : for(auto const& thread : threads){
128 6959 : thread->get_io_context().stop();
129 : }
130 : // Then the pending handlers of each are destroyed; the io_contexts are kept (see stop()
131 : // in workers.hpp)
132 4891 : auto* own_context = this_thread_mark().owner == this ? this_thread_mark().context : nullptr;
133 11850 : for(auto const& thread : threads){
134 6959 : if(&thread->get_io_context() == own_context) continue;
135 6884 : thread->stop();
136 6884 : thread->shutdown();
137 : }
138 :
139 4891 : std::scoped_lock<std::mutex> lock(mutex_);
140 4891 : std::move(threads.begin(), threads.end(), std::back_inserter(stopped_threads_));
141 4891 : --stopping_;
142 4891 : stopped_.notify_all();
143 4891 : LOG_DEBUG("all done!");
144 4906 : }
145 :
146 42 : int workers::wait(std::set<int> signals)
147 : {
148 42 : boost::asio::io_context context{1};
149 : // With no signal, it waits for stop()
150 42 : auto work = boost::asio::make_work_guard(context);
151 42 : boost::asio::signal_set signal_set(context);
152 90 : for(auto signal : signals){
153 48 : signal_set.add(signal);
154 : }
155 :
156 42 : int received = 0;
157 42 : signal_set.async_wait([&](const boost::system::error_code& ec, int signal_number){
158 21 : if(ec) return;
159 21 : LOG_INFO("received signal {}", signal_number);
160 21 : received = signal_number;
161 21 : context.stop();
162 : });
163 :
164 : {
165 42 : std::scoped_lock<std::mutex> lock(mutex_);
166 42 : if(!running_) return 0;
167 39 : waits_.push_back(&context);
168 42 : }
169 39 : context.run();
170 : {
171 39 : std::scoped_lock<std::mutex> lock(mutex_);
172 39 : std::erase(waits_, &context);
173 39 : }
174 39 : return received;
175 42 : }
176 :
177 941664 : boost::asio::io_context* workers::next_context()
178 : {
179 941664 : auto* table = table_.load();
180 941664 : if(!table) return nullptr;
181 286618 : auto next = next_io_context_.fetch_add(1, std::memory_order_relaxed);
182 286618 : return table->contexts[next % table->contexts.size()];
183 : }
184 :
185 941199 : boost::asio::any_io_executor workers::next_executor()
186 : {
187 941199 : auto* context = next_context();
188 941199 : if(!context) return {};
189 286159 : return context->get_executor();
190 : }
191 :
192 14288 : std::function<boost::asio::any_io_executor()> workers::provider()
193 : {
194 43478 : return [this] { return next_executor(); };
195 : }
196 :
197 465 : boost::asio::io_context& workers::get_next_io_context()
198 : {
199 465 : auto* context = next_context();
200 465 : if(!context) throw std::logic_error("thinger::asio::workers: not running");
201 459 : return *context;
202 : }
203 :
204 5500 : boost::asio::io_context& workers::get_thread_io_context()
205 : {
206 : // One of its threads: its io_context lives as long as the workers
207 5500 : const auto& mark = this_thread_mark();
208 5500 : if(mark.owner == this) return *mark.context;
209 :
210 : // Any other thread: the first thread of the pool
211 961 : auto* table = table_.load();
212 961 : if(!table) throw std::logic_error("thinger::asio::workers: not running");
213 955 : return *table->contexts.front();
214 : }
215 :
216 3898 : bool workers::in_worker_thread() const
217 : {
218 3898 : return this_thread_mark().owner == this && running_;
219 : }
220 :
221 : }
|
