Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion docs/spec/core/backend.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
52 changes: 51 additions & 1 deletion include/morph/net/socket_server.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,12 @@
#pragma once
#include <array>
#include <atomic>
#include <cerrno>
#include <chrono>
#include <core/async/Task.hpp>
#include <core/net/EventLoop.hpp>
#include <core/net/IListener.hpp>
#include <core/net/NetError.hpp>
#include <core/net/Sockets.hpp>
#include <cstddef>
#include <cstdint>
Expand All @@ -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.
Expand Down Expand Up @@ -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<void> acceptFlow(std::shared_ptr<Core> self, ::core::net::IListener* accepting,
std::uint64_t startedAt) {
for (;;) {
Expand All @@ -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.
Expand Down
91 changes: 91 additions & 0 deletions tests/net/test_socket_server.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<sockaddr*>(&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]") {
Expand Down Expand Up @@ -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<morph::backend::RemoteServer>(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.
Expand Down
Loading