Socket Reactor

The SocketReactor multiplexes many receiver sockets on a single select() event-loop thread and dispatches each socket’s read + user callback onto a shared ThreadPool, instead of dedicating one thread (one espp::Task) to every receiving socket. This collapses “N receiver threads” into “1 loop thread + a small fixed pool”, which can additionally be shared across several subsystems.

To keep per-socket handling correct under level-triggered select(), the reactor is one-shot: when a socket is reported readable it is disarmed and a job is submitted to the pool, and it is re-armed only after that job completes. This guarantees at most one in-flight handler per socket (so per-socket ordering is preserved and there is no concurrent recv on one fd), while different sockets’ handlers run concurrently on the pool. If the pool is saturated the socket is simply left for the next select() to re-report - the datagram stays buffered, giving natural backpressure with no loss.

Registration changes, re-arming, and stop are made immediately responsive by a loopback UDP “wakeup” socket that is also in the select set.

The reactor drives:

  • UDP receivers via add_udp_receiver() (binds the socket, calls the receive callback on a pool worker, and sends any returned response), replacing a per-socket UdpSocket::start_receiving() thread.

  • TCP listeners via add_tcp_listener() (accepts connections and hands each new client to a callback) and TCP streams via add_tcp_stream() (reads a connected socket, invokes a data callback, and auto-unregisters on disconnect) - so a whole TCP server runs with no accept thread and no thread-per-client.

A low-level add_fd() / remove() pair is also available.

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 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 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<Dscp>(0-63); out-of-range values are rejected with a warning.

Note

Lifetime: registered sockets and callbacks must outlive their registration. stop() (and the destructor) waits for any in-flight handler to finish, so the guaranteed-safe teardown order is to stop() / destroy the reactor first, then destroy the sockets. remove() is asynchronous with respect to a handler that is already running for that id.

Note

The select() backend uses fd_set, so on POSIX/lwip a registered file descriptor must be below FD_SETSIZE; registration rejects an fd at or above that limit. On lwip, socket fds occupy [FD_SETSIZE - CONFIG_LWIP_MAX_SOCKETS, FD_SETSIZE), so this is only hit if those limits are raised past the compiled FD_SETSIZE.

Code examples for the reactor are provided in the socket example folder (the “Socket reactor”, “Reactor priority bands”, and “TCP reactor” scenarios).

API Reference

Header File

Classes

class SocketReactor : public espp::BaseComponent

A single-threaded select() event loop that multiplexes many receiver sockets and dispatches their read handling onto a shared thread pool.

Instead of dedicating one background thread (one espp::Task) to every receiving socket, register the sockets with a single SocketReactor. One loop thread waits in ::select() on all registered file descriptors at once; when a socket becomes readable the reactor hands that socket’s read + user callback to a espp::ThreadPool worker. This collapses “N receiver threads” down to “1 loop thread + a small fixed pool”, which can additionally be shared across several subsystems.

To keep per-socket handling correct under level-triggered select(), the reactor is one-shot: when a socket is reported readable it is disarmed (removed from the interest set) and a job is submitted; the socket is re-armed only after that job completes. This guarantees at most one in-flight handler per socket (so per-socket ordering is preserved and there is no concurrent recv on the same fd), while different sockets’ handlers run concurrently on the pool.

Registration changes, re-arming, and stop are made immediately responsive by a loopback UDP “wakeup” socket that is also in the select set: poking it interrupts select() at once.

Each registration carries a priority band (espp::QosBand, default Normal). When one 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 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 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.

Socket Reactor UDP Example

  espp::SocketReactor reactor({.log_level = espp::Logger::Verbosity::WARN});

  espp::UdpSocket server_a({.log_level = espp::Logger::Verbosity::WARN});
  espp::UdpSocket server_b({.log_level = espp::Logger::Verbosity::WARN});
  auto id_a = reactor.add_udp_receiver(
      server_a,
      {.port = port_a, .buffer_size = kMaxPacketSize, .on_receive_callback = echo_reversed});
  auto id_b = reactor.add_udp_receiver(
      server_b,
      {.port = port_b, .buffer_size = kMaxPacketSize, .on_receive_callback = echo_reversed});

Socket Reactor TCP Example

    reactor.add_tcp_listener(server, [&](std::unique_ptr<espp::TcpSocket> client) {
      ++accepted;
      espp::TcpSocket *conn = nullptr;
      {
        std::lock_guard<std::mutex> lk(clients_mutex);
        clients.push_back(std::move(client));
        conn = clients.back().get();
      }
      // Register the accepted connection as a stream that echoes bytes back.
      reactor.add_tcp_stream(
          *conn, [](espp::TcpSocket &connection, ByteVector &data) { connection.transmit(data); },
          kMaxPacketSize, [&closed]() { closed = true; });
    });

