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.

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” and “TCP reactor” scenarios).

API Reference

Header File

components/socket/include/socket_reactor.hpp

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.

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; });
    });

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.

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.

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

Returns:

A registration Id, or INVALID_ID on failure.

Id add_tcp_listener(espp::TcpSocket &listener, const AcceptCallback &on_accept)

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.

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 = {})

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.

Returns:

A registration Id, or INVALID_ID on failure.

Id add_fd(sock_type_t fd, ReadHandler handler)

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.

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.

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.