Line data Source code
1 : #ifndef THINGER_ASIO_WORKERS
2 : #define THINGER_ASIO_WORKERS
3 :
4 : #include <atomic>
5 : #include <condition_variable>
6 : #include <csignal>
7 : #include <functional>
8 : #include <memory>
9 : #include <mutex>
10 : #include <set>
11 : #include <string>
12 : #include <thread>
13 : #include <vector>
14 : #include <boost/asio/any_io_executor.hpp>
15 : #include <boost/asio/io_context.hpp>
16 : #include <boost/system/error_code.hpp>
17 : #include "worker_thread.hpp"
18 :
19 : namespace thinger::asio {
20 :
21 : // A pool of threads, each running an io_context of its own, plus the isolated threads asked
22 : // for. An ordinary object: the application owns it, starts it and stops it (nothing starts
23 : // or stops it by itself), and hands it by reference to the servers and clients that run on
24 : // it (pool_server, async_client). They own no thread, and the application keeps the workers
25 : // alive longer than them. Several instances may live in one process; they share nothing.
26 : //
27 : // Never destroy a workers from one of its own threads.
28 : class workers {
29 : public:
30 : workers();
31 : ~workers();
32 :
33 : workers(const workers&) = delete;
34 : workers& operator=(const workers&) = delete;
35 :
36 : /// start `threads` threads in the pool: `already_started` if it runs already (no thread
37 : /// is added), `invalid_argument` for 0 threads. Their io_contexts are kept until the
38 : /// workers are destroyed, also when they start again after stop()
39 : boost::system::error_code start(size_t threads = std::thread::hardware_concurrency());
40 :
41 : /// stop every thread, the pool's and the isolated ones, destroying their pending
42 : /// operations, and release the wait() calls in progress. From any thread, more than once
43 : /// (a warning once stopped). Stop the servers and clients on it first. Their io_contexts
44 : /// live on, stopped, until the workers are destroyed: a client destroyed afterwards
45 : /// still closes its connections on them. Called from one of its threads, that thread
46 : /// ends once its handler returns, and the destructor joins it.
47 : void stop();
48 :
49 : /// whether the workers run
50 2579 : bool running() const { return running_; }
51 :
52 : /// the executor of the next thread of the pool (round-robin), from any thread, without
53 : /// a lock; a null executor when the workers do not run
54 : boost::asio::any_io_executor next_executor();
55 :
56 : /// next_executor(), as a callable (for a listener)
57 : std::function<boost::asio::any_io_executor()> provider();
58 :
59 : /// the executor of a new thread of its own, out of the pool, named `name`: it runs until
60 : /// the workers stop. A null executor when the workers do not run
61 : boost::asio::any_io_executor isolated_executor(std::string name);
62 :
63 : /// whether the caller thread is one of the threads these workers started (of the pool
64 : /// or isolated) and they still run: a thread-local, without a lock
65 : bool in_worker_thread() const;
66 :
67 : /// block until one of `signals` arrives, and return it; 0 when stop() released it
68 : /// instead, or at once if the workers do not run. The workers keep running: what to do
69 : /// with the signal (a server's shutdown(), then stop()) is the application's. With an
70 : /// empty set, it waits for stop(). While waiting, the signals do not take their default
71 : /// action; afterwards they do again.
72 : int wait(std::set<int> signals = {SIGINT, SIGTERM, SIGQUIT});
73 :
74 : /// the io_context of the next thread of the pool (round-robin), without a lock. Throws
75 : /// std::logic_error when the workers do not run
76 : boost::asio::io_context& get_next_io_context();
77 :
78 : /// the io_context of the caller thread when it is one of these workers' threads (from a
79 : /// thread-local, without a lock), or else the one of the first thread of the pool, which
80 : /// throws std::logic_error when the workers do not run. It goes at 0.10
81 : boost::asio::io_context& get_thread_io_context();
82 :
83 : private:
84 : // The pool's io_contexts, published by start() and read without a lock: immutable, and
85 : // kept (as the io_contexts) until the workers are destroyed
86 : struct pool_table {
87 : std::vector<boost::asio::io_context*> contexts;
88 : };
89 :
90 : /// the io_context of the next thread of the pool, or null when not running
91 : boost::asio::io_context* next_context();
92 :
93 : /// start a thread of these workers, marked as theirs
94 : std::unique_ptr<worker_thread> start_thread(std::string name);
95 :
96 : /// guards start(), stop(), the threads and the waits
97 : mutable std::mutex mutex_;
98 :
99 : /// the stop() calls still joining threads, which the destructor waits for (one called
100 : /// from a thread of the workers returns after the wait()s it releases)
101 : size_t stopping_ = 0;
102 : std::condition_variable stopped_;
103 :
104 : /// flag for controlling if it is running
105 : std::atomic<bool> running_{false};
106 :
107 : /// the published pool table (null when not running), and every table published
108 : std::atomic<const pool_table*> table_{nullptr};
109 : std::vector<std::unique_ptr<const pool_table>> tables_;
110 :
111 : /// index of the next io_context of the round-robin (relaxed: it only spreads the load)
112 : std::atomic<unsigned> next_io_context_{0};
113 :
114 : /// threads running: of the pool, and the isolated ones
115 : std::vector<std::unique_ptr<worker_thread>> worker_threads_;
116 : std::vector<std::unique_ptr<worker_thread>> isolated_threads_;
117 :
118 : /// threads stopped, with their io_contexts (see stop())
119 : std::vector<std::unique_ptr<worker_thread>> stopped_threads_;
120 :
121 : /// the io_contexts of the wait() calls in progress, which stop() releases
122 : std::vector<boost::asio::io_context*> waits_;
123 : };
124 :
125 : }
126 :
127 : #endif
|