Socket Reactor Priority Bands 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});

Note

Lifetime. Registered sockets and callbacks must outlive their registration. stop (and the destructor) waits for any in-flight handler to finish, so the guaranteed-safe teardown order is: stop() / destroy the reactor first, then destroy the sockets. remove is *asynchronous* with respect to a handler that is already running for that id (it unregisters, but a handler mid-recv() will still finish); do not free a socket immediately after remove() while its handler may be executing - unregister and then rely on reactor teardown, or ensure the socket outlives the reactor. stop (and destroying the reactor) must NOT be called from within a handler: it waits for the calling handler to finish (and joins an owned pool), which would deadlock - stop from another thread. A stop() invoked from a handler is refused + logged.

Note

The select() backend uses fd_set, which on POSIX/lwip can only hold file descriptors with value < FD_SETSIZE. Registration rejects an fd at or above that limit. On lwip, socket fds occupy [FD_SETSIZE - CONFIG_LWIP_MAX_SOCKETS, FD_SETSIZE), so this is only hit if those limits are raised past the compiled FD_SETSIZE.

Public Types

using Id = std::uint32_t

Opaque handle for a registration, returned by the add_* methods and passed to remove. INVALID_ID (0) is never returned on success.

using ReadHandler = std::function<void()>

Low-level read handler, invoked on a thread-pool worker when the registered socket is readable.

Note

The reactor guarantees at most one in-flight invocation per registration (disarm-on-dispatch, re-arm on completion), so the handler need not guard against concurrent calls for the same socket. The handler is expected to read from the socket (draining it as appropriate) and process the data.

using AcceptCallback = std::function<void(std::unique_ptr<espp::TcpSocket> client)>

Called (on a pool worker) when a listening TcpSocket accepts a new client. The consumer takes ownership of client and typically registers it with add_tcp_stream() on this same reactor.

using StreamCallback = std::function<void(espp::TcpSocket &connection, std::vector<uint8_t> &data)>

Called (on a pool worker) with data read from a connected TcpSocket.

using CloseCallback = std::function<void()>

Called (on a pool worker) when a connected TcpSocket reaches EOF / closes. After this fires the stream is automatically unregistered.

using RemovedCallback = std::function<void()>

Removal-completion notification for remove(): see the two-argument remove() overload for the exact invocation guarantees.

Public Functions

explicit SocketReactor(const Config &config)

Construct the reactor (and, if configured, start it).

Parameters:

config – Configuration for the reactor.

~SocketReactor()

Stop the loop and unregister everything.

bool start()

Start the select() loop and (if owned) the thread pool.

Returns:

true if the reactor is running.

void stop()

Stop the loop, wait for any in-flight handlers to finish, and (if owned) stop the thread pool.

bool is_running() const
Returns:

true if the loop is running.

Id add_udp_receiver(espp::UdpSocket &socket, const espp::UdpSocket::ReceiveConfig &receive_config)

Bind socket per receive_config and register it to receive on this reactor. When a datagram arrives the reactor calls socket.receive() and invokes receive_config.on_receive_callback with the data and sender on a pool worker; if the callback returns 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).

Parameters:

  • socket – A UdpSocket to bind and receive on. Must outlive the registration (call remove() before destroying it).

  • receive_config – Port / multicast / buffer_size / callback / band / dscp config.

Returns:

A registration Id, or INVALID_ID on failure.

Id add_tcp_listener(espp::TcpSocket &listener, const AcceptCallback &on_accept, QosBand band = QosBand::Normal)

Register a listening TcpSocket. When a connection is pending the reactor calls listener.accept() and invokes on_accept with the new client on a pool worker. The listener stays registered.

Parameters:

  • listener – A TcpSocket that has already been bind()+listen()’d. Must outlive the registration.

  • on_accept – Callback given ownership of each accepted client.

  • band – Priority band to dispatch accept handling at (see espp::QosBand; default Normal = pre-band behavior).

Returns:

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 = {}, QosBand band = QosBand::Normal)

Register a connected TcpSocket for reading. When it is readable the reactor reads up to buffer_size bytes and invokes on_data on a pool worker. On EOF / disconnect it invokes on_close (if set) and automatically unregisters the stream.

Parameters:

  • connection – A connected TcpSocket (e.g. from add_tcp_listener’s callback or TcpSocket::connect). Must outlive the registration.

  • on_data – Callback given the connection and the bytes read.

  • buffer_size – Max bytes to read per readable event.

  • on_close – Optional callback fired once when the peer closes.

  • band – Priority band to dispatch read handling at (see espp::QosBand; default Normal = pre-band behavior).

Returns:

A registration Id, or INVALID_ID on failure.

