Index/src / corosio / src / tcp_server.cpp

src/corosio/src/tcp_server.cpp

95.5% Lines (126 / 132) 62.6% Branches (119 / 190) 100.0% Functions (21 / 21)
[ ] prev/next file n p prev/next uncovered line
tcp_server.cpp
f(x) Functions (21)
Function Calls Lines Branches Blocks
boost::corosio::tcp_server::worker_base::worker_base() :21 403x 100.0% – 100.0% boost::corosio::tcp_server::worker_base::~worker_base() :22 403x 100.0% – 100.0% boost::corosio::tcp_server::impl::~impl() :24 294x 100.0% – 100.0% boost::corosio::tcp_server::impl::impl(boost::capy::execution_context&) :37 294x 100.0% 50.0% 100.0% boost::corosio::tcp_server::make_state(boost::capy::execution_context&, boost::capy::any_executor const&) :41 147x 37.5% 16.7% 25.0% boost::corosio::tcp_server::release(boost::corosio::tcp_server::state*) :57 7761x 100.0% 66.7% 100.0% boost::corosio::tcp_server::discard() :67 153x 100.0% 77.8% 95.0% boost::corosio::tcp_server::~tcp_server() :86 153x 100.0% – 100.0% boost::corosio::tcp_server::tcp_server(boost::corosio::tcp_server&&) :91 4x 75.0% – 100.0% boost::corosio::tcp_server::operator=(boost::corosio::tcp_server&&) :97 2x 100.0% 50.0% 100.0% boost::corosio::tcp_server::do_accept(boost::corosio::tcp_server::state_ref, boost::corosio::tcp_acceptor&) :109 135x 100.0% 63.4% 61.0% boost::corosio::tcp_server::bind(boost::corosio::endpoint) :138 147x 100.0% 75.0% 71.0% boost::corosio::tcp_server::local_endpoint(unsigned long) const :148 135x 100.0% 75.0% 100.0% boost::corosio::tcp_server::start() :156 137x 100.0% 80.0% 72.0% boost::corosio::tcp_server::start()::$_0::$_0($_0&&) :184 1080x 100.0% – 100.0% boost::corosio::tcp_server::start()::$_0::operator()() const :184 129x 100.0% 70.0% 100.0% boost::corosio::tcp_server::start()::$_0::~$_0() :184 1350x 100.0% – 100.0% boost::corosio::tcp_server::stop() :200 234x 100.0% 60.0% 57.0% boost::corosio::tcp_server::join() :216 65x 100.0% – 50.0% boost::corosio::tcp_server::join()::$_0::operator()() const :220 65x 100.0% 50.0% 100.0% boost::corosio::tcp_server::do_stop(boost::corosio::tcp_server::state_ref) :224 133x 100.0% 52.9% 57.0%
Line Branch TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Vinnie Falco ([email protected])
3 // Copyright (c) 2026 Steve Gerbino
4 //
5 // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 //
8 // Official repository: https://github.com/cppalliance/corosio
9 //
10
11 #include <boost/corosio/tcp_server.hpp>
12 #include <boost/corosio/detail/except.hpp>
13 #include <condition_variable>
14 #include <mutex>
15 #include <utility>
16
17 #include "src/detail/open_listener.hpp"
18
19 namespace boost::corosio {
20
21 403x tcp_server::worker_base::worker_base() = default;
22 403x tcp_server::worker_base::~worker_base() = default;
23
24 struct tcp_server::impl
25 {
26 std::mutex join_mutex;
27 std::condition_variable join_cv;
28 capy::execution_context& ctx;
29 std::vector<tcp_acceptor> ports;
30 std::stop_source stop;
31 // Guarded by join_mutex. Each listener is closed exactly once after
32 // the server is gone: by its loop's completion if the loop was still
33 // running when the server went, otherwise by the departing server.
34 std::vector<bool> loop_running;
35 147x bool discarded = false;
36
37
1/2
✓ Branch 0 taken 147 times.
✗ Branch 1 not taken.
 294x explicit impl(capy::execution_context& c) noexcept : ctx(c) {}
38 };
39
40 tcp_server::state*
41 147x tcp_server::make_state(
42 capy::execution_context& ctx, capy::any_executor const& ex)
43 {
44 147x auto* i = new impl(ctx);
45 try
46 {
47
1/2
✓ Branch 0 taken 147 times.
✗ Branch 1 not taken.
 147x return new state(i, ex);
48 ✗ }
49 catch (...)
50 {
51
0/2
✗ Branch 0 not taken.
✗ Branch 1 not taken.
 ✗ delete i;
52
0/2
✗ Branch 0 not taken.
✗ Branch 1 not taken.
 ✗ throw;
53 ✗ }
54 ✗ }
55
56 void
57 7761x tcp_server::release(state* s) noexcept
58 {
59
2/2
✓ Branch 0 taken 7614 times.
✓ Branch 1 taken 147 times.
 7761x if (s->refs.fetch_sub(1, std::memory_order_acq_rel) == 1)
60 {
61
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 147 times.
 147x delete s->impl_;
62
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 147 times.
 147x delete s;
63 147x }
64 7761x }
65
66 void
67 153x tcp_server::discard() noexcept
68 {
69
2/2
✓ Branch 0 taken 147 times.
✓ Branch 1 taken 6 times.
 153x if (!st_)
70 6x return;
71 147x auto& im = *st_->impl_;
72 147x std::vector<std::size_t> idle;
73 {
74
1/2
✓ Branch 0 taken 147 times.
✗ Branch 1 not taken.
 147x std::lock_guard lock(im.join_mutex);
75 147x im.discarded = true;
76
2/2
✓ Branch 0 taken 147 times.
✓ Branch 1 taken 143 times.
 290x for (std::size_t i = 0; i < im.ports.size(); ++i)
77
5/6
✓ Branch 0 taken 12 times.
✓ Branch 1 taken 131 times.
✓ Branch 2 taken 131 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 89 times.
✓ Branch 5 taken 54 times.
 232x if (i >= im.loop_running.size() || !im.loop_running[i])
78
1/2
✓ Branch 0 taken 89 times.
✗ Branch 1 not taken.
 89x idle.push_back(i);
79 147x }
80
2/2
✓ Branch 0 taken 89 times.
✓ Branch 1 taken 147 times.
 236x for (auto i : idle)
81 89x im.ports[i].close();
82
1/2
✓ Branch 0 taken 147 times.
✗ Branch 1 not taken.
 147x stop();
83 147x release(std::exchange(st_, nullptr));
84 153x }
85
86 153x tcp_server::~tcp_server()
87 2x {
88 151x discard();
89 153x }
90
91 4x tcp_server::tcp_server(tcp_server&& o) noexcept
92 4x : st_(std::exchange(o.st_, nullptr))
93 ✗ {
94 4x }
95
96 tcp_server&
97 2x tcp_server::operator=(tcp_server&& o) noexcept
98 {
99
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2x if (this != &o)
100 {
101 2x discard();
102 2x st_ = std::exchange(o.st_, nullptr);
103 2x }
104 2x return *this;
105 }
106
107 // Accept loop: wait for idle worker, accept connection, dispatch
108 capy::task<void>
109
11/26
✓ Branch 0 taken 135 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 135 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 135 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 135 times.
✓ Branch 8 taken 129 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
✓ Branch 12 taken 129 times.
✗ Branch 13 not taken.
✗ Branch 14 not taken.
✗ Branch 15 not taken.
✓ Branch 16 taken 129 times.
✓ Branch 17 taken 129 times.
✓ Branch 18 taken 129 times.
✗ Branch 19 not taken.
✓ Branch 20 taken 18782 times.
✓ Branch 21 taken 135 times.
✗ Branch 22 not taken.
✗ Branch 23 not taken.
✗ Branch 24 not taken.
✗ Branch 25 not taken.
 38233x tcp_server::do_accept(state_ref st, tcp_acceptor& acc)
110 135x {
111 // Analyzer can't trace value through coroutine await_transform
112 // NOLINTNEXTLINE(clang-analyzer-core.uninitialized.UndefReturn)
113
3/8
✓ Branch 0 taken 135 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 135 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 135 times.
 270x auto env = co_await capy::this_coro::environment;
114
2/2
✓ Branch 0 taken 129 times.
✓ Branch 1 taken 6529 times.
 6658x while (!env->stop_token.stop_requested())
115 {
116 // Wait for an idle worker before blocking on accept
117
9/12
✓ Branch 0 taken 6529 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 6529 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 478 times.
✓ Branch 5 taken 6051 times.
✓ Branch 6 taken 474 times.
✓ Branch 7 taken 4 times.
✓ Branch 8 taken 6525 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 13006 times.
✓ Branch 11 taken 19535 times.
 13058x auto* w = co_await pop_awaitable{*st};
118
2/2
✓ Branch 0 taken 19515 times.
✓ Branch 1 taken 20 times.
 19535x if (!w)
119 20x break;
120 // Not dropped if this frame is destroyed while accepting: that
121 // happens only when the listener's context is torn down, which
122 // the server cannot outlive.
123 19515x w->holds_.store(1, std::memory_order_relaxed);
124
12/16
✓ Branch 0 taken 6505 times.
✓ Branch 1 taken 13010 times.
✓ Branch 2 taken 19515 times.
✓ Branch 3 taken 13010 times.
✓ Branch 4 taken 19515 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 6505 times.
✓ Branch 7 taken 13010 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 6505 times.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
✓ Branch 12 taken 6505 times.
✓ Branch 13 taken 6505 times.
✓ Branch 14 taken 5780 times.
✓ Branch 15 taken 12285 times.
 26020x auto [ec] = co_await acc.accept(w->socket());
125
2/2
✓ Branch 0 taken 8672 times.
✓ Branch 1 taken 3613 times.
 12285x if (ec)
126 {
127
10/12
✓ Branch 0 taken 2892 times.
✓ Branch 1 taken 5780 times.
✓ Branch 2 taken 2892 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 2892 times.
✓ Branch 6 taken 2 times.
✓ Branch 7 taken 2 times.
✓ Branch 8 taken 2890 times.
✓ Branch 9 taken 2888 times.
✓ Branch 10 taken 2 times.
✓ Branch 11 taken 2890 times.
 11564x co_await hop_awaitable{*st};
128 2890x st->drop_hold(*w);
129 2890x continue;
130 }
131 3613x w->holds_.fetch_add(1, std::memory_order_relaxed);
132
1/2
✓ Branch 0 taken 3613 times.
✗ Branch 1 not taken.
 3613x w->run(launcher{*st, *w});
133 3613x st->drop_hold(*w);
134 25311x }
135 69507x }
136
137 std::error_code
138 147x tcp_server::bind(endpoint ep)
139 {
140 147x tcp_acceptor acc(st_->impl_->ctx);
141
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 143 times.
 147x if (auto ec = detail::open_listener(acc, ep, 128))
142 4x return ec;
143
1/2
✓ Branch 0 taken 143 times.
✗ Branch 1 not taken.
 143x st_->impl_->ports.push_back(std::move(acc));
144 143x return {};
145 147x }
146
147 endpoint
148 135x tcp_server::local_endpoint(std::size_t index) const noexcept
149 {
150
3/4
✓ Branch 0 taken 135 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 131 times.
 135x if (!st_ || index >= st_->impl_->ports.size())
151 4x return endpoint{};
152 131x return st_->impl_->ports[index].local_endpoint();
153 135x }
154
155 void
156 137x tcp_server::start()
157 {
158 // Idempotent - only start if not already running
159
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 135 times.
 137x if (st_->running)
160 2x return;
161
162 135x auto& im = *st_->impl_;
163 {
164 // Previous session must be fully stopped before restart
165 135x std::lock_guard lock(im.join_mutex);
166
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 133 times.
 135x if (st_->active_accepts != 0)
167
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2x detail::throw_logic_error(
168 "tcp_server::start: previous session not joined");
169 133x st_->active_accepts = im.ports.size();
170
1/2
✓ Branch 0 taken 133 times.
✗ Branch 1 not taken.
 133x im.loop_running.assign(im.ports.size(), true);
171 135x }
172 {
173 133x std::lock_guard<std::mutex> lock(st_->mutex);
174 133x st_->stopping = false;
175 133x }
176
177 133x st_->running = true;
178 133x im.stop = {}; // Fresh stop source
179 133x auto token = im.stop.get_token();
180
181 // The completion runs once the loop's frame is done with its
182 // listener, so a close from here cannot race the loop.
183
2/2
✓ Branch 0 taken 135 times.
✓ Branch 1 taken 133 times.
 268x for (std::size_t i = 0; i < im.ports.size(); ++i)
184
2/4
✓ Branch 0 taken 135 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 135 times.
 399x capy::run_async(st_->ex, token, [st = state_ref(st_), i]() {
185 129x auto& im = *st->impl_;
186 129x bool close = false;
187 {
188 129x std::lock_guard lock(im.join_mutex);
189 129x im.loop_running[i] = false;
190 129x close = im.discarded;
191
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 127 times.
 129x if (--st->active_accepts == 0)
192 127x im.join_cv.notify_all();
193 129x }
194
2/2
✓ Branch 0 taken 81 times.
✓ Branch 1 taken 48 times.
 129x if (close)
195 48x im.ports[i].close();
196
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 135 times.
 264x })(do_accept(state_ref(st_), im.ports[i]));
197 137x }
198
199 void
200 234x tcp_server::stop()
201 {
202 // Idempotent - only stop if running
203
3/4
✓ Branch 0 taken 234 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 101 times.
✓ Branch 3 taken 133 times.
 234x if (!st_ || !st_->running)
204 101x return;
205 133x st_->running = false;
206
207 // Stop accept loops, including those waiting for a worker
208 133x st_->impl_->stop.request_stop();
209 133x st_->stop_waiters();
210
211 // Launch cancellation coroutine on server executor
212
3/6
✓ Branch 0 taken 133 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 133 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 133 times.
✗ Branch 5 not taken.
 133x capy::run_async(st_->ex, std::stop_token{})(do_stop(state_ref(st_)));
213 234x }
214
215 void
216 65x tcp_server::join()
217 {
218 65x auto& im = *st_->impl_;
219 65x std::unique_lock lock(im.join_mutex);
220
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 130x im.join_cv.wait(lock, [this] { return st_->active_accepts == 0; });
221 65x }
222
223 capy::task<>
224
11/24
✓ Branch 0 taken 133 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 133 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 129 times.
✓ Branch 5 taken 4 times.
✓ Branch 6 taken 4 times.
✓ Branch 7 taken 129 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 129 times.
✗ Branch 11 not taken.
✗ Branch 12 not taken.
✗ Branch 13 not taken.
✓ Branch 14 taken 129 times.
✓ Branch 15 taken 129 times.
✓ Branch 16 taken 129 times.
✗ Branch 17 not taken.
✗ Branch 18 not taken.
✓ Branch 19 taken 133 times.
✗ Branch 20 not taken.
✗ Branch 21 not taken.
✗ Branch 22 not taken.
✗ Branch 23 not taken.
 657x tcp_server::do_stop(state_ref st)
225 133x {
226 // Requested outside the lock: a stop callback may return its worker,
227 // which takes the lock.
228 129x std::vector<std::stop_source> active;
229 {
230
1/2
✓ Branch 0 taken 129 times.
✗ Branch 1 not taken.
 129x std::lock_guard<std::mutex> lock(st->mutex);
231
2/2
✓ Branch 0 taken 31 times.
✓ Branch 1 taken 129 times.
 160x for (auto* w = st->active_head; w; w = w->next_)
232
1/2
✓ Branch 0 taken 31 times.
✗ Branch 1 not taken.
 31x active.push_back(w->stop_);
233 129x }
234
2/2
✓ Branch 0 taken 31 times.
✓ Branch 1 taken 129 times.
 160x for (auto& s : active)
235 31x s.request_stop();
236
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 129 times.
 262x co_return;
237 129x }
238
239 } // namespace boost::corosio
240