diff --git a/components/socket/CMakeLists.txt b/components/socket/CMakeLists.txt index 5bc6ad34a..be83d9708 100644 --- a/components/socket/CMakeLists.txt +++ b/components/socket/CMakeLists.txt @@ -1,4 +1,4 @@ idf_component_register( INCLUDE_DIRS "include" SRC_DIRS "src" - REQUIRES base_component task lwip) + REQUIRES base_component task thread_pool lwip) diff --git a/components/socket/README.md b/components/socket/README.md index f479673f8..c6c96f489 100644 --- a/components/socket/README.md +++ b/components/socket/README.md @@ -6,7 +6,9 @@ The network APIs provide a useful abstraction over POSIX sockets enabling easily starting client/server sockets and allowing their use with std::function callbacks for servers. -Currently, UDP and TCP sockets are supported. +Currently, UDP and TCP sockets are supported. A `SocketReactor` is also provided +for servicing many receiver sockets on a single `select()` loop plus a thread +pool, instead of one thread per socket. **Table of Contents** @@ -15,6 +17,7 @@ Currently, UDP and TCP sockets are supported. - [Base Socket](#base-socket) - [UDP Socket](#udp-socket) - [TCP Socket](#tcp-socket) + - [Socket Reactor](#socket-reactor) - [Example](#example) @@ -39,6 +42,8 @@ The `UdpSocket` API supports both one-shot sends and long-running receive tasks: * `start_receiving(...)` starts a task that continuously receives datagrams and can optionally send a callback-produced response * `stop_receiving()` cleanly stops a blocked receive task during teardown +* `bind(...)` binds the socket for a server without starting a thread, so it can + be driven by a `SocketReactor` (see below) instead of its own receive task ## TCP Socket @@ -55,6 +60,35 @@ The `TcpSocket` API covers both client and server patterns: * `bind(...)`, `listen(...)`, and `accept()` for server-side flows * `close()` / `reinit()` helpers for teardown and reconnect paths +## 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 turns "N receiver threads" into "1 loop thread + a small +fixed pool" that can be shared across subsystems. + +To stay correct under level-triggered `select()`, the reactor is one-shot: a +readable socket is disarmed, a job is submitted to the pool, and the socket is +re-armed only after that job completes - so there is at most one in-flight +handler per socket (per-socket ordering is preserved and there is no concurrent +`recv` on one fd), while different sockets run concurrently. A loopback UDP +"wakeup" socket keeps registration changes, re-arming, and stop responsive. + +The `SocketReactor` API drives: + +* `add_udp_receiver(...)` - binds a `UdpSocket` and receives on it, replacing a + per-socket `UdpSocket::start_receiving()` thread +* `add_tcp_listener(...)` - accepts connections and hands each new client to a + callback (no accept thread) +* `add_tcp_stream(...)` - reads a connected `TcpSocket`, invokes a data callback, + and auto-unregisters on disconnect (no thread-per-client) +* low-level `add_fd(...)` / `remove(...)` + +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). + ## Example The [example](./example) shows the use of the classes provided by the `socket` @@ -66,3 +100,5 @@ and reconnect behavior, including: * scope-based teardown while tasks are active or blocked * request/response callbacks and timeout handling * 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) diff --git a/components/socket/example/main/socket_example.cpp b/components/socket/example/main/socket_example.cpp index 1f4274731..49347709e 100644 --- a/components/socket/example/main/socket_example.cpp +++ b/components/socket/example/main/socket_example.cpp @@ -10,6 +10,7 @@ #include #include "logger.hpp" +#include "socket_reactor.hpp" #include "task.hpp" #include "tcp_socket.hpp" #include "udp_socket.hpp" @@ -495,6 +496,377 @@ ScenarioResult run_tcp_connect_failure_scenario() { } return pass("TCP connect failure", "connect failed as expected"); } + +ScenarioResult run_socket_reactor_scenario() { + // Two UDP receiver sockets, on two ports, both driven by ONE reactor (one + // select() loop thread + a shared 2-worker pool) - no dedicated thread per + // socket. Each server echoes the request back reversed. + constexpr size_t port_a = 5020; + constexpr size_t port_b = 5021; + auto echo_reversed = [](const ByteVector &data, const espp::Socket::Info &) { + return std::optional(reversed(data)); + }; + + //! [socket reactor 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 example] + + if (id_a == espp::SocketReactor::INVALID_ID || id_b == espp::SocketReactor::INVALID_ID) { + return fail("Socket reactor", "failed to register one or both UDP receivers"); + } + if (reactor.num_registered() != 2) { + return fail("Socket reactor", + fmt::format("expected 2 registrations, got {}", reactor.num_registered())); + } + + // Send a distinct request to each server and verify each reversed response. + 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(port_a, 0x10)) { + return fail("Socket reactor", "server A did not echo correctly via the reactor"); + } + if (!send_and_check(port_b, 0x80)) { + return fail("Socket reactor", "server B did not echo correctly via the reactor"); + } + + // Unregister one socket and confirm the count drops. + reactor.remove(id_a); + if (!wait_until([&] { return reactor.num_registered() == 1; }, 500ms)) { + return fail("Socket reactor", fmt::format("expected 1 registration after remove, got {}", + reactor.num_registered())); + } + + return pass("Socket reactor", + "2 UDP sockets multiplexed on 1 select loop + shared pool, both echoed"); +} + +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 + // echoes bytes back - no accept thread and no thread-per-client. + constexpr size_t port = 6010; + auto request = make_payload(256, 0x40); + std::atomic_bool got_echo{false}; + std::atomic_int accepted{0}; + std::atomic_bool closed{false}; + + // Accepted server-side client sockets must outlive their reactor stream + // registration, so keep them alive here (populated from a pool worker). + std::mutex clients_mutex; + std::vector> clients; + + { + espp::SocketReactor reactor({.log_level = espp::Logger::Verbosity::WARN}); + + espp::TcpSocket server({.log_level = espp::Logger::Verbosity::WARN}); + if (!server.bind(port) || !server.listen(kMaxConnections)) { + return fail("TCP reactor", "failed to bind/listen"); + } + + //! [socket reactor tcp example] + reactor.add_tcp_listener(server, [&](std::unique_ptr client) { + ++accepted; + espp::TcpSocket *conn = nullptr; + { + std::lock_guard 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 tcp example] + + ByteVector echo; + espp::TcpSocket client({.log_level = espp::Logger::Verbosity::WARN}); + if (!client.connect({.ip_address = kLoopbackAddress, .port = port})) { + return fail("TCP reactor", "client failed to connect"); + } + client.transmit(request, {.wait_for_response = true, + .response_size = kMaxPacketSize, + .on_response_callback = + [&](const ByteVector &r) { + echo = r; + got_echo = true; + }, + .response_timeout = 1s}); + + if (!got_echo.load() || echo != request) { + return fail("TCP reactor", "did not receive correct echo via the reactor"); + } + if (accepted.load() != 1) { + return fail("TCP reactor", fmt::format("expected 1 accept, got {}", accepted.load())); + } + + // Close the client and verify the server-side stream notices the disconnect + // and auto-unregisters, leaving only the listener. + client.close(); + if (!wait_until([&] { return closed.load(); }, 1s)) { + return fail("TCP reactor", "server did not observe client disconnect"); + } + if (!wait_until([&] { return reactor.num_registered() == 1; }, 1s)) { + return fail("TCP reactor", fmt::format("expected 1 registration after disconnect, got {}", + reactor.num_registered())); + } + } + return pass("TCP reactor", "listener + stream echo on one reactor, auto-removed on disconnect"); +} + +// Helper: send `request` to a UDP echo server on `port` and check the reversed +// reply comes back. Returns true on a correct round-trip. +bool udp_echo_roundtrip(size_t port, const ByteVector &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 == reversed(request); +} + +ScenarioResult run_reactor_shared_pool_scenario() { + // One externally-owned ThreadPool shared with the reactor, multiplexing three + // UDP echo receivers, plus a dynamic remove() while running. + constexpr size_t base_port = 5030; + auto echo = [](const ByteVector &d, const espp::Socket::Info &) { + return std::optional(reversed(d)); + }; + auto pool = std::make_shared( + espp::ThreadPool::Config{.worker_count = 3, .log_level = espp::Logger::Verbosity::WARN}); + // Declare the sockets before the reactor so the reactor is destroyed first. + std::vector> servers; + { + espp::SocketReactor reactor({.thread_pool = pool, .log_level = espp::Logger::Verbosity::WARN}); + std::vector ids; + for (int i = 0; i < 3; ++i) { + servers.push_back(std::make_unique( + espp::UdpSocket::Config{.log_level = espp::Logger::Verbosity::WARN})); + auto id = reactor.add_udp_receiver( + *servers.back(), + {.port = base_port + i, .buffer_size = kMaxPacketSize, .on_receive_callback = echo}); + if (id == espp::SocketReactor::INVALID_ID) { + return fail("Reactor shared pool", fmt::format("failed to register receiver {}", i)); + } + ids.push_back(id); + } + if (reactor.num_registered() != 3) { + return fail("Reactor shared pool", "expected 3 registrations"); + } + for (int i = 0; i < 3; ++i) { + if (!udp_echo_roundtrip(base_port + i, make_payload(200, static_cast(i * 10 + 1)))) { + return fail("Reactor shared pool", fmt::format("receiver {} echo failed", i)); + } + } + // Dynamically drop the middle receiver; the others keep working. + reactor.remove(ids[1]); + if (!wait_until([&] { return reactor.num_registered() == 2; }, 500ms)) { + return fail("Reactor shared pool", "count did not drop to 2 after remove()"); + } + if (!udp_echo_roundtrip(base_port + 0, make_payload(200, 0x01)) || + !udp_echo_roundtrip(base_port + 2, make_payload(200, 0x15))) { + return fail("Reactor shared pool", "remaining receivers broke after remove()"); + } + reactor.stop(); // quiesce before the sockets/pool go out of scope + } + return pass("Reactor shared pool", "3 UDP receivers on a shared pool; dynamic remove works"); +} + +ScenarioResult run_reactor_lifecycle_scenario() { + espp::SocketReactor reactor({.auto_start = false, .log_level = espp::Logger::Verbosity::WARN}); + if (reactor.is_running()) { + return fail("Reactor lifecycle", "running before start()"); + } + // Invalid registrations are rejected with INVALID_ID. + if (reactor.add_fd(-1, []() {}) != espp::SocketReactor::INVALID_ID) { + return fail("Reactor lifecycle", "invalid fd was not rejected"); + } + if (reactor.add_fd(0, nullptr) != espp::SocketReactor::INVALID_ID) { + return fail("Reactor lifecycle", "null handler was not rejected"); + } + if (!reactor.start() || !reactor.is_running()) { + return fail("Reactor lifecycle", "start() failed"); + } + { + constexpr size_t port = 5040; + espp::UdpSocket server({.log_level = espp::Logger::Verbosity::WARN}); + auto id = reactor.add_udp_receiver( + server, {.port = port, + .buffer_size = kMaxPacketSize, + .on_receive_callback = [](const ByteVector &d, const espp::Socket::Info &) { + return std::optional(reversed(d)); + }}); + if (id == espp::SocketReactor::INVALID_ID) { + return fail("Reactor lifecycle", "register after manual start failed"); + } + if (!udp_echo_roundtrip(port, make_payload(128, 0x55))) { + return fail("Reactor lifecycle", "echo after manual start failed"); + } + reactor.stop(); + } + if (reactor.is_running()) { + return fail("Reactor lifecycle", "still running after stop()"); + } + return pass("Reactor lifecycle", + "auto_start=false then start(); invalid registrations rejected; stop() works"); +} + +ScenarioResult run_reactor_tcp_multiclient_scenario() { + // One reactor accepts and services two concurrent TCP clients (no accept + // thread, no thread-per-client). + constexpr size_t port = 6020; + std::atomic_int accepted{0}; + std::mutex clients_mutex; + std::vector> clients; + { + espp::SocketReactor reactor({.log_level = espp::Logger::Verbosity::WARN}); + espp::TcpSocket server({.log_level = espp::Logger::Verbosity::WARN}); + if (!server.bind(port) || !server.listen(4)) { + return fail("Reactor TCP multi-client", "failed to bind/listen"); + } + reactor.add_tcp_listener(server, [&](std::unique_ptr client) { + ++accepted; + espp::TcpSocket *conn = nullptr; + { + std::lock_guard lk(clients_mutex); + clients.push_back(std::move(client)); + conn = clients.back().get(); + } + reactor.add_tcp_stream( + *conn, [](espp::TcpSocket &c, ByteVector &data) { c.transmit(data); }, kMaxPacketSize); + }); + + // Two clients each round-trip a distinct payload concurrently. + auto client_roundtrip = [&](uint8_t seed) -> bool { + auto request = make_payload(200, seed); + ByteVector echo; + std::atomic_bool got{false}; + espp::TcpSocket client({.log_level = espp::Logger::Verbosity::WARN}); + if (!client.connect({.ip_address = kLoopbackAddress, .port = port})) { + return false; + } + client.transmit(request, {.wait_for_response = true, + .response_size = kMaxPacketSize, + .on_response_callback = + [&](const ByteVector &r) { + echo = r; + got = true; + }, + .response_timeout = 1s}); + return got.load() && echo == request; + }; + if (!client_roundtrip(0x11) || !client_roundtrip(0x22)) { + return fail("Reactor TCP multi-client", "one or both clients did not echo"); + } + if (!wait_until([&] { return accepted.load() == 2; }, 1s)) { + return fail("Reactor TCP multi-client", + fmt::format("expected 2 accepts, got {}", accepted.load())); + } + reactor.stop(); // quiesce before clients/server go out of scope + } + return pass("Reactor TCP multi-client", "2 concurrent clients accepted + echoed on one reactor"); +} + +ScenarioResult run_udp_send_overloads_scenario() { + // Exercise the string_view and span send overloads and verify the server sees + // the correct sender address/port. + constexpr size_t port = 5050; + std::atomic_bool saw_sender{false}; + std::string sender_addr; + espp::UdpSocket server({.log_level = espp::Logger::Verbosity::WARN}); + auto server_task_config = make_task_config("UdpOverloadServer"); + server.start_receiving( + server_task_config, + {.port = port, + .buffer_size = kMaxPacketSize, + .on_receive_callback = [&](const ByteVector &d, const espp::Socket::Info &sender) { + sender_addr = sender.address; + saw_sender = true; + return std::optional(reversed(d)); + }}); + + // string_view overload + { + std::string_view msg = "hello-string-view"; + ByteVector resp; + std::atomic_bool got{false}; + espp::UdpSocket client({.log_level = espp::Logger::Verbosity::WARN}); + client.send(msg, {.ip_address = kLoopbackAddress, + .port = port, + .wait_for_response = true, + .response_size = kMaxPacketSize, + .on_response_callback = + [&](const ByteVector &r) { + resp = r; + got = true; + }, + .response_timeout = 500ms}); + ByteVector expected(msg.begin(), msg.end()); + if (!got.load() || resp != reversed(expected)) { + return fail("UDP send overloads", "string_view send/echo failed"); + } + } + // span overload + { + auto payload = make_payload(64, 0x33); + ByteVector resp; + std::atomic_bool got{false}; + espp::UdpSocket client({.log_level = espp::Logger::Verbosity::WARN}); + client.send(std::span{payload.data(), payload.size()}, + {.ip_address = kLoopbackAddress, + .port = port, + .wait_for_response = true, + .response_size = kMaxPacketSize, + .on_response_callback = + [&](const ByteVector &r) { + resp = r; + got = true; + }, + .response_timeout = 500ms}); + if (!got.load() || resp != reversed(payload)) { + return fail("UDP send overloads", "span send/echo failed"); + } + } + server.stop_receiving(); + if (!saw_sender.load() || sender_addr != kLoopbackAddress) { + return fail("UDP send overloads", fmt::format("sender address wrong: '{}'", sender_addr)); + } + return pass("UDP send overloads", "string_view + span overloads round-trip; sender info correct"); +} } // namespace extern "C" void app_main(void) { @@ -506,7 +878,7 @@ extern "C" void app_main(void) { .log_level = espp::Logger::Verbosity::INFO}); std::vector results; - results.reserve(9); + results.reserve(15); auto run_and_record = [&results](auto &&scenario_runner, std::string_view name) { print_scenario_start(name); @@ -525,14 +897,36 @@ extern "C" void app_main(void) { run_and_record(run_tcp_response_reconnect_scenario, "TCP response/reconnect"); 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_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"); + run_and_record(run_reactor_tcp_multiclient_scenario, "Reactor TCP multi-client"); + run_and_record(run_udp_send_overloads_scenario, "UDP send overloads + sender info"); + + // ---- final summary ---- auto passed_count = std::count_if(results.begin(), results.end(), [](const auto &result) { return result.passed; }); + const auto failed_count = results.size() - static_cast(passed_count); + fmt::print("\n"); fmt::print(fg(fmt::terminal_color::cyan) | fmt::emphasis::bold, - "Socket example summary: {}/{} scenarios passed\n", passed_count, results.size()); + "======== Socket example summary: {}/{} scenarios passed ========\n", passed_count, + results.size()); for (const auto &result : results) { print_scenario_result(result); } + if (failed_count == 0) { + fmt::print(fg(fmt::terminal_color::green) | fmt::emphasis::bold, "ALL {} SCENARIOS PASSED\n", + results.size()); + } else { + fmt::print(fg(fmt::terminal_color::red) | fmt::emphasis::bold, "{} SCENARIO(S) FAILED:\n", + failed_count); + for (const auto &result : results) { + if (!result.passed) { + fmt::print(fg(fmt::terminal_color::red), " - {}: {}\n", result.name, result.detail); + } + } + } while (true) { std::this_thread::sleep_for(1s); diff --git a/components/socket/idf_component.yml b/components/socket/idf_component.yml index 51df95066..313cd6887 100644 --- a/components/socket/idf_component.yml +++ b/components/socket/idf_component.yml @@ -20,3 +20,4 @@ dependencies: version: '>=5.0' espp/base_component: '>=1.0' espp/task: '>=1.0' + espp/thread_pool: '>=1.0' diff --git a/components/socket/include/socket.hpp b/components/socket/include/socket.hpp index dbf69b9f1..864082ea8 100644 --- a/components/socket/include/socket.hpp +++ b/components/socket/include/socket.hpp @@ -152,6 +152,15 @@ class Socket : public BaseComponent { */ static bool is_valid_fd(sock_type_t socket_fd); + /** + * @brief Get the underlying native socket file descriptor / handle. + * @note Provided so an external event loop (e.g. SocketReactor) can add this + * socket to a select()/poll() set. The Socket retains ownership of the + * descriptor; do not close it directly. + * @return The socket file descriptor. + */ + sock_type_t native_handle() const { return socket_; } + /** * @brief Get the Socket::Info for the socket. * @details This will call getsockname() on the socket to get the diff --git a/components/socket/include/socket_reactor.hpp b/components/socket/include/socket_reactor.hpp new file mode 100644 index 000000000..c81e972ea --- /dev/null +++ b/components/socket/include/socket_reactor.hpp @@ -0,0 +1,261 @@ +#pragma once + +#include "socket_msvc.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "base_component.hpp" +#include "socket.hpp" +#include "task.hpp" +#include "tcp_socket.hpp" +#include "thread_pool.hpp" +#include "udp_socket.hpp" + +namespace espp { +/** + * @brief A single-threaded select() event loop that multiplexes many receiver + * sockets and dispatches their read handling onto a shared thread pool. + * + * @details Instead of dedicating one background thread (one @ref espp::Task) to + * every receiving socket, register the sockets with a single + * SocketReactor. One loop thread waits in @c ::select() on all + * registered file descriptors at once; when a socket becomes readable + * the reactor hands that socket's read + user callback to a + * @ref 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 + * @c select(), the reactor is one-shot: when a socket is reported + * readable it is @b disarmed (removed from the interest set) and a job + * is submitted; the socket is @b 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 + * @c 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 @c select() at once. + * + * @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() / + * destroy the reactor first, then destroy the sockets. @ref 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. @ref 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 @c fd_set, which on POSIX/lwip can only hold + * file descriptors with value < @c 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 @c FD_SETSIZE. + * + * \section socket_reactor_ex1 Socket Reactor UDP Example + * \snippet socket_example.cpp socket reactor example + * \section socket_reactor_ex2 Socket Reactor TCP Example + * \snippet socket_example.cpp socket reactor tcp example + */ +class SocketReactor : public BaseComponent { +public: + /// Opaque handle for a registration, returned by the add_* methods and + /// passed to @ref remove. INVALID_ID (0) is never returned on success. + using Id = std::uint32_t; + static constexpr Id INVALID_ID = 0; ///< Never a valid registration id. + + /** + * @brief 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 ReadHandler = std::function; + + /// Called (on a pool worker) when a listening TcpSocket accepts a new client. + /// The consumer takes ownership of @p client and typically registers it with + /// add_tcp_stream() on this same reactor. + using AcceptCallback = std::function client)>; + + /// Called (on a pool worker) with data read from a connected TcpSocket. + using StreamCallback = + std::function &data)>; + + /// Called (on a pool worker) when a connected TcpSocket reaches EOF / closes. + /// After this fires the stream is automatically unregistered. + using CloseCallback = std::function; + + /** + * @brief Configuration for the SocketReactor. + */ + struct Config { + std::shared_ptr thread_pool{ + nullptr}; ///< Pool to dispatch handlers on. If null, the reactor creates and owns one from + ///< @ref 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 @ref 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. + }; + + /** + * @brief Construct the reactor (and, if configured, start it). + * @param config Configuration for the reactor. + */ + explicit SocketReactor(const Config &config); + + /// Stop the loop and unregister everything. + ~SocketReactor(); + + /** + * @brief Start the select() loop and (if owned) the thread pool. + * @return true if the reactor is running. + */ + bool start(); + + /** + * @brief Stop the loop, wait for any in-flight handlers to finish, and (if + * owned) stop the thread pool. + */ + void stop(); + + /// @return true if the loop is running. + bool is_running() const; + + /** + * @brief Bind @p socket per @p receive_config and register it to receive on + * this reactor. When a datagram arrives the reactor calls + * @c socket.receive() and invokes @c 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. + * @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. + * @return A registration Id, or INVALID_ID on failure. + */ + Id add_udp_receiver(espp::UdpSocket &socket, + const espp::UdpSocket::ReceiveConfig &receive_config); + + /** + * @brief Register a listening TcpSocket. When a connection is pending the + * reactor calls @c listener.accept() and invokes @p on_accept with the + * new client on a pool worker. The listener stays registered. + * @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. + * @return A registration Id, or INVALID_ID on failure. + */ + Id add_tcp_listener(espp::TcpSocket &listener, const AcceptCallback &on_accept); + + /** + * @brief Register a connected TcpSocket for reading. When it is readable the + * reactor reads up to @p buffer_size bytes and invokes @p on_data on a + * pool worker. On EOF / disconnect it invokes @p on_close (if set) and + * automatically unregisters the stream. + * @param connection A connected TcpSocket (e.g. from add_tcp_listener's + * callback or TcpSocket::connect). Must outlive the registration. + * @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. + * @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 = {}); + + /** + * @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. + * @return A registration Id, or INVALID_ID on failure. + */ + Id add_fd(sock_type_t fd, ReadHandler handler); + + /** + * @brief 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. + * @param id The registration Id returned by an add_* method. + * @return true if the id was found. + */ + bool remove(Id id); + + /// @return the number of currently registered sockets. + size_t num_registered() const; + +protected: + struct Entry { + sock_type_t fd{static_cast(-1)}; ///< Watched file descriptor. + ReadHandler handler; ///< Handler run on the pool. + 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. + }; + + /// Validate an fd for registration: must be valid and (for the select() + /// backend, on POSIX/lwip) below FD_SETSIZE so FD_SET is not UB. Logs on + /// failure. + bool check_fd(sock_type_t fd) const; + + /// Allocate a fresh registration id (thread-safe, never INVALID_ID). + 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); + + /// One iteration of the select() loop (the loop task callback body). + bool loop_iteration(std::mutex &m, std::condition_variable &cv, bool &task_notified); + /// Run an entry's handler on the pool, then re-arm (or erase) it. + void dispatch(Id id); + /// Interrupt select() by poking the wakeup socket. + void wake(); + bool create_wakeup_socket(); + void close_wakeup_socket(); + static bool set_nonblocking(sock_type_t fd); + + Config config_; + std::shared_ptr pool_; + bool owns_pool_{false}; + std::unique_ptr loop_task_; + std::atomic running_{false}; + std::atomic in_flight_count_{0}; + + mutable std::mutex mutex_; ///< Guards entries_ and next_id_. + std::map entries_; + Id next_id_{INVALID_ID}; + + std::atomic wakeup_recv_{ + static_cast(-1)}; ///< Loopback UDP socket in the select set. Atomic so wake() + ///< (called from arbitrary threads) races safely with the + ///< create/close in start()/stop(). + struct sockaddr_in wakeup_addr_ {}; ///< Its own address, for self-sendto (stable after create). +}; +} // namespace espp diff --git a/components/socket/include/tcp_socket.hpp b/components/socket/include/tcp_socket.hpp index 2db3deeb6..b2583138c 100644 --- a/components/socket/include/tcp_socket.hpp +++ b/components/socket/include/tcp_socket.hpp @@ -6,6 +6,7 @@ #include #endif // _MSC_VER +#include #include #include #include @@ -235,7 +236,9 @@ class TcpSocket : public Socket { const std::chrono::seconds &interval = std::chrono::seconds{10}, int max_probes = 5); - bool connected_{false}; + // atomic: with SocketReactor a pool worker may run receive() (which clears + // connected_ on EOF) while another thread calls is_connected()/close(). + std::atomic connected_{false}; espp::Socket::Info remote_info_{}; }; } // namespace espp diff --git a/components/socket/include/udp_socket.hpp b/components/socket/include/udp_socket.hpp index f805d6f11..9c97c664f 100644 --- a/components/socket/include/udp_socket.hpp +++ b/components/socket/include/udp_socket.hpp @@ -158,6 +158,19 @@ class UdpSocket : public Socket { */ bool receive(size_t max_num_bytes, std::vector &data, Socket::Info &remote_info); + /** + * @brief Bind the socket as a server according to \p receive_config (bind to + * the port and, if requested, join the multicast group), without + * starting any receive thread. + * @note This is called for you by start_receiving(). Use it directly when + * driving the socket from an external event loop such as + * espp::SocketReactor, which reads the socket itself. + * @param receive_config ReceiveConfig describing the port / multicast setup. + * Its callback / buffer_size fields are not used by this method. + * @return true if the socket was bound (and joined the group, if multicast). + */ + bool bind(const ReceiveConfig &receive_config); + /** * @brief Configure a server socket and start a thread to continuously * receive and handle data coming in on that socket. diff --git a/components/socket/src/socket.cpp b/components/socket/src/socket.cpp index 583a8433f..660679950 100644 --- a/components/socket/src/socket.cpp +++ b/components/socket/src/socket.cpp @@ -24,7 +24,11 @@ struct sockaddr_in6 *Socket::Info::ipv6_ptr() { void Socket::Info::update() { if (raw.ss_family == PF_INET) { const auto *ipv4 = reinterpret_cast(&raw); - address = inet_ntoa(ipv4->sin_addr); + // inet_ntop into a local buffer (inet_ntoa returns a shared static buffer, + // which races when multiple threads - e.g. reactor pool workers - build + // Socket::Info concurrently). + char buf[INET_ADDRSTRLEN]; + address = inet_ntop(AF_INET, &ipv4->sin_addr, buf, sizeof(buf)) ? buf : ""; port = ntohs(ipv4->sin_port); #if !defined(ESP_PLATFORM) || LWIP_IPV6 } else if (raw.ss_family == PF_INET6) { @@ -48,7 +52,10 @@ void Socket::Info::from_sockaddr(const struct sockaddr_storage &source_address) void Socket::Info::from_sockaddr(const struct sockaddr_in &source_address) { memcpy(&raw, &source_address, sizeof(source_address)); - address = inet_ntoa(source_address.sin_addr); + // inet_ntop into a local buffer (inet_ntoa's shared static buffer is not + // thread-safe - see Info::update()). + char buf[INET_ADDRSTRLEN]; + address = inet_ntop(AF_INET, &source_address.sin_addr, buf, sizeof(buf)) ? buf : ""; port = ntohs(source_address.sin_port); } diff --git a/components/socket/src/socket_reactor.cpp b/components/socket/src/socket_reactor.cpp new file mode 100644 index 000000000..b4ad3c0d6 --- /dev/null +++ b/components/socket/src/socket_reactor.cpp @@ -0,0 +1,518 @@ +#include "socket_reactor.hpp" + +#include +#include + +#ifndef _MSC_VER +#include +#endif + +using namespace espp; + +namespace { +// True while the current thread is running a reactor dispatch handler. Used to +// detect (and refuse) a stop() called from within a handler, which would +// otherwise deadlock waiting for the calling handler to finish. +thread_local bool t_in_reactor_dispatch = false; +struct DispatchGuard { + bool prev; + DispatchGuard() + : prev(t_in_reactor_dispatch) { + t_in_reactor_dispatch = true; + } + ~DispatchGuard() { t_in_reactor_dispatch = prev; } +}; + +// Close a socket with the platform-correct call (Winsock sockets must use +// closesocket(), not ::close()). +void close_socket(sock_type_t fd) { +#if defined(_MSC_VER) + closesocket(fd); +#else + ::close(fd); +#endif +} +} // namespace + +SocketReactor::SocketReactor(const SocketReactor::Config &config) + : BaseComponent(config.loop_task_config.name, config.log_level) + , config_(config) { + if (config_.thread_pool) { + pool_ = config_.thread_pool; + owns_pool_ = false; + } else { + auto pool_config = config_.pool_config; + pool_config.log_level = config_.log_level; + pool_ = std::make_shared(pool_config); + owns_pool_ = true; + } + if (config_.auto_start) { + start(); + } +} + +SocketReactor::~SocketReactor() { + // Destroying the reactor from within one of its own handlers is a fatal + // lifetime violation: the handler is executing inside the object being freed, + // so there is no safe teardown (stop() would refuse, and we would then clear + // live loop/pool state - a use-after-free). Fail loudly instead. + if (t_in_reactor_dispatch) { + logger_.error("SocketReactor destroyed from within a reactor handler; terminating " + "(destroy/stop the reactor from another thread)."); + std::terminate(); + } + stop(); + std::lock_guard lock(mutex_); + entries_.clear(); +} + +bool SocketReactor::start() { + if (running_) { + return true; + } + if (!create_wakeup_socket()) { + logger_.error("Could not create wakeup socket, not starting"); + return false; + } + // Make sure the (owned) pool is running before we start dispatching to it. + if (owns_pool_ && !pool_->is_running()) { + pool_->start(); + } + running_ = true; + loop_task_ = espp::Task::make_unique({ + .callback = [this](std::mutex &m, std::condition_variable &cv, bool &task_notified) -> bool { + return loop_iteration(m, cv, task_notified); + }, + .task_config = config_.loop_task_config, + .log_level = config_.log_level, + }); + if (!loop_task_->start()) { + logger_.error("Could not start reactor loop task"); + running_ = false; + close_wakeup_socket(); + return false; + } + logger_.info("SocketReactor started"); + return true; +} + +void SocketReactor::stop() { + // stop() waits for in-flight handlers to finish (and, for an owned pool, joins + // the pool workers). Calling it from within a handler would therefore wait for + // the calling handler - a deadlock (and, for an owned pool, a self-join). + // Refuse rather than hang; the caller must stop the reactor from another thread. + if (t_in_reactor_dispatch) { + logger_.error("stop() called from within a reactor handler; refusing (it would deadlock). " + "Stop/destroy the reactor from another thread."); + return; + } + if (!running_ && !loop_task_) { + return; + } + // Signal the loop to exit and interrupt select() so it notices immediately. + running_ = false; + wake(); + if (loop_task_) { + loop_task_->stop(); // joins the loop thread + loop_task_.reset(); + } + // Wait for any handlers still running on the pool to finish before we tear + // down, so their captured `this` stays valid. The loop task is stopped, so no + // new dispatches begin and in_flight_count_ only decreases. For an owned pool + // pool_->stop() below also joins the workers; for a shared pool this wait is + // the only thing preventing a use-after-free, so it must not give up early - + // handlers are required to be finite (see the class documentation). + auto warn_at = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (in_flight_count_ > 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + if (std::chrono::steady_clock::now() > warn_at) { + logger_.warn("Still waiting for {} in-flight handler(s) to finish", in_flight_count_.load()); + warn_at += std::chrono::seconds(2); + } + } + if (owns_pool_ && pool_) { + pool_->stop(); + } + close_wakeup_socket(); + logger_.info("SocketReactor stopped"); +} + +bool SocketReactor::is_running() const { return running_; } + +bool SocketReactor::check_fd(sock_type_t fd) const { + if (!Socket::is_valid_fd(fd)) { + logger_.error("register: invalid socket fd"); + return false; + } +#if !defined(_MSC_VER) + // The select() backend uses fd_set, a bitmap indexed by fd value on + // POSIX/lwip; FD_SET(fd) with fd >= FD_SETSIZE is undefined behavior. (On + // Winsock, fd_set is a bounded array of SOCKETs, so the value is not the + // limit - the count is - and this check does not apply.) + if (fd >= FD_SETSIZE) { + logger_.error("register: socket fd {} is >= FD_SETSIZE ({}); raise FD_SETSIZE (and " + "CONFIG_LWIP_MAX_SOCKETS on ESP) to watch this many sockets", + static_cast(fd), static_cast(FD_SETSIZE)); + return false; + } +#endif + return true; +} + +SocketReactor::Id SocketReactor::allocate_id() { + std::lock_guard lock(mutex_); + // Monotonic ids, skipping INVALID_ID on wrap. + do { + ++next_id_; + } while (next_id_ == INVALID_ID); + return next_id_; +} + +void SocketReactor::insert_entry(Id id, sock_type_t fd, ReadHandler handler) { + { + std::lock_guard lock(mutex_); + entries_[id] = Entry{.fd = fd, .handler = std::move(handler)}; + } + wake(); // interrupt select() so the new fd is picked up +} + +SocketReactor::Id SocketReactor::add_fd(sock_type_t fd, SocketReactor::ReadHandler handler) { + if (!handler) { + logger_.error("add_fd: null handler"); + return INVALID_ID; + } + if (!check_fd(fd)) { + return INVALID_ID; + } + Id id = allocate_id(); + insert_entry(id, fd, std::move(handler)); + return id; +} + +SocketReactor::Id +SocketReactor::add_udp_receiver(espp::UdpSocket &socket, + const espp::UdpSocket::ReceiveConfig &receive_config) { + if (!socket.bind(receive_config)) { + logger_.error("add_udp_receiver: could not bind socket to port {}", receive_config.port); + return INVALID_ID; + } + const auto callback = receive_config.on_receive_callback; + const auto buffer_size = receive_config.buffer_size; + sock_type_t fd = socket.native_handle(); + auto handler = [this, &socket, callback, buffer_size]() { + std::vector data; + Socket::Info sender; + if (!socket.receive(buffer_size, data, sender)) { + // Transient (e.g. spurious wakeup) - the socket will be re-armed and we + // will try again on the next readable event. + return; + } + if (!callback) { + return; + } + auto maybe_response = callback(data, sender); + if (!maybe_response.has_value() || maybe_response->empty()) { + return; + } + auto *sender_address = sender.ipv4_ptr(); + auto &response = maybe_response.value(); + int sent = ::sendto(socket.native_handle(), reinterpret_cast(response.data()), + response.size(), 0, reinterpret_cast(sender_address), + sizeof(*sender_address)); + if (sent < 0) { + logger_.warn("Failed to send UDP response to {}", sender); + } + }; + return add_fd(fd, std::move(handler)); +} + +SocketReactor::Id SocketReactor::add_tcp_listener(espp::TcpSocket &listener, + const AcceptCallback &on_accept) { + sock_type_t fd = listener.native_handle(); + auto handler = [this, &listener, on_accept]() { + // select() reported the listener readable, so accept() returns immediately. + auto client = listener.accept(); + if (!client) { + return; // transient (e.g. the pending connection went away) + } + if (on_accept) { + on_accept(std::move(client)); + } + }; + return add_fd(fd, std::move(handler)); +} + +SocketReactor::Id SocketReactor::add_tcp_stream(espp::TcpSocket &connection, + const StreamCallback &on_data, size_t buffer_size, + const CloseCallback &on_close) { + sock_type_t fd = connection.native_handle(); + if (!check_fd(fd)) { + return INVALID_ID; + } + // Reserve the id first so the handler can unregister itself on disconnect. + Id id = allocate_id(); + auto handler = [this, &connection, on_data, on_close, buffer_size, id]() { + std::vector data; + if (connection.receive(data, buffer_size)) { + if (on_data) { + on_data(connection, data); + } + return; + } + // receive() returned false after a readable event: this is terminal - either + // a clean EOF (recv == 0, is_connected() now false) or a socket error such + // as a peer RST (recv < 0). Both mean stop watching this connection; + // treating a non-EOF error as "transient" here would busy-loop the reactor + // (select keeps reporting the fd readable). Fire on_close and unregister. + if (on_close) { + on_close(); + } + remove(id); + }; + insert_entry(id, fd, std::move(handler)); + return id; +} + +bool SocketReactor::remove(SocketReactor::Id id) { + bool found = false; + { + std::lock_guard lock(mutex_); + auto it = entries_.find(id); + if (it != entries_.end()) { + found = true; + if (it->second.in_flight) { + // A handler is running; defer erasure until dispatch() completes. + it->second.remove_requested = true; + } else { + entries_.erase(it); + } + } + } + if (found) { + wake(); + } + return found; +} + +size_t SocketReactor::num_registered() const { + std::lock_guard lock(mutex_); + return entries_.size(); +} + +void SocketReactor::dispatch(SocketReactor::Id id) { + // Decrement the in-flight count when this dispatch finishes by ANY path (early + // return, or an exception thrown from the handler), so stop()'s wait on + // in_flight_count_ can never hang. + struct CountGuard { + std::atomic &count; + ~CountGuard() { --count; } + } count_guard{in_flight_count_}; + + ReadHandler handler; + { + std::lock_guard lock(mutex_); + auto it = entries_.find(id); + if (it == entries_.end()) { + return; // removed before we got here + } + handler = it->second.handler; // copy so we can run it unlocked + } + // Run the user handler without holding the lock (it may call back into the + // reactor, block on a mutex, etc.). Mark the thread so a stop() invoked from + // within the handler is detected rather than deadlocking, and never let a + // handler exception escape into the pool worker (it would kill the worker and + // leave the entry stuck in_flight). + if (handler) { + DispatchGuard guard; +#if defined(__cpp_exceptions) && __cpp_exceptions + try { + handler(); + } catch (const std::exception &e) { + logger_.error("Exception in reactor handler: {}", e.what()); + } catch (...) { + logger_.error("Unknown exception in reactor handler"); + } +#else + // C++ exceptions are disabled (e.g. the ESP-IDF default), so a throwing + // handler would abort regardless; call it directly. The RAII CountGuard + // still keeps in_flight_count_ consistent for normal returns. + handler(); +#endif + } + bool wake_needed = false; + { + std::lock_guard lock(mutex_); + auto it = entries_.find(id); + if (it != entries_.end()) { + it->second.in_flight = false; + if (it->second.remove_requested) { + entries_.erase(it); + } else { + it->second.armed = true; // re-arm so the loop watches it again + wake_needed = true; + } + } + } + if (wake_needed) { + wake(); + } +} + +bool SocketReactor::loop_iteration(std::mutex &, std::condition_variable &, bool &) { + if (!running_) { + return true; // stop the task + } + + const sock_type_t wakeup_fd = wakeup_recv_.load(); + fd_set readfds; + fd_set exceptfds; + FD_ZERO(&readfds); + FD_ZERO(&exceptfds); + sock_type_t max_fd = wakeup_fd; + FD_SET(wakeup_fd, &readfds); + { + std::lock_guard lock(mutex_); + for (const auto &[id, entry] : entries_) { + if (entry.armed && !entry.in_flight && Socket::is_valid_fd(entry.fd)) { + FD_SET(entry.fd, &readfds); + FD_SET(entry.fd, &exceptfds); + if (entry.fd > max_fd) { + max_fd = entry.fd; + } + } + } + } + + struct timeval tv; + tv.tv_sec = std::chrono::duration_cast(config_.select_timeout).count(); + tv.tv_usec = config_.select_timeout.count() % 1000000; + int num_ready = ::select(static_cast(max_fd) + 1, &readfds, nullptr, &exceptfds, &tv); + if (num_ready < 0) { + // Interrupted or transient error; just loop again. + return !running_; + } + if (num_ready == 0) { + return !running_; // timeout + } + + // Drain the wakeup socket if it fired (registration/stop/re-arm signal). + if (FD_ISSET(wakeup_fd, &readfds)) { + char buf[64]; + while (::recvfrom(wakeup_fd, buf, sizeof(buf), 0, nullptr, nullptr) > 0) { + } + } + + // Collect readable entries, disarming each so it is not dispatched again + // until its handler completes. + std::vector ready; + { + std::lock_guard lock(mutex_); + for (auto &[id, entry] : entries_) { + if (entry.armed && !entry.in_flight && Socket::is_valid_fd(entry.fd) && + (FD_ISSET(entry.fd, &readfds) || FD_ISSET(entry.fd, &exceptfds))) { + entry.armed = false; + entry.in_flight = true; + ready.push_back(id); + } + } + } + + for (Id id : ready) { + ++in_flight_count_; + bool submitted = pool_->submit([this, id]() { dispatch(id); }); + if (!submitted) { + // Pool is saturated; revert and let the next select() re-report this fd + // (the data stays buffered in the socket - natural backpressure). + --in_flight_count_; + std::lock_guard lock(mutex_); + auto it = entries_.find(id); + if (it != entries_.end()) { + it->second.in_flight = false; + // Honor a remove() that arrived while this entry was marked in_flight, + // rather than blindly re-arming a logically-removed registration. + if (it->second.remove_requested) { + entries_.erase(it); + } else { + it->second.armed = true; + } + } + } + } + + return !running_; +} + +void SocketReactor::wake() { + const sock_type_t fd = wakeup_recv_.load(); + if (!Socket::is_valid_fd(fd)) { + return; + } + const char byte = 'x'; + ::sendto(fd, &byte, 1, 0, reinterpret_cast(&wakeup_addr_), + sizeof(wakeup_addr_)); +} + +bool SocketReactor::create_wakeup_socket() { + const sock_type_t fd = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + if (!Socket::is_valid_fd(fd)) { + logger_.error("Could not create wakeup socket"); + return false; + } +#if !defined(_MSC_VER) + // The wakeup fd is FD_SET into the select set every iteration, so it too must + // be below FD_SETSIZE (see check_fd). It is created before any registrations, + // so on lwip it takes a low-offset fd - but guard anyway. + if (fd >= FD_SETSIZE) { + logger_.error("wakeup socket fd {} is >= FD_SETSIZE ({})", static_cast(fd), + static_cast(FD_SETSIZE)); + close_socket(fd); + return false; + } +#endif + struct sockaddr_in addr {}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = 0; // ask the OS for an ephemeral port + if (::bind(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + logger_.error("Could not bind wakeup socket"); + close_socket(fd); + return false; + } + // Learn the assigned address/port so wake() can send to ourselves. + wakeup_addr_ = {}; + socklen_t len = sizeof(wakeup_addr_); + if (::getsockname(fd, reinterpret_cast(&wakeup_addr_), &len) < 0) { + logger_.error("Could not getsockname on wakeup socket"); + close_socket(fd); + return false; + } + // Must be non-blocking, otherwise the loop thread's drain-recvfrom() loop can + // block forever and stall the reactor. + if (!set_nonblocking(fd)) { + logger_.error("Could not set wakeup socket non-blocking"); + close_socket(fd); + return false; + } + wakeup_recv_ = fd; // publish only once fully set up + return true; +} + +void SocketReactor::close_wakeup_socket() { + const sock_type_t fd = wakeup_recv_.exchange(static_cast(-1)); + if (Socket::is_valid_fd(fd)) { + close_socket(fd); + } +} + +bool SocketReactor::set_nonblocking(sock_type_t fd) { +#if defined(_MSC_VER) + u_long mode = 1; + return ioctlsocket(fd, FIONBIO, &mode) == 0; +#else + int flags = ::fcntl(fd, F_GETFL, 0); + if (flags < 0) { + return false; + } + return ::fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; +#endif +} diff --git a/components/socket/src/udp_socket.cpp b/components/socket/src/udp_socket.cpp index 2ff8d9217..a09858963 100644 --- a/components/socket/src/udp_socket.cpp +++ b/components/socket/src/udp_socket.cpp @@ -170,25 +170,18 @@ bool UdpSocket::receive(size_t max_num_bytes, std::vector &data, return true; } -bool UdpSocket::start_receiving(Task::BaseConfig &task_config, - const UdpSocket::ReceiveConfig &receive_config) { - if (task_ && task_->is_started()) { - logger_.error("Server is alrady receiving"); - return false; - } +bool UdpSocket::bind(const UdpSocket::ReceiveConfig &receive_config) { if (!is_valid()) { - logger_.error("Socket invalid, cannot start receiving."); + logger_.error("Socket invalid, cannot bind."); return false; } - server_receive_callback_ = receive_config.on_receive_callback; - // bind - struct sockaddr_in server_addr {}; // configure the server socket accordingly - assume IPV4 and bind to the // any address "0.0.0.0" + struct sockaddr_in server_addr {}; server_addr.sin_addr.s_addr = htonl(INADDR_ANY); server_addr.sin_family = address_family_; server_addr.sin_port = htons(receive_config.port); - int err = bind(socket_, reinterpret_cast(&server_addr), sizeof(server_addr)); + int err = ::bind(socket_, reinterpret_cast(&server_addr), sizeof(server_addr)); if (err < 0) { logger_.error("Unable to bind: {}", error_string()); return false; @@ -205,6 +198,23 @@ bool UdpSocket::start_receiving(Task::BaseConfig &task_config, return false; } } + return true; +} + +bool UdpSocket::start_receiving(Task::BaseConfig &task_config, + const UdpSocket::ReceiveConfig &receive_config) { + if (task_ && task_->is_started()) { + logger_.error("Server is already receiving"); + return false; + } + if (!is_valid()) { + logger_.error("Socket invalid, cannot start receiving."); + return false; + } + server_receive_callback_ = receive_config.on_receive_callback; + if (!bind(receive_config)) { + return false; + } // set the callback function using namespace std::placeholders; // start the thread diff --git a/doc/Doxyfile b/doc/Doxyfile index 7860b6cf2..e43755a04 100755 --- a/doc/Doxyfile +++ b/doc/Doxyfile @@ -389,6 +389,7 @@ INPUT = \ $(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/spi/include/spi.hpp \ $(PROJECT_PATH)/components/st25dv/include/st25dv.hpp \ $(PROJECT_PATH)/components/st7123touch/include/st7123touch.hpp \ diff --git a/doc/en/network/index.rst b/doc/en/network/index.rst index 8b181f356..9a0ce2d16 100644 --- a/doc/en/network/index.rst +++ b/doc/en/network/index.rst @@ -11,6 +11,7 @@ Network APIs socket tcp_socket udp_socket + socket_reactor The network APIs provide a useful abstraction over POSIX sockets enabling easily starting client/server sockets and allowing their use with std::function diff --git a/doc/en/network/socket_reactor.rst b/doc/en/network/socket_reactor.rst new file mode 100644 index 000000000..3c2955abe --- /dev/null +++ b/doc/en/network/socket_reactor.rst @@ -0,0 +1,62 @@ +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 :doc:`ThreadPool <../core/thread_pool>`, 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``. + +.. ------------------------------- Example ------------------------------------- + +Code examples for the reactor are provided in the ``socket`` example folder (the +"Socket reactor" and "TCP reactor" scenarios). + +.. ---------------------------- API Reference ---------------------------------- + +API Reference +------------- + +.. include-build-file:: inc/socket_reactor.inc diff --git a/lib/autogenerate_bindings.py b/lib/autogenerate_bindings.py index 4b063b682..858bc1e81 100644 --- a/lib/autogenerate_bindings.py +++ b/lib/autogenerate_bindings.py @@ -462,6 +462,14 @@ def autogenerate() -> None: # NOTE: this must come after task and base_component since it depends on them include_dir + "thread_pool/include/thread_pool.hpp", + # NOTE: socket/include/socket_reactor.hpp is intentionally NOT generated here. + # srcmlcpp cannot parse SocketReactor (its Config/Entry brace-init defaults with + # parenthesized/cast expressions trip srcmlcpp's brace-init fixer), and its TCP + # callbacks take std::unique_ptr / TcpSocket& which litgen cannot bind + # (the same reason ftp_client_session is excluded). It is instead bound by hand in + # python_bindings/socket_reactor_bindings.cpp (registered via py_init_socket_reactor + # in module.cpp), and is fully available in the C++ host library. + # NOTE: this must come after vector2d.hpp and range_mapper.hpp since it depends on them! include_dir + "joystick/include/joystick.hpp", diff --git a/lib/espp.cmake b/lib/espp.cmake index 78da945bf..2630b0168 100644 --- a/lib/espp.cmake +++ b/lib/espp.cmake @@ -75,6 +75,7 @@ set(ESPP_SOURCES ${ESPP_COMPONENTS}/socket/src/socket.cpp ${ESPP_COMPONENTS}/socket/src/tcp_socket.cpp ${ESPP_COMPONENTS}/socket/src/udp_socket.cpp + ${ESPP_COMPONENTS}/socket/src/socket_reactor.cpp ${CMAKE_CURRENT_LIST_DIR}/espp.cpp ) @@ -103,6 +104,7 @@ set(ESPP_PYTHON_SOURCES ${ESPP_PYTHON_BINDINGS_DIR}/pybind_espp.cpp ${ESPP_PYTHON_BINDINGS_DIR}/cdr_bindings.cpp ${ESPP_PYTHON_BINDINGS_DIR}/rtps_bindings.cpp + ${ESPP_PYTHON_BINDINGS_DIR}/socket_reactor_bindings.cpp ${ESPP_SOURCES} ) diff --git a/lib/include/espp.hpp b/lib/include/espp.hpp index 37eeff76b..55dd2e21a 100644 --- a/lib/include/espp.hpp +++ b/lib/include/espp.hpp @@ -55,6 +55,9 @@ extern "C" { #include "thread_pool.hpp" #include "timer.hpp" #include "udp_socket.hpp" +// NOTE: socket_reactor.hpp must come after tcp_socket/udp_socket/thread_pool/task, +// which it depends on. +#include "socket_reactor.hpp" #include "vector2d.hpp" // state machine includes diff --git a/lib/python_bindings/module.cpp b/lib/python_bindings/module.cpp index a8140362b..8dbfc84b7 100644 --- a/lib/python_bindings/module.cpp +++ b/lib/python_bindings/module.cpp @@ -11,6 +11,10 @@ void py_init_module_espp(py::module &m); // and the module's classes are already registered. void py_init_cdr(py::module &m); void py_init_rtps(py::module &m); +// Hand-written bindings for espp::SocketReactor (litgen cannot parse it; see +// socket_reactor_bindings.cpp). Runs after py_init_module_espp so UdpSocket / Socket::Info / +// Logger::Verbosity are already registered. +void py_init_socket_reactor(py::module &m); // This builds the native python extension module `espp._espp`, which the // `espp` python package (python_bindings/espp/__init__.py) re-exports. @@ -24,4 +28,5 @@ PYBIND11_MODULE(_espp, m) { py_init_module_espp(m); py_init_cdr(m); py_init_rtps(m); + py_init_socket_reactor(m); } diff --git a/lib/python_bindings/socket_reactor_bindings.cpp b/lib/python_bindings/socket_reactor_bindings.cpp new file mode 100644 index 000000000..2f93fbfc0 --- /dev/null +++ b/lib/python_bindings/socket_reactor_bindings.cpp @@ -0,0 +1,117 @@ +// Hand-written pybind11 bindings for espp::SocketReactor. +// +// Why hand-written (see the note in autogenerate_bindings.py): litgen/srcmlcpp cannot parse +// SocketReactor (brace-init default members with parenthesized/cast expressions trip srcmlcpp's +// brace-init fixer), and its TCP callbacks take std::unique_ptr / TcpSocket& which +// litgen cannot bind. This shim exposes a clean, GIL-correct subset for Python: the reactor +// lifecycle and the UDP-receiver path (the TCP listener/stream paths remain C++-only for now). +// +// It is kept out of the generated pybind_espp.cpp so regeneration never clobbers it. + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "socket_reactor.hpp" +#include "udp_socket.hpp" + +namespace py = pybind11; +using espp::SocketReactor; + +namespace { + +// Adapt a Python callable `cb(data: bytes, sender: Socket.Info) -> Optional[bytes]` into the C++ +// receive callback. Like the rtps shim: capture the py::function in a shared_ptr so the reactor may +// copy the std::function off the GIL (only the shared_ptr refcount moves, which is GIL-free); the +// callable is invoked and finally destroyed under the GIL. +espp::Socket::receive_callback_fn wrap_receive_callback(const py::function &fn) { + if (!fn) { + return {}; + } + // Own the py::function via a shared_ptr with a GIL-acquiring deleter: the last + // reference may be released on a non-Python thread (a reactor pool worker + // erasing an entry after remove()), and ~py::function must run under the GIL. + auto cb = std::shared_ptr(new py::function(fn), [](py::function *f) { + py::gil_scoped_acquire gil; + delete f; + }); + return [cb](std::vector &data, + const espp::Socket::Info &sender) -> std::optional> { + py::gil_scoped_acquire gil; + // The handler runs on a reactor pool worker; a Python exception (or a bad + // return type) must not propagate out - it would crash/stall the worker. + // Report it and degrade to "no response". + try { + py::object result = + (*cb)(py::bytes(reinterpret_cast(data.data()), data.size()), sender); + if (result.is_none()) { + return std::nullopt; + } + // Accept either bytes or str as the response payload. + std::string s = py::cast(result); + return std::vector(s.begin(), s.end()); + } catch (py::error_already_set &e) { + // Reports the traceback via sys.unraisablehook and clears the error. + e.discard_as_unraisable("espp.SocketReactor receive callback"); + return std::nullopt; + } catch (const std::exception &e) { + py::print("espp.SocketReactor receive callback error:", e.what()); + return std::nullopt; + } + }; +} + +} // namespace + +void py_init_socket_reactor(py::module &m) { + py::class_( + m, "SocketReactor", py::dynamic_attr(), + "A select()-based event loop that multiplexes many receiver sockets onto a thread pool, " + "instead of one thread per socket.") + // Construct with an owned thread pool of `worker_count` workers. (Sharing an external + // pool is available in C++ but not exposed here.) + .def(py::init( + [](std::size_t worker_count, bool auto_start, espp::Logger::Verbosity log_level) { + SocketReactor::Config config; + config.pool_config.worker_count = worker_count; + config.auto_start = auto_start; + config.log_level = log_level; + return std::make_unique(config); + }), + py::arg("worker_count") = 2, py::arg("auto_start") = true, + py::arg("log_level") = espp::Logger::Verbosity::WARN) + // Release the GIL around start()/stop(): stop() blocks waiting for + // in-flight handlers, and a pool worker running a Python receive callback + // needs the GIL - holding it here would deadlock. + .def("start", &SocketReactor::start, py::call_guard(), + "Start the select() loop (and owned pool).") + .def("stop", &SocketReactor::stop, py::call_guard(), + "Stop the loop and wait for in-flight handlers to finish.") + .def("is_running", &SocketReactor::is_running) + .def("num_registered", &SocketReactor::num_registered) + .def("remove", &SocketReactor::remove, py::arg("id"), + "Unregister a socket by the id returned from add_udp_receiver().") + .def( + "add_udp_receiver", + [](SocketReactor &self, espp::UdpSocket &socket, std::size_t port, + std::size_t buffer_size, const py::function &callback) -> SocketReactor::Id { + espp::UdpSocket::ReceiveConfig rc; + rc.port = port; + rc.buffer_size = buffer_size; + rc.on_receive_callback = wrap_receive_callback(callback); + return self.add_udp_receiver(socket, rc); + }, + py::arg("socket"), py::arg("port"), py::arg("buffer_size"), py::arg("callback"), + "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).") + .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 new file mode 100644 index 000000000..64995a1ec --- /dev/null +++ b/pc/tests/socket_reactor.cpp @@ -0,0 +1,214 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "socket_reactor.hpp" +#include "tcp_socket.hpp" +#include "udp_socket.hpp" + +using namespace std::chrono_literals; +using ByteVector = std::vector; + +namespace { +constexpr auto kLoopback = "127.0.0.1"; +constexpr size_t kBufferSize = 1500; +constexpr auto WARN = espp::Logger::Verbosity::WARN; + +ByteVector make_payload(size_t n, uint8_t seed) { + ByteVector v(n); + for (size_t i = 0; i < n; ++i) { + v[i] = static_cast(seed + i); + } + return v; +} + +ByteVector reversed(ByteVector v) { + std::reverse(v.begin(), v.end()); + return v; +} + +template +bool wait_until(Predicate &&pred, std::chrono::milliseconds timeout = 1s, + std::chrono::milliseconds interval = 10ms) { + 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(); +} + +// Send `request` to a UDP echo server on `port` and return true if the reversed +// reply comes back. +bool udp_echo_roundtrip(size_t port, const ByteVector &request) { + ByteVector response; + std::atomic_bool got{false}; + espp::UdpSocket client({.log_level = WARN}); + client.send(request, {.ip_address = kLoopback, + .port = port, + .wait_for_response = true, + .response_size = kBufferSize, + .on_response_callback = + [&](const ByteVector &r) { + response = r; + got = true; + }, + .response_timeout = 500ms}); + return got.load() && response == reversed(request); +} +} // namespace + +// Mirrors python/socket_reactor_test.py, plus the TCP listener/stream paths +// (which are C++-only). Exit code 0 on full pass, 1 on any failure. +int main() { + espp::Logger logger({.tag = "SocketReactor Test", .level = espp::Logger::Verbosity::INFO}); + + int total_passed = 0; + int total_tests = 0; + auto check = [&](bool condition, const std::string &desc) -> bool { + ++total_tests; + if (condition) { + ++total_passed; + logger.info(" PASS: {}", desc); + } else { + logger.error(" FAIL: {}", desc); + } + return condition; + }; + + auto echo_reversed = [](const ByteVector &d, const espp::Socket::Info &) { + return std::optional(reversed(d)); + }; + + // ------------------------------------------------------------------------- + // 1. Lifecycle + UDP echo + multiple receivers + dynamic remove + // ------------------------------------------------------------------------- + logger.info("--- lifecycle + udp ---"); + { + // sockets declared before the reactor so the reactor is destroyed first + std::vector> servers; + espp::SocketReactor reactor({.auto_start = false, .log_level = WARN}); + check(!reactor.is_running(), "not running before start()"); + check(reactor.start() && reactor.is_running(), "start() -> running"); + check(reactor.num_registered() == 0, "no registrations initially"); + + std::vector ids; + for (int i = 0; i < 3; ++i) { + servers.push_back( + std::make_unique(espp::UdpSocket::Config{.log_level = WARN})); + auto id = reactor.add_udp_receiver(*servers.back(), {.port = 6111 + static_cast(i), + .buffer_size = kBufferSize, + .on_receive_callback = echo_reversed}); + check(id != espp::SocketReactor::INVALID_ID, "add_udp_receiver returns a valid id"); + ids.push_back(id); + } + check(reactor.num_registered() == 3, "three registrations"); + check(udp_echo_roundtrip(6111, make_payload(40, 0x01)), "receiver 0 echoes"); + check(udp_echo_roundtrip(6113, make_payload(40, 0x20)), "receiver 2 echoes"); + + check(reactor.remove(ids[1]), "remove() returns true for a valid id"); + check(wait_until([&] { return reactor.num_registered() == 2; }), "count drops after remove"); + check(!reactor.remove(999999), "remove() of unknown id returns false"); + check(udp_echo_roundtrip(6111, make_payload(40, 0x02)), "remaining receiver still echoes"); + + reactor.stop(); + check(!reactor.is_running(), "not running after stop()"); + } + + // ------------------------------------------------------------------------- + // 2. Input validation + // ------------------------------------------------------------------------- + logger.info("--- input validation ---"); + { + espp::SocketReactor reactor({.log_level = WARN}); + check(reactor.add_fd(-1, []() {}) == espp::SocketReactor::INVALID_ID, "invalid fd rejected"); + check(reactor.add_fd(0, nullptr) == espp::SocketReactor::INVALID_ID, "null handler rejected"); + } + + // ------------------------------------------------------------------------- + // 3. Shared external thread pool + // ------------------------------------------------------------------------- + logger.info("--- shared pool ---"); + { + auto pool = std::make_shared( + espp::ThreadPool::Config{.worker_count = 2, .log_level = WARN}); + espp::UdpSocket server({.log_level = WARN}); + { + espp::SocketReactor reactor({.thread_pool = pool, .log_level = WARN}); + auto id = reactor.add_udp_receiver( + server, {.port = 6120, .buffer_size = kBufferSize, .on_receive_callback = echo_reversed}); + check(id != espp::SocketReactor::INVALID_ID, "registered on a shared pool"); + check(udp_echo_roundtrip(6120, make_payload(64, 0x33)), "echoes via a shared pool"); + reactor.stop(); + } + } + + // ------------------------------------------------------------------------- + // 4. TCP listener + stream echo + disconnect (C++-only paths) + // ------------------------------------------------------------------------- + logger.info("--- tcp listener + stream ---"); + { + constexpr size_t port = 6130; + std::atomic_bool closed{false}; + std::mutex clients_mutex; + std::vector> clients; + { + espp::SocketReactor reactor({.log_level = WARN}); + espp::TcpSocket server({.log_level = WARN}); + check(server.bind(port) && server.listen(2), "TCP server bind + listen"); + reactor.add_tcp_listener(server, [&](std::unique_ptr client) { + espp::TcpSocket *conn = nullptr; + { + std::lock_guard lk(clients_mutex); + clients.push_back(std::move(client)); + conn = clients.back().get(); + } + reactor.add_tcp_stream( + *conn, [](espp::TcpSocket &c, ByteVector &data) { c.transmit(data); }, kBufferSize, + [&closed]() { closed = true; }); + }); + + espp::TcpSocket client({.log_level = WARN}); + check(client.connect({.ip_address = kLoopback, .port = port}), "TCP client connects"); + auto request = make_payload(128, 0x40); + ByteVector echo; + std::atomic_bool got{false}; + client.transmit(request, {.wait_for_response = true, + .response_size = kBufferSize, + .on_response_callback = + [&](const ByteVector &r) { + echo = r; + got = true; + }, + .response_timeout = 1s}); + check(got.load() && echo == request, "TCP echo via the reactor"); + client.close(); + check(wait_until([&] { return closed.load(); }), "server observed the client disconnect"); + check(wait_until([&] { return reactor.num_registered() == 1; }), + "stream auto-removed, only the listener remains"); + reactor.stop(); + } + } + + // ------------------------------------------------------------------------- + // Summary + // ------------------------------------------------------------------------- + logger.info(""); + if (total_passed == total_tests) { + logger.info("======== SocketReactor test: {}/{} checks passed ========", total_passed, + total_tests); + logger.info("ALL CHECKS PASSED"); + return 0; + } + logger.error("======== SocketReactor test: {}/{} checks passed ========", total_passed, + total_tests); + return 1; +} diff --git a/python/README.md b/python/README.md index 7cbdd3b27..4b659cd03 100644 --- a/python/README.md +++ b/python/README.md @@ -31,6 +31,14 @@ This section gives a brief overview of what the scripts in this folder do. `espp` library to create a UDP client and server. The server listens for incoming UDP packets and prints them to the console, while the client sends UDP packets to the server. +- `socket_reactor.py`: Demonstrates `espp.SocketReactor` - a single select() + event loop plus a thread pool that services many receiver sockets instead of + one thread per socket. It multiplexes two UDP echo receivers on one reactor and + round-trips a datagram through each. +- `socket_reactor_test.py`: Self-checking test for the `espp.SocketReactor` + Python binding (lifecycle, UDP receiver + echo, sender info, `None` responses, + multiple receivers, dynamic `remove()`). Exits 0 on full pass, 1 on any + failure. Mirrors the C++ `pc/tests/socket_reactor.cpp`. - `rtsp_client.py` and `rtsp_server.py`: These scripts demonstrate how to use the `espp` library to create an RTSP client and server. The server streams MJPEG video from a webcam or display capture. The camera path captures live diff --git a/python/socket_reactor.py b/python/socket_reactor.py new file mode 100644 index 000000000..5329bb3d1 --- /dev/null +++ b/python/socket_reactor.py @@ -0,0 +1,63 @@ +"""SocketReactor example. + +Demonstrates espp.SocketReactor: a single select()-based event loop plus a small +thread pool that services many receiver sockets, instead of one thread per +socket. Here one reactor multiplexes two UDP echo receivers (on two ports); each +received datagram is handled on a pool worker by a Python callback that returns +the reversed payload, which the reactor sends back to the sender. + +Run (from this folder, with the `espp` package installed - see README): + + python socket_reactor.py +""" + +import socket +import time + +import espp + +PORTS = [6101, 6102] + + +def make_echo_callback(port): + def on_receive(data, sender): + # Runs on a reactor thread-pool worker. Return bytes to reply, or None. + print(f" [port {port}] received {len(data)} bytes from " + f"{sender.address}:{sender.port}") + return bytes(reversed(data)) + return on_receive + + +def main(): + # One reactor (one select() loop + a 2-worker pool) for BOTH receivers. + reactor = espp.SocketReactor(worker_count=2, log_level=espp.Logger.Verbosity.warn) + + # Keep the receiver sockets alive for as long as they are registered. + servers = [] + for port in PORTS: + server = espp.UdpSocket(espp.UdpSocket.Config(espp.Logger.Verbosity.warn)) + rid = reactor.add_udp_receiver(server, port, 1024, make_echo_callback(port)) + if rid == espp.SocketReactor.INVALID_ID: + print(f"failed to register receiver on port {port}") + return 1 + servers.append(server) + print(f"reactor running with {reactor.num_registered()} UDP receivers on ports {PORTS}") + + # Send a datagram to each receiver and print the reversed reply. + for port in PORTS: + client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + client.settimeout(1.0) + payload = bytes(f"hello:{port}".encode()) + client.sendto(payload, ("127.0.0.1", port)) + reply, _ = client.recvfrom(1024) + print(f" [port {port}] sent {payload!r}, got reply {reply!r}") + client.close() + + time.sleep(0.2) + reactor.stop() # stop the loop and wait for in-flight handlers before teardown + print("reactor stopped") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/socket_reactor_test.py b/python/socket_reactor_test.py new file mode 100644 index 000000000..9b4c2fc44 --- /dev/null +++ b/python/socket_reactor_test.py @@ -0,0 +1,165 @@ +"""SocketReactor Python test. + +Exercises the espp.SocketReactor Python binding: lifecycle, UDP receiver +registration + echo round-trip, sender info, callbacks returning None (no +response), multiple receivers, dynamic remove, and input validation. A plain +Python UDP socket is used as the client. + +Exit code 0 on full pass, 1 on any failure. +""" + +import socket +import sys +import time +from typing import List, Optional, Tuple + +import espp + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +results: List[Tuple[str, bool]] = [] + + +def check(test: str, condition: bool, desc: str) -> bool: + ok = bool(condition) + print(f" {'PASS' if ok else 'FAIL'} [{test}]: {desc}") + results.append((test, ok)) + return ok + + +def udp_send_recv(port: int, payload: bytes, timeout: float = 1.0) -> Optional[bytes]: + """Send `payload` to 127.0.0.1:port and return the reply bytes (or None on timeout).""" + cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + cli.settimeout(timeout) + try: + cli.sendto(payload, ("127.0.0.1", port)) + reply, _ = cli.recvfrom(4096) + return reply + except socket.timeout: + return None + finally: + cli.close() + + +def wait_until(predicate, timeout: float = 1.0, interval: float = 0.01) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +VERB = espp.Logger.Verbosity.warn + +# --------------------------------------------------------------------------- +# 1. Lifecycle: auto_start=False -> start() -> is_running() -> stop() +# --------------------------------------------------------------------------- +name = "lifecycle" +print(f"--- {name} ---") +reactor = espp.SocketReactor(worker_count=2, auto_start=False, log_level=VERB) +check(name, not reactor.is_running(), "not running before start()") +check(name, reactor.start(), "start() returns True") +check(name, reactor.is_running(), "running after start()") +check(name, reactor.num_registered() == 0, "no registrations initially") + +# --------------------------------------------------------------------------- +# 2. UDP echo: a Python callback reverses the payload; the reactor sends it back +# --------------------------------------------------------------------------- +name = "udp echo" +print(f"--- {name} ---") +server_a = espp.UdpSocket(espp.UdpSocket.Config(VERB)) + + +def reverse_cb(data, sender): + return bytes(reversed(data)) + + +id_a = reactor.add_udp_receiver(server_a, 5111, 1500, reverse_cb) +check(name, id_a != espp.SocketReactor.INVALID_ID, "add_udp_receiver returns a valid id") +check(name, reactor.num_registered() == 1, "one registration") +payload = bytes(range(40)) +reply = udp_send_recv(5111, payload) +check(name, reply == bytes(reversed(payload)), "reversed echo received via the reactor") + +# --------------------------------------------------------------------------- +# 3. Sender info + a callback that returns None sends no response +# --------------------------------------------------------------------------- +name = "no response / sender info" +print(f"--- {name} ---") +server_b = espp.UdpSocket(espp.UdpSocket.Config(VERB)) +seen = {} + + +def sink_cb(data, sender): + seen["address"] = sender.address + seen["length"] = len(data) + return None # -> reactor sends nothing back + + +id_b = reactor.add_udp_receiver(server_b, 5112, 1500, sink_cb) +check(name, id_b != espp.SocketReactor.INVALID_ID, "second receiver registered") +check(name, reactor.num_registered() == 2, "two registrations") +no_reply = udp_send_recv(5112, b"hello", timeout=0.4) +check(name, no_reply is None, "no reply when the callback returns None") +check(name, wait_until(lambda: seen.get("length") == 5), "callback observed the 5-byte payload") +check(name, seen.get("address") == "127.0.0.1", f"sender address seen: {seen.get('address')}") + +# --------------------------------------------------------------------------- +# 4. Dynamic remove() +# --------------------------------------------------------------------------- +name = "remove" +print(f"--- {name} ---") +check(name, reactor.remove(id_a) is True, "remove() returns True for a valid id") +check(name, wait_until(lambda: reactor.num_registered() == 1), "count drops to 1 after remove()") +check(name, reactor.remove(999999) is False, "remove() of an unknown id returns False") +check(name, udp_send_recv(5111, b"x", timeout=0.3) is None, "removed receiver no longer echoes") + +# --------------------------------------------------------------------------- +# 5. A raising callback must not crash the worker; the reactor keeps serving. +# --------------------------------------------------------------------------- +name = "callback exception" +print(f"--- {name} ---") +raiser = espp.UdpSocket(espp.UdpSocket.Config(VERB)) + + +def raising_cb(data, sender): + raise ValueError("boom") + + +check(name, reactor.add_udp_receiver(raiser, 5113, 1500, raising_cb) != espp.SocketReactor.INVALID_ID, + "raising receiver registered") +# The datagram triggers the exception (reported via sys.unraisablehook); no reply, +# no crash. +check(name, udp_send_recv(5113, b"boom", timeout=0.4) is None, "raising callback yields no reply") +# The reactor is still alive: the earlier echo receiver on 5112's sibling keeps working. +echo_srv = espp.UdpSocket(espp.UdpSocket.Config(VERB)) +reactor.add_udp_receiver(echo_srv, 5114, 1500, reverse_cb) +payload2 = bytes(range(24)) +check(name, udp_send_recv(5114, payload2) == bytes(reversed(payload2)), + "reactor still serves other receivers after a callback exception") + +# --------------------------------------------------------------------------- +# 6. Stop +# --------------------------------------------------------------------------- +name = "stop" +print(f"--- {name} ---") +reactor.stop() +check(name, not reactor.is_running(), "not running after stop()") + +# --------------------------------------------------------------------------- +# Summary +# --------------------------------------------------------------------------- +passed = sum(1 for _, ok in results if ok) +total = len(results) +print() +print(f"======== SocketReactor python test: {passed}/{total} checks passed ========") +if passed != total: + for test_name, ok in results: + if not ok: + print(f" FAILED: {test_name}") + sys.exit(1) +print("ALL CHECKS PASSED") +sys.exit(0)