Index/src / corosio / src / tcp_server.cpp

src/corosio/src/tcp_server.cpp

96.0% Lines (95 / 99) 85.7% Branches (54 / 63) 100.0% Functions (18 / 18)
[ ] prev/next file n p prev/next uncovered line
tcp_server.cpp
f(x) Functions (18)
Function Calls Lines Branches Blocks
boost::corosio::tcp_server::worker_base::worker_base() :21 189x 100.0% – 100.0% boost::corosio::tcp_server::worker_base::~worker_base() :22 189x 0.0% – 100.0% boost::corosio::tcp_server::impl::impl(boost::capy::execution_context&) :37 59x 100.0% – 100.0% boost::corosio::tcp_server::make_state(boost::capy::execution_context&, boost::capy::any_executor const&) :41 59x 42.9% 33.3% 50.0% boost::corosio::tcp_server::release(boost::corosio::tcp_server::state*) :57 757x 100.0% 66.7% 100.0% boost::corosio::tcp_server::discard() :67 62x 100.0% 100.0% 100.0% boost::corosio::tcp_server::~tcp_server() :86 61x 100.0% – 100.0% boost::corosio::tcp_server::tcp_server(boost::corosio::tcp_server&&) :91 2x 100.0% – 100.0% boost::corosio::tcp_server::operator=(boost::corosio::tcp_server&&) :97 1x 100.0% 50.0% 100.0% boost::corosio::tcp_server::do_accept(boost::corosio::tcp_server::state_ref, boost::corosio::tcp_acceptor&) :109 52x 100.0% 100.0% 42.1% boost::corosio::tcp_server::bind(boost::corosio::endpoint) :138 58x 100.0% 100.0% 85.7% boost::corosio::tcp_server::local_endpoint(unsigned long long) const :148 53x 100.0% 83.3% 100.0% boost::corosio::tcp_server::start() :156 53x 100.0% 100.0% 60.6% boost::corosio::tcp_server::start()::{lambda()#1}::operator()() const :184 49x 100.0% 80.0% 81.2% boost::corosio::tcp_server::stop() :200 87x 100.0% 83.3% 57.6% boost::corosio::tcp_server::join() :216 20x 100.0% 100.0% 71.4% boost::corosio::tcp_server::join()::{lambda()#1}::operator()() const :220 20x 100.0% 100.0% 100.0% boost::corosio::tcp_server::do_stop(boost::corosio::tcp_server::state_ref) :224 51x 100.0% 100.0% 42.1%
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 189x tcp_server::worker_base::worker_base() = default;
22 189x 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 bool discarded = false;
36
37 59x explicit impl(capy::execution_context& c) noexcept : ctx(c) {}
38 };
39
40 tcp_server::state*
41 59x tcp_server::make_state(
42 capy::execution_context& ctx, capy::any_executor const& ex)
43 {
44 59x auto* i = new impl(ctx);
45 try
46 {
47
1/1
✓ Branch 4 → 5 taken 59 times.
 59x return new state(i, ex);
48 }
49 ✗ catch (...)
50 {
51
0/2
✗ Branch 10 → 11 not taken.
✗ Branch 10 → 13 not taken.
 ✗ delete i;
52 ✗ throw;
53 ✗ }
54 }
55
56 void
57 757x tcp_server::release(state* s) noexcept
58 {
59
2/2
✓ Branch 4 → 5 taken 59 times.
✓ Branch 4 → 11 taken 698 times.
 1514x if (s->refs.fetch_sub(1, std::memory_order_acq_rel) == 1)
60 {
61
1/2
✓ Branch 5 → 6 taken 59 times.
✗ Branch 5 → 8 not taken.
 59x delete s->impl_;
62
1/2
✓ Branch 8 → 9 taken 59 times.
✗ Branch 8 → 11 not taken.
 59x delete s;
63 }
64 757x }
65
66 void
67 62x tcp_server::discard() noexcept
68 {
69
2/2
✓ Branch 2 → 3 taken 3 times.
✓ Branch 2 → 4 taken 59 times.
 62x if (!st_)
70 3x return;
71 59x auto& im = *st_->impl_;
72 59x std::vector<std::size_t> idle;
73 {
74 59x std::lock_guard lock(im.join_mutex);
75 59x im.discarded = true;
76
2/2
✓ Branch 17 → 6 taken 56 times.
✓ Branch 17 → 18 taken 59 times.
 115x for (std::size_t i = 0; i < im.ports.size(); ++i)
77
6/6
✓ Branch 7 → 8 taken 50 times.
✓ Branch 7 → 11 taken 6 times.
✓ Branch 10 → 11 taken 23 times.
✓ Branch 10 → 12 taken 27 times.
✓ Branch 13 → 14 taken 29 times.
✓ Branch 13 → 15 taken 27 times.
 56x if (i >= im.loop_running.size() || !im.loop_running[i])
78 29x idle.push_back(i);
79 59x }
80
2/2
✓ Branch 26 → 21 taken 29 times.
✓ Branch 26 → 27 taken 59 times.
 88x for (auto i : idle)
81 29x im.ports[i].close();
82 59x stop();
83 59x release(std::exchange(st_, nullptr));
84 59x }
85
86 61x tcp_server::~tcp_server()
87 {
88 61x discard();
89 61x }
90
91 2x tcp_server::tcp_server(tcp_server&& o) noexcept
92 2x : st_(std::exchange(o.st_, nullptr))
93 {
94 2x }
95
96 tcp_server&
97 1x tcp_server::operator=(tcp_server&& o) noexcept
98 {
99
1/2
✓ Branch 2 → 3 taken 1 time.
✗ Branch 2 → 6 not taken.
 1x if (this != &o)
100 {
101 1x discard();
102 1x st_ = std::exchange(o.st_, nullptr);
103 }
104 1x return *this;
105 }
106
107 // Accept loop: wait for idle worker, accept connection, dispatch
108 capy::task<void>
109
1/1
✓ Branch 6 → 7 taken 52 times.
 52x tcp_server::do_accept(state_ref st, tcp_acceptor& acc)
