feat(scheduling): priority bands in ThreadPool + SocketReactor, RT priorities on host Task

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"
   )

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

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`.

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<ByteVector>(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<uint8_t>(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");

components/socket/include/dscp.hpp

@@ -0,0 +1,70 @@
+#pragma once
+
+#include <cstdint>
+
+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 <sys/termios.h> 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<Dscp>(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<uint8_t>(dscp) << 2; }
+
+} // namespace espp

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<sock_type_t>(-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);

components/socket/include/udp_socket.hpp

@@ -7,7 +7,9 @@
 #include <string_view>
 #include <vector>
 
+#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<espp::Dscp> 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<Dscp>(0-63); out-of-range values are rejected
+                                      with a warning. */
   };
 
   struct SendConfig {