diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index ddeb1dd81..0965a775d 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -1550,6 +1550,19 @@ and ends — and every connection's socket, which resumes each connection's parked read the same way. `port()` reads `0` afterwards, and the port is free to rebind, matching `QtWebSocketServer`. +**A failed accept is retried or ends the flow, by whether it can recur.** +Exhaustion (`EMFILE`, `ENFILE`, `ENOBUFS`, `ENOMEM`) and any other failure the +next attempt may not repeat waits 50 ms and accepts again, so a process out of +descriptors keeps its listener and recovers when descriptors are freed. A +listener that can no longer accept at all ends the flow instead: one that has +stopped listening (`EINVAL`), a closed or foreign descriptor +(`EBADF`/`ENOTSOCK`) or a non-stream socket (`EOPNOTSUPP`) fails every attempt +the same way. core-cpp reports these as `BadHandle`/`Unsupported`, or, in +releases inside the supported range that predate that classification, as +`SystemError` carrying the `errno`; both are recognised. The server then drops +the listener, so `port()` reads `0` and a later `listen()` binds afresh. +Connections already accepted are untouched. + **The listener's non-blocking mode stops at the listener.** `TcpSocket`'s fd-adopting constructor clears `O_NONBLOCK` on every descriptor it takes ownership of, so a connection from `accept()` or `tryAccept()` is always @@ -2040,7 +2053,7 @@ keepalive to tell "idle" from "dead" apart). | `SocketServer(loop, server, port = 0, cfg = Config{})` | Fronts `RemoteServer& server` on `loop`, an `exec::IoLoop` that must outlive it. Does not start listening. | | `SocketServer(server, port = 0, cfg = Config{})` | The same, on a private `IoLoop` the first `listen()` creates. | | `listen()` | On the loop, and waited for: binds `127.0.0.1:port` with `SO_REUSEADDR`, hands the listener to the loop and starts the accept flow; returns success. `false` if the port cannot be bound or, for the loop-less constructor, the loop cannot be created (a process out of descriptors). | -| `port()` | Bound port (OS-assigned when constructed with `0`), or `0` when not listening. Atomic; callable from any thread. | +| `port()` | Bound port (OS-assigned when constructed with `0`), or `0` when not listening — including after the listener stopped being able to accept. Atomic; callable from any thread. | | `close()` | On the loop, and waited for (inline on the loop's own thread): closes the listener, reclaims every connection's models (`closeConnection`) and closes its socket. `port()` reads `0` afterwards and the port is free to rebind. Idempotent; also run by the destructor. Concurrent callers on a **live** object are safe — each call is one loop task, run one at a time. Racing `close()` against the *destructor* remains out of contract, as for any member call. | ## `executeInto` — settling the caller's own completion diff --git a/include/morph/net/socket_server.hpp b/include/morph/net/socket_server.hpp index 9a5fc9a54..3b58b9c98 100644 --- a/include/morph/net/socket_server.hpp +++ b/include/morph/net/socket_server.hpp @@ -3,10 +3,12 @@ #pragma once #include #include +#include #include #include #include #include +#include #include #include #include @@ -29,6 +31,35 @@ namespace morph::net { +namespace detail { + +/// @brief Whether a failed `accept()` says the listening socket itself is +/// unusable, so every later attempt fails the same way and retrying +/// cannot help. +/// +/// A socket that is no longer listening answers `EINVAL`; a closed or foreign +/// descriptor `EBADF`/`ENOTSOCK`; a non-stream socket `EOPNOTSUPP`. core-cpp +/// classifies these as `BadHandle` and `Unsupported`, but releases inside the +/// supported version range still report every failed accept as `SystemError` +/// with the `errno` in `systemCode`, so both spellings are recognised. +/// Exhaustion (`EMFILE`, `ENFILE`, `ENOBUFS`, `ENOMEM`) is not in the list: it +/// clears once descriptors or memory are freed, so it is retried. +/// @param error The error `IListener::accept()` reported. +/// @return `true` when the listener can never accept again. +[[nodiscard]] inline bool acceptListenerIsDead(::core::net::NetError const& error) noexcept { + using ::core::net::NetErrorCode; + if (error.code == NetErrorCode::BadHandle || error.code == NetErrorCode::Unsupported) { + return true; + } + if (error.code != NetErrorCode::SystemError) { + return false; + } + int const code = error.systemCode; + return code == EINVAL || code == EBADF || code == ENOTSOCK || code == EOPNOTSUPP; +} + +} // namespace detail + /// @brief Configuration for `SocketServer`. struct SocketServerConfig { /// @brief Pending-connection backlog passed to the listening socket. @@ -262,7 +293,22 @@ class SocketServer { loop.loop().spawn(clientFlow(shared_from_this(), client)); } - /// Accepts until the listener it was started with is closed. + /// Drops a listener that can no longer accept, so `port()` reads `0` + /// and a later `listen()` can bind again. Connections already accepted + /// keep running: only the listening socket is gone. + /// + /// Only the accept flow calls it, and only while its generation is + /// current, so `listener` is still the one it was started with: the + /// one place that resets it, `close()`, moves the generation first. + void listenerLost() { + note("SocketServer::listenerLost"); + listener->close(); + listener.reset(); + port.store(0); + } + + /// Accepts until the listener it was started with is closed or stops + /// being able to accept. static ::core::async::Task acceptFlow(std::shared_ptr self, ::core::net::IListener* accepting, std::uint64_t startedAt) { for (;;) { @@ -274,6 +320,10 @@ class SocketServer { if (next.error().code == ::core::net::NetErrorCode::Cancelled) { co_return; } + if (::morph::net::detail::acceptListenerIsDead(next.error())) { + self->listenerLost(); + co_return; + } // Out of descriptors, or another failure the next attempt // may not repeat: wait a little rather than spin on a // listener that stays readable. diff --git a/tests/net/test_socket_server.cpp b/tests/net/test_socket_server.cpp index 8170d556b..1b9b7e571 100644 --- a/tests/net/test_socket_server.cpp +++ b/tests/net/test_socket_server.cpp @@ -217,6 +217,27 @@ int highestOpenFd(int limit) { return highest; } +// The descriptor listening on 127.0.0.1:port in this process, or -1. Matches on +// SO_ACCEPTCONN as well as the port, because every connection the server +// accepts shares the listener's local port. +int listeningFdFor(std::uint16_t port) { + for (int fd = 0; fd < 4096; ++fd) { + sockaddr_in addr{}; + socklen_t len = sizeof(addr); + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast): POSIX takes a sockaddr*. + if (::getsockname(fd, reinterpret_cast(&addr), &len) != 0 || addr.sin_family != AF_INET || + ntohs(addr.sin_port) != port) { + continue; + } + int accepting = 0; + socklen_t optLen = sizeof(accepting); + if (::getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, &accepting, &optLen) == 0 && accepting != 0) { + return fd; + } + } + return -1; +} + } // namespace TEST_CASE("SocketServer: register -> execute -> reply round trip with a raw client", "[net][socket_server]") { @@ -1185,6 +1206,76 @@ TEST_CASE("SocketServer::close() releases the listening port", "[net][socket_ser REQUIRE(rebound.port() == port); } +TEST_CASE("acceptListenerIsDead: a failure that recurs on every accept ends the flow, any other is retried", + "[net][socket_server]") { + using ::core::net::NetErrorCode; + auto const dead = [](NetErrorCode code, int systemCode) { + return morph::net::detail::acceptListenerIsDead(::core::net::NetError{.code = code, .systemCode = systemCode}); + }; + // Dead: newer core-cpp's own classification, and the SystemError + errno + // spelling the pinned release reports for the same conditions. + CHECK(dead(NetErrorCode::BadHandle, EBADF)); + CHECK(dead(NetErrorCode::Unsupported, 0)); + for (int const errnoValue : {EINVAL, EBADF, ENOTSOCK, EOPNOTSUPP}) { + CAPTURE(errnoValue); + CHECK(dead(NetErrorCode::SystemError, errnoValue)); + } + // Retried: exhaustion clears once descriptors or memory are freed. + for (int const errnoValue : {EMFILE, ENFILE, ENOBUFS, ENOMEM}) { + CAPTURE(errnoValue); + CHECK_FALSE(dead(NetErrorCode::SystemError, errnoValue)); + } + // Retried: one peer's failure, not the listener's -- even carrying an + // errno that would mean a dead listener under SystemError. + CHECK_FALSE(dead(NetErrorCode::PermissionDenied, EINVAL)); + CHECK_FALSE(dead(NetErrorCode::HostUnreach, 0)); +} + +TEST_CASE("SocketServer: a listener that stops listening ends the accept flow instead of retrying it", + "[net][socket_server]") { + // shutdown() on a listening socket leaves the descriptor open but no longer + // listening, so every accept() on it fails at once and for good (EINVAL). + // The flow must give the listener up -- port() reads 0 -- rather than + // re-arming its back-off forever, and the server must be able to listen + // again. A connection accepted before the listener died keeps working. + morph::exec::ThreadPoolExecutor pool{2}; + auto server = std::make_shared(pool); + morph::net::SocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + std::uint16_t const port = wsServer.port(); + REQUIRE(port != 0U); + + RawWsClient survivor{port}; + survivor.send(morph::wire::makeRegister("NetEchoModel")); + auto const registered = survivor.receive(); + REQUIRE(registered.kind == "ok"); + + int const listeningFd = listeningFdFor(port); + REQUIRE(listeningFd >= 0); + REQUIRE(::shutdown(listeningFd, SHUT_RDWR) == 0); + + REQUIRE(morph::testing::waitUntil([&wsServer] { return wsServer.port() == 0U; }, + morph::testing::WaitBudget{std::chrono::seconds{5}})); + + morph::wire::Envelope execReq; + execReq.kind = "execute"; + execReq.callId = 1; + execReq.modelId = registered.modelId; + execReq.modelType = "NetEchoModel"; + execReq.actionType = "NetEchoAction"; + execReq.body = R"({"value":7})"; + survivor.send(execReq); + auto const reply = survivor.receive(); + REQUIRE(reply.kind == "ok"); + REQUIRE(reply.body == "7"); + + REQUIRE(wsServer.listen()); + REQUIRE(wsServer.port() != 0U); + RawWsClient fresh{wsServer.port()}; + fresh.send(morph::wire::makeRegister("NetEchoModel")); + REQUIRE(fresh.receive().kind == "ok"); +} + TEST_CASE("SocketServer: teardown racing a connecting client still finishes promptly", "[net][socket_server]") { // A client arriving while close() is already under way, so the accept flow // may be anywhere between an accept and starting the connection's flow.