110 {
111 // Analyzer can't trace value through coroutine await_transform
112 // NOLINTNEXTLINE(clang-analyzer-core.uninitialized.UndefReturn)
113 auto env = co_await capy::this_coro::environment;
114 while (!env->stop_token.stop_requested())
115 {
116 // Wait for an idle worker before blocking on accept
117 auto* w = co_await pop_awaitable{*st};
118 if (!w)
119 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 w->holds_.store(1, std::memory_order_relaxed);
124 auto [ec] = co_await acc.accept(w->socket());
125 if (ec)
126 {
127 co_await hop_awaitable{*st};
128 st->drop_hold(*w);
129 continue;
130 }
131 w->holds_.fetch_add(1, std::memory_order_relaxed);
132 w->run(launcher{*st, *w});
133 st->drop_hold(*w);
134 }
135 104x }
136
137 std::error_code
138 58x tcp_server::bind(endpoint ep)
139 {
140
1/1
✓ Branch 2 → 3 taken 58 times.
 58x tcp_acceptor acc(st_->impl_->ctx);
141
2/2
✓ Branch 5 → 6 taken 2 times.
✓ Branch 5 → 7 taken 56 times.
 58x if (auto ec = detail::open_listener(acc, ep, 128))
142 2x return ec;
143
1/1
✓ Branch 8 → 9 taken 56 times.
 56x st_->impl_->ports.push_back(std::move(acc));
144 56x return {};
145 58x }
146
147 endpoint
148 53x tcp_server::local_endpoint(std::size_t index) const noexcept
149 {
150
5/6
✓ Branch 2 → 3 taken 53 times.
✗ Branch 2 → 5 not taken.
✓ Branch 4 → 5 taken 3 times.
✓ Branch 4 → 6 taken 50 times.
✓ Branch 7 → 8 taken 3 times.
✓ Branch 7 → 9 taken 50 times.
 53x if (!st_ || index >= st_->impl_->ports.size())
151 3x return endpoint{};
152 50x return st_->impl_->ports[index].local_endpoint();
153 }
154
155 void
156 53x tcp_server::start()
157 {
158 // Idempotent - only start if not already running
159
2/2
✓ Branch 2 → 3 taken 1 time.
✓ Branch 2 → 4 taken 52 times.
 53x if (st_->running)
160 1x return;
161
162 52x auto& im = *st_->impl_;
163 {
164 // Previous session must be fully stopped before restart
165
1/1
✓ Branch 4 → 5 taken 52 times.
 52x std::lock_guard lock(im.join_mutex);
166
2/2
✓ Branch 5 → 6 taken 1 time.
✓ Branch 5 → 7 taken 51 times.
 52x if (st_->active_accepts != 0)
167 1x detail::throw_logic_error(
168 "tcp_server::start: previous session not joined");
169 51x st_->active_accepts = im.ports.size();
170
1/1
✓ Branch 9 → 10 taken 51 times.
 51x im.loop_running.assign(im.ports.size(), true);
171 52x }
172 {
173
1/1
✓ Branch 11 → 12 taken 51 times.
 51x std::lock_guard<std::mutex> lock(st_->mutex);
174 51x st_->stopping = false;
175 51x }
176
177 51x st_->running = true;
178
1/1
✓ Branch 13 → 14 taken 51 times.
 51x im.stop = {}; // Fresh stop source
179 51x 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 34 → 18 taken 52 times.
✓ Branch 34 → 35 taken 51 times.
 103x for (std::size_t i = 0; i < im.ports.size(); ++i)
184
0/2
✗ Branch 60 → 61 not taken.
✗ Branch 60 → 62 not taken.
 104x capy::run_async(st_->ex, token, [st = state_ref(st_), i]() {
185 49x auto& im = *st->impl_;
186 49x bool close = false;
187 {
188
1/1
✓ Branch 3 → 4 taken 49 times.
 49x std::lock_guard lock(im.join_mutex);
189
1/1
✓ Branch 4 → 5 taken 49 times.
 49x im.loop_running[i] = false;
190 49x close = im.discarded;
191
2/2
✓ Branch 7 → 8 taken 48 times.
✓ Branch 7 → 9 taken 1 time.
 49x if (--st->active_accepts == 0)
192 48x im.join_cv.notify_all();
193 49x }
194
2/2
✓ Branch 10 → 11 taken 24 times.
✓ Branch 10 → 13 taken 25 times.
 49x if (close)
195 24x im.ports[i].close();
196
2/2
✓ Branch 24 → 25 taken 52 times.
✓ Branch 25 → 26 taken 52 times.
 101x })(do_accept(state_ref(st_), im.ports[i]));
197 51x }
198
199 void
200 87x tcp_server::stop()
201 {
202 // Idempotent - only stop if running
203
3/4
✓ Branch 2 → 3 taken 87 times.
✗ Branch 2 → 4 not taken.
✓ Branch 3 → 4 taken 36 times.
✓ Branch 3 → 5 taken 51 times.
 87x if (!st_ || !st_->running)
204 36x return;
205 51x st_->running = false;
206
207 // Stop accept loops, including those waiting for a worker
208 51x st_->impl_->stop.request_stop();
209 51x st_->stop_waiters();
210
211 // Launch cancellation coroutine on server executor
212
2/2
✓ Branch 11 → 12 taken 51 times.
✓ Branch 12 → 13 taken 51 times.
 51x capy::run_async(st_->ex, std::stop_token{})(do_stop(state_ref(st_)));
213 }
214
215 void
216 20x tcp_server::join()
217 {
218 20x auto& im = *st_->impl_;
219
1/1
✓ Branch 2 → 3 taken 20 times.
 20x std::unique_lock lock(im.join_mutex);
220
1/1
✓ Branch 3 → 4 taken 20 times.
 40x im.join_cv.wait(lock, [this] { return st_->active_accepts == 0; });
221 20x }
222
223 capy::task<>
224
1/1
✓ Branch 6 → 7 taken 51 times.
 51x tcp_server::do_stop(state_ref st)
225 {
226 // Requested outside the lock: a stop callback may return its worker,
227 // which takes the lock.
228 std::vector<std::stop_source> active;
229 {
230 std::lock_guard<std::mutex> lock(st->mutex);
231 for (auto* w = st->active_head; w; w = w->next_)
232 active.push_back(w->stop_);
233 }
234 for (auto& s : active)
235 s.request_stop();
236 co_return;
237 102x }
238
239 } // namespace boost::corosio
240