diff --git a/README.md b/README.md index 59c8d59..a6e4c19 100644 --- a/README.md +++ b/README.md @@ -293,10 +293,10 @@ The external-interop test requires outbound UDP/443; if your network blocks it t ## Dependencies - **Crafter.Thread**: Thread pool management for asynchronous operations. -- **msquic** (native target only) — fetched and built automatically as a Crafter `ExternalDependency` (no system install required). The build clones `microsoft/msquic` recursively into the per-project external cache, configures it via CMake (`QUIC_TLS_LIB=quictls`, tests/tools/perf disabled), and links the produced `libmsquic` into the QUIC and HTTP/3 modules. Skipped entirely on browser builds. +- **msquic** (native target only) — fetched and built automatically as a Crafter `ExternalDependency` (no system install required). The build clones `microsoft/msquic` at a pinned release (v2.6.3), with the submodules that release records, into the per-project external cache, configures it via CMake (`QUIC_TLS_LIB=openssl` against the system OpenSSL, tests/tools/perf disabled), and links the produced `libmsquic` into the QUIC and HTTP/3 modules. Skipped entirely on browser builds. - On Linux msquic links against `libnuma` (provided by the `numactl` package on most distros). - The self-signed-cert path used by tests/dev shells out to the `openssl` CLI; install `openssl` if you intend to use `QUICServerCredentials{selfSigned = true}`. The same path also produces the cert hash that browser peers need for `serverCertificateHashes`. -- **libssl / libcrypto** (OpenSSL 3, native target only) — a *system* package, unlike msquic, linked via `-lssl -lcrypto`. Backs `Crafter.Network:TLS`, i.e. `https://` on the HTTP/1.1 client and listener. It is not built here on purpose: OpenSSL 3 is present on every platform this targets, and vendoring it would mean shipping a second TLS stack next to the one msquic already links (quictls, whose symbols stay inside `libmsquic.so`). Install your distro's OpenSSL development package (`openssl` on Arch, `libssl-dev` on Debian/Ubuntu). Certificate generation for `TLSServerCredentials{selfSigned = true}` happens in-process through this library — no `openssl` CLI needed. +- **libssl / libcrypto** (OpenSSL 3, native target only) — a *system* package, unlike msquic, linked via `-lssl -lcrypto`. Backs `Crafter.Network:TLS`, i.e. `https://` on the HTTP/1.1 client and listener. It is not built here on purpose, and msquic uses it too, so the process has one OpenSSL: msquic's QUIC support needs OpenSSL 3.5 or newer (Arch has it, as does the Flatpak runtime crafter-build builds in). msquic's own bundled fork (quictls) is not used: it is an older OpenSSL that still reads the system's `openssl.cnf`, and a config written for the system's newer OpenSSL (the Flatpak runtime's loads a PKCS#11 engine, with `config_diagnostics` on) made its initialisation fail. Install your distro's OpenSSL development package (`openssl` on Arch, `libssl-dev` on Debian/Ubuntu). Where that OpenSSL is older than 3.5 (Debian 12 and Ubuntu 24.04 ship 3.0), msquic doesn't build against it: build with `--no-quic`, which keeps TCP, TLS and HTTP/1.1. Certificate generation for `TLSServerCredentials{selfSigned = true}` happens in-process through this library — no `openssl` CLI needed. - Plaintext HTTP/1.1 needs nothing beyond POSIX sockets: `ClientHTTP1`/`ListenerHTTP1` built without credentials pull in no msquic and touch no TLS code path. - **Browser build** has no extra dependencies beyond Crafter.Build's `wasi-browser` runtime: HTTP delegates to the browser's `fetch()`, QUIC to its `WebTransport`. The JS glue lives in `additional/network-env.js` and is shipped alongside the produced `.wasm`. diff --git a/additional/network-env.js b/additional/network-env.js index 58b2678..9e20955 100644 --- a/additional/network-env.js +++ b/additional/network-env.js @@ -199,20 +199,21 @@ function crafterNetworkWtOpenStream(handle, unidirectional) { function crafterNetworkWtStreamWrite(streamId, bufPtr, bufLen, finish, callbackId) { const record = __wtStreams.get(streamId); + // The completion's second argument says whether the stream took the write. if (!record) { - if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId); + if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId, 0); return; } const data = bufLen > 0 ? __readBytes(bufPtr, bufLen) : new Uint8Array(0); const doWrite = () => { const p = data.length > 0 ? record.writer.write(data) : Promise.resolve(); p.then(() => { - if (finish) { try { return record.writer.close(); } catch (_) {} } + if (finish) return record.writer.close(); }).then(() => { - if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId); + if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId, 1); }).catch((err) => { console.error(`Crafter.Network: write(${streamId}) failed`, err); - if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId); + if (callbackId !== 0) __wasm().CrafterNetworkOnWtStreamWriteComplete(callbackId, 0); }); }; if (record.opened) doWrite(); diff --git a/examples/SimpleClient/main.cpp b/examples/SimpleClient/main.cpp index 7097e00..0026dd6 100644 --- a/examples/SimpleClient/main.cpp +++ b/examples/SimpleClient/main.cpp @@ -99,7 +99,11 @@ namespace { // returns and the echo handler runs. static QUICStream s = conn->OpenStream(/*unidirectional=*/false); constexpr std::string_view payload = "hello from crafter.network"; - s.SendAsync(payload.data(), static_cast(payload.size()), /*finish=*/true, []{ + s.SendAsync(payload.data(), static_cast(payload.size()), /*finish=*/true, [](bool sent) { + if (!sent) { + std::println(std::cout, "[Crafter.Network] WT stream write failed"); + return; + } std::println(std::cout, "[Crafter.Network] WT stream write OK"); s.RecieveUntilCloseAsync([](std::vector bytes) { std::string text(bytes.begin(), bytes.end()); diff --git a/implementations/Crafter.Network-ClientQUIC-Browser.cpp b/implementations/Crafter.Network-ClientQUIC-Browser.cpp index 9989569..182cae3 100644 --- a/implementations/Crafter.Network-ClientQUIC-Browser.cpp +++ b/implementations/Crafter.Network-ClientQUIC-Browser.cpp @@ -101,8 +101,8 @@ namespace Crafter::NetworkBrowser { static std::unordered_map m; return m; } - inline std::unordered_map>& WriteCallbacks() { - static std::unordered_map> m; + inline std::unordered_map>& WriteCallbacks() { + static std::unordered_map> m; return m; } inline std::int32_t NextWriteCallbackId() { @@ -200,9 +200,9 @@ QUICStream::~QUICStream() { } void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish, - std::function onSent) { + std::function onSent) { if (!impl || impl->state.handle == 0 || impl->state.closed) { - if (onSent) onSent(); + if (onSent) onSent(false); return; } std::int32_t cbId = 0; @@ -402,13 +402,14 @@ extern "C" { TryDispatchRecv(s); } + // `sent` is nonzero when the write (and the close, with finish) resolved. __attribute__((export_name("CrafterNetworkOnWtStreamWriteComplete"))) - void CrafterNetworkOnWtStreamWriteComplete(std::int32_t callbackId) { + void CrafterNetworkOnWtStreamWriteComplete(std::int32_t callbackId, std::int32_t sent) { auto it = WriteCallbacks().find(callbackId); if (it == WriteCallbacks().end()) return; auto cb = std::move(it->second); WriteCallbacks().erase(it); - if (cb) cb(); + if (cb) cb(sent != 0); } __attribute__((export_name("CrafterNetworkOnWtDatagram"))) diff --git a/implementations/Crafter.Network-ClientQUIC.cpp b/implementations/Crafter.Network-ClientQUIC.cpp index d0cd214..5962da7 100644 --- a/implementations/Crafter.Network-ClientQUIC.cpp +++ b/implementations/Crafter.Network-ClientQUIC.cpp @@ -123,6 +123,53 @@ struct QUICStream::Impl { bool shutdownComplete = false; bool sendInFlight = false; + // Up to `count` received bytes, waiting for them; fewer once the peer + // has closed its side and the rest is taken. + std::vector TakeUpTo(std::uint32_t count) { + std::vector out; + out.reserve(count); + std::unique_lock lk(mtx); + while (out.size() < count) { + cv.wait(lk, [&]{ return !pending.empty() || peerSendClosed || shutdownComplete; }); + Take(out, count); + if (pending.empty() && (peerSendClosed || shutdownComplete)) break; + } + return out; + } + + // Exactly `count` received bytes, once they have all arrived; nothing, + // leaving what did arrive where it is, once the peer has closed its side + // short of them. One hold of the lock throughout, so no other receive + // takes bytes from among them. + std::optional> TakeExactly(std::uint32_t count) { + std::unique_lock lk(mtx); + auto arrived = [&] { + std::size_t bytes = 0; + for (const std::vector& chunk : pending) bytes += chunk.size(); + return bytes; + }; + cv.wait(lk, [&]{ return arrived() >= count || peerSendClosed || shutdownComplete; }); + if (arrived() < count) return std::nullopt; + std::vector out; + out.reserve(count); + Take(out, count); + return out; + } + + // Moves pending bytes onto `out` until it holds `count`. Under mtx. + void Take(std::vector& out, std::size_t count) { + while (!pending.empty() && out.size() < count) { + auto& chunk = pending.front(); + std::size_t want = std::min(chunk.size(), count - out.size()); + out.insert(out.end(), chunk.begin(), chunk.begin() + want); + if (want == chunk.size()) { + pending.pop_front(); + } else { + chunk.erase(chunk.begin(), chunk.begin() + want); + } + } + } + // Lifetime: msquic delivers stream callbacks (RECEIVE, SHUTDOWN_COMPLETE, …) // possibly AFTER the owning QUICStream wrapper is destroyed (the wrapper only // initiates a graceful shutdown; completion is async). To make that safe, the @@ -401,48 +448,53 @@ std::vector QUICStream::RecieveUntilCloseSync() { std::vector QUICStream::RecieveUntilFullSync(std::uint32_t bufferSize) { if (!handle || !canReceive) throw QUICClosedException(); - std::vector out; - out.reserve(bufferSize); - while (out.size() < bufferSize) { - std::unique_lock lk(impl->mtx); - impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; }); - while (!impl->pending.empty() && out.size() < bufferSize) { - auto& chunk = impl->pending.front(); - std::size_t want = std::min(chunk.size(), bufferSize - out.size()); - out.insert(out.end(), chunk.begin(), chunk.begin() + want); - if (want == chunk.size()) { - impl->pending.pop_front(); - } else { - chunk.erase(chunk.begin(), chunk.begin() + want); - } - } - if (out.size() < bufferSize && (impl->peerSendClosed || impl->shutdownComplete)) { - throw QUICClosedException(); - } - } - return out; + // Closed early: what did arrive stays for the next receive. + std::optional> out = impl->TakeExactly(bufferSize); + if (!out) throw QUICClosedException(); + return std::move(*out); } void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish, - std::function onSent) { + std::function onSent) { // Copy now: the caller's buffer may not outlive the enqueued task. std::vector copy(static_cast(buffer), static_cast(buffer) + size); ThreadPool::Enqueue([this, copy = std::move(copy), finish, onSent = std::move(onSent)]() mutable { - try { this->SendSync(copy.data(), static_cast(copy.size()), finish); } - catch (...) { /* swallowed — callback still fires so the caller can move on */ } - if (onSent) onSent(); + // Nothing may leave a pool task; the callback hears how it went. + bool sent = true; + try { + this->SendSync(copy.data(), static_cast(copy.size()), finish); + } catch (...) { + sent = false; + } + if (onSent) onSent(sent); }); } +// The async receives deliver a closed stream as the browser build does: an +// empty chunk, as nothing may leave a pool task. void QUICStream::RecieveAsync(std::function)> cb) { - ThreadPool::Enqueue([this, cb]{ cb(this->RecieveSync()); }); + ThreadPool::Enqueue([this, cb]{ + std::vector chunk; + try { + chunk = this->RecieveSync(); + } catch (const QUICClosedException&) {} + cb(std::move(chunk)); + }); } void QUICStream::RecieveUntilCloseAsync(std::function)> cb) { - ThreadPool::Enqueue([this, cb]{ cb(this->RecieveUntilCloseSync()); }); + ThreadPool::Enqueue([this, cb]{ + std::vector data; + try { + data = this->RecieveUntilCloseSync(); + } catch (const QUICClosedException&) {} + cb(std::move(data)); + }); } void QUICStream::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> cb) { - ThreadPool::Enqueue([this, bufferSize, cb]{ cb(this->RecieveUntilFullSync(bufferSize)); }); + ThreadPool::Enqueue([this, bufferSize, cb]{ + cb(handle && canReceive ? impl->TakeUpTo(bufferSize) : std::vector{}); + }); } void QUICStream::PrependReceived(std::vector bytes) { diff --git a/implementations/Crafter.Network-ClientTCP.cpp b/implementations/Crafter.Network-ClientTCP.cpp index 31270e0..e4c791c 100755 --- a/implementations/Crafter.Network-ClientTCP.cpp +++ b/implementations/Crafter.Network-ClientTCP.cpp @@ -1,222 +1,288 @@ -//SPDX-License-Identifier: LGPL-3.0-only -//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® - -module; - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -module Crafter.Network:ClientTCP_impl; -import :ClientTCP; -import Crafter.Thread; -import std; - -using namespace Crafter; - -ClientTCP::ClientTCP(int socketid) : socketid(socketid) -{ - -} - -ClientTCP::ClientTCP(const char* hostName, std::uint16_t port) -{ - // getaddrinfo rather than gethostbyname: the latter is not thread safe, - // and it signals failure by returning null — which was then dereferenced - // straight into a crash on any unresolvable host. - addrinfo hints{}; - hints.ai_family = AF_INET; - hints.ai_socktype = SOCK_STREAM; - addrinfo* resolved = nullptr; - const std::string service = std::to_string(port); - const int status = getaddrinfo(hostName, service.c_str(), &hints, &resolved); - if (status != 0 || resolved == nullptr) { - throw std::runtime_error(std::string("Could not resolve host '") + hostName + "': " - + gai_strerror(status)); - } - host = nullptr; - serv_addr = *reinterpret_cast(resolved->ai_addr); - freeaddrinfo(resolved); - - Connect(); -} - -ClientTCP::ClientTCP(std::string hostName, std::uint16_t port): ClientTCP(hostName.c_str(), port) -{ - -} - -// The moved-from object gives up ownership; the socket itself stays open — -// closing it here (as this used to) meant moving a ClientTCP silently -// dropped the connection it was carrying. -ClientTCP::ClientTCP(ClientTCP&& other) noexcept : socketid(other.socketid) { - other.socketid = -1; -} - -ClientTCP::~ClientTCP() -{ - if(socketid != -1) { - shutdown(socketid, SHUT_RDWR); - close(socketid); - } -} - -void ClientTCP::Connect() { - if((socketid = socket(AF_INET, SOCK_STREAM, 0)) == -1){ - throw std::runtime_error(std::string("Could not open socket: ") + std::strerror(errno)); - } - - if(connect(socketid,(sockaddr*)&serv_addr, sizeof(sockaddr)) == -1){ - // Report the failure instead of handing back a socket that is not - // connected to anything — every later send/recv on it would fail - // with a far less obvious error. - const std::string reason = std::strerror(errno); - close(socketid); - socketid = -1; - throw std::runtime_error("Could not connect to server: " + reason); - } -} - -void ClientTCP::Stop() { - shutdown(socketid, SHUT_RDWR); - close(socketid); - socketid = -1; -} - -void ClientTCP::Send(const void* buffer, std::uint32_t size) const { - // send() is free to accept less than it was offered, and does so - // routinely once a buffer outgrows the socket's send buffer — a - // multi-megabyte HTTP body, say. Loop until it is all handed over. - const char* data = reinterpret_cast(buffer); - std::uint32_t sent = 0; - while (sent < size) { - const auto status = send(socketid, data + sent, size - sent, MSG_NOSIGNAL); - if (status == 0) { - throw SocketClosedException(); - } else if (status < 0) { - if (errno == EINTR) continue; - throw std::runtime_error(std::strerror(errno)); - } - sent += static_cast(status); - } -} -std::vector ClientTCP::RecieveSync(std::uint32_t bufferSize) const { - std::vector totalBuffer(bufferSize); - int read = recv(socketid, totalBuffer.data(), bufferSize, 0); - if(read < 0) { - throw std::runtime_error(std::strerror(errno)); - } else if(read == 0) { - throw SocketClosedException(); - } - return totalBuffer; -} -int ClientTCP::RecieveSync(std::uint32_t bufferSize, void* buffer) const { - return recv(socketid, reinterpret_cast(buffer), bufferSize, 0); -} - -std::vector ClientTCP::RecieveSync() const { - int count; - ioctl(socketid, FIONREAD, &count); - if(count == 0){ - count = 1024; - } - std::vector buffer(count); - int read = recv(socketid, buffer.data(), count, 0); - if(read < 0) { - throw std::runtime_error(std::strerror(errno)); - } else if(read == 0) { - throw SocketClosedException(); - } else if(count != read) { - buffer.resize(read); - } - return buffer; -} - -std::vector ClientTCP::RecieveUntilCloseSync() const { - int count; - ioctl(socketid, FIONREAD, &count); - if(count == 0){ - count = 1024; - } - std::vector buffer(count); - int read = recv(socketid, buffer.data(), count, 0); - if(read < 0) { - throw std::runtime_error(std::strerror(errno)); - } else if(read == 0) { - throw SocketClosedException(); - } - - while(true) { - ioctl(socketid, FIONREAD, &count); - if(count == 0){ - count = 1024; - } - unsigned int oldSize = buffer.size(); - buffer.resize(buffer.size()+count); - int read = recv(socketid, buffer.data()+oldSize, count, 0); - if(read < 0) { - throw std::runtime_error(std::strerror(errno)); - } else if(read == 0) { - return buffer; - } else if(count != read) { - buffer.resize(read); - } - } - return buffer; -} - -std::vector ClientTCP::RecieveUntilFullSync(std::uint32_t bufferSize) const { - std::vector buffer(bufferSize); - int read = 0; - while(read < bufferSize) { - int newRead = recv(socketid, buffer.data()+read, bufferSize-read, 0); - if(newRead < 0) { - throw std::runtime_error(std::strerror(errno)); - } else if(newRead == 0) { - throw SocketClosedException(); - } - read+=newRead; - } - buffer.resize(read); - return buffer; -} - -void ClientTCP::RecieveAsync(std::uint32_t bufferSize, std::function)> recieveCallback) const { - ThreadPool::Enqueue([recieveCallback, this, bufferSize](){ - recieveCallback(this->RecieveSync(bufferSize)); - }); -} - -void ClientTCP::RecieveAsync(std::uint32_t bufferSize, std::function recieveCallback, char* buffer) const { - ThreadPool::Enqueue([recieveCallback, this, bufferSize, buffer](){ - recieveCallback(this->RecieveSync(bufferSize, buffer)); - }); -} - -void ClientTCP::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> recieveCallback) const { - ThreadPool::Enqueue([recieveCallback, this, bufferSize](){ - recieveCallback(this->RecieveUntilFullSync(bufferSize)); - }); -} - -void ClientTCP::RecieveAsync(std::function)> recieveCallback) const { - ThreadPool::Enqueue([this, recieveCallback](){ - recieveCallback(this->RecieveSync()); - }); -} - -void ClientTCP::RecieveUntilCloseAsync(std::function)> recieveCallback) const { - ThreadPool::Enqueue([this, recieveCallback](){ - recieveCallback(this->RecieveUntilCloseSync()); - }); -} +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +module Crafter.Network:ClientTCP_impl; +import :ClientTCP; +import Crafter.Thread; +import std; + +using namespace Crafter; + +namespace { + // recv, retried when a signal interrupts it: a receive waits on through + // a signal rather than failing. + ssize_t RecieveRetrying(int socketid, void* buffer, std::size_t size) { + ssize_t read; + do { + read = recv(socketid, buffer, size, 0); + } while (read < 0 && errno == EINTR); + return read; + } + + // A failed recv. EAGAIN is the receive timeout (SO_RCVTIMEO) running out. + [[noreturn]] void ThrowRecieveError() { + if (errno == EAGAIN || errno == EWOULDBLOCK) throw SocketTimeoutException(); + throw std::runtime_error(std::strerror(errno)); + } +} + +ClientTCP::ClientTCP(int socketid) : socketid(socketid) +{ + +} + +ClientTCP::ClientTCP(const char* hostName, std::uint16_t port, std::chrono::milliseconds connectTimeout) +{ + // getaddrinfo rather than gethostbyname: the latter is not thread safe, + // and it signals failure by returning null — which was then dereferenced + // straight into a crash on any unresolvable host. + addrinfo hints{}; + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + addrinfo* resolved = nullptr; + const std::string service = std::to_string(port); + const int status = getaddrinfo(hostName, service.c_str(), &hints, &resolved); + if (status != 0 || resolved == nullptr) { + throw std::runtime_error(std::string("Could not resolve host '") + hostName + "': " + + gai_strerror(status)); + } + serv_addr = *reinterpret_cast(resolved->ai_addr); + freeaddrinfo(resolved); + + Connect(connectTimeout); +} + +ClientTCP::ClientTCP(std::string hostName, std::uint16_t port, std::chrono::milliseconds connectTimeout): ClientTCP(hostName.c_str(), port, connectTimeout) +{ + +} + +// The moved-from object gives up ownership; the socket itself stays open — +// closing it here (as this used to) meant moving a ClientTCP silently +// dropped the connection it was carrying. +ClientTCP::ClientTCP(ClientTCP&& other) noexcept : socketid(other.socketid) { + other.socketid = -1; +} + +ClientTCP::~ClientTCP() +{ + if(socketid != -1) { + shutdown(socketid, SHUT_RDWR); + close(socketid); + } +} + +void ClientTCP::Connect(std::chrono::milliseconds timeout) { + if((socketid = socket(AF_INET, SOCK_STREAM, 0)) == -1){ + throw std::runtime_error(std::string("Could not open socket: ") + std::strerror(errno)); + } + + // With a timeout the connect runs non-blocking, so poll can bound the + // wait; the socket is made blocking again once it is connected. + const bool bounded = timeout > std::chrono::milliseconds::zero(); + const int flags = fcntl(socketid, F_GETFL, 0); + if (bounded) { + fcntl(socketid, F_SETFL, flags | O_NONBLOCK); + } + int failure = 0; + if(connect(socketid,(sockaddr*)&serv_addr, sizeof(sockaddr)) == -1){ + failure = errno; + if (failure == EINPROGRESS) { + pollfd connecting { .fd = socketid, .events = POLLOUT, .revents = 0 }; + const int waitMs = static_cast(std::min(timeout.count(), std::numeric_limits::max())); + int ready; + do { + ready = poll(&connecting, 1, waitMs); + } while (ready == -1 && errno == EINTR); + if (ready == 0) { + close(socketid); + socketid = -1; + throw SocketTimeoutException(); + } + if (ready == -1) { + failure = errno; + } else { + // The connect's own result. + socklen_t length = sizeof(failure); + getsockopt(socketid, SOL_SOCKET, SO_ERROR, &failure, &length); + } + } + } + if (failure != 0) { + // Report the failure instead of handing back a socket that is not + // connected to anything — every later send/recv on it would fail + // with a far less obvious error. + close(socketid); + socketid = -1; + throw std::runtime_error(std::string("Could not connect to server: ") + std::strerror(failure)); + } + if (bounded) { + fcntl(socketid, F_SETFL, flags); + } +} + +void ClientTCP::Stop() { + shutdown(socketid, SHUT_RDWR); + close(socketid); + socketid = -1; +} + +void ClientTCP::SetNoDelay(bool enabled) { + const int value = enabled ? 1 : 0; + if (setsockopt(socketid, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)) == -1) { + throw std::runtime_error(std::string("Could not set TCP_NODELAY: ") + std::strerror(errno)); + } +} + +void ClientTCP::SetReceiveTimeout(std::chrono::milliseconds timeout) { + const auto seconds = std::chrono::duration_cast(timeout); + timeval value {}; + value.tv_sec = static_cast(seconds.count()); + value.tv_usec = static_cast(std::chrono::duration_cast(timeout - seconds).count()); + if (setsockopt(socketid, SOL_SOCKET, SO_RCVTIMEO, &value, sizeof(value)) == -1) { + throw std::runtime_error(std::string("Could not set the receive timeout: ") + std::strerror(errno)); + } +} + +void ClientTCP::Send(const void* buffer, std::uint32_t size) const { + // send() is free to accept less than it was offered, and does so + // routinely once a buffer outgrows the socket's send buffer — a + // multi-megabyte HTTP body, say. Loop until it is all handed over. + const char* data = reinterpret_cast(buffer); + std::uint32_t sent = 0; + while (sent < size) { + const auto status = send(socketid, data + sent, size - sent, MSG_NOSIGNAL); + if (status == 0) { + throw SocketClosedException(); + } else if (status < 0) { + if (errno == EINTR) continue; + throw std::runtime_error(std::strerror(errno)); + } + sent += static_cast(status); + } +} +std::vector ClientTCP::RecieveSync(std::uint32_t bufferSize) const { + // recv's 0 means a closed peer, but not for a receive of no bytes. + if (bufferSize == 0) return {}; + std::vector buffer(bufferSize); + const ssize_t read = RecieveRetrying(socketid, buffer.data(), bufferSize); + if (read < 0) ThrowRecieveError(); + if (read == 0) throw SocketClosedException(); + buffer.resize(static_cast(read)); + return buffer; +} +std::uint32_t ClientTCP::RecieveSync(std::uint32_t bufferSize, void* buffer) const { + if (bufferSize == 0) return 0; + const ssize_t read = RecieveRetrying(socketid, buffer, bufferSize); + if (read < 0) ThrowRecieveError(); + if (read == 0) throw SocketClosedException(); + return static_cast(read); +} + +std::vector ClientTCP::RecieveSync() const { + int count = 0; + ioctl(socketid, FIONREAD, &count); + if (count <= 0) { + count = 1024; + } + std::vector buffer(static_cast(count)); + const ssize_t read = RecieveRetrying(socketid, buffer.data(), buffer.size()); + if (read < 0) ThrowRecieveError(); + if (read == 0) throw SocketClosedException(); + buffer.resize(static_cast(read)); + return buffer; +} + +std::vector ClientTCP::RecieveUntilCloseSync() const { + std::vector buffer; + while (true) { + int count = 0; + ioctl(socketid, FIONREAD, &count); + if (count <= 0) { + count = 1024; + } + const std::size_t oldSize = buffer.size(); + buffer.resize(oldSize + static_cast(count)); + const ssize_t read = RecieveRetrying(socketid, buffer.data() + oldSize, static_cast(count)); + if (read < 0) ThrowRecieveError(); + buffer.resize(oldSize + static_cast(read)); + if (read == 0) { + // A peer that closes before sending anything is reported the + // way the other receives report it. + if (buffer.empty()) throw SocketClosedException(); + return buffer; + } + } +} + +std::vector ClientTCP::RecieveUntilFullSync(std::uint32_t bufferSize) const { + std::vector buffer(bufferSize); + std::size_t read = 0; + while (read < bufferSize) { + const ssize_t newRead = RecieveRetrying(socketid, buffer.data() + read, bufferSize - read); + if (newRead < 0) ThrowRecieveError(); + if (newRead == 0) throw SocketClosedException(); + read += static_cast(newRead); + } + return buffer; +} + +namespace { + // Runs `receive` on the ThreadPool and hands its result to `onReceived`, + // or what it threw to `onError` (dropped without one): a pool task must + // not let an exception escape, or it ends the program. The callback runs + // outside the try, so its own exceptions aren't taken for a failed + // receive. + template + void EnqueueReceive(std::function receive, std::function onReceived, std::function onError) { + ThreadPool::Enqueue([receive = std::move(receive), onReceived = std::move(onReceived), onError = std::move(onError)]() { + std::optional result; + try { + result = receive(); + } catch (...) { + if (onError) onError(std::current_exception()); + return; + } + onReceived(std::move(*result)); + }); + } +} + +void ClientTCP::RecieveAsync(std::uint32_t bufferSize, std::function)> recieveCallback, std::function onError) const { + EnqueueReceive>([this, bufferSize] { return RecieveSync(bufferSize); }, std::move(recieveCallback), std::move(onError)); +} + +void ClientTCP::RecieveAsync(std::uint32_t bufferSize, std::function recieveCallback, char* buffer, std::function onError) const { + EnqueueReceive([this, bufferSize, buffer] { return RecieveSync(bufferSize, buffer); }, std::move(recieveCallback), std::move(onError)); +} + +void ClientTCP::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> recieveCallback, std::function onError) const { + EnqueueReceive>([this, bufferSize] { return RecieveUntilFullSync(bufferSize); }, std::move(recieveCallback), std::move(onError)); +} + +void ClientTCP::RecieveAsync(std::function)> recieveCallback, std::function onError) const { + EnqueueReceive>([this] { return RecieveSync(); }, std::move(recieveCallback), std::move(onError)); +} + +void ClientTCP::RecieveUntilCloseAsync(std::function)> recieveCallback, std::function onError) const { + EnqueueReceive>([this] { return RecieveUntilCloseSync(); }, std::move(recieveCallback), std::move(onError)); +} diff --git a/interfaces/Crafter.Network-ClientQUIC.cppm b/interfaces/Crafter.Network-ClientQUIC.cppm index c32765c..4333bd8 100644 --- a/interfaces/Crafter.Network-ClientQUIC.cppm +++ b/interfaces/Crafter.Network-ClientQUIC.cppm @@ -126,19 +126,25 @@ namespace Crafter { // Read until the peer closes the send-side, accumulating all chunks. std::vector RecieveUntilCloseSync(); - // Read exactly bufferSize bytes; throws if the peer closes early. + // Read exactly bufferSize bytes. Throws if the peer closes early, + // leaving what did arrive for the next receive. std::vector RecieveUntilFullSync(std::uint32_t bufferSize); #endif // Send a buffer. If finish=true, the send-side is closed after the - // buffer is delivered. onSent fires once the transport has accepted - // the buffer (native) or the WritableStream writer has resolved - // (browser). Available on both native and browser builds. + // buffer is delivered. onSent fires with true once the transport has + // accepted the buffer (native) or the WritableStream writer has + // resolved (browser), and with false when the stream was closed or + // the send failed. Available on both native and browser builds. void SendAsync(const void* buffer, std::uint32_t size, bool finish, - std::function onSent); + std::function onSent); // Async receive variants. Dispatched on Crafter.Thread's ThreadPool - // (native) or driven by a per-stream JS reader loop (browser). + // (native) or driven by a per-stream JS reader loop (browser). A + // stream that has closed gives the callback an empty chunk, and + // RecieveUntilFullAsync on one that closes early gives what arrived: + // the browser build has no exceptions to report it with, and on the + // ThreadPool an exception would end the program. void RecieveAsync(std::function)> recieveCallback); void RecieveUntilCloseAsync(std::function)> recieveCallback); void RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> recieveCallback); diff --git a/interfaces/Crafter.Network-ClientTCP.cppm b/interfaces/Crafter.Network-ClientTCP.cppm index 2b48558..181c830 100755 --- a/interfaces/Crafter.Network-ClientTCP.cppm +++ b/interfaces/Crafter.Network-ClientTCP.cppm @@ -1,59 +1,89 @@ -//SPDX-License-Identifier: LGPL-3.0-only -//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® - -module; -#ifndef CRAFTER_NETWORK_BROWSER -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#endif -export module Crafter.Network:ClientTCP; -import std; - -#ifndef CRAFTER_NETWORK_BROWSER -namespace Crafter { - export class SocketClosedException : public std::exception { - public: - const char* what() const noexcept override { - return "Socket closed"; - } -}; - - export class ClientTCP { - public: - int socketid; - ClientTCP(int socket); - ClientTCP(const char* host, std::uint16_t port); - ClientTCP(std::string host, std::uint16_t port); - ~ClientTCP(); - ClientTCP(const ClientTCP&) = delete; - ClientTCP(ClientTCP&& other) noexcept; - void Connect(); - void Stop(); - void Send(const void* buffer, std::uint32_t size) const; - std::vector RecieveSync() const; - std::vector RecieveUntilCloseSync() const; - std::vector RecieveUntilFullSync(std::uint32_t bufferSize) const; - std::vector RecieveSync(std::uint32_t bufferSize) const; - int RecieveSync(std::uint32_t bufferSize, void* buffer) const; - void RecieveAsync(std::function)> recieveCallback) const; - void RecieveUntilCloseAsync(std::function)> recieveCallback) const; - void RecieveAsync(std::uint32_t bufferSize, std::function)> recieveCallback) const; - void RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> recieveCallback) const; - void RecieveAsync(std::uint32_t bufferSize, std::function recieveCallback, char* buffer) const; - private: - hostent* host; - sockaddr_in serv_addr; - }; -} +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; +#ifndef CRAFTER_NETWORK_BROWSER +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +export module Crafter.Network:ClientTCP; +import std; + +#ifndef CRAFTER_NETWORK_BROWSER +namespace Crafter { + export class SocketClosedException : public std::exception { + public: + const char* what() const noexcept override { + return "Socket closed"; + } +}; + + // A connect or a receive waited longer than its timeout. + export class SocketTimeoutException : public std::exception { + public: + const char* what() const noexcept override { + return "Socket timed out"; + } +}; + + export class ClientTCP { + public: + int socketid; + ClientTCP(int socket); + // A non-zero connectTimeout makes a connect nobody answers throw + // SocketTimeoutException after that long, where it would otherwise + // wait out the kernel's retries, about two minutes. A refused connect + // fails at once either way. + ClientTCP(const char* host, std::uint16_t port, std::chrono::milliseconds connectTimeout = std::chrono::milliseconds::zero()); + ClientTCP(std::string host, std::uint16_t port, std::chrono::milliseconds connectTimeout = std::chrono::milliseconds::zero()); + ~ClientTCP(); + ClientTCP(const ClientTCP&) = delete; + ClientTCP(ClientTCP&& other) noexcept; + void Connect(std::chrono::milliseconds timeout = std::chrono::milliseconds::zero()); + void Stop(); + // Turns Nagle's algorithm off (TCP_NODELAY), so a small write goes + // out at once instead of waiting for the ACK of the one before it. + // For small messages that must arrive promptly, such as a stream of + // frames. + void SetNoDelay(bool enabled); + // A receive that waits longer than this throws + // SocketTimeoutException, losing what a whole-buffer receive had read + // so far. Zero, the default, waits for good. + void SetReceiveTimeout(std::chrono::milliseconds timeout); + void Send(const void* buffer, std::uint32_t size) const; + // The receives return what arrived, at least one byte; asked for none, + // none, at once, without looking at the socket. They throw + // SocketClosedException once the peer has closed, and + // SocketTimeoutException when a receive timeout runs out. + std::vector RecieveSync() const; + std::vector RecieveUntilCloseSync() const; + std::vector RecieveUntilFullSync(std::uint32_t bufferSize) const; + std::vector RecieveSync(std::uint32_t bufferSize) const; + // Into `buffer`; returns how many bytes it put there. + std::uint32_t RecieveSync(std::uint32_t bufferSize, void* buffer) const; + // The same receives on Crafter.Thread's ThreadPool. What arrived goes + // to the callback; what a receive throws goes to onError instead (an + // empty one drops it), as an exception left to escape a pool task + // would end the program. + void RecieveAsync(std::function)> recieveCallback, std::function onError) const; + void RecieveUntilCloseAsync(std::function)> recieveCallback, std::function onError) const; + void RecieveAsync(std::uint32_t bufferSize, std::function)> recieveCallback, std::function onError) const; + void RecieveUntilFullAsync(std::uint32_t bufferSize, std::function)> recieveCallback, std::function onError) const; + // Into `buffer`, which must outlive the receive; the callback gets the count. + void RecieveAsync(std::uint32_t bufferSize, std::function recieveCallback, char* buffer, std::function onError) const; + private: + sockaddr_in serv_addr; + }; +} #endif \ No newline at end of file diff --git a/interfaces/Crafter.Network.cppm b/interfaces/Crafter.Network.cppm index 92cd7a2..66ea843 100755 --- a/interfaces/Crafter.Network.cppm +++ b/interfaces/Crafter.Network.cppm @@ -5,18 +5,24 @@ export module Crafter.Network; export import :ClientTCP; export import :ListenerTCP; +export import :HTTP; +#ifndef CRAFTER_NETWORK_NO_QUIC +// HTTP/3, QUIC and WebTransport: over msquic natively, over the browser's +// WebTransport in wasm. A native build with --no-quic leaves them out. export import :ClientHTTP; export import :ListenerHTTP; -export import :HTTP; export import :ClientQUIC; export import :ListenerQUIC; export import :WebTransport; +#endif #ifndef CRAFTER_NETWORK_BROWSER +#ifndef CRAFTER_NETWORK_NO_QUIC // Exposed so user code can build WebTransport clients by hand against a // ClientQUIC until we ship a ClientWebTransport wrapper. Most callers do // not need the HTTP/3 frame helpers directly. Excluded from the browser // build — HTTP3 uses throw and the wasm target runs with -fno-exceptions. export import :HTTP3; +#endif // HTTP/1.1 over TCP, for peers that are not ready for HTTP/3. Native only: // in the browser this job is already done by fetch() behind :ClientHTTP, // and these partitions use exceptions and POSIX sockets. diff --git a/project.cpp b/project.cpp index 6403c65..4060c1d 100644 --- a/project.cpp +++ b/project.cpp @@ -7,36 +7,52 @@ namespace fs = std::filesystem; using namespace Crafter; extern "C" Configuration CrafterBuildProject(std::span args) { - constexpr std::array networkInterfaces = { + // TCP, HTTP/1.1 and TLS: what every native build has. + constexpr std::array networkInterfaces = { "interfaces/Crafter.Network", "interfaces/Crafter.Network-ClientTCP", "interfaces/Crafter.Network-ListenerTCP", - "interfaces/Crafter.Network-ClientHTTP", - "interfaces/Crafter.Network-ListenerHTTP", "interfaces/Crafter.Network-HTTP", "interfaces/Crafter.Network-HTTP1", "interfaces/Crafter.Network-Stream", "interfaces/Crafter.Network-TLS", "interfaces/Crafter.Network-ClientHTTP1", "interfaces/Crafter.Network-ListenerHTTP1", + }; + // HTTP/3, QUIC and WebTransport, over msquic. --no-quic leaves them out. + constexpr std::array quicInterfaces = { + "interfaces/Crafter.Network-ClientHTTP", + "interfaces/Crafter.Network-ListenerHTTP", "interfaces/Crafter.Network-HTTP3", "interfaces/Crafter.Network-ClientQUIC", "interfaces/Crafter.Network-ListenerQUIC", "interfaces/Crafter.Network-WebTransport", }; - std::vector depArgs(args.begin(), args.end()); - Configuration* thread = GitProject({ - .source = { .url = "https://forgejo.catcrafts.net/Catcrafts/Crafter.Thread.git" }, - .args = depArgs, - }); + // --no-quic is this project's own flag, so it stays out of the + // dependencies' args: Crafter.Thread would otherwise build a second copy + // of itself, hashed apart from the one the consumer imports. + std::vector depArgs; + for (std::string_view arg : args) { + if (arg != "--no-quic") depArgs.emplace_back(arg); + } + // Crafter.Thread comes from forgejo, or with --local from the checkout + // next to this one. --local stays in depArgs, so Crafter.Thread resolves + // its own dependencies the same way. + const bool useLocal = std::ranges::find(args, std::string_view("--local")) != args.end(); + Configuration* thread = useLocal + ? LocalProject({ .projectFile = "../Crafter.Thread/project.cpp", .args = depArgs }) + : GitProject({ + .source = { .url = "https://forgejo.catcrafts.net/Catcrafts/Crafter.Thread.git" }, + .args = depArgs, + }); Configuration cfg; cfg.path = "./"; cfg.name = "crafter-network"; cfg.outputName = "crafter-network"; cfg.type = ConfigurationType::LibraryStatic; - ApplyStandardArgs(cfg, args); + ArgQuery opts = ApplyStandardArgs(cfg, args); cfg.dependencies = { thread }; // Browser path: any wasm32-* target gets the browser network stack @@ -74,44 +90,65 @@ extern "C" Configuration CrafterBuildProject(std::span a return cfg; } - constexpr std::array networkImplementations = { + constexpr std::array networkImplementations = { "implementations/Crafter.Network-ClientTCP", "implementations/Crafter.Network-ListenerTCP", - "implementations/Crafter.Network-ClientHTTP", - "implementations/Crafter.Network-ListenerHTTP", "implementations/Crafter.Network-Stream", "implementations/Crafter.Network-TLS", "implementations/Crafter.Network-ClientHTTP1", "implementations/Crafter.Network-ListenerHTTP1", + }; + constexpr std::array quicImplementations = { + "implementations/Crafter.Network-ClientHTTP", + "implementations/Crafter.Network-ListenerHTTP", "implementations/Crafter.Network-ClientQUIC", "implementations/Crafter.Network-ListenerQUIC", "implementations/Crafter.Network-WebTransport", }; - // msquic — provides the QUIC transport used by ClientQUIC / ListenerQUIC. - // Cloned + built via CMake into the per-project external cache; no system - // package required. Submodules (quictls / clog / etc.) come via the - // recursive clone Crafter.Build performs. We disable msquic's own tests, - // tools and perf binaries since we only need the library. - ExternalDependency& msquic = cfg.externalDependencies.emplace_back(); - msquic.name = "msquic"; - msquic.source.url = "https://github.com/microsoft/msquic.git"; - msquic.source.branch = "main"; - msquic.builder = ExternalBuilder::CMake; - msquic.options = { - "-DQUIC_TLS_LIB=quictls", - "-DQUIC_BUILD_TEST=OFF", - "-DQUIC_BUILD_TOOLS=OFF", - "-DQUIC_BUILD_PERF=OFF", - "-DQUIC_BUILD_SHARED=ON", - }; - msquic.includeDirs = { "src/inc" }; - // msquic's CMakeLists overrides CMAKE_LIBRARY_OUTPUT_DIRECTORY with - // QUIC_OUTPUT_DIR (defaults to bin/$), so libmsquic.so lands in - // a subdir of the cmake build dir rather than at its root. Point the - // linker at the actual output location. - msquic.libDirs = { "bin/Release" }; - msquic.libs = { "msquic" }; + // --no-quic: TCP, HTTP/1.1 and TLS only, for consumers that need no more. + // msquic takes a recursive clone and a CMake build, so leaving it out is + // most of a fresh build. The browser build above has no msquic to leave + // out and ignores the flag. + const bool quic = !opts.Has("--no-quic"); + if (quic) { + // msquic — provides the QUIC transport used by ClientQUIC / ListenerQUIC. + // Cloned + built via CMake into the per-project external cache; no system + // package required. Pinned to a release (v2.6.3): main moves, and its + // submodules (quictls / clog / etc.) with it. Crafter.Build checks the + // submodules out at the revisions the pinned commit records. We disable + // msquic's own tests, tools and perf binaries since we only need the + // library. + ExternalDependency& msquic = cfg.externalDependencies.emplace_back(); + msquic.name = "msquic"; + msquic.source.url = "https://github.com/microsoft/msquic.git"; + msquic.source.commit = "16ca4d8d785d34fb22285e7c652e477ac16475a5"; + msquic.builder = ExternalBuilder::CMake; + // TLS from the system's OpenSSL (3.5 or newer, which has the QUIC + // API), the same libssl the HTTP/1.1 TLS code links: one OpenSSL in + // the process. The quictls fork msquic otherwise builds and bundles is + // an older OpenSSL that still reads the system's openssl.cnf, and the + // Flatpak runtime's (an engine section for the runtime's own OpenSSL, + // config_diagnostics on) makes its initialisation fail: MsQuicOpen2 + // returned QUIC_STATUS_TLS_ERROR (0x7e) there. + msquic.options = { + "-DQUIC_TLS_LIB=openssl", + "-DQUIC_USE_EXTERNAL_OPENSSL=ON", + "-DQUIC_BUILD_TEST=OFF", + "-DQUIC_BUILD_TOOLS=OFF", + "-DQUIC_BUILD_PERF=OFF", + "-DQUIC_BUILD_SHARED=ON", + }; + msquic.includeDirs = { "src/inc" }; + // msquic's CMakeLists overrides CMAKE_LIBRARY_OUTPUT_DIRECTORY with + // QUIC_OUTPUT_DIR (defaults to bin/$), so libmsquic.so lands in + // a subdir of the cmake build dir rather than at its root. Point the + // linker at the actual output location. + msquic.libDirs = { "bin/Release" }; + msquic.libs = { "msquic" }; + } else { + cfg.defines.push_back({"CRAFTER_NETWORK_NO_QUIC", ""}); + } // libssl/libcrypto — the TLS transport behind :TLS, i.e. https:// on the // HTTP/1.1 client and listener. A system package rather than a built @@ -121,36 +158,48 @@ extern "C" Configuration CrafterBuildProject(std::span a cfg.linkFlags.push_back("-lssl"); cfg.linkFlags.push_back("-lcrypto"); - std::array ifaces; - std::ranges::copy(networkInterfaces, ifaces.begin()); - std::array impls; - std::ranges::copy(networkImplementations, impls.begin()); + std::vector ifaces(networkInterfaces.begin(), networkInterfaces.end()); + std::vector impls(networkImplementations.begin(), networkImplementations.end()); + if (quic) { + ifaces.insert(ifaces.end(), quicInterfaces.begin(), quicInterfaces.end()); + impls.insert(impls.end(), quicImplementations.begin(), quicImplementations.end()); + } cfg.GetInterfacesAndImplementations(ifaces, impls); - // Linux-only: msquic + POSIX socket backends. The browser path above - // returns early, so wasm builds skip these. Each test links the local - // crafter-network static lib via .Dependencies({ &cfg }). - if (cfg.target == "x86_64-pc-linux-gnu") { - cfg.AddTest("ShouldEchoWebTransport").Dependencies({ &cfg }); - cfg.AddTest("ShouldFallbackUnknownRoutes").Dependencies({ &cfg }); - cfg.AddTest("ShouldHonourQUICSettings").Dependencies({ &cfg }); + // Linux-only: the POSIX socket backends, and msquic unless --no-quic. + // The browser path above returns early, so wasm builds skip these. Each + // test links the local crafter-network static lib via + // .Dependencies({ &cfg }). Run the suite with --no-quic as well. Any + // Linux triple: the toolchain's own is x86_64-unknown-linux-gnu. + if (cfg.target.find("linux") != std::string::npos) { + cfg.AddTest("ShouldConfigureTCP").Dependencies({ &cfg }); + cfg.AddTest("ShouldRecieveTCP").Dependencies({ &cfg }); cfg.AddTest("ShouldInteropCurlHTTP1").Dependencies({ &cfg }); cfg.AddTest("ShouldInteropCurlHTTPS1").Dependencies({ &cfg }); - cfg.AddTest("ShouldNotDropEarlyStreams").Dependencies({ &cfg }); cfg.AddTest("ShouldParseHTTP1").Dependencies({ &cfg }); cfg.AddTest("ShouldRequireClientCertificateHTTPS1").Dependencies({ &cfg }); - cfg.AddTest("ShouldSend").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveHTTP1").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveHTTPS1").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveKeepaliveHTTP1").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveLargeHTTP1").Dependencies({ &cfg }); - cfg.AddTest("ShouldSendRecieveHTTP").Dependencies({ &cfg }); - cfg.AddTest("ShouldSendRecieveKeepaliveHTTP").Dependencies({ &cfg }); - cfg.AddTest("ShouldSendRecieveLargeHTTP").Dependencies({ &cfg }); - cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg }); - cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg }); cfg.AddTest("ShouldSurviveAbuseHTTP1").Dependencies({ &cfg }); - cfg.AddTest("ShouldSurviveConnectionChurn").Dependencies({ &cfg }); + if (quic) { + cfg.AddTest("ShouldEchoWebTransport").Dependencies({ &cfg }); + cfg.AddTest("ShouldFallbackUnknownRoutes").Dependencies({ &cfg }); + cfg.AddTest("ShouldHonourQUICSettings").Dependencies({ &cfg }); + cfg.AddTest("ShouldNotDropEarlyStreams").Dependencies({ &cfg }); + cfg.AddTest("ShouldSend").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendRecieveHTTP").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendRecieveKeepaliveHTTP").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendRecieveLargeHTTP").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg }); + cfg.AddTest("ShouldReportClosedQUICStream").Dependencies({ &cfg }); + cfg.AddTest("ShouldSurviveConnectionChurn").Dependencies({ &cfg }); + } else { + // Only a --no-quic build can show that msquic stayed out. + cfg.AddTest("ShouldBuildWithoutQUIC").Dependencies({ &cfg }); + } } return cfg; diff --git a/tests/ShouldBuildWithoutQUIC/main.cpp b/tests/ShouldBuildWithoutQUIC/main.cpp new file mode 100644 index 0000000..e2243c3 --- /dev/null +++ b/tests/ShouldBuildWithoutQUIC/main.cpp @@ -0,0 +1,62 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// A build with --no-quic leaves out the QUIC, HTTP/3 and WebTransport +// partitions, and msquic with them, for consumers that only need TCP or +// HTTP/1.1: building msquic means cloning it with its submodules and running +// its CMake build. This test is only built that way. It checks that msquic is +// not in the process and that TCP and HTTP/1.1 still work. + +#include +#include +#include +#include +#include + +import Crafter.Network; +import std; +using namespace Crafter; + +namespace { + +int failures = 0; + +void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; +} + +} // namespace + +int main() { + std::ifstream maps("/proc/self/maps"); + const std::string mapped((std::istreambuf_iterator(maps)), std::istreambuf_iterator()); + Check(!mapped.empty() && mapped.find("msquic") == std::string::npos, "libmsquic is not loaded"); + + const int listener = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address {}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + bind(listener, reinterpret_cast(&address), sizeof(address)); + listen(listener, 1); + socklen_t length = sizeof(address); + getsockname(listener, reinterpret_cast(&address), &length); + + ClientTCP client("127.0.0.1", ntohs(address.sin_port)); + const int peer = accept(listener, nullptr, nullptr); + client.Send("ping", 4); + char echoed[4] {}; + recv(peer, echoed, sizeof(echoed), MSG_WAITALL); + Check(std::string_view(echoed, 4) == "ping", "a ClientTCP still connects and sends over loopback"); + + // HTTP/1.1 stays in this build. + HTTP1::MessageParser parser(HTTP1::MessageKind::Request); + const std::string_view wire = "GET / HTTP/1.1\r\nHost: h\r\n\r\n"; + parser.Feed(wire.data(), wire.size()); + Check(parser.Complete() && parser.TakeRequest().path == "/", "the HTTP/1.1 parser is still there"); + + close(peer); + close(listener); + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE; +} diff --git a/tests/ShouldConfigureTCP/main.cpp b/tests/ShouldConfigureTCP/main.cpp new file mode 100644 index 0000000..4f71e91 --- /dev/null +++ b/tests/ShouldConfigureTCP/main.cpp @@ -0,0 +1,150 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// ClientTCP's socket options: +// - SetNoDelay turns Nagle's algorithm off, so a small message goes out at +// once instead of waiting for the ACK of the one before it, which a client +// streaming small frames needs. +// - A receive timeout makes a receive that waits too long throw +// SocketTimeoutException. +// - A connect timeout does the same for a connect nobody answers, where +// connect() would otherwise sit through the kernel's SYN retries, about +// two minutes. +// Before, a ClientTCP had none of these, so a stalled peer blocked the +// reading thread for good. +// +// Everything runs over loopback. The unanswered connect goes to a listener +// whose accept queue is full: Linux drops the SYNs of further connects to it. + +#include +#include +#include +#include +#include +#include + +import Crafter.Network; +import std; +using namespace Crafter; +using namespace std::chrono_literals; + +namespace { + +int failures = 0; + +void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; +} + +// A socket listening on 127.0.0.1, on a port the kernel picks. +struct LoopbackListener { + int fd = -1; + std::uint16_t port = 0; + + explicit LoopbackListener(int backlog) { + fd = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address {}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + bind(fd, reinterpret_cast(&address), sizeof(address)); + listen(fd, backlog); + socklen_t length = sizeof(address); + getsockname(fd, reinterpret_cast(&address), &length); + port = ntohs(address.sin_port); + } + ~LoopbackListener() { close(fd); } +}; + +long long Milliseconds(std::chrono::steady_clock::duration duration) { + return std::chrono::duration_cast(duration).count(); +} + +} // namespace + +int main() { + // ── SetNoDelay ─────────────────────────────────────────────────────────── + { + LoopbackListener listener(4); + ClientTCP client("127.0.0.1", listener.port); + auto noDelay = [&] { + int value = 0; + socklen_t size = sizeof(value); + getsockopt(client.socketid, IPPROTO_TCP, TCP_NODELAY, &value, &size); + return value != 0; + }; + Check(!noDelay(), "a new connection uses Nagle's algorithm"); + client.SetNoDelay(true); + Check(noDelay(), "SetNoDelay(true) turns it off"); + client.SetNoDelay(false); + Check(!noDelay(), "SetNoDelay(false) turns it back on"); + } + + // ── Receive timeout ────────────────────────────────────────────────────── + { + LoopbackListener listener(4); + ClientTCP client("127.0.0.1", listener.port); + const int peer = accept(listener.fd, nullptr, nullptr); + client.SetReceiveTimeout(150ms); + + const auto start = std::chrono::steady_clock::now(); + bool timedOut = false; + try { + client.RecieveUntilFullSync(16); + } catch (const SocketTimeoutException&) { + timedOut = true; + } + const auto waited = std::chrono::steady_clock::now() - start; + Check(timedOut, "a receive from a peer that sends nothing throws SocketTimeoutException"); + Check(waited >= 140ms && waited < 1s, std::format("after about the timeout ({} ms)", Milliseconds(waited))); + + send(peer, "hello", 5, 0); + const std::vector received = client.RecieveUntilFullSync(5); + Check(std::string_view(received.data(), received.size()) == "hello", + "the connection still receives what arrives after a timeout"); + close(peer); + } + + // ── Connect timeout ────────────────────────────────────────────────────── + { + // Connects queue up until the accept queue is full, which with a + // backlog of 0 is after one or two; the next one goes unanswered. + LoopbackListener full(0); + std::vector queued; + bool timedOut = false; + std::chrono::steady_clock::duration waited {}; + for (int i = 0; i < 8 && !timedOut; ++i) { + const auto start = std::chrono::steady_clock::now(); + try { + queued.emplace_back("127.0.0.1", full.port, 200ms); + } catch (const SocketTimeoutException&) { + timedOut = true; + waited = std::chrono::steady_clock::now() - start; + } + } + Check(timedOut, std::format("a connect nobody answers throws SocketTimeoutException (after {} queued)", queued.size())); + Check(waited >= 190ms && waited < 1s, std::format("after about the timeout ({} ms)", Milliseconds(waited))); + } + + // ── A refused connect still fails at once ──────────────────────────────── + { + std::uint16_t closedPort = 0; + { + LoopbackListener gone(1); + closedPort = gone.port; + } + const auto start = std::chrono::steady_clock::now(); + bool refused = false; + try { + ClientTCP client("127.0.0.1", closedPort, 2s); + } catch (const SocketTimeoutException&) { + } catch (const std::runtime_error&) { + refused = true; + } + Check(refused && std::chrono::steady_clock::now() - start < 1s, + "a connect to a closed port fails at once, not after the timeout"); + } + + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE; +} diff --git a/tests/ShouldRecieveTCP/main.cpp b/tests/ShouldRecieveTCP/main.cpp new file mode 100644 index 0000000..cdf7eae --- /dev/null +++ b/tests/ShouldRecieveTCP/main.cpp @@ -0,0 +1,272 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// ClientTCP's receive calls return exactly the bytes that arrived: +// - RecieveSync(bufferSize) returned the whole buffer when fewer bytes had +// arrived, so the caller read garbage after them. +// - RecieveUntilCloseSync kept the unused end of its first buffer, and then +// cut everything it had read down to the size of the last read. +// - A signal arriving while a receive waited made it throw "Interrupted +// system call", where the receive should simply wait on. +// - RecieveSync(bufferSize, buffer) returned recv's -1 or 0 where the other +// receives throw: it now returns the count and throws the same way. A +// receive of no bytes returns at once: recv's 0 for it isn't a close. +// - The *Async receives let a closed peer's exception escape the thread +// pool's task, which ended the whole program; it now goes to onError, +// and without an onError it's dropped. +// +// Everything runs over loopback. + +#include +#include +#include +#include +#include +#include +#include + +import Crafter.Network; +import Crafter.Thread; +import std; +using namespace Crafter; +using namespace std::chrono_literals; + +namespace { + +int failures = 0; + +void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; +} + +// A connected pair: a ClientTCP and the socket the listener accepted for it. +struct Connection { + int listener = -1; + int peer = -1; + std::optional client; + + Connection() { + listener = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address {}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + bind(listener, reinterpret_cast(&address), sizeof(address)); + listen(listener, 1); + socklen_t length = sizeof(address); + getsockname(listener, reinterpret_cast(&address), &length); + client.emplace("127.0.0.1", ntohs(address.sin_port)); + peer = accept(listener, nullptr, nullptr); + } + ~Connection() { + if (peer != -1) close(peer); + close(listener); + } + void Send(std::string_view bytes) const { + send(peer, bytes.data(), bytes.size(), 0); + } + void ClosePeer() { + close(peer); + peer = -1; + } +}; + +std::string_view Text(const std::vector& bytes) { + return std::string_view(bytes.data(), bytes.size()); +} + +// What a receive threw, by kind. +std::string Kind(std::exception_ptr error) { + try { + std::rethrow_exception(error); + } catch (const SocketClosedException&) { + return "closed"; + } catch (const SocketTimeoutException&) { + return "timeout"; + } catch (const std::exception& e) { + return std::string("other: ") + e.what(); + } +} + +// An async receive's outcome. Shared with the callbacks, so it outlives +// whichever thread sets it. +struct Outcome { + std::mutex mutex; + std::condition_variable done; + std::optional value; + + void Set(std::string what) { + { + std::lock_guard lock(mutex); + value = std::move(what); + } + done.notify_all(); + } + std::string Wait() { + std::unique_lock lock(mutex); + done.wait_for(lock, 2s, [&] { return value.has_value(); }); + return value.value_or("nothing within 2 s"); + } +}; + +void IgnoreSignal(int) {} + +} // namespace + +int main() { + // ── A short read returns only what arrived ─────────────────────────────── + { + Connection connection; + connection.Send("hello"); + std::this_thread::sleep_for(20ms); + const std::vector received = connection.client->RecieveSync(64); + Check(Text(received) == "hello", std::format("RecieveSync(64) returns the 5 bytes that arrived (got {})", received.size())); + } + + // ── Until close: every chunk, in order, nothing more ───────────────────── + { + Connection connection; + std::thread sender([&] { + connection.Send("abc"); + std::this_thread::sleep_for(30ms); + connection.Send("defgh"); + std::this_thread::sleep_for(30ms); + connection.Send("ij"); + std::this_thread::sleep_for(30ms); + connection.ClosePeer(); + }); + const std::vector received = connection.client->RecieveUntilCloseSync(); + sender.join(); + Check(Text(received) == "abcdefghij", + std::format("RecieveUntilCloseSync returns the three chunks it was sent ({} bytes)", received.size())); + } + + // ── A signal doesn't break a waiting receive ───────────────────────────── + { + // No SA_RESTART, so the signal makes recv return EINTR. + struct sigaction action {}; + action.sa_handler = IgnoreSignal; + sigemptyset(&action.sa_mask); + sigaction(SIGUSR1, &action, nullptr); + + Connection connection; + std::string outcome; + std::thread reader([&] { + try { + outcome = Text(connection.client->RecieveUntilFullSync(4)); + } catch (const std::exception& e) { + outcome = std::string("threw: ") + e.what(); + } + }); + std::this_thread::sleep_for(50ms); + pthread_kill(reader.native_handle(), SIGUSR1); + std::this_thread::sleep_for(50ms); + connection.Send("data"); + reader.join(); + Check(outcome == "data", std::format("a receive interrupted by a signal waits on for its data ({})", outcome)); + } + + // ── Into a buffer: the count, and the same exceptions ──────────────────── + { + Connection connection; + connection.Send("hello"); + std::this_thread::sleep_for(20ms); + std::array buffer {}; + const std::uint32_t got = connection.client->RecieveSync(buffer.size(), buffer.data()); + Check(got == 5 && std::string_view(buffer.data(), got) == "hello", std::format("RecieveSync(64, buffer) returns the 5 bytes that arrived ({})", got)); + connection.ClosePeer(); + std::string closed; + try { + (void)connection.client->RecieveSync(buffer.size(), buffer.data()); + } catch (...) { + closed = Kind(std::current_exception()); + } + Check(closed == "closed", std::format("and throws SocketClosedException once the peer has closed ({})", closed.empty() ? "returned" : closed)); + } + { + Connection connection; + std::array buffer {}; + std::string outcome = "returned"; + try { + const bool empty = connection.client->RecieveSync(0).empty() && connection.client->RecieveSync(0, buffer.data()) == 0; + if (!empty) outcome = "bytes"; + } catch (...) { + outcome = Kind(std::current_exception()); + } + Check(outcome == "returned", std::format("a receive of no bytes returns nothing on a live connection ({})", outcome)); + } + { + Connection connection; + connection.client->SetReceiveTimeout(50ms); + std::array buffer {}; + std::string timedOut; + try { + (void)connection.client->RecieveSync(buffer.size(), buffer.data()); + } catch (...) { + timedOut = Kind(std::current_exception()); + } + Check(timedOut == "timeout", std::format("and SocketTimeoutException when nothing comes in time ({})", timedOut.empty() ? "returned" : timedOut)); + } + + // ── Async: data to the callback, failures to onError ───────────────────── + ThreadPool::Start(3); + { + Connection connection; + connection.Send("hi"); + auto outcome = std::make_shared(); + connection.client->RecieveAsync(64, [outcome](std::vector data) { outcome->Set(std::string(Text(data))); }, + [outcome](std::exception_ptr error) { outcome->Set(Kind(error)); }); + const std::string got = outcome->Wait(); + Check(got == "hi", std::format("RecieveAsync hands the data to its callback ({})", got)); + } + { + Connection connection; + connection.ClosePeer(); + auto outcome = std::make_shared(); + connection.client->RecieveAsync(64, [outcome](std::vector) { outcome->Set("data"); }, + [outcome](std::exception_ptr error) { outcome->Set(Kind(error)); }); + const std::string got = outcome->Wait(); + Check(got == "closed", std::format("and a closed peer to onError, the program carrying on ({})", got)); + } + { + Connection connection; + connection.Send("ab"); + connection.ClosePeer(); + auto outcome = std::make_shared(); + connection.client->RecieveUntilFullAsync(4, [outcome](std::vector) { outcome->Set("data"); }, + [outcome](std::exception_ptr error) { outcome->Set(Kind(error)); }); + const std::string got = outcome->Wait(); + Check(got == "closed", std::format("RecieveUntilFullAsync reports a peer that closes early to onError ({})", got)); + } + { + Connection connection; + connection.Send("xyz"); + std::array buffer {}; + auto outcome = std::make_shared(); + connection.client->RecieveAsync(buffer.size(), [outcome, &buffer](std::uint32_t count) { outcome->Set(std::string(buffer.data(), count)); }, + buffer.data(), [outcome](std::exception_ptr error) { outcome->Set(Kind(error)); }); + const std::string got = outcome->Wait(); + Check(got == "xyz", std::format("the buffer RecieveAsync hands its callback the count ({})", got)); + } + { + Connection connection; + connection.ClosePeer(); + auto outcome = std::make_shared(); + connection.client->RecieveUntilCloseAsync([outcome](std::vector) { outcome->Set("data"); }, + [outcome](std::exception_ptr error) { outcome->Set(Kind(error)); }); + const std::string got = outcome->Wait(); + Check(got == "closed", std::format("RecieveUntilCloseAsync on a peer that sent nothing goes to onError ({})", got)); + } + { + Connection connection; + connection.ClosePeer(); + connection.client->RecieveAsync(64, [](std::vector) {}, {}); + // Calling the empty onError ended the program from the pool's task. + std::this_thread::sleep_for(300ms); + Check(true, "without an onError, a closed peer's exception is dropped, the program carrying on"); + } + ThreadPool::Stop(); + + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE; +} diff --git a/tests/ShouldReportClosedQUICStream/main.cpp b/tests/ShouldReportClosedQUICStream/main.cpp new file mode 100644 index 0000000..c365172 --- /dev/null +++ b/tests/ShouldReportClosedQUICStream/main.cpp @@ -0,0 +1,135 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// A QUIC stream's async calls report a closed stream to their callbacks, the +// same way on native as in the browser build, which has no exceptions: +// - a receive on a closed stream gets an empty chunk. Natively it used to +// let QUICClosedException escape the ThreadPool task, which ended the +// program; +// - RecieveUntilFullAsync on a stream that closes early gets what arrived; +// RecieveUntilFullSync still throws, and leaves those bytes for the next +// receive instead of dropping them; +// - SendAsync tells its callback whether the stream took the buffer. It +// used to call it the same way either way. +// A server answers a stream with "abc" and closes its side. + +#include + +import Crafter.Network; +import Crafter.Thread; +import std; +using namespace Crafter; + +namespace { + +int failures = 0; + +void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; +} + +// A callback's result, shared with it so it outlives whichever thread sets it. +template +struct Outcome { + std::mutex mutex; + std::condition_variable done; + std::optional value; + + void Set(T result) { + { + std::lock_guard lock(mutex); + value = std::move(result); + } + done.notify_all(); + } + std::optional Wait() { + std::unique_lock lock(mutex); + done.wait_for(lock, std::chrono::seconds(5), [&] { return value.has_value(); }); + return value; + } +}; + +std::string Text(const std::vector& bytes) { + return std::string(bytes.begin(), bytes.end()); +} + +} // namespace + +int main() { + ThreadPool::Start(); + constexpr std::uint16_t port = 9188; + constexpr std::string_view alpn = "crafter/test-closed-stream"; + + QUICServerCredentials serverCreds; + serverCreds.selfSigned = true; + ListenerQUIC listener(port, std::string(alpn), serverCreds, [&](ClientQUIC* peer) { + peer->OnStream([](QUICStream stream) { + try { + (void)stream.RecieveSync(); + stream.SendSync("abc", 3, true); + } catch (...) {} + }); + }); + listener.ListenAsyncAsync(); + + try { + // ── A stream that was never opened ─────────────────────────────────── + { + QUICStream closed; + auto received = std::make_shared>(); + closed.RecieveAsync([received](std::vector chunk) { received->Set(Text(chunk)); }); + const std::optional got = received->Wait(); + Check(got && got->empty(), "a receive on a closed stream gets an empty chunk, and the program carries on"); + auto sent = std::make_shared>(); + closed.SendAsync("x", 1, false, [sent](bool ok) { sent->Set(ok); }); + const std::optional ok = sent->Wait(); + Check(ok && !*ok, "a send on it tells its callback the stream didn't take it"); + } + + QUICClientCredentials clientCreds; + clientCreds.insecureNoServerValidation = true; + ClientQUIC client("localhost", port, std::string(alpn), clientCreds); + + // ── The server answers, then closes its side ───────────────────────── + { + QUICStream stream = client.OpenStream(); + auto sent = std::make_shared>(); + stream.SendAsync("hi", 2, false, [sent](bool ok) { sent->Set(ok); }); + const std::optional ok = sent->Wait(); + Check(ok && *ok, "a send the stream takes tells its callback so"); + + auto first = std::make_shared>(); + stream.RecieveUntilFullAsync(10, [first](std::vector chunk) { first->Set(Text(chunk)); }); + const std::optional got = first->Wait(); + Check(got == "abc", std::format("RecieveUntilFullAsync on a stream that closes early gets what arrived ({})", got.value_or("nothing"))); + + auto after = std::make_shared>(); + stream.RecieveAsync([after](std::vector chunk) { after->Set(Text(chunk)); }); + const std::optional next = after->Wait(); + Check(next && next->empty(), "a receive after the close gets an empty chunk"); + } + { + QUICStream stream = client.OpenStream(); + stream.SendSync("hi", 2, false); + bool threw = false; + try { + (void)stream.RecieveUntilFullSync(10); + } catch (const QUICClosedException&) { + threw = true; + } + const std::vector left = stream.RecieveUntilCloseSync(); + Check(threw && Text(left) == "abc", + std::format("RecieveUntilFullSync throws on the early close and leaves the bytes for the next receive ({})", Text(left))); + } + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + // As the other QUIC tests: skip the static teardown, which waits for + // the accepted peer's idle timeout. _Exit doesn't flush. + std::fflush(stdout); + std::_Exit(failures == 0 ? 0 : 1); + } catch (const std::exception& e) { + std::println("FAIL {}", e.what()); + std::fflush(stdout); + std::_Exit(1); + } +}