TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Michael Vandeberg
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 : #ifndef BOOST_COROSIO_TCP_SERVER_HPP
12 : #define BOOST_COROSIO_TCP_SERVER_HPP
13 :
14 : #include <boost/corosio/detail/config.hpp>
15 : #include <boost/corosio/detail/except.hpp>
16 : #include <boost/corosio/tcp_acceptor.hpp>
17 : #include <boost/corosio/tcp_socket.hpp>
18 : #include <boost/corosio/io_context.hpp>
19 : #include <boost/corosio/endpoint.hpp>
20 : #include <boost/capy/task.hpp>
21 : #include <boost/capy/concept/execution_context.hpp>
22 : #include <boost/capy/concept/context_awaitable.hpp>
23 : #include <boost/capy/concept/executor.hpp>
24 : #include <boost/capy/ex/any_executor.hpp>
25 : #include <boost/capy/detail/abandonable.hpp>
26 : #include <boost/capy/detail/await_suspend_helper.hpp>
27 : #include <boost/capy/ex/abandon.hpp>
28 : #include <boost/capy/ex/frame_alloc_mixin.hpp>
29 : #include <boost/capy/ex/frame_allocator.hpp>
30 : #include <boost/capy/ex/await_context.hpp>
31 : #include <boost/capy/ex/recycling_memory_resource.hpp>
32 : #include <boost/capy/ex/run_async.hpp>
33 : #include <boost/capy/ex/work_guard.hpp>
34 :
35 : #include <atomic>
36 : #include <coroutine>
37 : #include <memory>
38 : #include <memory_resource>
39 : #include <mutex>
40 : #include <ranges>
41 : #include <utility>
42 : #include <vector>
43 :
44 : namespace boost::corosio {
45 :
46 : #ifdef _MSC_VER
47 : #pragma warning(push)
48 : #pragma warning(disable : 4251) // class needs to have dll-interface
49 : #endif
50 :
51 : /** Manages a pool of reusable workers that handle incoming TCP connections.
52 :
53 : This class manages a pool of reusable worker objects that handle
54 : incoming connections. When a connection arrives, an idle worker
55 : is dispatched to handle it. After the connection completes, the
56 : worker returns to the pool for reuse, avoiding allocation overhead
57 : per connection.
58 :
59 : Workers are set via @ref set_workers as a forward range of
60 : pointer-like objects (e.g., `unique_ptr<worker_base>`). The server
61 : takes ownership of the container via type erasure.
62 :
63 : @par Thread Safety
64 : Distinct objects: Safe.
65 : Shared objects: Unsafe.
66 :
67 : @par Lifecycle
68 : The server operates in three states:
69 :
70 : - **Stopped**: Initial state, or after @ref join completes.
71 : - **Running**: After @ref start, actively accepting connections.
72 : - **Stopping**: After @ref stop, draining active work.
73 :
74 : State transitions:
75 : @code
76 : [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped]
77 : @endcode
78 :
79 : @par Running the Server
80 : @par !example running_the_server
81 :
82 : @par Graceful Shutdown
83 : To shut down gracefully, call @ref stop then drain the `io_context`:
84 : @par !example graceful_shutdown
85 :
86 : @par Restart After Stop
87 : The server can be restarted after a complete shutdown cycle.
88 : You must drain the `io_context`, call @ref join, and restart the
89 : `io_context` itself (`ioc.restart()`) before restarting:
90 : @par !example restart_after_stop
91 :
92 : @par WARNING: What NOT to Do
93 : - Do NOT call @ref join from inside a worker coroutine (deadlock).
94 : - Do NOT call @ref join from a thread running `ioc.run()` (deadlock).
95 : - Do NOT call @ref start without completing @ref join after @ref stop.
96 : - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead.
97 :
98 : @par Example
99 : @par !example custom_worker
100 :
101 : @see worker_base, set_workers, launcher
102 : */
103 : class BOOST_COROSIO_DECL tcp_server
104 : {
105 : public:
106 : class worker_base; ///< Abstract base for connection handlers.
107 : class launcher; ///< Move-only handle to launch worker coroutines.
108 :
109 : private:
110 : struct waiter
111 : {
112 : waiter* next;
113 : capy::continuation cont;
114 : worker_base* w;
115 : };
116 :
117 : struct impl;
118 :
119 : /** Everything the server's coroutines reach.
120 :
121 : Accept loops, launchers and the coroutines that start
122 : connections each hold a reference, so moving or destroying the
123 : server never strands them. The lists are guarded by `mutex`,
124 : because workers return from whichever thread their coroutine
125 : finished on.
126 : */
127 : struct state
128 : {
129 : std::atomic<std::size_t> refs{1};
130 : impl* impl_;
131 : capy::any_executor ex;
132 : std::mutex mutex;
133 : waiter* waiters = nullptr;
134 : worker_base* idle_head = nullptr; // forward list
135 : worker_base* active_head = nullptr; // doubly linked
136 : worker_base* active_tail = nullptr;
137 : std::size_t active_accepts = 0; // guarded by the join mutex
138 : std::size_t lent = 0; // borrowed workers, guarded by mutex
139 : std::shared_ptr<void> storage; // owns the worker container
140 : bool running = false;
141 : // Set by stop(); accept loops waiting for a worker give up.
142 : bool stopping = false; // guarded by mutex
143 :
144 HIT 169 : state(impl* i, capy::any_executor const& e) noexcept : impl_(i), ex(e)
145 : {
146 169 : }
147 :
148 : // The list operations below require `mutex`.
149 :
150 1198 : void idle_push(worker_base* w) noexcept
151 : {
152 1198 : w->next_ = idle_head;
153 1198 : idle_head = w;
154 1198 : }
155 :
156 1158 : worker_base* idle_pop() noexcept
157 : {
158 1158 : auto* w = idle_head;
159 1158 : if (w)
160 : {
161 787 : idle_head = w->next_;
162 787 : ++lent;
163 : }
164 1158 : return w;
165 : }
166 :
167 1000 : void active_push(worker_base* w) noexcept
168 : {
169 1000 : w->next_ = nullptr;
170 1000 : w->prev_ = active_tail;
171 1000 : if (active_tail)
172 703 : active_tail->next_ = w;
173 : else
174 297 : active_head = w;
175 1000 : active_tail = w;
176 1000 : }
177 :
178 1130 : void active_remove(worker_base* w) noexcept
179 : {
180 : // Skip if not in active list (e.g., after failed accept)
181 1130 : if (w != active_head && w->prev_ == nullptr)
182 130 : return;
183 1000 : if (w->prev_)
184 268 : w->prev_->next_ = w->next_;
185 : else
186 732 : active_head = w->next_;
187 1000 : if (w->next_)
188 608 : w->next_->prev_ = w->prev_;
189 : else
190 392 : active_tail = w->prev_;
191 1000 : w->prev_ = nullptr; // Mark as not in active list
192 : }
193 :
194 : /// Return @p w to the pool, handing it to a waiting accept loop
195 : /// if there is one.
196 1130 : void push(worker_base& w) noexcept
197 : {
198 1130 : waiter* wake = nullptr;
199 : {
200 1130 : std::lock_guard<std::mutex> lock(mutex);
201 1130 : active_remove(&w);
202 1130 : if (waiters)
203 : {
204 345 : wake = waiters;
205 345 : waiters = wake->next;
206 345 : wake->w = &w;
207 : }
208 : else
209 : {
210 785 : idle_push(&w);
211 785 : --lent;
212 : }
213 1130 : }
214 1130 : if (wake)
215 345 : ex.post(wake->cont);
216 1130 : }
217 :
218 : /// Drop one hold on @p w, returning it to the pool with the last.
219 2144 : void drop_hold(worker_base& w) noexcept
220 : {
221 4288 : if (w.holds_.fetch_sub(1, std::memory_order_acq_rel) == 1)
222 1130 : push(w);
223 2144 : }
224 :
225 : /// Mark the pool as stopping and release every accept loop
226 : /// waiting for a worker, with none.
227 157 : void stop_waiters() noexcept
228 : {
229 157 : waiter* list = nullptr;
230 : {
231 157 : std::lock_guard<std::mutex> lock(mutex);
232 157 : stopping = true;
233 157 : list = std::exchange(waiters, nullptr);
234 157 : }
235 183 : while (list)
236 : {
237 26 : auto* next = list->next;
238 26 : ex.post(list->cont);
239 26 : list = next;
240 : }
241 157 : }
242 : };
243 :
244 : /// Add a reference to @p s.
245 2489 : static void add_ref(state* s) noexcept
246 : {
247 2489 : s->refs.fetch_add(1, std::memory_order_relaxed);
248 2489 : }
249 :
250 : /// Drop a reference to @p s, freeing it with the last one.
251 : static void release(state* s) noexcept;
252 :
253 : /// An owned reference to the server's state.
254 : class state_ref
255 : {
256 : state* p_ = nullptr;
257 :
258 : public:
259 1475 : explicit state_ref(state* p) noexcept : p_(p)
260 : {
261 1475 : add_ref(p_);
262 1475 : }
263 :
264 : state_ref(state_ref const& o) noexcept : p_(o.p_)
265 : {
266 : if (p_)
267 : add_ref(p_);
268 : }
269 :
270 3944 : state_ref(state_ref&& o) noexcept : p_(std::exchange(o.p_, nullptr)) {}
271 :
272 : state_ref& operator=(state_ref const&) = delete;
273 : state_ref& operator=(state_ref&&) = delete;
274 :
275 5419 : ~state_ref()
276 : {
277 5419 : if (p_)
278 1475 : release(p_);
279 5419 : }
280 :
281 2742 : state* operator->() const noexcept
282 : {
283 2742 : return p_;
284 : }
285 :
286 3276 : state& operator*() const noexcept
287 : {
288 3276 : return *p_;
289 : }
290 : };
291 :
292 : static state*
293 : make_state(capy::execution_context& ctx, capy::any_executor const& ex);
294 :
295 : state* st_;
296 :
297 : template<capy::Executor Ex>
298 : struct launch_wrapper
299 : {
300 : // frame_alloc_mixin routes the frame through the thread-local
301 : // recycling allocator, so a warmed per-connection launch does
302 : // not hit the global allocator.
303 : struct promise_type : capy::frame_alloc_mixin
304 : {
305 : Ex ex; // Executor stored directly in frame (outlives child tasks)
306 : // Owns the connection's chain: abandoning it resumes, or has
307 : // the executor destroy, this frame to unwind the task.
308 : capy::chain_owner owner_;
309 : capy::await_context env_;
310 : /// Embedded post node: the start posts this instead of the
311 : /// bare handle, which would heap-allocate a wrapper op.
312 : capy::continuation cont_;
313 :
314 : // For regular coroutines: first arg is executor, second is stop token
315 : template<class E, class S, class... Args>
316 : requires capy::Executor<std::decay_t<E>>
317 : promise_type(E e, S s, Args&&...)
318 : : ex(std::move(e))
319 : , env_{
320 : capy::executor_ref(ex), std::move(s),
321 : capy::get_current_frame_allocator(), &owner_}
322 : {
323 : }
324 :
325 : // For lambda coroutines: first arg is closure, second is executor, third is stop token
326 : template<class Closure, class E, class S, class... Args>
327 : requires(!capy::Executor<std::decay_t<Closure>> &&
328 : capy::Executor<std::decay_t<E>>)
329 1000 : promise_type(Closure&&, E e, S s, Args&&...)
330 1000 : : ex(std::move(e))
331 1000 : , env_{
332 1000 : capy::executor_ref(ex), std::move(s),
333 2000 : capy::get_current_frame_allocator(), &owner_}
334 : {
335 1000 : }
336 :
337 1000 : launch_wrapper get_return_object() noexcept
338 : {
339 : return {
340 1000 : std::coroutine_handle<promise_type>::from_promise(*this)};
341 : }
342 1000 : std::suspend_always initial_suspend() noexcept
343 : {
344 1000 : return {};
345 : }
346 983 : std::suspend_never final_suspend() noexcept
347 : {
348 983 : return {};
349 : }
350 983 : void return_void() noexcept {}
351 MIS 0 : void unhandled_exception()
352 : {
353 : // LCOV_EXCL_START: terminating by contract is not a
354 : // coverable outcome.
355 : std::terminate();
356 : // LCOV_EXCL_STOP
357 : }
358 :
359 : // Inject await_context for ContextAwaitable
360 : template<capy::ContextAwaitable Awaitable>
361 HIT 1978 : auto await_transform(Awaitable&& a)
362 : {
363 : using AwaitableT = std::decay_t<Awaitable>;
364 : struct adapter
365 : {
366 : AwaitableT aw;
367 : capy::await_context const* env;
368 :
369 1978 : bool await_ready()
370 : {
371 1978 : return aw.await_ready();
372 : }
373 1969 : decltype(auto) await_resume()
374 : {
375 1969 : return aw.await_resume();
376 : }
377 :
378 : // Where the returned handle would be stored in a frame
379 : // another thread may already have resumed and freed,
380 : // it is resumed here instead, as a task does.
381 1978 : auto await_suspend(std::coroutine_handle<promise_type> h)
382 : {
383 1978 : return capy::detail::symmetric_transfer(
384 1978 : capy::detail::call_await_suspend(&aw, h, env));
385 : }
386 : };
387 1978 : return adapter{std::forward<Awaitable>(a), &env_};
388 : }
389 : };
390 :
391 : std::coroutine_handle<promise_type> h;
392 :
393 1000 : launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
394 1000 : : h(handle)
395 : {
396 1000 : }
397 :
398 1000 : ~launch_wrapper()
399 : {
400 : // An unstarted root owns no chain yet: destroying it is
401 : // the whole teardown.
402 1000 : if (h)
403 4 : h.destroy();
404 1000 : }
405 :
406 : launch_wrapper(launch_wrapper&& o) noexcept
407 : : h(std::exchange(o.h, nullptr))
408 : {
409 : }
410 :
411 : launch_wrapper(launch_wrapper const&) = delete;
412 : launch_wrapper& operator=(launch_wrapper const&) = delete;
413 : launch_wrapper& operator=(launch_wrapper&&) = delete;
414 : };
415 :
416 : /// One hold on a worker, dropped when this is destroyed.
417 : class worker_hold
418 : {
419 : state_ref st_;
420 : worker_base* w_;
421 :
422 : public:
423 1000 : worker_hold(state& st, worker_base& w) noexcept : st_(&st), w_(&w) {}
424 :
425 2992 : worker_hold(worker_hold&& o) noexcept
426 2992 : : st_(std::move(o.st_))
427 2992 : , w_(std::exchange(o.w_, nullptr))
428 : {
429 2992 : }
430 :
431 : worker_hold(worker_hold const&) = delete;
432 : worker_hold& operator=(worker_hold const&) = delete;
433 : worker_hold& operator=(worker_hold&&) = delete;
434 :
435 3992 : ~worker_hold()
436 : {
437 3992 : if (w_)
438 1000 : st_->drop_hold(*w_);
439 3992 : }
440 :
441 986 : state& server() const noexcept
442 : {
443 986 : return *st_;
444 : }
445 : };
446 :
447 : // The holds travel in the frame so that they are also released
448 : // when the frame is destroyed before it starts or finishes. One
449 : // parameter, not three: compilers disagree on the order parameter
450 : // copies are destroyed, but members always go in reverse, so the
451 : // task, whose parameters may still use the worker, goes first.
452 : template<class Executor>
453 : struct session
454 : {
455 : worker_hold hold;
456 : // Keeps the session's context running until the connection ends.
457 : capy::work_guard<Executor> work;
458 : capy::task<void> t;
459 : };
460 :
461 : // Named functor to avoid incomplete lambda type in coroutine promise
462 : template<class Executor>
463 : struct launch_coro
464 : {
465 : launch_wrapper<Executor>
466 1000 : operator()(Executor, std::stop_token, session<Executor> in)
467 : {
468 : // Executor and stop token stored in promise via constructor.
469 : // A local, so a frame destroyed unfinished destroys the task
470 : // while the promise and its environment still exist.
471 : auto s = std::move(in);
472 : // Resumed with the task unfinished: the chain was abandoned,
473 : // and leaving destroys the task, which unwinds it.
474 : if (!co_await capy::detail::abandonable<capy::task<void>>{s.t})
475 : co_return;
476 : // The connection's work ends with its task: the hop below is
477 : // the server's, and its context may no longer be running.
478 : s.work.reset();
479 : // The hold returns the worker from the server's executor.
480 : co_await hop_awaitable{s.hold.server(), true};
481 2000 : }
482 : };
483 :
484 : /// Resume on the server's executor.
485 : class hop_awaitable
486 : {
487 : state& st_;
488 : bool root_;
489 : capy::continuation cont_;
490 :
491 : public:
492 : // `root` is true when the awaiting coroutine is its chain's root.
493 1104 : explicit hop_awaitable(state& st, bool root = false) noexcept
494 1104 : : st_(st)
495 1104 : , root_(root)
496 : {
497 1104 : }
498 :
499 1104 : bool await_ready() const noexcept
500 : {
501 1104 : return false;
502 : }
503 :
504 : std::coroutine_handle<>
505 1104 : await_suspend(
506 : std::coroutine_handle<> h, capy::await_context const* env) noexcept
507 : {
508 1104 : cont_.h = h;
509 : // A root has no owner: the launcher's would unwind it on the
510 : // session's executor, which may already be gone. Given up
511 : // here, it is destroyed in place.
512 1104 : if (!root_)
513 118 : cont_.owner = env->owner;
514 1104 : return st_.ex.dispatch(cont_);
515 : }
516 :
517 1099 : void await_resume() noexcept {}
518 : };
519 :
520 : class pop_awaitable
521 : {
522 : state& st_;
523 : waiter wait_;
524 :
525 : public:
526 1158 : pop_awaitable(state& st) noexcept : st_(st), wait_{} {}
527 :
528 1158 : bool await_ready() const noexcept
529 : {
530 1158 : return false;
531 : }
532 :
533 : bool
534 1158 : await_suspend(std::coroutine_handle<> h, capy::await_context const* env) noexcept
535 : {
536 1158 : std::lock_guard<std::mutex> lock(st_.mutex);
537 1158 : if (st_.stopping)
538 : {
539 MIS 0 : wait_.w = nullptr;
540 0 : return false;
541 : }
542 HIT 1158 : if (auto* w = st_.idle_pop())
543 : {
544 787 : wait_.w = w;
545 787 : return false;
546 : }
547 371 : wait_.cont.h = h;
548 371 : wait_.cont.owner = env->owner;
549 371 : wait_.w = nullptr;
550 371 : wait_.next = st_.waiters;
551 371 : st_.waiters = &wait_;
552 371 : return true;
553 1158 : }
554 :
555 : /// Return the worker, or null once the server is stopping.
556 1154 : worker_base* await_resume() noexcept
557 : {
558 : // Set by await_suspend, or by the push that woke us.
559 1154 : return wait_.w;
560 : }
561 : };
562 :
563 : static capy::task<void> do_accept(state_ref st, tcp_acceptor& acc);
564 : static capy::task<> do_stop(state_ref st);
565 :
566 : public:
567 : /** Handles one accepted connection using a socket the derived class owns.
568 :
569 : Derive from this class to implement custom connection handling.
570 : Each worker owns a socket and is reused across multiple
571 : connections to avoid per-connection allocation.
572 :
573 : @par Thread Safety
574 : run() and socket() execute on the server's executor.
575 :
576 : @see tcp_server, launcher
577 : */
578 : class BOOST_COROSIO_DECL worker_base
579 : {
580 : // Ordered largest to smallest for optimal packing
581 : std::stop_source stop_; // ~16 bytes
582 : worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists
583 : worker_base* prev_ = nullptr; // 8 bytes - used only by active list
584 : // One for the accept loop until run() returns, one for the
585 : // launcher and its connection; the last one returns the worker.
586 : std::atomic<unsigned> holds_{0};
587 :
588 : friend class tcp_server;
589 :
590 : public:
591 : /// Construct a worker.
592 : worker_base();
593 :
594 : /// Destroy the worker.
595 : virtual ~worker_base();
596 :
597 : /** Handle an accepted connection.
598 :
599 : Called when this worker is dispatched to handle a new
600 : connection. The implementation must invoke the launcher
601 : exactly once to start the handling coroutine.
602 :
603 : @par Exception Safety
604 : An exception escaping this function ends the process. A
605 : worker that cannot start the connection should handle the
606 : failure here; destroying the launcher without invoking it
607 : returns the worker to the pool.
608 :
609 : The worker is not given another connection before this
610 : returns, even if the connection it started has finished.
611 :
612 : @param launch Handle to start the connection coroutine.
613 : */
614 : virtual void run(launcher launch) = 0;
615 :
616 : /// Return the socket used for connections.
617 : virtual corosio::tcp_socket& socket() = 0;
618 : };
619 :
620 : /** Starts a worker's connection-handling coroutine and returns the
621 : worker to the idle pool automatically.
622 :
623 : Passed to @ref worker_base::run to start the connection-handling
624 : coroutine. The launcher ensures the worker returns to the idle
625 : pool when the coroutine completes or if starting fails.
626 :
627 : The launcher must be invoked exactly once via `operator()`.
628 : If destroyed without invoking, the worker is returned to the
629 : idle pool automatically. So is the worker of a coroutine
630 : abandoned before it completes, as when the execution context
631 : holding it is destroyed.
632 :
633 : @see worker_base::run
634 : */
635 : class BOOST_COROSIO_DECL launcher
636 : {
637 : state* st_;
638 : worker_base* w_;
639 :
640 : friend class tcp_server;
641 :
642 1014 : launcher(state& st, worker_base& w) noexcept : st_(&st), w_(&w)
643 : {
644 1014 : add_ref(st_);
645 1014 : }
646 :
647 : public:
648 : /// Return the worker to the pool if not started.
649 1026 : ~launcher()
650 : {
651 1026 : if (w_)
652 14 : st_->drop_hold(*w_);
653 1026 : if (st_)
654 1014 : release(st_);
655 1026 : }
656 :
657 : /** Move construct, transferring the borrowed worker.
658 :
659 : @param o The launcher to take the worker from. It is left
660 : holding none, so only one of the two returns it.
661 : */
662 12 : launcher(launcher&& o) noexcept
663 12 : : st_(std::exchange(o.st_, nullptr))
664 12 : , w_(std::exchange(o.w_, nullptr))
665 : {
666 12 : }
667 : /// Copy construction is disabled; a launcher holds a borrowed worker it must return exactly once.
668 : launcher(launcher const&) = delete;
669 : /// Copy assignment is disabled; a launcher holds a borrowed worker it must return exactly once.
670 : launcher& operator=(launcher const&) = delete;
671 : /// Move assignment is disabled; a launcher is moved, never reassigned.
672 : launcher& operator=(launcher&&) = delete;
673 :
674 : /** Start the connection-handling coroutine.
675 :
676 : Starts the given coroutine on the specified executor, which
677 : counts it as work until it ends. That execution context's
678 : `run()` therefore does not return while a connection is
679 : active. When the coroutine completes, or is abandoned before
680 : completing, the worker is automatically returned to the idle
681 : pool.
682 :
683 : @pre When @p ex belongs to another execution context, the
684 : server's context is not destroyed until the coroutine has
685 : finished. The coroutine holds the worker, whose socket
686 : belongs to the server's context. The coroutine's own
687 : context may be destroyed first, but only while the
688 : coroutine is not suspended in an operation of the server's
689 : context. An operation on the worker's socket is one such
690 : operation. Destroying
691 : that context then abandons the coroutine and returns its
692 : worker. Otherwise stop the server and run both contexts
693 : until their `run()` returns first.
694 :
695 : @pre @p task was created on a chain whose frame allocator
696 : outlives the connection. The task's frame is allocated
697 : where it is created, and it lives until the connection
698 : ends. Inside @ref worker_base::run that allocator is the
699 : accept loop's, the frame allocator of the server's
700 : execution context. By default that is the process-wide
701 : recycling resource, so a task created there meets this.
702 :
703 : @tparam Executor Executor type satisfying capy::Executor.
704 :
705 : @param ex The executor to run the coroutine on.
706 : @param task The coroutine to execute.
707 :
708 : @throws std::logic_error If this launcher was already invoked.
709 : */
710 : template<class Executor>
711 1002 : void operator()(Executor const& ex, capy::task<void> task)
712 : {
713 1002 : if (!w_)
714 2 : detail::throw_logic_error(); // launcher already invoked
715 :
716 1000 : auto* w = std::exchange(w_, nullptr);
717 :
718 : // A stop_source allocates shared state on construction;
719 : // reuse the worker's across connections and replace it
720 : // only once a stop has actually been delivered (the state
721 : // is then latched for good). Replaced before the worker is
722 : // listed active, under the lock stop() copies it under, so
723 : // that copy never races the replacement or misses it.
724 1000 : std::stop_token st;
725 1000 : bool stopped_before = false;
726 : {
727 1000 : std::lock_guard<std::mutex> lock(st_->mutex);
728 1000 : if (w->stop_.stop_requested())
729 MIS 0 : w->stop_ = {};
730 HIT 1000 : st = w->stop_.get_token();
731 1000 : st_->active_push(w);
732 1000 : stopped_before = st_->stopping;
733 1000 : }
734 : // A stop that came before this launch never saw the worker
735 : // listed. The token is not handed out yet, so no callback
736 : // runs here.
737 1000 : if (stopped_before)
738 10 : w->stop_.request_stop();
739 :
740 1000 : worker_hold hold(*st_, *w);
741 :
742 : // The connection can outlive whatever chain invoked this
743 : // launcher, so its frame comes from the process-wide
744 : // recycling resource rather than that chain's allocator.
745 : struct allocator_scope
746 : {
747 : std::pmr::memory_resource* prev =
748 : capy::get_current_frame_allocator();
749 1000 : allocator_scope() noexcept
750 1000 : {
751 1000 : capy::set_current_frame_allocator(
752 : capy::get_recycling_memory_resource());
753 1000 : }
754 1000 : ~allocator_scope()
755 : {
756 1000 : capy::set_current_frame_allocator(prev);
757 1000 : }
758 : };
759 3000 : launch_wrapper<Executor> wrapper = [&] {
760 1000 : allocator_scope scope;
761 : return launch_coro<Executor>{}(
762 : ex, st,
763 : session<Executor>{
764 1000 : std::move(hold), capy::work_guard<Executor>(ex),
765 3000 : std::move(task)});
766 1000 : }();
767 :
768 : // Executor and stop token stored in promise via
769 : // constructor. Post through the frame-embedded
770 : // continuation — posting the bare handle would allocate a
771 : // wrapper op. The frame stays suspended until the
772 : // executor resumes it, so the node outlives the queue.
773 1000 : auto h = wrapper.h;
774 1000 : h.promise().owner_.bind(h, h.promise().ex);
775 1000 : h.promise().cont_.h = h;
776 : // Released only once posted: a post that throws leaves the
777 : // frame to the wrapper, which destroys it and with it the
778 : // hold, the work, and the task.
779 1000 : ex.post(h.promise().cont_);
780 996 : wrapper.h = nullptr;
781 1008 : }
782 : };
783 :
784 : /** Construct a TCP server.
785 :
786 : @tparam Ctx Execution context type satisfying ExecutionContext.
787 : @tparam Ex Executor type satisfying Executor.
788 :
789 : @param ctx The execution context for socket operations.
790 : @param ex The executor for dispatching coroutines.
791 :
792 : @par Example
793 : @par !example tcp_server
794 : */
795 : template<capy::ExecutionContext Ctx, capy::Executor Ex>
796 169 : tcp_server(Ctx& ctx, Ex ex)
797 169 : : st_(make_state(ctx, capy::any_executor(std::move(ex))))
798 : {
799 169 : }
800 :
801 : public:
802 : /** Destroy the server, stopping it.
803 :
804 : A running server is stopped and its listeners closed, so its
805 : accept loops end. The loops and any active workers finish on
806 : the server's executor without reaching the destroyed object;
807 : the workers stay alive until they do.
808 :
809 : @par Thread Safety
810 : Not thread safe.
811 : */
812 : ~tcp_server();
813 :
814 : /// Copy construction is disabled; the server owns its worker storage.
815 : tcp_server(tcp_server const&) = delete;
816 : /// Copy assignment is disabled; the server owns its worker storage.
817 : tcp_server& operator=(tcp_server const&) = delete;
818 :
819 : /** Move construct from another server.
820 :
821 : @param o The source server. After the move, @p o is
822 : in a valid but unspecified state.
823 : */
824 : tcp_server(tcp_server&& o) noexcept;
825 :
826 : /** Move assign from another server.
827 :
828 : @param o The source server. After the move, @p o is
829 : in a valid but unspecified state.
830 :
831 : @return `*this`.
832 : */
833 : tcp_server& operator=(tcp_server&& o) noexcept;
834 :
835 : /** Bind to a local endpoint.
836 :
837 : Creates an acceptor listening on the specified endpoint.
838 : Multiple endpoints can be bound by calling this method
839 : multiple times before @ref start.
840 :
841 : @pre No session is active: the server was never started, or
842 : it was stopped and joined.
843 :
844 : @param ep The local endpoint to bind to.
845 :
846 : @return An error code indicating success, or the reason binding
847 : failed.
848 : */
849 : [[nodiscard]] std::error_code bind(endpoint ep);
850 :
851 : /** Set the worker pool.
852 :
853 : Replaces any existing workers with the given range. Any
854 : previous workers are released and the idle/active lists
855 : are cleared before populating with new workers.
856 :
857 : @pre No worker is waiting on or serving a connection: the
858 : server was never started, or it was stopped and joined
859 : and every connection has finished.
860 :
861 : @par Exception Safety
862 : The server keeps its current workers if this throws.
863 :
864 : @throws std::logic_error If a worker is waiting on or serving
865 : a connection.
866 :
867 : @tparam Range Forward range of pointer-like objects to worker_base.
868 :
869 : @param workers Range of workers to manage. Each element must
870 : support `std::to_address()` yielding `worker_base*`.
871 :
872 : @par Example
873 : @par !example set_workers
874 : */
875 : template<std::ranges::forward_range Range>
876 : requires std::convertible_to<
877 : decltype(std::to_address(
878 : std::declval<std::ranges::range_value_t<Range>&>())),
879 : worker_base*>
880 183 : void set_workers(Range&& workers)
881 : {
882 : // The previous workers are released after the lock.
883 183 : std::shared_ptr<void> previous;
884 183 : std::lock_guard<std::mutex> lock(st_->mutex);
885 183 : if (st_->lent != 0)
886 6 : detail::throw_logic_error(
887 : "tcp_server::set_workers: a worker is still in use");
888 :
889 : using StorageType = std::decay_t<Range>;
890 177 : auto* p = new StorageType(std::forward<Range>(workers));
891 177 : std::shared_ptr<void> storage(
892 177 : p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
893 :
894 177 : st_->idle_head = nullptr;
895 177 : st_->active_head = nullptr;
896 177 : st_->active_tail = nullptr;
897 177 : previous = std::exchange(st_->storage, std::move(storage));
898 590 : for (auto&& elem : *p)
899 413 : st_->idle_push(std::to_address(elem));
900 189 : }
901 :
902 : /** Start accepting connections.
903 :
904 : Starts accept loops for all bound endpoints. Incoming
905 : connections are dispatched to idle workers from the pool.
906 :
907 : Calling `start()` on an already-running server has no effect.
908 :
909 : @pre At least one endpoint bound via @ref bind.
910 : @pre Workers provided via @ref set_workers.
911 : @pre If restarting, @ref join must have completed first, and the
912 : `io_context` must be restarted (`ioc.restart()`).
913 :
914 : @par Effects
915 : Creates one accept coroutine per bound endpoint. Each coroutine
916 : runs on the server's executor, waiting for connections and
917 : dispatching them to idle workers.
918 :
919 : @par Restart Sequence
920 : To restart after stopping, complete the full shutdown cycle:
921 : @par !example start
922 :
923 : @par Thread Safety
924 : Not thread safe.
925 :
926 : @throws std::logic_error If a previous session has not been
927 : joined (accept loops still active).
928 : */
929 : void start();
930 :
931 : /** Return the local endpoint for the i-th bound port.
932 :
933 : @param index Zero-based index into the list of bound ports.
934 :
935 : @return The local endpoint, or a default-constructed endpoint
936 : if @p index is out of range or the acceptor is not open.
937 : */
938 : endpoint local_endpoint(std::size_t index = 0) const noexcept;
939 :
940 : /** Stop accepting connections.
941 :
942 : Requests the accept loops' stop token and requests cancellation
943 : of active workers via their stop tokens. The acceptors are not
944 : closed. A suspended accept completes once more before its loop
945 : observes the stop token and ends.
946 :
947 : This function returns immediately; it does not wait for workers
948 : to finish. Pending I/O operations complete asynchronously.
949 :
950 : Calling `stop()` on a non-running server has no effect.
951 :
952 : @par Effects
953 : - Requests stop on the accept loops' stop token. The acceptors
954 : are not closed; a pending accept completes once more before
955 : the accept loop ends.
956 : - Requests stop on each active worker's stop token.
957 : - Workers observing their stop token should exit promptly.
958 : - A worker reuses its stop token across connections until a stop
959 : is requested. Work that keeps a connection's token past that
960 : connection's end therefore also observes this stop.
961 :
962 : @par Postconditions
963 : The server accepts no new connections. Active workers continue
964 : until they observe their stop token or complete naturally.
965 :
966 : @par What Happens Next
967 : After calling `stop()`:
968 : 1. Let `ioc.run()` return (drains pending completions).
969 : 2. Call @ref join to wait for accept loops to finish.
970 : 3. Only then may the server be restarted. It may be destroyed at
971 : any time; see @ref ~tcp_server.
972 :
973 : @par Thread Safety
974 : Not thread safe.
975 :
976 : @see join, start
977 : */
978 : void stop();
979 :
980 : /** Block until all accept loops complete.
981 :
982 : Blocks the calling thread until all accept coroutines started
983 : by @ref start have finished executing. This synchronizes the
984 : shutdown sequence, ensuring the server is fully stopped before
985 : restarting or destroying it.
986 :
987 : @pre @ref stop was called and `ioc.run()` returned.
988 :
989 : @par Postconditions
990 : All accept loops have completed. The server is in the stopped
991 : state and may be restarted via @ref start.
992 :
993 : @par Example (Correct Usage)
994 : @par !example correct_usage
995 :
996 : @par WARNING: Deadlock Scenario
997 : Calling `join()` from inside a worker coroutine deadlocks:
998 :
999 : @par !example deadlock_scenarios
1000 :
1001 : @par Thread Safety
1002 : May be called from any thread. It deadlocks if called
1003 : from within the `io_context` event loop or from a worker coroutine.
1004 :
1005 : @see stop, start
1006 : */
1007 : void join();
1008 :
1009 : private:
1010 : void discard() noexcept;
1011 : };
1012 :
1013 : #ifdef _MSC_VER
1014 : #pragma warning(pop)
1015 : #endif
1016 :
1017 : } // namespace boost::corosio
1018 :
1019 : #endif
|
