asynchrony 2.3.1
Add asynchrony to your C++ applications using standard C++20
Loading...
Searching...
No Matches
roundrobin_pool.hpp
1/*
2 roundrobin pool : Add asynchrony to your apps
3
4 BSD 3-Clause License
5
6 Copyright (c) 2021, Siddiq Software LLC
7 All rights reserved.
8
9 Redistribution and use in source and binary forms, with or without
10 modification, are permitted provided that the following conditions are met:
11
12 1. Redistributions of source code must retain the above copyright notice, this
13 list of conditions and the following disclaimer.
14
15 2. Redistributions in binary form must reproduce the above copyright notice,
16 this list of conditions and the following disclaimer in the documentation
17 and/or other materials provided with the distribution.
18
19 3. Neither the name of the copyright holder nor the names of its
20 contributors may be used to endorse or promote products derived from
21 this software without specific prior written permission.
22
23 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
24 AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
25 IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
26 DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
27 FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
28 DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
29 SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
30 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
32 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
33 */
34
35#pragma once
36#ifndef ROUNDROBIN_POOL_HPP
37#define ROUNDROBIN_POOL_HPP
38
39#include <concepts>
40#include <deque>
41#include "simple_worker.hpp"
42
43
44namespace siddiqsoft
45{
88 template <typename T, uint16_t N = 0>
89 requires std::is_move_constructible_v<T>
91 {
92 public:
95
97 auto operator=(roundrobin_pool&&) = delete;
98
101
103 auto operator=(roundrobin_pool&) = delete;
104
105
123 roundrobin_pool(std::function<void(T&&)> c)
124 {
125 // Calculate size first
126 workersSize = (N > 0) ? N : std::thread::hardware_concurrency();
127
128 // Then create workers
129 for (unsigned i = 0; i < workersSize; i++) {
130 workers.emplace_back(c);
131 }
132 }
133
158 void queue(T&& item)
159 {
160 // Add into the thread's internal queue using round-robin index
161 // Atomic fetch_add to ensures each thread gets a unique index
162 // and cast to size_t to avoid type mismatch issues
163 size_t idx = nextWorkerIndex();
164 workers.at(idx).queue(std::move(item));
165 // Increment the queue counter with release semantics for visibility
166 queueCounter.fetch_add(1, std::memory_order_release);
167 }
168
169#if defined(NLOHMANN_JSON_VERSION_MAJOR)
185 nlohmann::json toJson() const
186 {
187 return {{"_typver", "siddiqsoft.asynchrony-lib.roundrobin_pool/0.10"},
188 {"workersSize", workersSize},
189 {"queueCounter", queueCounter.load(std::memory_order_acquire)}};
190 }
191#endif
192
193#ifdef _DEBUG
194 public:
196 std::atomic_uint64_t queueCounter {0};
197#else
198 private:
200 std::atomic_uint64_t queueCounter {0};
201#endif
202
203 private:
213 std::deque<simple_worker<T>> workers {};
214
216 uint64_t workersSize {};
217
235 size_t nextWorkerIndex()
236 {
237 if (workersSize == 0) return 0;
238 return static_cast<size_t>(queueCounter.load(std::memory_order_acquire) % workersSize);
239 }
240 };
241
242#if defined(NLOHMANN_JSON_VERSION_MAJOR)
253 template <typename T, uint16_t N = 0>
254 static void to_json(nlohmann::json& dest, const siddiqsoft::roundrobin_pool<T, N>& src)
255 {
256 dest = src.toJson();
257 }
258#endif
259
260} // namespace siddiqsoft
261#endif // !ROUNDROBIN_POOL_HPP
roundrobin_pool(std::function< void(T &&)> c)
Constructs a round-robin thread pool.
roundrobin_pool(roundrobin_pool &)=delete
Copy constructor (deleted - pools are not copyable).
void queue(T &&item)
Queue a work item for processing.
auto operator=(roundrobin_pool &&)=delete
Move assignment operator (deleted - pools are not movable).
auto operator=(roundrobin_pool &)=delete
Copy assignment operator (deleted - pools are not copyable).
roundrobin_pool(roundrobin_pool &&)=delete
Move constructor (deleted - pools are not movable).