Id add_fd(sock_type_t fd, ReadHandler handler, QosBand band = QosBand::Normal)

Low-level registration: watch fd for readability and run handler (on a pool worker) each time it is readable.

Parameters:

  • fd – A valid socket file descriptor (see Socket::native_handle()).

  • handler – Handler that reads/processes the socket.

  • band – Priority band to dispatch the handler at (see espp::QosBand; default Normal = pre-band behavior).

Returns:

A registration Id, or INVALID_ID on failure.

bool remove(Id id)

Unregister a socket. Safe to call from any thread, including from within a running handler. If a handler for this id is currently in-flight, the entry is erased once it completes.

Parameters:

id – The registration Id returned by an add_* method.

Returns:

true if the id was found.

bool remove(Id id, RemovedCallback on_removed)

Unregister a socket and be notified when the removal has fully completed - i.e. the registration is erased AND any handler that was running or pending for it has finished, so no reactor code can reference the socket/fd anymore. Use this to know when it is safe to destroy the socket (the reactor’s handlers hold references to it, and remove() itself never blocks).

Invocation guarantees for on_removed:

  • Invoked EXACTLY ONCE, and only when this call returns true (an unknown id returns false and never invokes it).

  • Thread: the CALLER’s thread (synchronously, before remove() returns) when no handler is in flight at remove() time; otherwise the pool worker that finishes the in-flight handler, or the reactor loop thread when a pending dispatch is reverted (pool saturated). Callers must be prepared for any of the three.

  • Runs with NO reactor lock held: it may re-enter the reactor (add_*, remove), but must return promptly (it can run on a worker or the loop) and must not call stop() when it runs from a worker/loop context.

  • Residual fd caveat: an already-blocked select() may still have the fd in its interest set for one iteration (the reactor wakes itself on removal, so the window is tiny). Closing the fd from the callback is safe; if the OS immediately reuses the fd number, the worst case is one bounded spurious wake of the new registration (handlers must already tolerate spurious readiness - see add_udp_receiver()’s receive bound).

  • Calling remove() again for an id whose removal is still pending chains the callbacks (both fire on completion).

Parameters:

  • id – The registration Id returned by an add_* method.

  • on_removed – Invoked once the removal has fully completed (may be empty, making this identical to remove(id)).

Returns:

true if the id was found.

size_t num_registered() const
Returns:

the number of currently registered sockets.

inline const std::string &get_name() const

Get the name of the component

Note

This is the tag of the logger

Returns:

A const reference to the name of the component

inline void set_log_tag(const std::string_view &tag)

Set the tag for the logger

Parameters:

tag – The tag to use for the logger

inline espp::Logger::Verbosity get_log_level() const

Get the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Returns:

The verbosity level of the logger

inline void set_log_level(espp::Logger::Verbosity level)

Set the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Parameters:

level – The verbosity level to use for the logger

inline void set_log_verbosity(espp::Logger::Verbosity level)

Set the log verbosity for the logger

See also

set_log_level

See also

Logger::Verbosity

See also

Logger::set_verbosity

Note

This is a convenience method that calls set_log_level

Parameters:

level – The verbosity level to use for the logger

inline espp::Logger::Verbosity get_log_verbosity() const

Get the log verbosity for the logger

See also

get_log_level

See also

Logger::Verbosity

See also

Logger::get_verbosity

Note

This is a convenience method that calls get_log_level

Returns:

The verbosity level of the logger

inline void set_log_rate_limit(std::chrono::duration<float> rate_limit)

Set the rate limit for the logger

See also

Logger::set_rate_limit

Note

Only calls to the logger that have _rate_limit suffix will be rate limited

Parameters:

rate_limit – The rate limit to use for the logger

Public Static Attributes

static constexpr Id INVALID_ID = 0

Never a valid registration id.

struct Config

Configuration for the SocketReactor.

Public Members

std::shared_ptr<espp::ThreadPool> thread_pool{nullptr}

Pool to dispatch handlers on. If null, the reactor creates and owns one from pool_config; provide a shared pool to share workers across subsystems.

espp::ThreadPool::Config pool_config  {.worker_count = 2,.worker_task_config = {.name = "SocketReactor pool",.stack_size_bytes = 4096,.priority = 5}}

Used only when thread_pool is null.

espp::Task::BaseConfig loop_task_config  {.name = "SocketReactor",.stack_size_bytes = 4096,.priority = 5,.core_id =-1}

Config for the single select() loop task.

std::chrono::microseconds select_timeout{std::chrono::seconds(1)}

Max time select() blocks; the wakeup socket makes registration/stop responsive regardless of this.

bool auto_start = {true}

Start the loop (and owned pool) on construction.

espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}

Logger verbosity.

Header File