diff --git a/components/esp32-timer-cam/example/CMakeLists.txt b/components/esp32-timer-cam/example/CMakeLists.txt index ef7664675e..6b3a66f6c6 100644 --- a/components/esp32-timer-cam/example/CMakeLists.txt +++ b/components/esp32-timer-cam/example/CMakeLists.txt @@ -17,12 +17,13 @@ set(EXTRA_COMPONENT_DIRS "../../../components/rtsp" "../../../components/socket" "../../../components/task" + "../../../components/thread_pool" "../../../components/wifi" ) set( COMPONENTS - "main esptool_py esp32-camera esp32-timer-cam mdns monitor nvs rtsp socket task wifi" + "main esptool_py esp32-camera esp32-timer-cam mdns monitor nvs rtsp socket task thread_pool wifi" CACHE STRING "List of components to include" ) diff --git a/components/socket/README.md b/components/socket/README.md index c6c96f4896..b2333b92c7 100644 --- a/components/socket/README.md +++ b/components/socket/README.md @@ -89,6 +89,22 @@ The thread pool may be owned (built from `Config`) or an external shared pool. Note: registered sockets must outlive their registration - `stop()` / destroy the reactor before destroying the sockets (`stop()` waits for in-flight handlers). +### Priority bands and DSCP + +Each registration carries an `espp::QosBand` (`Critical` / `High` / `Normal` / +`Low`; `Normal` by default, preserving the pre-band FIFO behavior): when several +sockets are readable in one `select()` round the ready set is dispatched +most-urgent-first, and each handler is submitted to the `ThreadPool` at its +band, so a `Critical` socket's handler overtakes queued lower-band handlers even +on a saturated pool. `UdpSocket::ReceiveConfig::band` sets it for UDP receivers; +`add_tcp_listener(...)` / `add_tcp_stream(...)` / `add_fd(...)` take a band +argument. UDP receivers can additionally set `UdpSocket::ReceiveConfig::dscp` +to mark their *transmitted* replies with a DSCP code point (applied as +`IP_TOS`, best-effort) using the typed `espp::Dscp` enum of standard DiffServ +names - e.g. `Dscp::Ef` (expedited forwarding, latency-critical), `Dscp::Cs1` +(low-priority data), `Dscp::Af41` - network / driver treatment for outgoing +traffic, orthogonal to the local `band` scheduling. + ## Example The [example](./example) shows the use of the classes provided by the `socket` @@ -102,3 +118,5 @@ and reconnect behavior, including: * reconnect behavior after TCP session shutdown * `SocketReactor` multiplexing UDP receivers and TCP listeners/streams on one select loop + thread pool (shared-pool, dynamic remove, and multi-client cases) +* `SocketReactor` priority bands: a Critical-band receiver (with DSCP-marked + replies) staying responsive during a Low-band flood diff --git a/components/socket/example/README.md b/components/socket/example/README.md index 869009be6b..3e81665106 100644 --- a/components/socket/example/README.md +++ b/components/socket/example/README.md @@ -15,6 +15,12 @@ The covered scenarios include: * TCP request/response followed by reconnect * TCP blocked-accept teardown * TCP connect failure to an unused port +* `SocketReactor` multiplexing UDP receivers and TCP listeners/streams on one + select loop + thread pool (shared-pool, dynamic remove, lifecycle/validation, + and multi-client cases) +* `SocketReactor` priority bands: Critical (DSCP EF) + Low banded UDP receivers, + with the Critical port verified responsive during a Low-band flood +* UDP send overloads and sender info At startup the example creates a small open Wi-Fi AP so the network stack is initialized, but the actual test traffic stays local to the device using @@ -40,4 +46,4 @@ See the Getting Started Guide for full steps to configure and use ESP-IDF to bui ## Example Output The serial log shows each scenario as it starts, a per-scenario pass/fail line, -and a final summary such as `Socket example summary: 9/9 scenarios passed`. +and a final summary such as `Socket example summary: 16/16 scenarios passed`. diff --git a/components/socket/example/main/socket_example.cpp b/components/socket/example/main/socket_example.cpp index 49347709ee..50b105cc75 100644 --- a/components/socket/example/main/socket_example.cpp +++ b/components/socket/example/main/socket_example.cpp @@ -566,6 +566,83 @@ ScenarioResult run_socket_reactor_scenario() { "2 UDP sockets multiplexed on 1 select loop + shared pool, both echoed"); } +ScenarioResult run_reactor_priority_bands_scenario() { + // Two UDP echo receivers on ONE reactor, registered at different QosBands. + // The Critical receiver additionally marks its transmitted replies with + // DSCP 46 (EF, "expedited forwarding"), exercising the IP_TOS setsockopt + // path on lwIP. Functional on-target check of the band-aware registration + + // dispatch: flood the Low-band port, then confirm the Critical-band port + // still answers within the timeout, and that Low is still serviced (no band + // is starved or unreachable). + constexpr size_t critical_port = 5030; + constexpr size_t low_port = 5031; + auto echo_reversed = [](const ByteVector &data, const espp::Socket::Info &) { + return std::optional(reversed(data)); + }; + + //! [socket reactor priority example] + espp::SocketReactor reactor({.log_level = espp::Logger::Verbosity::WARN}); + + espp::UdpSocket critical_server({.log_level = espp::Logger::Verbosity::WARN}); + espp::UdpSocket low_server({.log_level = espp::Logger::Verbosity::WARN}); + auto critical_id = reactor.add_udp_receiver(critical_server, + {.port = critical_port, + .buffer_size = kMaxPacketSize, + .on_receive_callback = echo_reversed, + .band = espp::QosBand::Critical, + .dscp = espp::Dscp::Ef}); // latency-critical replies + auto low_id = reactor.add_udp_receiver(low_server, {.port = low_port, + .buffer_size = kMaxPacketSize, + .on_receive_callback = echo_reversed, + .band = espp::QosBand::Low}); + //! [socket reactor priority example] + + if (critical_id == espp::SocketReactor::INVALID_ID || low_id == espp::SocketReactor::INVALID_ID) { + return fail("Reactor priority bands", "failed to register one or both banded UDP receivers"); + } + + // Flood the Low-band port with fire-and-forget packets so its receive + // handling occupies the shared pool... + espp::UdpSocket flooder({.log_level = espp::Logger::Verbosity::WARN}); + for (int i = 0; i < 8; ++i) { + flooder.send(make_payload(512, static_cast(i)), + {.ip_address = kLoopbackAddress, .port = low_port}); + } + + // ...then verify each banded port answers within the timeout. + auto send_and_check = [&](size_t port, uint8_t seed) -> bool { + auto request = make_payload(512, seed); + auto expected = reversed(request); + ByteVector response; + std::atomic_bool got_response{false}; + espp::UdpSocket client({.log_level = espp::Logger::Verbosity::WARN}); + client.send(request, {.ip_address = kLoopbackAddress, + .port = port, + .wait_for_response = true, + .response_size = kMaxPacketSize, + .on_response_callback = + [&](const ByteVector &r) { + response = r; + got_response = true; + }, + .response_timeout = 500ms}); + return got_response.load() && response == expected; + }; + + if (!send_and_check(critical_port, 0x20)) { + return fail("Reactor priority bands", + "Critical-band server did not echo during the Low-band flood"); + } + // The Low band must still be serviced too (banded dispatch must not starve + // or strand the least-urgent band). + if (!send_and_check(low_port, 0x90)) { + return fail("Reactor priority bands", "Low-band server did not echo after the flood"); + } + + return pass("Reactor priority bands", + "Critical (DSCP EF) + Low banded receivers echoed during and after a Low flood"); +} + ScenarioResult run_tcp_reactor_scenario() { // A TCP echo server built entirely on the reactor: one listener registration // accepts clients, and each accepted client is registered as a stream that @@ -898,6 +975,7 @@ extern "C" void app_main(void) { run_and_record(run_tcp_blocked_accept_teardown_scenario, "TCP blocked accept teardown"); run_and_record(run_tcp_connect_failure_scenario, "TCP connect failure"); run_and_record(run_socket_reactor_scenario, "Socket reactor (select + thread pool)"); + run_and_record(run_reactor_priority_bands_scenario, "Reactor priority bands (QosBand + DSCP)"); run_and_record(run_tcp_reactor_scenario, "TCP reactor (listener + streams)"); run_and_record(run_reactor_shared_pool_scenario, "Reactor shared pool + dynamic remove"); run_and_record(run_reactor_lifecycle_scenario, "Reactor lifecycle + input validation"); diff --git a/components/socket/include/dscp.hpp b/components/socket/include/dscp.hpp new file mode 100644 index 0000000000..1fb207aa02 --- /dev/null +++ b/components/socket/include/dscp.hpp @@ -0,0 +1,70 @@ +#pragma once + +#include + +namespace espp { + +/// @brief Standard DiffServ code points (DSCP) for IP traffic marking. +/// +/// A DSCP is the 6-bit field carried in the upper bits of the IP TOS / +/// Traffic Class byte (RFC 2474); network devices use it to prioritize, +/// queue, or drop traffic. The named values below are the IANA-registered +/// per-hop behaviors, so application code can write `Dscp::Ef` instead of a +/// magic number: +/// +/// - **CS0..CS7** - class selectors (RFC 2474): backwards-compatible with the +/// legacy IP precedence bits. CS0 (0) is default/best-effort forwarding; +/// higher classes are conventionally more important (CS6/CS7 are typically +/// reserved for network control traffic). +/// - **AF11..AF43** - assured forwarding (RFC 2597): four classes (AF1x +/// lowest priority .. AF4x highest), each with three drop precedences +/// (x1 = lowest drop probability .. x3 = highest). E.g. AF41 = high +/// priority, low drop probability. +/// - **EF** - expedited forwarding (RFC 3246): the standard marking for +/// low-latency, low-jitter traffic (e.g. voice / control loops). +/// - **VoiceAdmit** - capacity-admitted EF traffic (RFC 5865). +/// - **LE** - lower effort (RFC 8622): scavenger-class traffic that should +/// yield to everything else (e.g. bulk background transfers). +/// +/// Enumerator names are CamelCase (Ef, Cs5, Af41) rather than the all-caps +/// RFC names (EF, CS5, AF41) because platform headers define macros with the +/// all-caps names (e.g. newlib's defines CS5/CS6/CS7), which +/// would break the enum wherever both headers are included. +/// +/// A custom (non-standard) code point can still be expressed with +/// `static_cast(value)` for values 0-63; consumers reject out-of-range +/// values. +enum class Dscp : uint8_t { + Cs0 = 0, ///< Class selector 0 - default / best-effort forwarding. + Default = Cs0, ///< Alias for CS0. + Le = 1, ///< Lower effort / scavenger (RFC 8622). + Cs1 = 8, ///< Class selector 1 (conventionally low-priority data). + Af11 = 10, ///< Assured forwarding: class 1, low drop precedence. + Af12 = 12, ///< Assured forwarding: class 1, medium drop precedence. + Af13 = 14, ///< Assured forwarding: class 1, high drop precedence. + Cs2 = 16, ///< Class selector 2 (conventionally OAM / management). + Af21 = 18, ///< Assured forwarding: class 2, low drop precedence. + Af22 = 20, ///< Assured forwarding: class 2, medium drop precedence. + Af23 = 22, ///< Assured forwarding: class 2, high drop precedence. + Cs3 = 24, ///< Class selector 3 (conventionally call signaling). + Af31 = 26, ///< Assured forwarding: class 3, low drop precedence. + Af32 = 28, ///< Assured forwarding: class 3, medium drop precedence. + Af33 = 30, ///< Assured forwarding: class 3, high drop precedence. + Cs4 = 32, ///< Class selector 4 (conventionally real-time interactive). + Af41 = 34, ///< Assured forwarding: class 4, low drop precedence. + Af42 = 36, ///< Assured forwarding: class 4, medium drop precedence. + Af43 = 38, ///< Assured forwarding: class 4, high drop precedence. + Cs5 = 40, ///< Class selector 5 (conventionally broadcast video). + VoiceAdmit = 44, ///< Capacity-admitted EF traffic (RFC 5865). + Ef = 46, ///< Expedited forwarding (RFC 3246) - low-latency/low-jitter. + Cs6 = 48, ///< Class selector 6 (network control - use with care). + Cs7 = 56, ///< Class selector 7 (reserved network control). +}; + +/// @brief Convert a DSCP code point to the IP TOS / Traffic Class byte value +/// that carries it (DSCP occupies the upper 6 bits - RFC 2474). +/// @param dscp The code point to convert. +/// @return The TOS byte value (dscp << 2), e.g. Dscp::Ef -> 184 (0xB8). +constexpr uint8_t dscp_to_tos(Dscp dscp) { return static_cast(dscp) << 2; } + +} // namespace espp diff --git a/components/socket/include/socket_reactor.hpp b/components/socket/include/socket_reactor.hpp index c81e972ea3..19e36c175a 100644 --- a/components/socket/include/socket_reactor.hpp +++ b/components/socket/include/socket_reactor.hpp @@ -45,6 +45,19 @@ namespace espp { * responsive by a loopback UDP "wakeup" socket that is also in the * select set: poking it interrupts @c select() at once. * + * Each registration carries a priority band (@ref espp::QosBand, + * default Normal). When one @c select() wakeup reports several + * sockets readable at once, their handlers are submitted to the pool + * in band order (most urgent first) and each is submitted AT its band, + * so band-aware pools (see @ref espp::ThreadPool) run urgent sockets' + * handlers first. This also shapes the saturation policy: when the + * pool is (nearly) full, urgent sockets win the remaining queue slots + * while less urgent ones simply stay readable and are re-reported by + * the next @c select() (kernel-buffer backpressure, no data loss for + * TCP / bounded loss semantics identical to before for UDP). With the + * default band on every registration the dispatch order and behavior + * are unchanged from the pre-band reactor. + * * @note Lifetime. Registered sockets and callbacks must outlive their * registration. @ref stop (and the destructor) waits for any in-flight * handler to finish, so the guaranteed-safe teardown order is: stop() / @@ -68,6 +81,8 @@ namespace espp { * \snippet socket_example.cpp socket reactor example * \section socket_reactor_ex2 Socket Reactor TCP Example * \snippet socket_example.cpp socket reactor tcp example + * \section socket_reactor_ex3 Socket Reactor Priority Bands Example + * \snippet socket_example.cpp socket reactor priority example */ class SocketReactor : public BaseComponent { public: @@ -156,9 +171,14 @@ class SocketReactor : public BaseComponent { * data, it is sent back to the sender. * @note This replaces UdpSocket::start_receiving() (which spawns a dedicated * thread) - the reactor drives the socket instead. + * @note The receive_config's band field selects the priority band this + * socket's handling is dispatched at, and its dscp field (if set) is + * applied to the socket via IP_TOS here (best-effort, marks + * transmitted packets - see UdpSocket::ReceiveConfig). * @param socket A UdpSocket to bind and receive on. Must outlive the * registration (call remove() before destroying it). - * @param receive_config Port / multicast / buffer_size / callback config. + * @param receive_config Port / multicast / buffer_size / callback / band / + * dscp config. * @return A registration Id, or INVALID_ID on failure. */ Id add_udp_receiver(espp::UdpSocket &socket, @@ -171,9 +191,12 @@ class SocketReactor : public BaseComponent { * @param listener A TcpSocket that has already been bind()+listen()'d. Must * outlive the registration. * @param on_accept Callback given ownership of each accepted client. + * @param band Priority band to dispatch accept handling at (see + * espp::QosBand; default Normal = pre-band behavior). * @return A registration Id, or INVALID_ID on failure. */ - Id add_tcp_listener(espp::TcpSocket &listener, const AcceptCallback &on_accept); + Id add_tcp_listener(espp::TcpSocket &listener, const AcceptCallback &on_accept, + QosBand band = QosBand::Normal); /** * @brief Register a connected TcpSocket for reading. When it is readable the @@ -185,19 +208,23 @@ class SocketReactor : public BaseComponent { * @param on_data Callback given the connection and the bytes read. * @param buffer_size Max bytes to read per readable event. * @param on_close Optional callback fired once when the peer closes. + * @param band Priority band to dispatch read handling at (see espp::QosBand; + * default Normal = pre-band behavior). * @return A registration Id, or INVALID_ID on failure. */ Id add_tcp_stream(espp::TcpSocket &connection, const StreamCallback &on_data, size_t buffer_size, - const CloseCallback &on_close = {}); + const CloseCallback &on_close = {}, QosBand band = QosBand::Normal); /** * @brief Low-level registration: watch @p fd for readability and run * @p handler (on a pool worker) each time it is readable. * @param fd A valid socket file descriptor (see Socket::native_handle()). * @param handler Handler that reads/processes the socket. + * @param band Priority band to dispatch the handler at (see espp::QosBand; + * default Normal = pre-band behavior). * @return A registration Id, or INVALID_ID on failure. */ - Id add_fd(sock_type_t fd, ReadHandler handler); + Id add_fd(sock_type_t fd, ReadHandler handler, QosBand band = QosBand::Normal); /** * @brief Unregister a socket. Safe to call from any thread, including from @@ -215,6 +242,7 @@ class SocketReactor : public BaseComponent { struct Entry { sock_type_t fd{static_cast(-1)}; ///< Watched file descriptor. ReadHandler handler; ///< Handler run on the pool. + QosBand band{QosBand::Normal}; ///< Priority band for dispatch. bool armed{true}; ///< In the select set (not currently dispatched). bool in_flight{false}; ///< A pool job is currently running the handler. bool remove_requested{false}; ///< remove() was called while in-flight. @@ -229,7 +257,7 @@ class SocketReactor : public BaseComponent { Id allocate_id(); /// Insert an entry for a pre-allocated id and wake the loop. Used so a /// stream handler can capture its own id before the entry becomes reachable. - void insert_entry(Id id, sock_type_t fd, ReadHandler handler); + void insert_entry(Id id, sock_type_t fd, ReadHandler handler, QosBand band); /// One iteration of the select() loop (the loop task callback body). bool loop_iteration(std::mutex &m, std::condition_variable &cv, bool &task_notified); diff --git a/components/socket/include/udp_socket.hpp b/components/socket/include/udp_socket.hpp index 9c97c664f0..60dd413cb2 100644 --- a/components/socket/include/udp_socket.hpp +++ b/components/socket/include/udp_socket.hpp @@ -7,7 +7,9 @@ #include #include +#include "dscp.hpp" #include "logger.hpp" +#include "qos_band.hpp" #include "socket.hpp" #include "task.hpp" @@ -53,6 +55,20 @@ class UdpSocket : public Socket { on multi-homed hosts to bind multicast to a specific NIC (e.g. wired vs Wi-Fi). */ espp::Socket::receive_callback_fn on_receive_callback{ nullptr}; /**< Function containing business logic to handle data received. */ + espp::QosBand band{ + espp::QosBand::Normal}; /**< Priority band for dispatching this socket's receive handling + when registered on an espp::SocketReactor (unused by + start_receiving(), which owns a dedicated task). Normal (the + default) preserves the pre-band FIFO dispatch behavior. */ + std::optional dscp{}; /**< Optional DSCP code point to mark this socket's + TRANSMITTED packets with (applied as IP_TOS by + espp::SocketReactor at registration, best-effort). Affects + network / driver treatment of outgoing traffic (e.g. + Dscp::Ef "expedited forwarding" for latency-critical flows, + Dscp::Af41), NOT local scheduling - use `band` for that. + A non-standard code point can be expressed with + static_cast(0-63); out-of-range values are rejected + with a warning. */ }; struct SendConfig { diff --git a/components/socket/src/socket_reactor.cpp b/components/socket/src/socket_reactor.cpp index 337245ec43..865bb1fb3b 100644 --- a/components/socket/src/socket_reactor.cpp +++ b/components/socket/src/socket_reactor.cpp @@ -1,7 +1,9 @@ #include "socket_reactor.hpp" +#include #include #include +#include #ifndef _WIN32 #include @@ -168,15 +170,16 @@ SocketReactor::Id SocketReactor::allocate_id() { return next_id_; } -void SocketReactor::insert_entry(Id id, sock_type_t fd, ReadHandler handler) { +void SocketReactor::insert_entry(Id id, sock_type_t fd, ReadHandler handler, QosBand band) { { std::lock_guard lock(mutex_); - entries_[id] = Entry{.fd = fd, .handler = std::move(handler)}; + entries_[id] = Entry{.fd = fd, .handler = std::move(handler), .band = band}; } wake(); // interrupt select() so the new fd is picked up } -SocketReactor::Id SocketReactor::add_fd(sock_type_t fd, SocketReactor::ReadHandler handler) { +SocketReactor::Id SocketReactor::add_fd(sock_type_t fd, SocketReactor::ReadHandler handler, + QosBand band) { if (!handler) { logger_.error("add_fd: null handler"); return INVALID_ID; @@ -185,7 +188,7 @@ SocketReactor::Id SocketReactor::add_fd(sock_type_t fd, SocketReactor::ReadHandl return INVALID_ID; } Id id = allocate_id(); - insert_entry(id, fd, std::move(handler)); + insert_entry(id, fd, std::move(handler), band); return id; } @@ -199,6 +202,27 @@ SocketReactor::add_udp_receiver(espp::UdpSocket &socket, const auto callback = receive_config.on_receive_callback; const auto buffer_size = receive_config.buffer_size; sock_type_t fd = socket.native_handle(); + if (receive_config.dscp.has_value()) { + // Mark this socket's transmitted packets (e.g. echo responses) with the + // requested DSCP code point. The TOS byte carries the 6-bit DSCP in its + // upper bits (RFC 2474). Best-effort: network / driver treatment only, no + // effect on local scheduling (that is what `band` is for). + const uint8_t dscp = static_cast(receive_config.dscp.value()); + if (dscp > 63) { + // Named espp::Dscp values are always in range; a custom static_cast'd + // code point is documented as 0-63. Silently masking would apply a + // DIFFERENT code point (64 -> 0, 255 -> 63), so ignore invalid values. + logger_.warn("add_udp_receiver: invalid DSCP {} (valid range 0-63) on port {}; not applied", + dscp, receive_config.port); + } else { + const int tos = espp::dscp_to_tos(receive_config.dscp.value()); + if (::setsockopt(fd, IPPROTO_IP, IP_TOS, reinterpret_cast(&tos), sizeof(tos)) < + 0) { + logger_.warn("add_udp_receiver: could not set IP_TOS (DSCP {}) on port {}", dscp, + receive_config.port); + } + } + } auto handler = [this, &socket, callback, buffer_size]() { std::vector data; Socket::Info sender; @@ -223,11 +247,11 @@ SocketReactor::add_udp_receiver(espp::UdpSocket &socket, logger_.warn("Failed to send UDP response to {}", sender); } }; - return add_fd(fd, std::move(handler)); + return add_fd(fd, std::move(handler), receive_config.band); } SocketReactor::Id SocketReactor::add_tcp_listener(espp::TcpSocket &listener, - const AcceptCallback &on_accept) { + const AcceptCallback &on_accept, QosBand band) { sock_type_t fd = listener.native_handle(); auto handler = [this, &listener, on_accept]() { // select() reported the listener readable, so accept() returns immediately. @@ -239,12 +263,12 @@ SocketReactor::Id SocketReactor::add_tcp_listener(espp::TcpSocket &listener, on_accept(std::move(client)); } }; - return add_fd(fd, std::move(handler)); + return add_fd(fd, std::move(handler), band); } SocketReactor::Id SocketReactor::add_tcp_stream(espp::TcpSocket &connection, const StreamCallback &on_data, size_t buffer_size, - const CloseCallback &on_close) { + const CloseCallback &on_close, QosBand band) { sock_type_t fd = connection.native_handle(); if (!check_fd(fd)) { return INVALID_ID; @@ -269,7 +293,7 @@ SocketReactor::Id SocketReactor::add_tcp_stream(espp::TcpSocket &connection, } remove(id); }; - insert_entry(id, fd, std::move(handler)); + insert_entry(id, fd, std::move(handler), band); return id; } @@ -404,7 +428,7 @@ bool SocketReactor::loop_iteration(std::mutex &, std::condition_variable &, bool // Collect readable entries, disarming each so it is not dispatched again // until its handler completes. - std::vector ready; + std::vector> ready; { std::lock_guard lock(mutex_); for (auto &[id, entry] : entries_) { @@ -412,14 +436,20 @@ bool SocketReactor::loop_iteration(std::mutex &, std::condition_variable &, bool (FD_ISSET(entry.fd, &readfds) || FD_ISSET(entry.fd, &exceptfds))) { entry.armed = false; entry.in_flight = true; - ready.push_back(id); + ready.emplace_back(id, entry.band); } } } - for (Id id : ready) { + // Dispatch in band order (most urgent first; stable, so same-band sockets + // keep their registration order). Submitting urgent sockets first also means + // they win the remaining pool slots when the pool is nearly saturated. + std::stable_sort(ready.begin(), ready.end(), + [](const auto &a, const auto &b) { return a.second < b.second; }); + + for (const auto &[id, band] : ready) { ++in_flight_count_; - bool submitted = pool_->submit([this, id]() { dispatch(id); }); + bool submitted = pool_->submit([this, id]() { dispatch(id); }, band); if (!submitted) { // Pool is saturated; revert and let the next select() re-report this fd // (the data stays buffered in the socket - natural backpressure). diff --git a/components/task/README.md b/components/task/README.md index 655eda53c5..e16fd154d2 100644 --- a/components/task/README.md +++ b/components/task/README.md @@ -11,6 +11,24 @@ It also supports firing off syncrhonous (blocking) and asynchronous (non-blocking) functions in separate threads, with the option of configuring the core id and other esp-specific paramters. +## Priority semantics per platform + +- **ESP (FreeRTOS)**: `BaseConfig::priority` is the FreeRTOS task priority + (clamped to `configMAX_PRIORITIES - 1`) and is always applied. +- **Host platforms (Linux / macOS / Windows)**: the priority is stored, but only + applied to the OS thread when `BaseConfig::host_realtime` is set (default + `false`, preserving the historical behavior of running at the OS default + scheduling). With the opt-in: + - **Linux / macOS**: priority ≥ 1 is mapped linearly onto the `SCHED_FIFO` + real-time priority range (priority 0 resets to the default `SCHED_OTHER` + scheduler). On Linux this requires `CAP_SYS_NICE` or an `RLIMIT_RTPRIO` + allowance, and delivers hard preemption on `PREEMPT_RT` kernels; without + permission the task falls back gracefully to default scheduling with a + one-time warning. **Beware:** a `SCHED_FIFO` thread that spins can starve + the rest of the system. + - **Windows**: best-effort mapping onto `SetThreadPriority()` classes + (NORMAL / ABOVE_NORMAL / HIGHEST / TIME_CRITICAL). + ## Example The [example](./example) shows some various different ways of starting and diff --git a/components/task/example/main/task_example.cpp b/components/task/example/main/task_example.cpp index 62fde1248d..85bd34b75d 100644 --- a/components/task/example/main/task_example.cpp +++ b/components/task/example/main/task_example.cpp @@ -481,16 +481,38 @@ extern "C" void app_main(void) { return false; // keep running }; auto task = espp::Task({.callback = task_fn, - .task_config = {.name = "Reconfig Task", .priority = 5, .core_id = 0}, + .task_config = {.name = "Reconfig Task", + .priority = 5, + .core_id = 0, + // host-only opt-in: on Linux/macOS/Windows this would + // apply the priority to the OS thread (SCHED_FIFO / + // SetThreadPriority); ignored on ESP, where the + // FreeRTOS priority is always applied + .host_realtime = false}, .log_level = espp::Logger::Verbosity::DEBUG}); + // the configured priority is readable before (and while) the task runs + if (task.get_configured_priority() != 5) { + logger.error("get_configured_priority() != 5 after construction!"); + } task.start(); - fmt::println("Task started on core {} at priority {}", espp::Task::get_core_id(task), - espp::Task::get_priority(task)); + fmt::println("Task started on core {} at priority {} (configured {})", + espp::Task::get_core_id(task), espp::Task::get_priority(task), + task.get_configured_priority()); - // priority changes apply to the running task immediately + // priority changes apply to the running task immediately, and the + // configured priority tracks them (thread-safe to read from other threads) bool applied = task.set_priority(10); - fmt::println("set_priority(10) applied live: {}, priority now {}", applied, - espp::Task::get_priority(task)); + fmt::println("set_priority(10) applied live: {}, priority now {} (configured {})", applied, + espp::Task::get_priority(task), task.get_configured_priority()); + if (task.get_configured_priority() != 10) { + logger.error("get_configured_priority() != 10 after set_priority(10)!"); + } +#if defined(ESP_PLATFORM) + // on ESP the FreeRTOS scheduler must observe the live change as well + if (espp::Task::get_priority(task) != 10) { + logger.error("FreeRTOS priority did not follow set_priority(10)!"); + } +#endif // core affinity changes are stored but only take effect when the task is // (re)started on the default ESP-IDF FreeRTOS port (a task's core is fixed diff --git a/components/task/include/task.hpp b/components/task/include/task.hpp index aeb5c0d3fd..a9b005ed36 100644 --- a/components/task/include/task.hpp +++ b/components/task/include/task.hpp @@ -158,8 +158,27 @@ class Task : public espp::BaseComponent { struct BaseConfig { std::string name; /**< Name of the task */ size_t stack_size_bytes{4096}; /**< Stack Size (B) allocated to the task. */ - size_t priority{0}; /**< Priority of the task, 0 is lowest priority on ESP / FreeRTOS. */ + size_t priority{0}; /**< Priority of the task; 0 is the lowest, and espp uses the FreeRTOS + convention that ~25 is the highest useful "real-time" priority. + Platform semantics: on ESP this is the FreeRTOS task priority (clamped + to configMAX_PRIORITIES - 1) and is always applied. On host platforms + (Linux / macOS / Windows) the priority is stored but only applied to + the OS thread when host_realtime is set - see below. */ int core_id{-1}; /**< Core ID of the task, -1 means it is not pinned to any core. */ + bool host_realtime{ + false}; /**< Opt-in to applying the priority to the OS thread on HOST platforms (ignored + on ESP, where the FreeRTOS priority is always applied). When false (the + default) the thread uses the OS default scheduling, matching espp's historical + host behavior. When true: on Linux and macOS, priority 0 resets the thread to + the default scheduler (SCHED_OTHER) while priority >= 1 is mapped linearly + onto the SCHED_FIFO real-time priority range - giving true preemptive + priority scheduling when permitted; on Windows the priority is mapped + best-effort onto SetThreadPriority() classes (NORMAL / ABOVE_NORMAL / HIGHEST + / TIME_CRITICAL). + @warning A SCHED_FIFO thread that spins can starve the rest of the system. + On Linux this requires CAP_SYS_NICE or an RLIMIT_RTPRIO allowance (and + delivers hard preemption on PREEMPT_RT kernels); without permission the task + falls back gracefully to default scheduling with a one-time warning. */ }; /** @@ -276,17 +295,31 @@ class Task : public espp::BaseComponent { * @brief Set the priority of the task. * @details The new priority is always stored in the task's configuration, so * it will be used the next time the task is started. If the task is - * currently running (ESP only), the change is also applied to the - * live task immediately via vTaskPrioritySet(). - * @param priority New FreeRTOS priority (0 is lowest priority on ESP / - * FreeRTOS). It is clamped to [0, configMAX_PRIORITIES - 1] on ESP. + * currently running, the change is also applied to the live task + * immediately: via vTaskPrioritySet() on ESP, via + * pthread_setschedparam() (SCHED_FIFO for priority >= 1, default + * scheduling for priority 0) on Linux/macOS, and via + * SetThreadPriority() on Windows. See BaseConfig::priority for the + * per-platform semantics (including the unprivileged-Linux graceful + * fallback). + * @param priority New priority (0 is lowest; see BaseConfig::priority). It is + * clamped to [0, configMAX_PRIORITIES - 1] on ESP. * @return true if the change was applied to the currently-running task; false * if the task is not running (the new value still takes effect the - * next time the task is started) or the platform does not support - * changing a live task's priority. + * next time the task is started) or the platform / privileges did + * not allow changing the live task's scheduling. */ bool set_priority(size_t priority); + /** + * @brief Get the priority stored in the task's configuration. + * @details This is the value set at construction or via set_priority(); it + * is the priority the task will be started with (and, if the task + * is running, the priority that was last requested for it). + * @return The configured priority (0 is lowest; see BaseConfig::priority). + */ + size_t get_configured_priority() const { return priority_; } + /** * @brief Set the core affinity (core ID) of the task. * @details The new core ID is always stored in the task's configuration, so @@ -534,15 +567,57 @@ class Task : public espp::BaseComponent { */ void notify_and_join(); +#if !defined(ESP_PLATFORM) + /** + * @brief Apply \p priority to the (running) \p thread using the host OS + * scheduling API (best-effort; see BaseConfig::priority). + * @param thread The thread to apply the priority to; must be joinable. + * @param priority The espp priority to apply (0 = default scheduling). + * @return true if the OS accepted the scheduling change, false otherwise + * (e.g. insufficient privileges - the thread keeps running with + * default scheduling). + */ + bool apply_thread_priority(std::thread &thread, size_t priority); + + /** + * @brief Apply \p priority to the thread identified by the OS-native + * \p handle (best-effort; see BaseConfig::priority). Called by the + * worker thread itself (with its own handle) before its first + * callback invocation so the scheduling policy is guaranteed to be + * in place for the task's entire execution, and by + * apply_thread_priority() for live set_priority() changes. + * @param handle Native handle of the thread to apply the priority to. + * @param priority The espp priority to apply (0 = default scheduling). + * @return true if the OS accepted the scheduling change, false otherwise. + */ + bool apply_thread_priority_to_handle(std::thread::native_handle_type handle, size_t priority); +#endif + callback_variant callback_; ///< Variant of the callback function for the task. BaseConfig config_; ///< Configuration for the task. + /// Configured priority. Single source of truth after construction (initialized + /// from config_.priority): written by set_priority() and read concurrently by + /// the worker thread's startup priority application, start(), and + /// get_configured_priority() - atomic so a live set_priority() cannot race + /// those reads. + std::atomic priority_{0}; + std::atomic started_{false}; std::condition_variable cv_; bool notified_{false}; std::mutex cv_m_; std::mutex thread_mutex_; std::thread thread_; +#if !defined(ESP_PLATFORM) + /// Serializes the OS-level priority applications (the worker's startup + /// self-application vs a live set_priority()), so a startup application can + /// never overwrite a newer concurrent request with a stale value. Separate + /// from thread_mutex_ because notify_and_join() holds thread_mutex_ across + /// join() - the worker taking thread_mutex_ at startup could deadlock a + /// stop() issued right after start(). + std::mutex priority_apply_mutex_; +#endif #if defined(ESP_PLATFORM) std::atomic watchdog_started_{false}; task_id_t task_handle_{nullptr}; diff --git a/components/task/src/task.cpp b/components/task/src/task.cpp index ad4c84d721..436154903f 100644 --- a/components/task/src/task.cpp +++ b/components/task/src/task.cpp @@ -1,11 +1,146 @@ #include "task.hpp" +#if !defined(ESP_PLATFORM) +#if defined(_WIN32) +#include +#else +#include +#include +#include +#include +#endif +#endif // !ESP_PLATFORM + using namespace espp; +#if !defined(ESP_PLATFORM) +namespace { +#if !defined(_WIN32) +// espp follows the FreeRTOS convention for priorities: 0 is the lowest and +// ~25 (a configMAX_PRIORITIES-like ceiling) is the highest useful priority. +// Host-side scheduling maps espp priorities into the native range using this +// ceiling. +constexpr size_t ESPP_PRIORITY_CEILING = 25; +#endif +// Warn only once per process when real-time scheduling is unavailable +// (e.g. unprivileged Linux without CAP_SYS_NICE / RLIMIT_RTPRIO). +std::atomic rt_unavailable_warned{false}; +} // namespace + +bool Task::apply_thread_priority(std::thread &thread, size_t priority) { + if (!thread.joinable()) { + return false; + } + return apply_thread_priority_to_handle(thread.native_handle(), priority); +} + +bool Task::apply_thread_priority_to_handle(std::thread::native_handle_type handle, + size_t priority) { +#if defined(__linux__) || defined(__APPLE__) + struct sched_param param = {}; + if (priority == 0) { + // espp priority 0 = default (non-realtime) scheduling. SCHED_OTHER only + // accepts the static priority range [min, max] of that policy (a single + // value, 0, on Linux; the default is the middle of the range on macOS). + const int other_min = sched_get_priority_min(SCHED_OTHER); + const int other_max = sched_get_priority_max(SCHED_OTHER); + if (other_min < 0 || other_max < other_min) { + // sched_get_priority_min/max return -1 on failure; don't feed a bogus + // (negative) priority to pthread_setschedparam. + return false; + } + param.sched_priority = (other_min + other_max) / 2; + const int err = pthread_setschedparam(handle, SCHED_OTHER, ¶m); + if (err != 0) { + logger_.debug("Could not reset task '{}' to default scheduling: {}", config_.name, + strerror(err)); + return false; + } + return true; + } + // espp priority >= 1: map linearly onto the SCHED_FIFO real-time priority + // range. Priority 1 -> the minimum RT priority, ESPP_PRIORITY_CEILING (or + // above) -> the maximum. + const int fifo_min = sched_get_priority_min(SCHED_FIFO); + const int fifo_max = sched_get_priority_max(SCHED_FIFO); + if (fifo_min < 0 || fifo_max < fifo_min) { + return false; + } + const size_t clamped = std::min(priority, ESPP_PRIORITY_CEILING); + const int span = fifo_max - fifo_min; + param.sched_priority = fifo_min + static_cast((clamped - 1) * static_cast(span) / + (ESPP_PRIORITY_CEILING - 1)); + const int err = pthread_setschedparam(handle, SCHED_FIFO, ¶m); + if (err != 0) { + // Most commonly EPERM on Linux: real-time scheduling needs CAP_SYS_NICE or + // an RLIMIT_RTPRIO allowance. This must never fail the task - fall back to + // default scheduling and warn once per process. The fallback is an + // explicit SCHED_OTHER reset: if a previous application succeeded (and + // e.g. a privilege / RLIMIT_RTPRIO change made this one fail), the thread + // may currently be running SCHED_FIFO, and leaving it there would + // contradict the documented fallback and keep a potentially starving RT + // thread alive. + const int other_min = sched_get_priority_min(SCHED_OTHER); + const int other_max = sched_get_priority_max(SCHED_OTHER); + struct sched_param other_param = {}; + other_param.sched_priority = + (other_min >= 0 && other_max >= other_min) ? (other_min + other_max) / 2 : 0; + const int reset_err = pthread_setschedparam(handle, SCHED_OTHER, &other_param); + if (reset_err != 0) { + logger_.warn("Could not apply SCHED_FIFO priority {} to task '{}' ({}), and resetting to " + "default scheduling also failed ({}); the thread keeps its previous scheduling " + "policy", + param.sched_priority, config_.name, strerror(err), strerror(reset_err)); + } else if (!rt_unavailable_warned.exchange(true)) { +#if defined(__APPLE__) + logger_.warn("Could not apply SCHED_FIFO priority {} to task '{}' ({}); running without " + "realtime priority", + param.sched_priority, config_.name, strerror(err)); +#else + logger_.warn("Could not apply SCHED_FIFO priority {} to task '{}' ({}); running without " + "realtime priority; grant CAP_SYS_NICE or configure RLIMIT_RTPRIO for RT " + "scheduling (e.g. PREEMPT_RT)", + param.sched_priority, config_.name, strerror(err)); +#endif + } + return false; + } + logger_.debug("Applied SCHED_FIFO priority {} to task '{}'", param.sched_priority, config_.name); + return true; +#elif defined(_WIN32) + // Best-effort mapping of the espp priority range onto the Windows thread + // priority classes. + int win_priority = THREAD_PRIORITY_NORMAL; + if (priority >= 17) { + win_priority = THREAD_PRIORITY_TIME_CRITICAL; + } else if (priority >= 9) { + win_priority = THREAD_PRIORITY_HIGHEST; + } else if (priority >= 1) { + win_priority = THREAD_PRIORITY_ABOVE_NORMAL; + } + if (!SetThreadPriority(static_cast(handle), win_priority)) { + if (!rt_unavailable_warned.exchange(true)) { + logger_.warn("Could not apply thread priority {} to task '{}'; running without elevated " + "priority", + win_priority, config_.name); + } + return false; + } + return true; +#else + // Unknown host platform: priorities are stored but not applied. + (void)handle; + (void)priority; + return false; +#endif +} +#endif // !ESP_PLATFORM + Task::Task(const Task::Config &config) : BaseComponent(config.task_config.name, config.log_level) , callback_(config.callback) - , config_(config.task_config) {} + , config_(config.task_config) + , priority_(config.task_config.priority) {} std::unique_ptr Task::make_unique(const Task::Config &config) { return std::make_unique(config); @@ -37,7 +172,16 @@ bool Task::start() { return false; } thread_config.stack_size = config_.stack_size_bytes; - thread_config.prio = config_.priority; + // clamp to the valid FreeRTOS priority range, exactly like set_priority(): + // an out-of-range configured value must not make startup fail + size_t start_priority = priority_.load(); + if (start_priority >= configMAX_PRIORITIES) { + logger_.warn("Configured priority ({}) >= configMAX_PRIORITIES ({}), clamping", start_priority, + configMAX_PRIORITIES); + start_priority = configMAX_PRIORITIES - 1; + priority_ = start_priority; + } + thread_config.prio = start_priority; // this will set the config for the next created thread auto err = esp_pthread_set_cfg(&thread_config); if (err == ESP_ERR_NO_MEM) { @@ -70,6 +214,12 @@ bool Task::start() { std::lock_guard lock(thread_mutex_); // create and start the std::thread thread_ = std::thread(&Task::thread_function, this); + // On ESP the priority was applied via esp_pthread above; on host platforms + // (when host_realtime is opted in) the new thread applies the priority to + // itself at the top of thread_function(), BEFORE the first callback + // invocation - applying it from here would race the thread's startup and a + // short callback could run entirely (or even exit) at the default + // priority. } logger_.debug("Task started"); return true; @@ -227,8 +377,10 @@ bool Task::set_priority(size_t priority) { priority = configMAX_PRIORITIES - 1; } #endif - // always store the new priority so it is used on the next start() - config_.priority = priority; + // always store the new priority so it is used on the next start() (atomic: + // the worker thread's startup priority application and + // get_configured_priority() read it concurrently) + priority_ = priority; logger_.debug("Set priority to {} for task '{}'", priority, config_.name); #if defined(ESP_PLATFORM) // if the task is running, apply the change to the live task as well @@ -237,6 +389,18 @@ bool Task::set_priority(size_t priority) { vTaskPrioritySet(handle, static_cast(priority)); return true; } +#else + // if the task is running and host real-time scheduling was opted in, apply + // the change to the live thread as well (best-effort; see + // BaseConfig::host_realtime for the per-platform semantics). Apply the + // freshest stored value under priority_apply_mutex_ so concurrent + // applications (including the worker's startup self-application) always + // converge on the last stored priority. + if (started_ && config_.host_realtime) { + std::lock_guard lock(thread_mutex_); + std::lock_guard apply_lock(priority_apply_mutex_); + return apply_thread_priority(thread_, priority_.load()); + } #endif return false; } @@ -290,6 +454,23 @@ std::string Task::get_info(const Task &task) { void Task::thread_function() { #if defined(ESP_PLATFORM) task_handle_ = get_current_id(); +#else + // Apply the (opted-in) host scheduling policy to ourselves before the first + // callback invocation, so the priority reliably covers the task's entire + // execution (best-effort: an unprivileged failure falls back to default + // scheduling with a one-time warning). + if (config_.host_realtime) { + // Serialize with a live set_priority() and read the stored priority + // INSIDE the lock: otherwise this startup application could overwrite a + // newer concurrent request with a stale value (thread runs at the old + // priority while get_configured_priority() reports the new one). + std::lock_guard apply_lock(priority_apply_mutex_); +#if defined(_WIN32) + apply_thread_priority_to_handle(GetCurrentThread(), priority_.load()); +#elif defined(__linux__) || defined(__APPLE__) + apply_thread_priority_to_handle(pthread_self(), priority_.load()); +#endif + } #endif // ESP_PLATFORM while (started_) { bool should_stop = false; diff --git a/components/thread_pool/README.md b/components/thread_pool/README.md index 95d45ce11c..cf4d0a54b1 100644 --- a/components/thread_pool/README.md +++ b/components/thread_pool/README.md @@ -13,4 +13,26 @@ It is implemented with `espp::Task` workers and `std::condition_variable` synchr - Optional blocking submit mode for backpressure - Manual `start()` / `stop()` control; `start()` returns `true` if all workers launched successfully (or the pool was already running), `false` if any worker failed to start and the pool was rolled back to stopped state - Graceful stop (stops workers; queued jobs may not be executed) -- Thread-safe stats for submitted / executed / rejected jobs +- Thread-safe stats for submitted / executed / rejected jobs (total and per band) +- **Priority bands** (`espp::QosBand`: `Critical` / `High` / `Normal` / `Low`): + `submit(job, band)` / `try_submit(job, band)` enqueue into one FIFO queue per + band and workers always pop the most urgent non-empty band first. The + band-less `submit(job)` overload uses `Normal`, so existing code is + unaffected. +- **Aging (anti-starvation)**: a queued job whose wait exceeds + `Config::aging_threshold` (default 100 ms) is promoted up one band (to the + back of the next band's queue), so a busy high band cannot starve lower + bands indefinitely. Set to 0 for strict band priority. +- **Per-band workers** (opt-in via `Config::band_worker_counts`): band *k* gets + its own workers at `Config::band_task_priorities[k]`, each servicing bands + 0..k (its own band and every more urgent band). The deepest (least urgent) + configured band's workers service *every* band, so no band is ever + unreachable — even with aging disabled. On ESP the priorities are + FreeRTOS task priorities and are always applied; on host platforms + (Linux/macOS) they map to `SCHED_FIFO` real-time priorities but are **only + applied when `Config::band_workers_realtime` is set** — by default the + workers run at the OS default scheduling and band ordering is enforced at + the queue level only. `SCHED_FIFO` on Linux requires `CAP_SYS_NICE` or an + `RLIMIT_RTPRIO` allowance (and a spinning job can starve the system — see + the `Config` docs); without permission the workers fall back gracefully to + default scheduling with a one-time warning. diff --git a/components/thread_pool/example/README.md b/components/thread_pool/example/README.md index a34dd22cbb..19a8620f35 100644 --- a/components/thread_pool/example/README.md +++ b/components/thread_pool/example/README.md @@ -15,6 +15,7 @@ operation to more advanced concurrent and multi-pool scenarios. | 7 | Concurrent submission | multi-thread `submit()` + `try_submit()` | | 8 | Chained pools | a job in `pool_a` submitting work into `pool_b` | | 9 | Self-submit | a running job submitting back to its own pool | +| 10 | Priority bands | `submit(job, QosBand)` — queued `Critical` jobs overtake queued `Low` jobs; per-band `stats()` counters | Each test logs individual `PASS` / `FAIL` results inline. At the end of the run a summary is printed: @@ -25,7 +26,7 @@ a summary is printed: PASS submit: normal dispatch + queue_size + stats ... ================================================= -9/9 tests passed +10/10 tests passed All tests passed! ``` diff --git a/components/thread_pool/example/main/thread_pool_example.cpp b/components/thread_pool/example/main/thread_pool_example.cpp index 3d33a0dfb8..0f9a757d34 100644 --- a/components/thread_pool/example/main/thread_pool_example.cpp +++ b/components/thread_pool/example/main/thread_pool_example.cpp @@ -22,7 +22,7 @@ extern "C" void app_main(void) { struct TestResult { std::string name; - bool passed; + bool passed{false}; }; std::vector results; @@ -49,23 +49,25 @@ extern "C" void app_main(void) { .worker_count = 3, .max_queue_size = 0, .auto_start = false, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); - passed &= check(name, !pool.is_running(), "pool should not be running before start()"); + passed &= check(name, !pool.is_running(), "pool should not be running before start()"); passed &= check(name, pool.worker_count() == 3, "worker_count() should be 3"); - passed &= check(name, pool.start(), "start() should return true on first call"); + passed &= check(name, pool.start(), "start() should return true on first call"); passed &= check(name, pool.is_running(), "pool should be running after start()"); - passed &= check(name, pool.start(), "start() should return true when already running (no-op)"); - passed &= check(name, pool.is_running(), "pool should still be running after duplicate start()"); + passed &= check(name, pool.start(), "start() should return true when already running (no-op)"); + passed &= + check(name, pool.is_running(), "pool should still be running after duplicate start()"); pool.stop(); passed &= check(name, !pool.is_running(), "pool should not be running after stop()"); @@ -92,12 +94,13 @@ extern "C" void app_main(void) { .worker_count = 2, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -111,16 +114,17 @@ extern "C" void app_main(void) { ++accepted_count; } } - passed &= check(name, accepted_count == total_jobs, "all jobs should be accepted (unbounded queue)"); + passed &= + check(name, accepted_count == total_jobs, "all jobs should be accepted (unbounded queue)"); wait_for_jobs(done_cv, done_mutex, completed_jobs, total_jobs); auto s = pool.stats(); logger.info(" stats: {}", s); passed &= check(name, s.submitted == total_jobs, "submitted count should equal total_jobs"); - passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); - passed &= check(name, s.rejected == 0, "rejected count should be 0"); - passed &= check(name, pool.queue_size() == 0, "queue should be empty after all jobs finish"); + passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); + passed &= check(name, s.rejected == 0, "rejected count should be 0"); + passed &= check(name, pool.queue_size() == 0, "queue should be empty after all jobs finish"); pool.stop(); //! [submit example] @@ -169,12 +173,13 @@ extern "C" void app_main(void) { .max_queue_size = 2, .auto_start = true, .block_on_submit_when_full = false, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -240,12 +245,13 @@ extern "C" void app_main(void) { .max_queue_size = 2, .auto_start = true, .block_on_submit_when_full = true, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -259,15 +265,16 @@ extern "C" void app_main(void) { ++accepted_count; } } - passed &= check(name, accepted_count == total_jobs, "all jobs should be accepted (blocking submit)"); + passed &= + check(name, accepted_count == total_jobs, "all jobs should be accepted (blocking submit)"); wait_for_jobs(done_cv, done_mutex, completed_jobs, total_jobs); auto s = pool.stats(); logger.info(" stats: {}", s); passed &= check(name, s.submitted == total_jobs, "submitted count should equal total_jobs"); - passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); - passed &= check(name, s.rejected == 0, "rejected count should be 0"); + passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); + passed &= check(name, s.rejected == 0, "rejected count should be 0"); pool.stop(); //! [blocking submit example] @@ -287,20 +294,21 @@ extern "C" void app_main(void) { .worker_count = 1, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); pool.stop(); bool accepted = pool.submit([]() {}); - passed &= check(name, !accepted, "submit() after stop() should return false"); + passed &= check(name, !accepted, "submit() after stop() should return false"); passed &= check(name, pool.stats().submitted == 0, "submitted count should be 0"); - passed &= check(name, pool.stats().rejected == 1, "rejected count should be 1"); + passed &= check(name, pool.stats().rejected == 1, "rejected count should be 1"); auto s = pool.stats(); logger.info(" stats: {}", s); @@ -321,12 +329,13 @@ extern "C" void app_main(void) { .worker_count = 2, .max_queue_size = 0, .auto_start = false, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -378,8 +387,7 @@ extern "C" void app_main(void) { constexpr int num_submit_threads = 3; constexpr int num_try_submit_threads = 2; constexpr int jobs_per_thread = 10; - constexpr int total_jobs = - (num_submit_threads + num_try_submit_threads) * jobs_per_thread; + constexpr int total_jobs = (num_submit_threads + num_try_submit_threads) * jobs_per_thread; std::atomic completed_jobs{0}; std::atomic total_accepted{0}; @@ -387,12 +395,13 @@ extern "C" void app_main(void) { .worker_count = 4, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -468,12 +477,13 @@ extern "C" void app_main(void) { .worker_count = 2, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "pool_b_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "pool_b_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -481,12 +491,13 @@ extern "C" void app_main(void) { .worker_count = 2, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "pool_a_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "pool_a_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -508,9 +519,9 @@ extern "C" void app_main(void) { auto sb = pool_b.stats(); logger.info(" pool_a stats: {}", sa); logger.info(" pool_b stats: {}", sb); - passed &= check(name, sa.executed == num_a_jobs, "pool_a should execute all A jobs"); + passed &= check(name, sa.executed == num_a_jobs, "pool_a should execute all A jobs"); passed &= check(name, sb.executed == total_b_jobs, "pool_b should execute all chained B jobs"); - passed &= check(name, sb.rejected == 0, "pool_b should not reject any jobs"); + passed &= check(name, sb.rejected == 0, "pool_b should not reject any jobs"); pool_a.stop(); pool_b.stop(); @@ -538,12 +549,13 @@ extern "C" void app_main(void) { .worker_count = 2, .max_queue_size = 0, .auto_start = true, - .worker_task_config = { - .name = "tp_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, .log_level = espp::Logger::Verbosity::WARN, }); @@ -575,6 +587,109 @@ extern "C" void app_main(void) { results.push_back({name, passed}); } + // ------------------------------------------------------------------------- + // 10. Priority bands: Critical jobs overtake queued Low jobs + // ------------------------------------------------------------------------- + { + const std::string name = "priority bands: Critical overtakes Low"; + //! [priority bands example] + logger.info("--- {} ---", name); + bool passed = true; + + // Gate the single worker on a barrier so the queue contents (and therefore + // the dispatch order) are fully deterministic. + std::mutex barrier_mutex; + std::condition_variable barrier_cv; + bool release_worker = false; + std::mutex started_mutex; + std::condition_variable started_cv; + std::atomic jobs_started{0}; + + std::mutex done_mutex; + std::condition_variable done_cv; + std::atomic completed_jobs{0}; + std::mutex order_mutex; + std::vector execution_order; + + espp::ThreadPool pool({ + .worker_count = 1, + .auto_start = true, + // 0 = strict band priority; the default (100ms) also promotes + // long-waiting jobs up one band to prevent starvation ("aging"). + .aging_threshold = std::chrono::milliseconds(0), + .worker_task_config = + { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + // Occupy the worker, then queue Low jobs followed by a Critical one. + pool.submit([&]() { + { + std::lock_guard lock(started_mutex); + ++jobs_started; + } + started_cv.notify_one(); + std::unique_lock lock(barrier_mutex); + barrier_cv.wait(lock, [&]() { return release_worker; }); + }); + { + std::unique_lock lock(started_mutex); + started_cv.wait(lock, [&]() { return jobs_started.load() >= 1; }); + } + + for (int i = 0; i < 2; ++i) { + pool.submit( + [&, i]() { + { + std::lock_guard lock(order_mutex); + execution_order.push_back("low" + std::to_string(i)); + } + ++completed_jobs; + done_cv.notify_one(); + }, + espp::QosBand::Low); + } + pool.submit( + [&]() { + { + std::lock_guard lock(order_mutex); + execution_order.push_back("critical"); + } + ++completed_jobs; + done_cv.notify_one(); + }, + espp::QosBand::Critical); + + // Release the worker: the Critical job must run before the queued Low jobs. + { + std::lock_guard lock(barrier_mutex); + release_worker = true; + } + barrier_cv.notify_all(); + wait_for_jobs(done_cv, done_mutex, completed_jobs, 3); + + { + std::lock_guard lock(order_mutex); + passed &= check(name, execution_order.size() == 3 && execution_order[0] == "critical", + "Critical job should run before the queued Low jobs"); + } + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.band_submitted[static_cast(espp::QosBand::Critical)] == 1, + "one job accounted to the Critical band"); + passed &= check(name, s.band_submitted[static_cast(espp::QosBand::Low)] == 2, + "two jobs accounted to the Low band"); + + pool.stop(); + //! [priority bands example] + results.push_back({name, passed}); + } + // ------------------------------------------------------------------------- // Summary // ------------------------------------------------------------------------- diff --git a/components/thread_pool/include/qos_band.hpp b/components/thread_pool/include/qos_band.hpp new file mode 100644 index 0000000000..569c3f7797 --- /dev/null +++ b/components/thread_pool/include/qos_band.hpp @@ -0,0 +1,30 @@ +#pragma once + +#include +#include + +namespace espp { + +/** + * @brief Bucketed priority band ("quality of service" class) for schedulable + * work such as espp::ThreadPool jobs and espp::SocketReactor + * registrations. + * + * Bands are deliberately coarse (four fixed buckets, Linux-scheduler style + * "bands" rather than fine-grained priorities): lower numeric value = more + * urgent. QosBand::Normal is the default everywhere and preserves the + * pre-band, single-FIFO-queue behavior when no other band is used. + */ +enum class QosBand : std::uint8_t { + Critical = 0, ///< Most urgent. Serviced before all other bands. + High = 1, ///< Urgent, but yields to Critical. + Normal = 2, ///< Default band; matches the pre-band FIFO behavior. + Low = 3, ///< Background / bulk work; serviced only when the more urgent + ///< bands are empty (subject to aging, see espp::ThreadPool). +}; + +/// Number of QosBand values; bands are indexed 0..kNumQosBands-1 (0 = most +/// urgent). +inline constexpr std::size_t kNumQosBands = 4; + +} // namespace espp diff --git a/components/thread_pool/include/thread_pool.hpp b/components/thread_pool/include/thread_pool.hpp index 4e3302a9e3..d4f0dd3145 100644 --- a/components/thread_pool/include/thread_pool.hpp +++ b/components/thread_pool/include/thread_pool.hpp @@ -1,6 +1,8 @@ #pragma once +#include #include +#include #include #include #include @@ -12,18 +14,63 @@ #include #include "base_component.hpp" +#include "qos_band.hpp" #include "task.hpp" namespace espp { /** - * @brief A thread pool that dispatches submitted jobs to a fixed set of worker threads. + * @brief A thread pool that dispatches submitted jobs to a fixed set of worker + * threads, with bucketed priority bands (espp::QosBand). * - * Workers are implemented as espp::Task instances. Jobs are queued and - * consumed in FIFO order. The queue can be optionally bounded; when full, - * new submissions are either rejected immediately or blocked until space + * Workers are implemented as espp::Task instances. Jobs are queued into one of + * four priority bands (QosBand::Critical/High/Normal/Low) and consumed in FIFO + * order within a band, with more urgent (lower-index) bands always drained + * first. Submitting without a band uses QosBand::Normal, which preserves the + * original single-FIFO-queue behavior exactly. The queue can be optionally + * bounded (Config::max_queue_size bounds the TOTAL across all bands); when + * full, new submissions are either rejected immediately or blocked until space * becomes available, depending on the configuration. * + * **Aging (starvation guard).** With strict band priority a continuous stream + * of urgent jobs could starve less urgent bands forever. To prevent this, when + * a worker looks for work it first promotes the front entry of any band whose + * wait time exceeds Config::aging_threshold up one band (to the back of the + * next more-urgent band, with its aging clock restarted). Aging is + * deliberately approximate: only band fronts are examined (O(bands) per pop, + * no full-queue scans), and because bands are FIFO this is sufficient - the + * front is always the longest-waiting entry of its band. aging_threshold is + * an ELIGIBILITY interval, not an at-most wait bound: promotion is evaluated + * only when a worker next dequeues work, so a hop can happen arbitrarily + * later than the threshold if every worker is stuck in a long in-flight job, + * and after promotion the entry still waits behind the destination band's + * backlog at that moment. What IS guaranteed: since promoted entries enter + * ahead of all later arrivals, every queued entry makes progress toward + * Critical under any sustained load, so nothing starves. Set aging_threshold + * to 0 for strict (starvation-permitting) band priority. + * + * **Worker bands (true OS preemption, opt-in).** By default all + * Config::worker_count workers are identical and service every band. Setting + * any element of Config::band_worker_counts non-zero switches to per-band + * workers instead: band_worker_counts[k] workers are created for band k, each + * running at the espp::Task priority Config::band_task_priorities[k] (a + * FreeRTOS priority on ESP; mapped to a SCHED_FIFO real-time priority on + * Linux/macOS - see espp::Task::BaseConfig::priority). A band-k worker + * services bands 0..k, i.e. its own band and every MORE urgent band. This + * means a Critical job can be picked up by any worker, while a band-k worker + * never runs work less urgent than band k - with one exception: the deepest + * (least urgent) configured band's workers service EVERY band, so that no + * band is unreachable when the configuration has no Low-band worker (see + * Config::band_worker_counts). The latency guarantee this buys: + * because every worker drains band 0 first and the band-0 workers run at the + * highest OS priority, a newly arrived Critical job - WHEN the Critical queue + * is otherwise empty - waits at most the remaining duration of one + * already-running job before a high-OS-priority worker picks it up (and on a + * preemptive OS - e.g. FreeRTOS or Linux PREEMPT_RT with granted RT + * scheduling - that worker preempts lower-priority ones the moment it becomes + * runnable). With a Critical backlog the new job additionally waits behind + * the earlier Critical jobs, which drain FIFO across all workers first. + * * \section thread_pool_ex1 Lifecycle: start / stop / is_running / worker_count * \snippet thread_pool_example.cpp lifecycle example * \section thread_pool_ex2 Submit Jobs @@ -42,29 +89,88 @@ namespace espp { * \snippet thread_pool_example.cpp chained pools example * \section thread_pool_ex9 Self-Submit * \snippet thread_pool_example.cpp self-submit example + * \section thread_pool_ex10 Priority Bands + * \snippet thread_pool_example.cpp priority bands example */ class ThreadPool : public espp::BaseComponent { public: /// @brief A callable job that can be submitted to the pool. using Job = std::function; + /// @brief Number of priority bands (see espp::QosBand). + static constexpr std::size_t kNumBands = kNumQosBands; + /// @brief Snapshot of pool activity counters. struct Stats { std::uint64_t submitted = 0; ///< Total jobs accepted into the queue. - std::uint64_t executed = 0; ///< Total jobs successfully executed. + std::uint64_t executed = 0; ///< Total jobs executed (counted when a worker begins executing + ///< the job, so a job's own side effects always observe it). std::uint64_t rejected = 0; ///< Total jobs rejected (invalid job, stopped/stopping, or queue ///< full) or dropped (due to stop, the enqueued jobs were dropped). + std::array band_submitted{}; ///< Jobs accepted per band (by the band + ///< they were submitted to). + std::array band_executed{}; ///< Jobs executed per band (by the band + ///< they were popped from, i.e. after + ///< any aging promotions). + std::array band_aged{}; ///< Aging promotions OUT of each band (an + ///< entry moved from band i to band i-1). + std::array band_rejected{}; ///< Jobs rejected per band (by the band + ///< they were submitted to; on stop(), + ///< dropped queued jobs count against + ///< the band they were queued in). }; /// @brief Configuration parameters for constructing a ThreadPool. struct Config { - std::size_t worker_count = 1; ///< Number of worker threads to spawn. - std::size_t max_queue_size = 0; ///< Maximum pending jobs (0 = unbounded). + std::size_t worker_count = 1; ///< Number of worker threads to spawn (ignored when + ///< band_worker_counts is set - see below). + std::size_t max_queue_size = 0; ///< Maximum pending jobs TOTAL across all bands (0 = + ///< unbounded). bool auto_start = true; ///< Start workers immediately on construction. bool block_on_submit_when_full = false; ///< If true, submit() blocks when the queue is full instead of rejecting. + std::chrono::milliseconds aging_threshold{ + 100}; ///< Starvation guard: a queued job whose wait exceeds this is promoted up one band + ///< (approximate, front-of-band only - see the class docs). 0 disables aging + ///< (strict band priority). + std::array band_worker_counts{}; ///< Opt-in per-band worker counts + ///< (index = QosBand). All zero (the + ///< default) = disabled: worker_count + ///< identical workers service all + ///< bands. When any element is + ///< non-zero, band k gets + ///< band_worker_counts[k] workers at + ///< band_task_priorities[k], each + ///< servicing bands 0..k. The + ///< deepest configured band's + ///< workers service ALL bands (with + ///< a warning if that band is not + ///< Low), so every band is always + ///< reachable even with + ///< aging_threshold == 0. + std::array band_task_priorities{ + 10, 7, 5, 1}; ///< espp::Task priorities for per-band workers (only used when + ///< band_worker_counts is set). Defaults descend from Critical to Low. On + ///< ESP these are FreeRTOS priorities and are always applied; on host + ///< platforms (Linux/macOS) they map to SCHED_FIFO real-time priorities + ///< (see espp::Task::BaseConfig::priority) and are only applied when + ///< band_workers_realtime is set. + bool band_workers_realtime = + false; ///< Opt-in for OS real-time scheduling of per-band workers on HOST platforms + ///< (sets espp::Task::BaseConfig::host_realtime on each worker). When false (the + ///< default), host per-band workers run at default (non-realtime) + ///< scheduling: band ordering is still honored at the queue level (workers pop + ///< the most urgent band first), but the OS scheduler does not preempt in favor + ///< of the more urgent bands' workers. When true, band_task_priorities are + ///< applied as SCHED_FIFO real-time priorities. + ///< @warning SCHED_FIFO workers can starve the rest of the system if a job spins; + ///< on Linux this additionally requires CAP_SYS_NICE or an RLIMIT_RTPRIO + ///< allowance (e.g. under PREEMPT_RT), otherwise the Task falls back to default + ///< scheduling with a one-time warning. Ignored on ESP, where FreeRTOS + ///< priorities are always applied. espp::Task::BaseConfig worker_task_config = { - ///< Base configuration applied to every worker task. + ///< Base configuration applied to every worker task. (For per-band workers the priority + ///< field is overridden by band_task_priorities.) .name = "thread_pool_worker", .stack_size_bytes = 4096, .priority = 5, @@ -94,7 +200,7 @@ class ThreadPool : public espp::BaseComponent { /// @return true if workers are active, false otherwise. bool is_running() const; - /// @brief Submit a job, optionally blocking when the queue is full. + /// @brief Submit a job at QosBand::Normal, optionally blocking when the queue is full. /// /// Blocks if Config::block_on_submit_when_full is true and the queue has /// reached its capacity limit. Otherwise behaves identically to try_submit(). @@ -102,14 +208,31 @@ class ThreadPool : public espp::BaseComponent { /// @return true if the job was accepted, false if it was rejected. bool submit(Job &&job); - /// @brief Attempt to submit a job without blocking. + /// @brief Submit a job at the given priority band, optionally blocking when the queue is full. + /// + /// Blocks if Config::block_on_submit_when_full is true and the queue has + /// reached its capacity limit. Otherwise behaves identically to try_submit(). + /// @param job Callable to enqueue; moved into the queue on acceptance. + /// @param band Priority band to enqueue the job at (see espp::QosBand). + /// @return true if the job was accepted, false if it was rejected. + bool submit(Job &&job, QosBand band); + + /// @brief Attempt to submit a job at QosBand::Normal without blocking. /// /// Returns immediately with false when the queue is full. /// @param job Callable to enqueue; moved into the queue on acceptance. /// @return true if the job was accepted, false if it was rejected. bool try_submit(Job &&job); - /// @brief Return the number of jobs currently waiting in the queue. + /// @brief Attempt to submit a job at the given priority band without blocking. + /// + /// Returns immediately with false when the queue is full. + /// @param job Callable to enqueue; moved into the queue on acceptance. + /// @param band Priority band to enqueue the job at (see espp::QosBand). + /// @return true if the job was accepted, false if it was rejected. + bool try_submit(Job &&job, QosBand band); + + /// @brief Return the number of jobs currently waiting in the queue (all bands). /// @return Pending job count. std::size_t queue_size() const; @@ -118,21 +241,34 @@ class ThreadPool : public espp::BaseComponent { std::size_t worker_count() const; /// @brief Return a snapshot of the pool's activity counters. - /// @return Stats struct with submitted, executed, and rejected counts. + /// @return Stats struct with total and per-band submitted / executed / rejected / aged counts. Stats stats() const; private: - bool worker_task_fn(); + /// @brief An enqueued job plus its enqueue timestamp (for aging). + struct Entry { + Job job; + std::chrono::steady_clock::time_point enqueued_at{}; + }; + + bool worker_task_fn(std::size_t max_band); + + bool submit_impl(Job &&job, QosBand band, bool allow_blocking_when_full); - bool submit_impl(Job &&job, bool allow_blocking_when_full); + /// Promote the front entry of any band whose wait exceeds the aging + /// threshold up one band. Must be called with queue_mutex_ held. + /// @return true if any entry was promoted. + bool age_bands_locked(); Config config_; + bool per_band_workers_{false}; ///< True when Config::band_worker_counts is in use. std::mutex lifecycle_mutex_; mutable std::mutex queue_mutex_; std::condition_variable queue_has_work_cv_; std::condition_variable queue_has_space_cv_; - std::deque queue_; + std::array, kNumBands> queues_; ///< One FIFO per band; index = QosBand. + std::size_t total_queued_{0}; ///< Sum of all band queue sizes. std::vector> workers_; std::atomic running_{false}; @@ -141,6 +277,10 @@ class ThreadPool : public espp::BaseComponent { std::atomic submitted_{0}; std::atomic executed_{0}; std::atomic rejected_{0}; + std::array, kNumBands> band_submitted_{}; + std::array, kNumBands> band_executed_{}; + std::array, kNumBands> band_aged_{}; + std::array, kNumBands> band_rejected_{}; }; } // namespace espp diff --git a/components/thread_pool/include/thread_pool_format_helpers.hpp b/components/thread_pool/include/thread_pool_format_helpers.hpp index f301b10fb5..174c043806 100644 --- a/components/thread_pool/include/thread_pool_format_helpers.hpp +++ b/components/thread_pool/include/thread_pool_format_helpers.hpp @@ -11,7 +11,14 @@ template <> struct fmt::formatter { template auto format(espp::ThreadPool::Stats const &s, FormatContext &ctx) const { return fmt::format_to(ctx.out(), - "ThreadPool::Stats{{submitted: {}, executed: {}, rejected: {}}}", - s.submitted, s.executed, s.rejected); + "ThreadPool::Stats{{submitted: {}, executed: {}, rejected: {}, " + "band_submitted: [{}, {}, {}, {}], band_executed: [{}, {}, {}, {}], " + "band_aged: [{}, {}, {}, {}], band_rejected: [{}, {}, {}, {}]}}", + s.submitted, s.executed, s.rejected, s.band_submitted[0], + s.band_submitted[1], s.band_submitted[2], s.band_submitted[3], + s.band_executed[0], s.band_executed[1], s.band_executed[2], + s.band_executed[3], s.band_aged[0], s.band_aged[1], s.band_aged[2], + s.band_aged[3], s.band_rejected[0], s.band_rejected[1], + s.band_rejected[2], s.band_rejected[3]); } }; diff --git a/components/thread_pool/src/thread_pool.cpp b/components/thread_pool/src/thread_pool.cpp index d42a0ba38f..90202dc852 100644 --- a/components/thread_pool/src/thread_pool.cpp +++ b/components/thread_pool/src/thread_pool.cpp @@ -1,24 +1,77 @@ #include "thread_pool.hpp" +#include #include using namespace espp; ThreadPool::ThreadPool(const Config &config) - : BaseComponent("ThreadPool", config.log_level), config_(config) { - if (config_.worker_count == 0) { - config_.worker_count = 1; - } - - workers_.reserve(config_.worker_count); - for (std::size_t i = 0; i < config_.worker_count; ++i) { - auto worker_config = config_.worker_task_config; - worker_config.name = config_.worker_task_config.name + "_" + std::to_string(i); - workers_.emplace_back(espp::Task::make_unique({ - .callback = [this]() { return worker_task_fn(); }, - .task_config = worker_config, - .log_level = config_.log_level, - })); + : BaseComponent("ThreadPool", config.log_level) + , config_(config) { + per_band_workers_ = + std::any_of(config_.band_worker_counts.begin(), config_.band_worker_counts.end(), + [](std::size_t count) { return count > 0; }); + + if (per_band_workers_) { + // Per-band workers: band k gets band_worker_counts[k] workers at + // band_task_priorities[k], each servicing bands 0..k (its own band and + // every more urgent band). +#if !defined(ESP_PLATFORM) + // On host platforms Task priority >= 1 maps to SCHED_FIFO real-time + // scheduling, which can starve the system - only apply it when explicitly + // opted in. Band ordering is still honored at the queue level regardless + // (workers pop the most urgent band first). + if (!config_.band_workers_realtime) { + logger_.info("Per-band workers running without OS real-time priorities (band ordering is " + "queue-level only); set Config::band_workers_realtime to opt in to SCHED_FIFO"); + } +#endif + // Guarantee every band is reachable: the deepest (least urgent) configured + // band's workers service ALL bands, not just 0..k. Otherwise a + // configuration like {1,0,0,0} would leave Normal/Low submissions queued + // forever (with aging_threshold == 0 nothing would ever promote them, and + // a bounded blocking submit could then deadlock). + std::size_t deepest_band = 0; + for (std::size_t band = 0; band < kNumBands; ++band) { + if (config_.band_worker_counts[band] > 0) { + deepest_band = band; + } + } + if (deepest_band != kNumBands - 1) { + logger_.warn("No Low-band worker configured; the band-{} workers will service every band so " + "no submission is unreachable", + deepest_band); + } + for (std::size_t band = 0; band < kNumBands; ++band) { + for (std::size_t i = 0; i < config_.band_worker_counts[band]; ++i) { + auto worker_config = config_.worker_task_config; + worker_config.name = + config_.worker_task_config.name + "_b" + std::to_string(band) + "_" + std::to_string(i); + worker_config.priority = config_.band_task_priorities[band]; + // Only meaningful on host (ESP always applies the FreeRTOS priority). + worker_config.host_realtime = config_.band_workers_realtime; + const std::size_t max_band = (band == deepest_band) ? (kNumBands - 1) : band; + workers_.emplace_back(espp::Task::make_unique({ + .callback = [this, max_band]() { return worker_task_fn(max_band); }, + .task_config = worker_config, + .log_level = config_.log_level, + })); + } + } + } else { + if (config_.worker_count == 0) { + config_.worker_count = 1; + } + workers_.reserve(config_.worker_count); + for (std::size_t i = 0; i < config_.worker_count; ++i) { + auto worker_config = config_.worker_task_config; + worker_config.name = config_.worker_task_config.name + "_" + std::to_string(i); + workers_.emplace_back(espp::Task::make_unique({ + .callback = [this]() { return worker_task_fn(kNumBands - 1); }, + .task_config = worker_config, + .log_level = config_.log_level, + })); + } } if (config_.auto_start) { @@ -70,8 +123,12 @@ void ThreadPool::stop() { { std::lock_guard lock(queue_mutex_); - rejected_ += static_cast(queue_.size()); - queue_.clear(); + rejected_ += static_cast(total_queued_); + for (std::size_t band = 0; band < kNumBands; ++band) { + band_rejected_[band] += static_cast(queues_[band].size()); + queues_[band].clear(); + } + total_queued_ = 0; } queue_has_work_cv_.notify_all(); @@ -85,80 +142,191 @@ void ThreadPool::stop() { bool ThreadPool::is_running() const { return running_.load(); } bool ThreadPool::submit(Job &&job) { - return submit_impl(std::move(job), config_.block_on_submit_when_full); + return submit_impl(std::move(job), QosBand::Normal, config_.block_on_submit_when_full); +} + +bool ThreadPool::submit(Job &&job, QosBand band) { + return submit_impl(std::move(job), band, config_.block_on_submit_when_full); +} + +bool ThreadPool::try_submit(Job &&job) { + return submit_impl(std::move(job), QosBand::Normal, false); } -bool ThreadPool::try_submit(Job &&job) { return submit_impl(std::move(job), false); } +bool ThreadPool::try_submit(Job &&job, QosBand band) { + return submit_impl(std::move(job), band, false); +} + +bool ThreadPool::submit_impl(Job &&job, QosBand band, bool allow_blocking_when_full) { + auto band_index = static_cast(band); + if (band_index >= kNumBands) { + logger_.warn("Invalid band {}, clamping to Low", band_index); + band_index = static_cast(QosBand::Low); + } -bool ThreadPool::submit_impl(Job &&job, bool allow_blocking_when_full) { if (!job) { rejected_++; + band_rejected_[band_index]++; return false; } std::unique_lock lock(queue_mutex_); if (!running_.load() || stopping_) { rejected_++; + band_rejected_[band_index]++; return false; } if (config_.max_queue_size > 0) { if (allow_blocking_when_full) { - queue_has_space_cv_.wait(lock, [&]() { - return stopping_ || queue_.size() < config_.max_queue_size; - }); + queue_has_space_cv_.wait( + lock, [&]() { return stopping_ || total_queued_ < config_.max_queue_size; }); if (stopping_) { rejected_++; + band_rejected_[band_index]++; return false; } - } else if (queue_.size() >= config_.max_queue_size) { + } else if (total_queued_ >= config_.max_queue_size) { rejected_++; + band_rejected_[band_index]++; return false; } } - queue_.push_back(std::move(job)); + queues_[band_index].push_back(Entry{std::move(job), std::chrono::steady_clock::now()}); + ++total_queued_; submitted_++; + band_submitted_[band_index]++; lock.unlock(); - queue_has_work_cv_.notify_one(); + if (per_band_workers_) { + // Workers are heterogeneous (each sees only bands 0..k); notify_one could + // wake a worker that cannot service this band, so wake them all. + queue_has_work_cv_.notify_all(); + } else { + queue_has_work_cv_.notify_one(); + } return true; } std::size_t ThreadPool::queue_size() const { std::lock_guard lock(queue_mutex_); - return queue_.size(); + return total_queued_; } std::size_t ThreadPool::worker_count() const { return workers_.size(); } ThreadPool::Stats ThreadPool::stats() const { - return { + Stats s{ .submitted = submitted_.load(), .executed = executed_.load(), .rejected = rejected_.load(), }; + for (std::size_t band = 0; band < kNumBands; ++band) { + s.band_submitted[band] = band_submitted_[band].load(); + s.band_executed[band] = band_executed_[band].load(); + s.band_aged[band] = band_aged_[band].load(); + s.band_rejected[band] = band_rejected_[band].load(); + } + return s; +} + +bool ThreadPool::age_bands_locked() { + bool promoted = false; + const auto now = std::chrono::steady_clock::now(); + for (std::size_t band = 1; band < kNumBands; ++band) { + if (queues_[band].empty()) { + continue; + } + // Bands are FIFO, so the front is always the longest-waiting entry of its + // band - checking only the front is sufficient (and O(bands) per pop). + if (now - queues_[band].front().enqueued_at < config_.aging_threshold) { + continue; + } + Entry entry = std::move(queues_[band].front()); + queues_[band].pop_front(); + // Restart the aging clock: an entry ages up at most one band per + // aging_threshold interval. It enters the more urgent band at the back, + // ahead of all of that band's later arrivals (FIFO), which is what + // guarantees progress under sustained load. + entry.enqueued_at = now; + queues_[band - 1].push_back(std::move(entry)); + band_aged_[band]++; + promoted = true; + } + return promoted; } -bool ThreadPool::worker_task_fn() { +bool ThreadPool::worker_task_fn(std::size_t max_band) { Job job; + std::size_t band = 0; + bool promoted = false; + const bool aging_enabled = config_.aging_threshold.count() > 0; { std::unique_lock lock(queue_mutex_); - queue_has_work_cv_.wait(lock, [&]() { return stopping_ || !queue_.empty(); }); + auto have_visible_work = [&]() { + const auto visible_end = queues_.begin() + static_cast(max_band) + 1; + return std::any_of(queues_.begin(), visible_end, + [](const auto &queue) { return !queue.empty(); }); + }; + auto pred = [&]() { return stopping_.load() || have_visible_work(); }; + if (aging_enabled && per_band_workers_) { + // A band-restricted worker cannot see the less urgent bands, but it is + // still responsible for aging them (promotion is what eventually makes a + // starving entry visible to some worker); wake periodically to do so. + queue_has_work_cv_.wait_for(lock, config_.aging_threshold, pred); + } else { + queue_has_work_cv_.wait(lock, pred); + } if (stopping_) { return true; } - job = std::move(queue_.front()); - queue_.pop_front(); + if (aging_enabled) { + promoted = age_bands_locked(); + } + + // Take from the most urgent (lowest index) non-empty band this worker + // services. + std::size_t found_band = kNumBands; + for (std::size_t b = 0; b <= max_band; ++b) { + if (!queues_[b].empty()) { + found_band = b; + break; + } + } + if (found_band == kNumBands) { + // Nothing visible (e.g. a timed wakeup purely to age other bands). + if (promoted && per_band_workers_) { + lock.unlock(); + queue_has_work_cv_.notify_all(); + } + return false; + } + band = found_band; + job = std::move(queues_[band].front().job); + queues_[band].pop_front(); + --total_queued_; + } + + if (promoted && per_band_workers_) { + // A promotion may have made work visible to a sleeping band-restricted + // worker that submit() targeted at a band it cannot see. + queue_has_work_cv_.notify_all(); } if (config_.max_queue_size > 0) { queue_has_space_cv_.notify_one(); } - job(); + // Count the job as executed BEFORE invoking it: anything the job makes + // observable (e.g. signalling a waiter that then reads stats()) must already + // see this job accounted for. Counting after the call would race with such + // observers (the worker can be descheduled between the job body finishing + // and the increment - especially when workers run at real-time priority). executed_++; + band_executed_[band]++; + job(); return false; } diff --git a/components/xiao-esp32s3-sense/example/CMakeLists.txt b/components/xiao-esp32s3-sense/example/CMakeLists.txt index 8a80fe5628..3edc629cbf 100644 --- a/components/xiao-esp32s3-sense/example/CMakeLists.txt +++ b/components/xiao-esp32s3-sense/example/CMakeLists.txt @@ -14,13 +14,14 @@ set(EXTRA_COMPONENT_DIRS "../../../components/rtsp" "../../../components/socket" "../../../components/task" + "../../../components/thread_pool" "../../../components/wifi" "../../../components/xiao-esp32s3-sense" ) set( COMPONENTS - "main esptool_py cli esp32-camera mdns monitor rtsp task wifi xiao-esp32s3-sense" + "main esptool_py cli esp32-camera mdns monitor rtsp task thread_pool wifi xiao-esp32s3-sense" CACHE STRING "List of components to include" ) diff --git a/doc/Doxyfile b/doc/Doxyfile index 2a21ae9ff2..f8df001064 100755 --- a/doc/Doxyfile +++ b/doc/Doxyfile @@ -415,10 +415,11 @@ INPUT = \ $(PROJECT_PATH)/components/serialization/include/serialization.hpp \ $(PROJECT_PATH)/components/seeed-studio-round-display/include/seeed-studio-round-display.hpp \ $(PROJECT_PATH)/components/smartpanlee-sc01-plus/include/smartpanlee-sc01-plus.hpp \ + $(PROJECT_PATH)/components/socket/include/dscp.hpp \ $(PROJECT_PATH)/components/socket/include/socket.hpp \ - $(PROJECT_PATH)/components/socket/include/udp_socket.hpp \ - $(PROJECT_PATH)/components/socket/include/tcp_socket.hpp \ $(PROJECT_PATH)/components/socket/include/socket_reactor.hpp \ + $(PROJECT_PATH)/components/socket/include/tcp_socket.hpp \ + $(PROJECT_PATH)/components/socket/include/udp_socket.hpp \ $(PROJECT_PATH)/components/spi/include/spi.hpp \ $(PROJECT_PATH)/components/st25dv/include/st25dv.hpp \ $(PROJECT_PATH)/components/st7123touch/include/st7123touch.hpp \ @@ -434,6 +435,7 @@ INPUT = \ $(PROJECT_PATH)/components/task/include/task.hpp \ $(PROJECT_PATH)/components/task/include/run_on_core.hpp \ $(PROJECT_PATH)/components/thermistor/include/thermistor.hpp \ + $(PROJECT_PATH)/components/thread_pool/include/qos_band.hpp \ $(PROJECT_PATH)/components/thread_pool/include/thread_pool.hpp \ $(PROJECT_PATH)/components/timer/include/high_resolution_timer.hpp \ $(PROJECT_PATH)/components/timer/include/timer.hpp \ diff --git a/doc/en/core/task.rst b/doc/en/core/task.rst index f6be0369be..bb20fa21ec 100644 --- a/doc/en/core/task.rst +++ b/doc/en/core/task.rst @@ -13,6 +13,27 @@ It also supports firing off syncrhonous (blocking) and asynchronous (non-blocking) functions in separate threads, with the option of configuring the core id and other esp-specific paramters. +Priority semantics per platform +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +On ESP, :cpp:member:`espp::Task::BaseConfig::priority` is the FreeRTOS task +priority (clamped to ``configMAX_PRIORITIES - 1``) and is always applied. On +host platforms (Linux / macOS / Windows) the priority is stored, but only +applied to the OS thread when :cpp:member:`espp::Task::BaseConfig::host_realtime` +is set (default ``false``, preserving the historical behavior of running at the +OS default scheduling). With the opt-in, Linux and macOS map priority ≥ 1 +linearly onto the ``SCHED_FIFO`` real-time priority range (priority 0 resets to +the default ``SCHED_OTHER`` scheduler), and Windows maps it best-effort onto +``SetThreadPriority()`` classes. + +.. warning:: + + A ``SCHED_FIFO`` thread that spins can starve the rest of the system. On + Linux, real-time scheduling requires ``CAP_SYS_NICE`` or an + ``RLIMIT_RTPRIO`` allowance (and delivers hard preemption on ``PREEMPT_RT`` + kernels); without permission the task falls back gracefully to default + scheduling with a one-time warning. + Code examples for the task API are provided in the `task` example folder. .. ------------------------------- Example ------------------------------------- diff --git a/doc/en/core/thread_pool.rst b/doc/en/core/thread_pool.rst index f5ce0d47ef..9a5d3b71bf 100644 --- a/doc/en/core/thread_pool.rst +++ b/doc/en/core/thread_pool.rst @@ -6,12 +6,48 @@ ThreadPool The :cpp:class:`espp::ThreadPool` component provides a reusable pool of worker tasks for executing queued jobs asynchronously. Workers are implemented as -:cpp:class:`espp::Task` instances and pull work from an internal job queue -(backed by ``std::deque``) whose maximum size is optionally enforced by +:cpp:class:`espp::Task` instances and pull work from internal job queues whose +combined maximum size is optionally enforced by :cpp:member:`espp::ThreadPool::Config::max_queue_size`. Submissions can either reject immediately when the queue is full or block until space is available, depending on configuration. +Priority bands +^^^^^^^^^^^^^^ + +Jobs can be submitted at one of four :cpp:enum:`espp::QosBand` priority bands — +``Critical`` / ``High`` / ``Normal`` / ``Low`` — via +``submit(job, band)`` / ``try_submit(job, band)``. Internally the pool keeps one +FIFO queue per band and workers always pop the most urgent non-empty band +first; the band-less ``submit(job)`` overload uses ``Normal``, so code that does +not use bands behaves exactly as before. Per-band submitted / executed / aged / rejected +counters are reported through :cpp:member:`espp::ThreadPool::Stats`. + +To keep a busy high band from starving lower bands, a queued job whose wait +exceeds :cpp:member:`espp::ThreadPool::Config::aging_threshold` (default 100 ms) +is *aged*: promoted up one band, to the back of that band's queue. Setting the +threshold to 0 disables aging (strict band priority). + +Per-band workers +^^^^^^^^^^^^^^^^ + +By default all workers are identical and serve every band. Setting +:cpp:member:`espp::ThreadPool::Config::band_worker_counts` opts into dedicated +per-band workers: band *k* gets its own workers running at +:cpp:member:`espp::ThreadPool::Config::band_task_priorities` [k], each servicing +bands 0..k (its own band and every more urgent band), so a ``Critical`` job +never waits behind more than one in-flight lower-band job. The deepest (least +urgent) configured band's workers service *every* band, so no band is ever +unreachable — even with aging disabled. + +On ESP the per-band priorities are FreeRTOS task priorities and are always +applied. On host platforms (Linux / macOS) they map onto ``SCHED_FIFO`` +real-time priorities, but are **only applied when** +:cpp:member:`espp::ThreadPool::Config::band_workers_realtime` **is set** — by +default host workers run at the OS default scheduling and band ordering is +enforced at the queue level only. See the :cpp:class:`espp::Task` documentation +for the host real-time scheduling requirements and caveats. + Code examples for the thread pool API are provided in the ``thread_pool`` example folder. @@ -27,3 +63,4 @@ API Reference ------------- .. include-build-file:: inc/thread_pool.inc +.. include-build-file:: inc/qos_band.inc diff --git a/doc/en/network/socket_reactor.rst b/doc/en/network/socket_reactor.rst index 3c2955abea..e1777064fa 100644 --- a/doc/en/network/socket_reactor.rst +++ b/doc/en/network/socket_reactor.rst @@ -33,6 +33,29 @@ The reactor drives: A low-level ``add_fd()`` / ``remove()`` pair is also available. +Priority bands and DSCP +----------------------- + +Each registration carries an :cpp:enum:`espp::QosBand` (``Critical`` / ``High`` +/ ``Normal`` / ``Low``; ``Normal`` by default, preserving the pre-band FIFO +behavior): when several sockets are readable in one ``select()`` round the +ready set is dispatched most-urgent-first, and each handler is submitted to the +:doc:`ThreadPool <../core/thread_pool>` *at its band*, so a ``Critical`` +socket's handler overtakes already-queued lower-band handlers even on a +saturated pool. ``UdpSocket::ReceiveConfig::band`` sets the band for UDP +receivers; ``add_tcp_listener()`` / ``add_tcp_stream()`` / ``add_fd()`` take a +band argument. + +UDP receivers can additionally set ``UdpSocket::ReceiveConfig::dscp`` to mark +their *transmitted* replies with a DSCP code point (applied as ``IP_TOS`` at +registration, best-effort), using the typed :cpp:enum:`espp::Dscp` enum of +standard DiffServ names - e.g. ``Dscp::Ef`` (expedited forwarding for +latency-critical flows), ``Dscp::Cs1`` (low-priority data), ``Dscp::Af41`` +(high-priority assured forwarding). This affects network / driver treatment of +outgoing traffic and is orthogonal to the local ``band`` scheduling. A custom +code point can be expressed with ``static_cast(0-63)``; out-of-range +values are rejected with a warning. + .. note:: Lifetime: registered sockets and callbacks must outlive their registration. @@ -52,7 +75,7 @@ A low-level ``add_fd()`` / ``remove()`` pair is also available. .. ------------------------------- Example ------------------------------------- Code examples for the reactor are provided in the ``socket`` example folder (the -"Socket reactor" and "TCP reactor" scenarios). +"Socket reactor", "Reactor priority bands", and "TCP reactor" scenarios). .. ---------------------------- API Reference ---------------------------------- @@ -60,3 +83,4 @@ API Reference ------------- .. include-build-file:: inc/socket_reactor.inc +.. include-build-file:: inc/dscp.inc diff --git a/lib/autogenerate_bindings.py b/lib/autogenerate_bindings.py index 09d2e8b07a..0cd3ee9adf 100644 --- a/lib/autogenerate_bindings.py +++ b/lib/autogenerate_bindings.py @@ -454,6 +454,11 @@ def autogenerate() -> None: include_dir + "math/include/vector2d.hpp", # have to set class template options include_dir + "ndef/include/ndef.hpp", include_dir + "pid/include/pid.hpp", + # NOTE: must come before socket / thread_pool: their bindings use QosBand + # values as default arguments (defaults are converted at def time). + include_dir + "thread_pool/include/qos_band.hpp", + + include_dir + "socket/include/dscp.hpp", include_dir + "socket/include/socket.hpp", include_dir + "socket/include/tcp_socket.hpp", include_dir + "socket/include/udp_socket.hpp", diff --git a/lib/python_bindings/espp/__init__.pyi b/lib/python_bindings/espp/__init__.pyi index 85ee11ac2c..4aa00bb521 100644 --- a/lib/python_bindings/espp/__init__.pyi +++ b/lib/python_bindings/espp/__init__.pyi @@ -2,7 +2,9 @@ # mypy: disable-error-code="type-arg" -from typing import overload, List +import datetime +import enum +from typing import overload, List, Optional NumberType = (int, float, np.number) @@ -2731,6 +2733,60 @@ class Pid: #################### #################### +#################### #################### + +class QosBand(enum.IntEnum): + """* + * @brief Priority band for queued work. Critical is the most urgent and Low the + * least; Normal is the default for band-less submissions. + + """ + Critical = enum.auto() # (= 0) #*< Most urgent. + High = enum.auto() # (= 1) + Normal = enum.auto() # (= 2) #*< Default for band-less submissions. + Low = enum.auto() # (= 3) #*< Least urgent. + +#################### #################### + + +#################### #################### + +class Dscp(enum.IntEnum): + """* + * @brief Standard DiffServ code points (DSCP) for IP traffic marking - the 6-bit field + * in the IP TOS / Traffic Class byte (RFC 2474). E.g. Dscp.Ef = expedited forwarding + * for latency-critical flows; Dscp.Cs1 = low-priority data; Dscp.Af41 = high-priority + * assured forwarding with low drop probability. + + """ + Cs0 = 0 #*< Class selector 0 - default / best-effort forwarding. + Default = 0 #*< Alias for CS0. + Le = 1 #*< Lower effort / scavenger (RFC 8622). + Cs1 = 8 #*< Class selector 1 (conventionally low-priority data). + Af11 = 10 #*< Assured forwarding: class 1, low drop precedence. + Af12 = 12 #*< Assured forwarding: class 1, medium drop precedence. + Af13 = 14 #*< Assured forwarding: class 1, high drop precedence. + Cs2 = 16 #*< Class selector 2 (conventionally OAM / management). + Af21 = 18 #*< Assured forwarding: class 2, low drop precedence. + Af22 = 20 #*< Assured forwarding: class 2, medium drop precedence. + Af23 = 22 #*< Assured forwarding: class 2, high drop precedence. + Cs3 = 24 #*< Class selector 3 (conventionally call signaling). + Af31 = 26 #*< Assured forwarding: class 3, low drop precedence. + Af32 = 28 #*< Assured forwarding: class 3, medium drop precedence. + Af33 = 30 #*< Assured forwarding: class 3, high drop precedence. + Cs4 = 32 #*< Class selector 4 (conventionally real-time interactive). + Af41 = 34 #*< Assured forwarding: class 4, low drop precedence. + Af42 = 36 #*< Assured forwarding: class 4, medium drop precedence. + Af43 = 38 #*< Assured forwarding: class 4, high drop precedence. + Cs5 = 40 #*< Class selector 5 (conventionally broadcast video). + VoiceAdmit = 44 #*< Capacity-admitted EF traffic (RFC 5865). + Ef = 46 #*< Expedited forwarding (RFC 3246) - low-latency/low-jitter. + Cs6 = 48 #*< Class selector 6 (network control - use with care). + Cs7 = 56 #*< Class selector 7 (reserved network control). + +#################### #################### + + #################### #################### @@ -3245,6 +3301,8 @@ class UdpSocket: receive its traffic. Empty/"0.0.0.0" lets the OS pick the default interface; set it on multi-homed hosts to bind multicast to a specific NIC (e.g. wired vs Wi-Fi). on_receive_callback: Socket.receive_callback_fn = Socket.receive_callback_fn(None) #*< Function containing business logic to handle data received. + band: QosBand = QosBand.Normal #*< Priority band for dispatching this socket's receive handling when registered on an espp.SocketReactor (unused by start_receiving()). + dscp: Optional[Dscp] = None #*< Optional espp.Dscp code point (e.g. Dscp.Ef) to mark this socket's TRANSMITTED packets with (applied as IP_TOS by espp.SocketReactor at registration, best-effort). def __init__( self, port: int = int(), @@ -3252,7 +3310,9 @@ class UdpSocket: is_multicast_endpoint: bool = bool(False), multicast_group: str = str(""), multicast_interface: str = str(""), - on_receive_callback: Socket.receive_callback_fn = Socket.receive_callback_fn(None) + on_receive_callback: Socket.receive_callback_fn = Socket.receive_callback_fn(None), + band: QosBand = QosBand.Normal, + dscp: Optional[Dscp] = None ) -> None: """Auto-generated default constructor with named params""" pass @@ -3478,12 +3538,14 @@ class Task: stack_size_bytes: int = int(4096) #*< Stack Size (B) allocated to the task. priority: int = int(0) #*< Priority of the task, 0 is lowest priority on ESP / FreeRTOS. core_id: int = int(-1) #*< Core ID of the task, -1 means it is not pinned to any core. + host_realtime: bool = False #*< Opt-in to applying the priority to the OS thread on host platforms (SCHED_FIFO on Linux/macOS; ignored on ESP). def __init__( self, name: str = "", stack_size_bytes: int = int(4096), priority: int = int(0), - core_id: int = int(-1) + core_id: int = int(-1), + host_realtime: bool = False ) -> None: """Auto-generated default constructor with named params""" pass @@ -3568,6 +3630,16 @@ class Task: """ pass + def get_configured_priority(self) -> int: + """* + * @brief Get the priority stored in the task's configuration. + * @details This is the value set at construction or via set_priority(); it + * is the priority the task will be started with (and, if the task + * is running, the priority that was last requested for it). + * @return The configured priority (0 is lowest; see BaseConfig.priority). + """ + pass + def set_core_id(self, core_id: int) -> bool: """* * @brief Set the core affinity (core ID) of the task. @@ -4099,6 +4171,10 @@ class ThreadPool: executed: std.int = 0 #/< Total jobs successfully executed. rejected: std.int = 0 #/< Total jobs rejected (invalid job, stopped/stopping, or queue #/< full) or dropped (due to stop, the enqueued jobs were dropped). + band_submitted: List[int] #/< Jobs accepted per band (index = QosBand). + band_executed: List[int] #/< Jobs executed per band (by the band they were popped from, i.e. after any aging promotions). + band_aged: List[int] #/< Aging promotions OUT of each band (an entry moved from band i to band i-1). + band_rejected: List[int] #/< Jobs rejected per band (by the band they were submitted to). def __init__( self, submitted: std.int = 0, @@ -4122,6 +4198,10 @@ class ThreadPool: .core_id = -1, ) log_level: Logger.Verbosity = Logger.Verbosity.WARN #/< Logger verbosity level. + aging_threshold: datetime.timedelta #/< Starvation guard: a queued job whose wait exceeds this is promoted up one band (default 100ms; 0 disables aging - strict band priority). Pass a datetime.timedelta or float seconds. + band_worker_counts: List[int] #/< Opt-in per-band worker counts (index = QosBand); all zero (the default) = disabled: identical workers service all bands. + band_task_priorities: List[int] #/< Task priorities for per-band workers (default [10, 7, 5, 1]; only used when band_worker_counts is set). + band_workers_realtime: bool = False #/< Opt-in for OS real-time scheduling of per-band workers on host platforms (SCHED_FIFO; see Task.BaseConfig.host_realtime). def __init__( self, worker_count: std.int = 1, @@ -4134,7 +4214,11 @@ class ThreadPool: .priority = 5, .core_id = -1, ), - log_level: Logger.Verbosity = Logger.Verbosity.WARN + log_level: Logger.Verbosity = Logger.Verbosity.WARN, + aging_threshold: datetime.timedelta = ..., + band_worker_counts: List[int] = ..., + band_task_priorities: List[int] = ..., + band_workers_realtime: bool = False ) -> None: """Auto-generated default constructor with named params""" pass @@ -4161,8 +4245,9 @@ class ThreadPool: """ pass + @overload def submit(self, job: Job) -> bool: - """/ @brief Submit a job, optionally blocking when the queue is full. + """/ @brief Submit a job at QosBand.Normal, optionally blocking when the queue is full. / / Blocks if Config::block_on_submit_when_full is True and the queue has / reached its capacity limit. Otherwise behaves identically to try_submit(). @@ -4171,8 +4256,18 @@ class ThreadPool: """ pass + @overload + def submit(self, job: Job, band: QosBand) -> bool: + """/ @brief Submit a job at the given priority band, optionally blocking when the queue is full. + / @param job Callable to enqueue; moved into the queue on acceptance. + / @param band Priority band to enqueue the job at. + / @return True if the job was accepted, False if it was rejected. + """ + pass + + @overload def try_submit(self, job: Job) -> bool: - """/ @brief Attempt to submit a job without blocking. + """/ @brief Attempt to submit a job at QosBand.Normal without blocking. / / Returns immediately with False when the queue is full. / @param job Callable to enqueue; moved into the queue on acceptance. @@ -4180,6 +4275,15 @@ class ThreadPool: """ pass + @overload + def try_submit(self, job: Job, band: QosBand) -> bool: + """/ @brief Attempt to submit a job at the given priority band without blocking. + / @param job Callable to enqueue; moved into the queue on acceptance. + / @param band Priority band to enqueue the job at. + / @return True if the job was accepted, False if it was rejected. + """ + pass + def queue_size(self) -> std.int: """/ @brief Return the number of jobs currently waiting in the queue. / @return Pending job count. @@ -4194,12 +4298,12 @@ class ThreadPool: def stats(self) -> ThreadPool.Stats: """/ @brief Return a snapshot of the pool's activity counters. - / @return Stats struct with submitted, executed, and rejected counts. + / @return Stats struct with submitted, executed, and rejected counts (total and per band). """ pass - def __init__(self) -> None: - """Auto-generated default constructor""" + def __init__(self, config: ThreadPool.Config) -> None: + """/ @brief Construct the pool with the given configuration.""" pass diff --git a/lib/python_bindings/pybind_espp.cpp b/lib/python_bindings/pybind_espp.cpp index 3486654c83..1a6ae5b2c9 100644 --- a/lib/python_bindings/pybind_espp.cpp +++ b/lib/python_bindings/pybind_espp.cpp @@ -1606,6 +1606,54 @@ void py_init_module_espp(py::module &m) { "structure containing gains, etc.\n"); //////////////////// //////////////////// + //////////////////// //////////////////// + // Bound before the socket / thread_pool sections, whose bindings use + // QosBand values as default arguments (defaults are converted at def time). + py::enum_( + m, "QosBand", + "*\n * @brief Priority band for queued work. Critical is the most urgent and Low the " + "least;\n * Normal is the default for band-less submissions.\n") + .value("Critical", espp::QosBand::Critical) + .value("High", espp::QosBand::High) + .value("Normal", espp::QosBand::Normal) + .value("Low", espp::QosBand::Low); + //////////////////// //////////////////// + + //////////////////// //////////////////// + // Bound before the socket section, whose bindings use std::optional + // parameters. + py::enum_( + m, "Dscp", + "*\n * @brief Standard DiffServ code points (DSCP) for IP traffic marking - the 6-bit " + "field\n * in the IP TOS / Traffic Class byte (RFC 2474). E.g. Dscp.Ef = expedited " + "forwarding\n * for latency-critical flows; Dscp.Cs1 = low-priority data; Dscp.Af41 = " + "high-priority\n * assured forwarding with low drop probability.\n") + .value("Cs0", espp::Dscp::Cs0) + .value("Default", espp::Dscp::Default) + .value("Le", espp::Dscp::Le) + .value("Cs1", espp::Dscp::Cs1) + .value("Af11", espp::Dscp::Af11) + .value("Af12", espp::Dscp::Af12) + .value("Af13", espp::Dscp::Af13) + .value("Cs2", espp::Dscp::Cs2) + .value("Af21", espp::Dscp::Af21) + .value("Af22", espp::Dscp::Af22) + .value("Af23", espp::Dscp::Af23) + .value("Cs3", espp::Dscp::Cs3) + .value("Af31", espp::Dscp::Af31) + .value("Af32", espp::Dscp::Af32) + .value("Af33", espp::Dscp::Af33) + .value("Cs4", espp::Dscp::Cs4) + .value("Af41", espp::Dscp::Af41) + .value("Af42", espp::Dscp::Af42) + .value("Af43", espp::Dscp::Af43) + .value("Cs5", espp::Dscp::Cs5) + .value("VoiceAdmit", espp::Dscp::VoiceAdmit) + .value("Ef", espp::Dscp::Ef) + .value("Cs6", espp::Dscp::Cs6) + .value("Cs7", espp::Dscp::Cs7); + //////////////////// //////////////////// + //////////////////// //////////////////// auto pyClassSocket = py::class_(m, "Socket", py::dynamic_attr(), @@ -1904,7 +1952,9 @@ void py_init_module_espp(py::module &m) { bool is_multicast_endpoint = {false}, std::string multicast_group = {""}, std::string multicast_interface = {""}, - espp::Socket::receive_callback_fn on_receive_callback = {nullptr}) { + espp::Socket::receive_callback_fn on_receive_callback = {nullptr}, + espp::QosBand band = {espp::QosBand::Normal}, + std::optional dscp = {}) { auto r_ctor_ = std::make_unique(); r_ctor_->port = port; r_ctor_->buffer_size = buffer_size; @@ -1912,13 +1962,17 @@ void py_init_module_espp(py::module &m) { r_ctor_->multicast_group = multicast_group; r_ctor_->multicast_interface = multicast_interface; r_ctor_->on_receive_callback = on_receive_callback; + r_ctor_->band = band; + r_ctor_->dscp = dscp; return r_ctor_; }), py::arg("port") = size_t(), py::arg("buffer_size") = size_t(), py::arg("is_multicast_endpoint") = bool{false}, py::arg("multicast_group") = std::string{""}, py::arg("multicast_interface") = std::string{""}, - py::arg("on_receive_callback") = espp::Socket::receive_callback_fn{nullptr}) + py::arg("on_receive_callback") = espp::Socket::receive_callback_fn{nullptr}, + py::arg("band") = espp::QosBand::Normal, + py::arg("dscp") = std::optional{}) .def_readwrite("port", &espp::UdpSocket::ReceiveConfig::port, "*< Port number to bind to / receive from.") .def_readwrite("buffer_size", &espp::UdpSocket::ReceiveConfig::buffer_size, @@ -1936,7 +1990,14 @@ void py_init_module_espp(py::module &m) { "to a specific NIC (e.g. wired vs Wi-Fi).") .def_readwrite("on_receive_callback", &espp::UdpSocket::ReceiveConfig::on_receive_callback, - "*< Function containing business logic to handle data received."); + "*< Function containing business logic to handle data received.") + .def_readwrite("band", &espp::UdpSocket::ReceiveConfig::band, + "*< Priority band for dispatching this socket's receive handling when " + "registered on an espp.SocketReactor (unused by start_receiving()).") + .def_readwrite("dscp", &espp::UdpSocket::ReceiveConfig::dscp, + "*< Optional espp.Dscp code point (e.g. Dscp.Ef) to mark this " + "socket's TRANSMITTED packets with (applied as IP_TOS by " + "espp.SocketReactor at registration, best-effort)."); auto pyClassUdpSocket_ClassSendConfig = py::class_(pyClassUdpSocket, "SendConfig", py::dynamic_attr(), "") @@ -2107,23 +2168,29 @@ void py_init_module_espp(py::module &m) { "to be used as a configuration struct in other classes\n * that may have a " "Task as a member.\n") .def(py::init<>([](std::string name = std::string(), size_t stack_size_bytes = {4096}, - size_t priority = {0}, int core_id = {-1}) { + size_t priority = {0}, int core_id = {-1}, + bool host_realtime = {false}) { auto r_ctor_ = std::make_unique(); r_ctor_->name = name; r_ctor_->stack_size_bytes = stack_size_bytes; r_ctor_->priority = priority; r_ctor_->core_id = core_id; + r_ctor_->host_realtime = host_realtime; return r_ctor_; }), py::arg("name") = std::string(), py::arg("stack_size_bytes") = size_t{4096}, - py::arg("priority") = size_t{0}, py::arg("core_id") = int{-1}) + py::arg("priority") = size_t{0}, py::arg("core_id") = int{-1}, + py::arg("host_realtime") = false) .def_readwrite("name", &espp::Task::BaseConfig::name, "*< Name of the task") .def_readwrite("stack_size_bytes", &espp::Task::BaseConfig::stack_size_bytes, "*< Stack Size (B) allocated to the task.") .def_readwrite("priority", &espp::Task::BaseConfig::priority, "*< Priority of the task, 0 is lowest priority on ESP / FreeRTOS.") .def_readwrite("core_id", &espp::Task::BaseConfig::core_id, - "*< Core ID of the task, -1 means it is not pinned to any core."); + "*< Core ID of the task, -1 means it is not pinned to any core.") + .def_readwrite("host_realtime", &espp::Task::BaseConfig::host_realtime, + "*< Opt-in to applying the priority to the OS thread on host " + "platforms (SCHED_FIFO on Linux/macOS; ignored on ESP)."); auto pyClassTask_ClassConfig = py::class_( pyClassTask, "Config", py::dynamic_attr(), @@ -2180,6 +2247,12 @@ void py_init_module_espp(py::module &m) { "False\n * if the task is not running (the new value still takes effect " "the\n * next time the task is started) or the platform does not support\n " "* changing a live task's priority.\n") + .def("get_configured_priority", &espp::Task::get_configured_priority, + "*\n * @brief Get the priority stored in the task's configuration.\n * @details " + "This is the value set at construction or via set_priority(); it\n * is the " + "priority the task will be started with (and, if the task\n * is running, " + "the priority that was last requested for it).\n * @return The configured priority " + "(0 is lowest; see BaseConfig.priority).\n") .def( "set_core_id", &espp::Task::set_core_id, py::arg("core_id"), "*\n * @brief Set the core affinity (core ID) of the task.\n * @details The new core " @@ -2633,31 +2706,51 @@ void py_init_module_espp(py::module &m) { .def_readwrite("executed", &espp::ThreadPool::Stats::executed, "/< Total jobs successfully executed.") .def_readwrite("rejected", &espp::ThreadPool::Stats::rejected, - "/< Total jobs rejected (invalid job, stopped/stopping, or queue"); + "/< Total jobs rejected (invalid job, stopped/stopping, or queue") + .def_readwrite("band_submitted", &espp::ThreadPool::Stats::band_submitted, + "/< Jobs accepted per band (by the band they were submitted to).") + .def_readwrite("band_executed", &espp::ThreadPool::Stats::band_executed, + "/< Jobs executed per band (by the band they were popped from, i.e. " + "after any aging promotions).") + .def_readwrite("band_aged", &espp::ThreadPool::Stats::band_aged, + "/< Aging promotions OUT of each band (an entry moved from band i to " + "band i-1).") + .def_readwrite("band_rejected", &espp::ThreadPool::Stats::band_rejected, + "/< Jobs rejected per band (by the band they were submitted to)."); auto pyClassThreadPool_ClassConfig = py::class_( pyClassThreadPool, "Config", py::dynamic_attr(), "/ @brief Configuration parameters for constructing a ThreadPool.") - .def(py::init<>([](std::size_t worker_count = 1, std::size_t max_queue_size = 0, - bool auto_start = true, bool block_on_submit_when_full = false, - espp::Task::BaseConfig worker_task_config = - { - ///< Base configuration applied to every worker task. - .name = "thread_pool_worker", - .stack_size_bytes = 4096, - .priority = 5, - .core_id = -1, - }, - espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN) { - auto r_ctor_ = std::make_unique(); - r_ctor_->worker_count = worker_count; - r_ctor_->max_queue_size = max_queue_size; - r_ctor_->auto_start = auto_start; - r_ctor_->block_on_submit_when_full = block_on_submit_when_full; - r_ctor_->worker_task_config = worker_task_config; - r_ctor_->log_level = log_level; - return r_ctor_; - }), + .def(py::init<>( + [](std::size_t worker_count = 1, std::size_t max_queue_size = 0, + bool auto_start = true, bool block_on_submit_when_full = false, + espp::Task::BaseConfig worker_task_config = + { + ///< Base configuration applied to every worker task. + .name = "thread_pool_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN, + std::chrono::milliseconds aging_threshold = std::chrono::milliseconds{100}, + std::array band_worker_counts = {}, + std::array band_task_priorities = {10, 7, + 5, 1}, + bool band_workers_realtime = false) { + auto r_ctor_ = std::make_unique(); + r_ctor_->worker_count = worker_count; + r_ctor_->max_queue_size = max_queue_size; + r_ctor_->auto_start = auto_start; + r_ctor_->block_on_submit_when_full = block_on_submit_when_full; + r_ctor_->worker_task_config = worker_task_config; + r_ctor_->log_level = log_level; + r_ctor_->aging_threshold = aging_threshold; + r_ctor_->band_worker_counts = band_worker_counts; + r_ctor_->band_task_priorities = band_task_priorities; + r_ctor_->band_workers_realtime = band_workers_realtime; + return r_ctor_; + }), py::arg("worker_count") = 1, py::arg("max_queue_size") = 0, py::arg("auto_start") = true, py::arg("block_on_submit_when_full") = false, py::arg("worker_task_config") = @@ -2668,7 +2761,12 @@ void py_init_module_espp(py::module &m) { .priority = 5, .core_id = -1, }, - py::arg("log_level") = espp::Logger::Verbosity::WARN) + py::arg("log_level") = espp::Logger::Verbosity::WARN, + py::arg("aging_threshold") = std::chrono::milliseconds{100}, + py::arg("band_worker_counts") = std::array{}, + py::arg("band_task_priorities") = + std::array{10, 7, 5, 1}, + py::arg("band_workers_realtime") = false) .def_readwrite("worker_count", &espp::ThreadPool::Config::worker_count, "/< Number of worker threads to spawn.") .def_readwrite("max_queue_size", &espp::ThreadPool::Config::max_queue_size, @@ -2680,7 +2778,20 @@ void py_init_module_espp(py::module &m) { "/< If True, submit() blocks when the queue is full instead of rejecting.") .def_readwrite("worker_task_config", &espp::ThreadPool::Config::worker_task_config, "") .def_readwrite("log_level", &espp::ThreadPool::Config::log_level, - "/< Logger verbosity level."); + "/< Logger verbosity level.") + .def_readwrite("aging_threshold", &espp::ThreadPool::Config::aging_threshold, + "/< Starvation guard: a queued job whose wait exceeds this is promoted " + "up one band. 0 disables aging (strict band priority).") + .def_readwrite("band_worker_counts", &espp::ThreadPool::Config::band_worker_counts, + "/< Opt-in per-band worker counts (index = QosBand); all zero = " + "disabled (identical workers service all bands).") + .def_readwrite("band_task_priorities", &espp::ThreadPool::Config::band_task_priorities, + "/< Task priorities for per-band workers (only used when " + "band_worker_counts is set).") + .def_readwrite("band_workers_realtime", + &espp::ThreadPool::Config::band_workers_realtime, + "/< Opt-in for OS real-time scheduling of per-band workers on host " + "platforms (SCHED_FIFO; see Task.BaseConfig.host_realtime)."); } // end of inner classes & enums of ThreadPool pyClassThreadPool.def(py::init()) @@ -2697,17 +2808,41 @@ void py_init_module_espp(py::module &m) { .def("is_running", &espp::ThreadPool::is_running, "/ @brief Query whether the pool is currently running.\n/ @return True if workers are " "active, False otherwise.") - .def("submit", &espp::ThreadPool::submit, py::arg("job"), - "/ @brief Submit a job, optionally blocking when the queue is full.\n/\n/ Blocks if " - "Config::block_on_submit_when_full is True and the queue has\n/ reached its capacity " - "limit. Otherwise behaves identically to try_submit().\n/ @param job Callable to " - "enqueue; moved into the queue on acceptance.\n/ @return True if the job was accepted, " - "False if it was rejected.", + .def("submit", + static_cast( + &espp::ThreadPool::submit), + py::arg("job"), + "/ @brief Submit a job at QosBand.Normal, optionally blocking when the queue is " + "full.\n/\n/ Blocks if Config::block_on_submit_when_full is True and the queue has\n/ " + "reached its capacity limit. Otherwise behaves identically to try_submit().\n/ @param " + "job Callable to enqueue; moved into the queue on acceptance.\n/ @return True if the " + "job was accepted, False if it was rejected.", + py::call_guard()) + .def("submit", + static_cast( + &espp::ThreadPool::submit), + py::arg("job"), py::arg("band"), + "/ @brief Submit a job at the given priority band, optionally blocking when the queue " + "is full.\n/ @param job Callable to enqueue; moved into the queue on acceptance.\n/ " + "@param band Priority band to enqueue the job at.\n/ @return True if the job was " + "accepted, False if it was rejected.", py::call_guard()) - .def("try_submit", &espp::ThreadPool::try_submit, py::arg("job"), - "/ @brief Attempt to submit a job without blocking.\n/\n/ Returns immediately with " - "False when the queue is full.\n/ @param job Callable to enqueue; moved into the queue " - "on acceptance.\n/ @return True if the job was accepted, False if it was rejected.") + .def("try_submit", + static_cast( + &espp::ThreadPool::try_submit), + py::arg("job"), + "/ @brief Attempt to submit a job at QosBand.Normal without blocking.\n/\n/ Returns " + "immediately with False when the queue is full.\n/ @param job Callable to enqueue; " + "moved into the queue on acceptance.\n/ @return True if the job was accepted, False if " + "it was rejected.") + .def("try_submit", + static_cast( + &espp::ThreadPool::try_submit), + py::arg("job"), py::arg("band"), + "/ @brief Attempt to submit a job at the given priority band without blocking.\n/ " + "@param job Callable to enqueue; moved into the queue on acceptance.\n/ @param band " + "Priority band to enqueue the job at.\n/ @return True if the job was accepted, False " + "if it was rejected.") .def("queue_size", &espp::ThreadPool::queue_size, "/ @brief Return the number of jobs currently waiting in the queue.\n/ @return Pending " "job count.") diff --git a/lib/python_bindings/socket_reactor_bindings.cpp b/lib/python_bindings/socket_reactor_bindings.cpp index 2f93fbfc08..5fed766b4a 100644 --- a/lib/python_bindings/socket_reactor_bindings.cpp +++ b/lib/python_bindings/socket_reactor_bindings.cpp @@ -18,6 +18,8 @@ #include #include +#include "dscp.hpp" +#include "qos_band.hpp" #include "socket_reactor.hpp" #include "udp_socket.hpp" @@ -100,17 +102,23 @@ void py_init_socket_reactor(py::module &m) { .def( "add_udp_receiver", [](SocketReactor &self, espp::UdpSocket &socket, std::size_t port, - std::size_t buffer_size, const py::function &callback) -> SocketReactor::Id { + std::size_t buffer_size, const py::function &callback, espp::QosBand band, + std::optional dscp) -> SocketReactor::Id { espp::UdpSocket::ReceiveConfig rc; rc.port = port; rc.buffer_size = buffer_size; rc.on_receive_callback = wrap_receive_callback(callback); + rc.band = band; + rc.dscp = dscp; return self.add_udp_receiver(socket, rc); }, py::arg("socket"), py::arg("port"), py::arg("buffer_size"), py::arg("callback"), + py::arg("band") = espp::QosBand::Normal, py::arg("dscp") = std::optional{}, "Bind `socket` to `port` and receive on it via the reactor. `callback(data: bytes, " - "sender) -> Optional[bytes]`; a returned bytes is sent back to the sender. Returns a " - "registration id (0 == INVALID_ID on failure).") + "sender) -> Optional[bytes]`; a returned bytes is sent back to the sender. `band` " + "selects the espp.QosBand this socket's handlers are dispatched at; `dscp` (an " + "espp.Dscp, e.g. Dscp.Ef) optionally marks transmitted replies (IP_TOS, best-effort). " + "Returns a registration id (0 == INVALID_ID on failure).") .def_property_readonly_static( "INVALID_ID", [](py::object) { return SocketReactor::INVALID_ID; }, "The id value returned by add_* on failure."); diff --git a/pc/tests/socket_reactor.cpp b/pc/tests/socket_reactor.cpp index 64995a1ece..741d7e5a37 100644 --- a/pc/tests/socket_reactor.cpp +++ b/pc/tests/socket_reactor.cpp @@ -1,6 +1,8 @@ #include +#include #include #include +#include #include #include #include @@ -198,6 +200,261 @@ int main() { } } + // ------------------------------------------------------------------------- + // 5. Priority bands: Critical socket stays responsive under a Low-band flood + // ------------------------------------------------------------------------- + logger.info("--- priority bands + dscp ---"); + { + constexpr size_t low_port = 6140; + constexpr size_t crit_port = 6141; + constexpr int num_critical_msgs = 10; + std::atomic flood_processed{0}; + std::atomic crit_received{0}; + std::atomic stop_flood{false}; + + // sockets declared before the reactor so the reactor is destroyed first + espp::UdpSocket low_server({.log_level = WARN}); + espp::UdpSocket crit_server({.log_level = WARN}); + { + // Single pool worker so dispatch order is observable: when both sockets + // are readable in one select() round, the Critical one must be handled + // first. + espp::SocketReactor reactor({.pool_config = {.worker_count = 1, + .worker_task_config = {.name = "reactor pool", + .stack_size_bytes = 4096, + .priority = 5}}, + .log_level = WARN}); + + auto low_id = reactor.add_udp_receiver( + low_server, + {.port = low_port, + .buffer_size = kBufferSize, + .on_receive_callback = [&](const ByteVector &, + const espp::Socket::Info &) -> std::optional { + // simulate per-packet work so the flood keeps the pool busy + std::this_thread::sleep_for(5ms); + ++flood_processed; + return std::nullopt; + }, + .band = espp::QosBand::Low, + .dscp = espp::Dscp::Cs1}); // "low-priority data" + auto crit_id = reactor.add_udp_receiver( + crit_server, + {.port = crit_port, + .buffer_size = kBufferSize, + .on_receive_callback = [&](const ByteVector &, + const espp::Socket::Info &) -> std::optional { + ++crit_received; + return std::nullopt; + }, + .band = espp::QosBand::Critical, + .dscp = espp::Dscp::Ef}); // "expedited forwarding" + check(low_id != espp::SocketReactor::INVALID_ID, "Low-band receiver registered (with dscp)"); + check(crit_id != espp::SocketReactor::INVALID_ID, + "Critical-band receiver registered (with dscp)"); + +#if !defined(_WIN32) + // Verify the DSCP marking actually landed on the sockets: IP_TOS carries + // the DSCP in its upper 6 bits, so read it back with getsockopt(). + auto read_tos = [](espp::UdpSocket &s) { + int tos = -1; + socklen_t len = sizeof(tos); + if (::getsockopt(s.native_handle(), IPPROTO_IP, IP_TOS, reinterpret_cast(&tos), + &len) < 0) { + return -1; + } + return tos; + }; + check(read_tos(low_server) == espp::dscp_to_tos(espp::Dscp::Cs1), + "IP_TOS on the Low socket reflects Dscp::Cs1"); + check(read_tos(crit_server) == espp::dscp_to_tos(espp::Dscp::Ef), + "IP_TOS on the Critical socket reflects Dscp::Ef"); + // Out-of-range DSCP: registration must still succeed, but the invalid + // value must be ignored (TOS left at the OS default), not masked into a + // different code point. + espp::UdpSocket bad_dscp_server({.log_level = WARN}); + auto bad_dscp_id = reactor.add_udp_receiver( + bad_dscp_server, + {.port = 6142, + .buffer_size = kBufferSize, + .on_receive_callback = [](const ByteVector &, const espp::Socket::Info &) + -> std::optional { return std::nullopt; }, + .dscp = static_cast(200)}); + check(bad_dscp_id != espp::SocketReactor::INVALID_ID, + "registration with an out-of-range DSCP still succeeds"); + check(read_tos(bad_dscp_server) == 0, "out-of-range DSCP is ignored (TOS stays default)"); + // bad_dscp_server is scoped inside the reactor's block, so make sure its + // registration is fully gone before it goes out of scope + reactor.remove(bad_dscp_id); + check(wait_until([&] { return reactor.num_registered() == 2; }, 2s), + "out-of-range-DSCP registration removed"); +#endif + + // Flood the Low-band socket from a background thread for the duration. + std::thread flood([&]() { + espp::UdpSocket client({.log_level = WARN}); + auto payload = make_payload(100, 0x55); + while (!stop_flood.load()) { + client.send(payload, {.ip_address = kLoopback, .port = low_port}); + std::this_thread::sleep_for(1ms); + } + }); + + check(wait_until([&] { return flood_processed.load() >= 5; }, 5s), + "flood is being processed"); + const int flood_before = flood_processed.load(); + + // Sparse Critical messages; each must be dispatched with bounded delay + // even though the (single-worker) pool is saturated by the flood. + espp::UdpSocket crit_client({.log_level = WARN}); + int delivered = 0; + std::chrono::milliseconds worst_latency{0}; + for (int i = 0; i < num_critical_msgs; ++i) { + const int before = crit_received.load(); + const auto t0 = std::chrono::steady_clock::now(); + crit_client.send(make_payload(32, static_cast(i)), + {.ip_address = kLoopback, .port = crit_port}); + if (wait_until([&] { return crit_received.load() > before; }, 2s, 1ms)) { + ++delivered; + auto latency = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0); + worst_latency = std::max(worst_latency, latency); + } + } + logger.info(" {} critical messages delivered, worst latency {}ms (flood processed: {})", + delivered, worst_latency.count(), flood_processed.load()); + check(delivered == num_critical_msgs, + "all Critical messages dispatched promptly during the flood"); + // Band-aware dispatch bounds a Critical message's wait to roughly one + // in-flight 5ms Low handler + scheduling noise. A band-less FIFO would + // queue each Critical behind the ever-growing Low backlog (1ms arrival + // vs 5ms service), blowing far past this bound within a few messages. + check(worst_latency < 500ms, "worst Critical latency bounded during the flood (<500ms)"); + check(flood_processed.load() >= flood_before + 5, + "Low-band flood kept making progress alongside Critical traffic"); + + stop_flood = true; + flood.join(); + reactor.stop(); + } + } + + // ------------------------------------------------------------------------- + // 6. Priority bands: deterministic queue-jump - Critical overtakes an + // already-queued Low backlog (this FAILS under band-less FIFO dispatch) + // ------------------------------------------------------------------------- + logger.info("--- priority bands: deterministic queue-jump ordering ---"); + { + constexpr size_t blocker_port = 6150; + constexpr size_t crit_port = 6151; + constexpr std::array low_ports = {6152, 6153, 6154, 6155}; + + // sockets declared before the reactor so the reactor is destroyed first + espp::UdpSocket blocker_server({.log_level = WARN}); + espp::UdpSocket crit_server({.log_level = WARN}); + std::array, low_ports.size()> low_servers; + { + std::mutex gate_mtx; + std::condition_variable gate_cv; + bool release = false; + std::atomic blocker_running{false}; + std::mutex order_mtx; + std::vector order; + + // External SINGLE-worker pool with aging disabled, so (a) the pop order + // is strictly by band and (b) queue_size() lets the test observe when + // the backlog is fully queued. + auto pool = std::make_shared(espp::ThreadPool::Config{ + .worker_count = 1, + .aging_threshold = std::chrono::milliseconds(0), + .worker_task_config = {.name = "ordering pool", .stack_size_bytes = 4096, .priority = 5}, + .log_level = WARN}); + espp::SocketReactor reactor({.thread_pool = pool, .log_level = WARN}); + + // Register the Low receivers FIRST (lowest fds, earliest submissions): + // a band-less FIFO dispatch would therefore run them all first. + for (size_t i = 0; i < low_ports.size(); ++i) { + low_servers[i] = + std::make_unique(espp::UdpSocket::Config{.log_level = WARN}); + auto id = reactor.add_udp_receiver( + *low_servers[i], + {.port = low_ports[i], + .buffer_size = kBufferSize, + .on_receive_callback = [&](const ByteVector &, + const espp::Socket::Info &) -> std::optional { + std::lock_guard lk(order_mtx); + order.push_back("low"); + return std::nullopt; + }, + .band = espp::QosBand::Low}); + check(id != espp::SocketReactor::INVALID_ID, "Low backlog receiver registered"); + } + auto crit_id = reactor.add_udp_receiver( + crit_server, + {.port = crit_port, + .buffer_size = kBufferSize, + .on_receive_callback = [&](const ByteVector &, + const espp::Socket::Info &) -> std::optional { + std::lock_guard lk(order_mtx); + order.push_back("critical"); + return std::nullopt; + }, + .band = espp::QosBand::Critical}); + auto blocker_id = reactor.add_udp_receiver( + blocker_server, + {.port = blocker_port, + .buffer_size = kBufferSize, + .on_receive_callback = [&](const ByteVector &, + const espp::Socket::Info &) -> std::optional { + blocker_running = true; + std::unique_lock lk(gate_mtx); + gate_cv.wait(lk, [&] { return release; }); + return std::nullopt; + }}); + check(crit_id != espp::SocketReactor::INVALID_ID && + blocker_id != espp::SocketReactor::INVALID_ID, + "Critical + blocker receivers registered"); + + espp::UdpSocket client({.log_level = WARN}); + // 1. Occupy the single pool worker with the gated blocker handler. + client.send(make_payload(8, 0x01), {.ip_address = kLoopback, .port = blocker_port}); + check(wait_until([&] { return blocker_running.load(); }, 5s), + "blocker handler occupies the single pool worker"); + // 2. Queue the whole Low backlog while the worker is blocked... + for (auto port : low_ports) { + client.send(make_payload(8, 0x02), {.ip_address = kLoopback, .port = port}); + } + check(wait_until([&] { return pool->queue_size() == low_ports.size(); }, 5s), + "all Low handlers queued behind the blocker"); + // 3. ...then submit the Critical packet LAST. + client.send(make_payload(8, 0x03), {.ip_address = kLoopback, .port = crit_port}); + check(wait_until([&] { return pool->queue_size() == low_ports.size() + 1; }, 5s), + "Critical handler queued last"); + // 4. Release the worker: strict band pop order must run Critical FIRST, + // even though it was the last submission (FIFO would run it last). + { + std::lock_guard lk(gate_mtx); + release = true; + } + gate_cv.notify_all(); + check(wait_until( + [&] { + std::lock_guard lk(order_mtx); + return order.size() == low_ports.size() + 1; + }, + 5s), + "all queued handlers executed after release"); + { + std::lock_guard lk(order_mtx); + check(!order.empty() && order.front() == "critical", + "Critical overtook the entire already-queued Low backlog"); + check(std::count(order.begin(), order.end(), "low") == static_cast(low_ports.size()), + "every Low handler still executed (none lost)"); + } + reactor.stop(); + } + } + // ------------------------------------------------------------------------- // Summary // ------------------------------------------------------------------------- diff --git a/pc/tests/thread_pool.cpp b/pc/tests/thread_pool.cpp index fcf3543660..bbaf32b409 100644 --- a/pc/tests/thread_pool.cpp +++ b/pc/tests/thread_pool.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -6,10 +7,28 @@ #include #include +#include "task.hpp" #include "thread_pool.hpp" using namespace std::chrono_literals; +namespace { +// Poll `pred` until it returns true or `timeout` elapses; returns the final +// value of pred(). Used for bounded, non-flaky waits on background progress. +template +bool wait_until(Predicate &&pred, std::chrono::milliseconds timeout, + std::chrono::milliseconds interval = std::chrono::milliseconds(1)) { + auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (pred()) { + return true; + } + std::this_thread::sleep_for(interval); + } + return pred(); +} +} // namespace + int main() { espp::Logger logger({.tag = "ThreadPool Test", .level = espp::Logger::Verbosity::INFO}); @@ -138,6 +157,15 @@ int main() { } check(rejected == 3, "3 try_submit calls rejected when queue full"); check(pool.stats().rejected == 3, "stats.rejected == 3"); + // band-less try_submit rejects at Normal; the rejection must be + // attributed to that band in the per-band counters + check(pool.stats().band_rejected[static_cast(espp::QosBand::Normal)] == 3, + "stats.band_rejected[Normal] == 3"); + // a band-aware rejection is attributed to ITS band + check(!pool.try_submit(espp::ThreadPool::Job([] {}), espp::QosBand::Critical), + "banded try_submit also rejected when queue full"); + check(pool.stats().band_rejected[static_cast(espp::QosBand::Critical)] == 1, + "stats.band_rejected[Critical] == 1"); { std::lock_guard lk(barrier_mtx); @@ -394,6 +422,371 @@ int main() { pool.stop(); } + // --------------------------------------------------------------------------- + // 10. Task host priority: start()/set_priority() succeed without privileges + // --------------------------------------------------------------------------- + logger.info("--- task: host priority application (graceful fallback) ---"); + { + std::atomic iterations{0}; + espp::Task task({.callback = [&]() -> bool { + ++iterations; + std::this_thread::sleep_for(1ms); + return false; + }, + .task_config = {.name = "prio_task", + .stack_size_bytes = 4096, + .priority = 10, + .core_id = -1, + .host_realtime = true}}); + check(task.get_configured_priority() == 10, "configured priority round-trips from config"); + // Must succeed even when RT scheduling is not permitted (unprivileged CI): + // the priority application falls back gracefully and never fails start(). + check(task.start(), "start() succeeds with an RT-range priority, unprivileged"); + check(wait_until([&] { return iterations.load() > 0; }, 2s), "task callback runs"); + // Live priority changes are best-effort (return value depends on platform + // privileges); the stored value must always round-trip. + task.set_priority(3); + check(task.get_configured_priority() == 3, "set_priority(3) round-trips while running"); + task.set_priority(0); // demote back to default scheduling + check(task.get_configured_priority() == 0, "set_priority(0) round-trips while running"); + check(task.stop(), "task stops cleanly"); + } + + // --------------------------------------------------------------------------- + // 11. Priority ordering: Critical overtakes queued Low jobs (strict priority) + // --------------------------------------------------------------------------- + logger.info("--- priority: Critical overtakes queued Low ---"); + { + std::mutex gate_mtx; + std::condition_variable gate_cv; + bool release = false; + std::atomic blocker_started{0}; + std::atomic done{0}; + std::mutex order_mtx; + std::vector order; + + espp::ThreadPool pool({ + .worker_count = 1, + .auto_start = true, + .aging_threshold = 0ms, // strict band priority for determinism + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + + // Gate the single worker so submission order is fully deterministic. + pool.submit(espp::ThreadPool::Job([&]() { + ++blocker_started; + std::unique_lock lk(gate_mtx); + gate_cv.wait(lk, [&] { return release; }); + })); + check(wait_until([&] { return blocker_started.load() >= 1; }, 2s), "blocker job is executing"); + + for (int i = 0; i < 3; ++i) { + pool.submit(espp::ThreadPool::Job([&, i]() { + std::lock_guard lk(order_mtx); + order.push_back("low" + std::to_string(i)); + ++done; + }), + espp::QosBand::Low); + } + pool.submit(espp::ThreadPool::Job([&]() { + std::lock_guard lk(order_mtx); + order.push_back("critical"); + ++done; + }), + espp::QosBand::Critical); + + { + std::lock_guard lk(gate_mtx); + release = true; + } + gate_cv.notify_all(); + check(wait_until([&] { return done.load() >= 4; }, 2s), "all banded jobs completed"); + { + std::lock_guard lk(order_mtx); + check(order.size() == 4 && order[0] == "critical", + "Critical job ran before the queued Low jobs"); + check(order.size() == 4 && order[1] == "low0" && order[2] == "low1" && order[3] == "low2", + "Low jobs kept FIFO order within their band"); + } + pool.stop(); + } + + // --------------------------------------------------------------------------- + // 12. Default-band equivalence: no-band submits behave FIFO exactly as before + // --------------------------------------------------------------------------- + logger.info("--- priority: default band preserves FIFO ---"); + { + std::mutex gate_mtx; + std::condition_variable gate_cv; + bool release = false; + std::atomic blocker_started{0}; + std::atomic done{0}; + std::mutex order_mtx; + std::vector order; + constexpr int N = 10; + + espp::ThreadPool pool({ + .worker_count = 1, + .auto_start = true, + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + + pool.submit(espp::ThreadPool::Job([&]() { + ++blocker_started; + std::unique_lock lk(gate_mtx); + gate_cv.wait(lk, [&] { return release; }); + })); + check(wait_until([&] { return blocker_started.load() >= 1; }, 2s), "blocker job is executing"); + + for (int i = 0; i < N; ++i) { + pool.submit(espp::ThreadPool::Job([&, i]() { + std::lock_guard lk(order_mtx); + order.push_back(i); + ++done; + })); + } + { + std::lock_guard lk(gate_mtx); + release = true; + } + gate_cv.notify_all(); + check(wait_until([&] { return done.load() >= N; }, 2s), "all no-band jobs completed"); + { + std::lock_guard lk(order_mtx); + bool fifo = order.size() == N; + for (int i = 0; fifo && i < N; ++i) { + fifo = (order[i] == i); + } + check(fifo, "no-band submits execute in exact FIFO submission order"); + } + auto s = pool.stats(); + check(s.band_submitted[static_cast(espp::QosBand::Normal)] == N + 1, + "no-band submits are accounted to the Normal band"); + pool.stop(); + } + + // --------------------------------------------------------------------------- + // 13. Aging: a Low job runs under a continuous stream of Normal jobs + // --------------------------------------------------------------------------- + logger.info("--- priority: aging rescues a Low job under Normal load ---"); + { + std::mutex gate_mtx; + std::condition_variable gate_cv; + bool release = false; + std::atomic blocker_started{0}; + std::atomic low_done{false}; + std::atomic normals_done{0}; + std::atomic stop_stream{false}; + + espp::ThreadPool pool({ + .worker_count = 1, + .auto_start = true, + .aging_threshold = 20ms, + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + + // Gate the worker so the Low job provably queues BEHIND a busy Normal band + // (an idle worker would otherwise pick the Low job up immediately). + pool.submit(espp::ThreadPool::Job([&]() { + ++blocker_started; + std::unique_lock lk(gate_mtx); + gate_cv.wait(lk, [&] { return release; }); + })); + check(wait_until([&] { return blocker_started.load() >= 1; }, 2s), "blocker job is executing"); + + // The Low job goes in first, plus a primed backlog of Normal jobs longer + // than the aging threshold (5 x 5ms > 20ms)... + pool.submit(espp::ThreadPool::Job([&]() { low_done = true; }), espp::QosBand::Low); + auto normal_job = espp::ThreadPool::Job([&]() { + std::this_thread::sleep_for(5ms); + ++normals_done; + }); + for (int i = 0; i < 5; ++i) { + pool.submit(espp::ThreadPool::Job(normal_job)); + } + // ...then more Normal jobs are produced faster than the single worker + // consumes them, so the Normal band never empties. Without aging the Low + // job would starve indefinitely; with aging it must run in bounded time. + std::thread producer([&]() { + while (!stop_stream.load()) { + pool.submit(espp::ThreadPool::Job(normal_job)); + std::this_thread::sleep_for(2ms); + } + }); + { + std::lock_guard lk(gate_mtx); + release = true; + } + gate_cv.notify_all(); + + bool rescued = wait_until([&] { return low_done.load(); }, 5s, 5ms); + stop_stream = true; + producer.join(); + check(rescued, "aged Low job ran while the Normal stream was still active"); + check(normals_done.load() > 0, "Normal stream made progress concurrently"); + auto s = pool.stats(); + logger.info(" stats: {}", s); + check(s.band_aged[static_cast(espp::QosBand::Low)] >= 1, + "stats recorded an aging promotion out of the Low band"); + pool.stop(); + } + + // --------------------------------------------------------------------------- + // 14. Per-band stats + // --------------------------------------------------------------------------- + logger.info("--- priority: per-band stats ---"); + { + std::mutex gate_mtx; + std::condition_variable gate_cv; + bool release = false; + std::atomic blocker_started{0}; + std::atomic done{0}; + + espp::ThreadPool pool({ + .worker_count = 1, + .auto_start = true, + .aging_threshold = 0ms, // keep jobs in their submitted bands + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + + pool.submit(espp::ThreadPool::Job([&]() { // Normal-band blocker + ++blocker_started; + std::unique_lock lk(gate_mtx); + gate_cv.wait(lk, [&] { return release; }); + })); + check(wait_until([&] { return blocker_started.load() >= 1; }, 2s), "blocker job is executing"); + + auto count_done = espp::ThreadPool::Job([&]() { ++done; }); + pool.submit(espp::ThreadPool::Job(count_done), espp::QosBand::Critical); + pool.submit(espp::ThreadPool::Job(count_done), espp::QosBand::High); + pool.submit(espp::ThreadPool::Job(count_done), espp::QosBand::Normal); + pool.submit(espp::ThreadPool::Job(count_done), espp::QosBand::Low); + + { + std::lock_guard lk(gate_mtx); + release = true; + } + gate_cv.notify_all(); + check(wait_until([&] { return done.load() >= 4; }, 2s), "one job per band completed"); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + check(s.band_submitted[0] == 1 && s.band_submitted[1] == 1 && s.band_submitted[2] == 2 && + s.band_submitted[3] == 1, + "band_submitted counts each band (blocker is Normal)"); + check(s.band_executed == s.band_submitted, "band_executed matches band_submitted (no aging)"); + check(s.band_aged[0] == 0 && s.band_aged[1] == 0 && s.band_aged[2] == 0 && s.band_aged[3] == 0, + "no aging promotions recorded with aging disabled"); + check(s.submitted == 5 && s.executed == 5 && s.rejected == 0, "totals consistent"); + pool.stop(); + } + + // --------------------------------------------------------------------------- + // 15. Worker bands: per-band workers, mixed load, nothing lost + // --------------------------------------------------------------------------- + logger.info("--- priority: per-band workers (worker bands) ---"); + { + constexpr int num_critical = 20; + constexpr int num_normal = 40; + constexpr int num_low = 20; + constexpr int total = num_critical + num_normal + num_low; + std::atomic done{0}; + std::mutex lat_mtx; + std::vector critical_latencies; + std::vector low_latencies; + + espp::ThreadPool pool({ + .auto_start = true, + .band_worker_counts = {{1, 1, 2, 1}}, // 1 Critical, 1 High, 2 Normal, 1 Low worker + .band_workers_realtime = true, // exercise the host SCHED_FIFO path (best-effort; falls + // back gracefully without CAP_SYS_NICE/RLIMIT_RTPRIO) + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + check(pool.worker_count() == 5, "per-band worker counts create 1+1+2+1 workers"); + + auto submit_timed = [&](espp::QosBand band, std::vector *lat) { + auto t0 = std::chrono::steady_clock::now(); + pool.submit(espp::ThreadPool::Job([&, t0, lat]() { + auto waited = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0); + if (lat != nullptr) { + std::lock_guard lk(lat_mtx); + lat->push_back(waited); + } + std::this_thread::sleep_for(2ms); + ++done; + }), + band); + }; + + // Interleave the load: mostly Normal with periodic Critical / Low. + int c = 0, n = 0, l = 0; + while (c < num_critical || n < num_normal || l < num_low) { + if (n < num_normal) { + submit_timed(espp::QosBand::Normal, nullptr); + ++n; + } + if ((n % 2) == 0 && c < num_critical) { + submit_timed(espp::QosBand::Critical, &critical_latencies); + ++c; + } + if ((n % 2) == 1 && l < num_low) { + submit_timed(espp::QosBand::Low, &low_latencies); + ++l; + } + } + + check(wait_until([&] { return done.load() >= total; }, 10s, 5ms), + "all jobs completed on per-band workers (none lost)"); + auto s = pool.stats(); + logger.info(" stats: {}", s); + check(s.submitted == total && s.executed == total && s.rejected == 0, + "stats: everything submitted was executed, nothing rejected"); + { + std::lock_guard lk(lat_mtx); + auto max_of = [](const std::vector &v) { + return v.empty() ? std::chrono::microseconds(0) : *std::max_element(v.begin(), v.end()); + }; + logger.info(" critical: n={} max wait={}us; low: n={} max wait={}us", + critical_latencies.size(), max_of(critical_latencies).count(), + low_latencies.size(), max_of(low_latencies).count()); + check(critical_latencies.size() == num_critical && low_latencies.size() == num_low, + "latency recorded for every Critical and Low job"); + } + pool.stop(); + } + + // --------------------------------------------------------------------------- + // Per-band workers: every band stays reachable even in strict mode + // --------------------------------------------------------------------------- + logger.info("--- priority: deepest band worker covers all bands (strict, no aging) ---"); + { + // Only a Critical worker configured, and aging disabled: without the + // coverage fallback the Normal and Low submissions below would sit queued + // forever. The deepest configured band's workers must service every band. + std::atomic done{0}; + espp::ThreadPool pool({ + .auto_start = true, + .aging_threshold = std::chrono::milliseconds(0), // strict band priority + .band_worker_counts = {{1, 0, 0, 0}}, // ONLY a Critical worker + .worker_task_config = + {.name = "tp_worker", .stack_size_bytes = 4096, .priority = 5, .core_id = -1}, + }); + check(pool.worker_count() == 1, "single Critical worker created"); + pool.submit(espp::ThreadPool::Job([&]() { ++done; }), espp::QosBand::Critical); + pool.submit(espp::ThreadPool::Job([&]() { ++done; })); // Normal (default) + pool.submit(espp::ThreadPool::Job([&]() { ++done; }), espp::QosBand::Low); + check(wait_until([&] { return done.load() == 3; }, 5s), + "Critical, Normal, and Low jobs all execute with only a Critical worker (no band " + "unreachable)"); + pool.stop(); + } + // --------------------------------------------------------------------------- // Summary // ---------------------------------------------------------------------------