diff --git a/project.cpp b/project.cpp index a0f4f9d..e23efb9 100644 --- a/project.cpp +++ b/project.cpp @@ -204,6 +204,7 @@ extern "C" Configuration CrafterBuildProject(std::span a cfg.AddTest("ShouldSendRecieveLargeHTTP").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg }); cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg }); + cfg.AddTest("ShouldSendWithoutBlockingQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldReportClosedQUICStream").Dependencies({ &cfg }); cfg.AddTest("ShouldReportListenerPortQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldReturnFromSendSyncQUIC").Dependencies({ &cfg }); diff --git a/tests/ShouldSendWithoutBlockingQUIC/main.cpp b/tests/ShouldSendWithoutBlockingQUIC/main.cpp new file mode 100644 index 0000000..2e9fd93 --- /dev/null +++ b/tests/ShouldSendWithoutBlockingQUIC/main.cpp @@ -0,0 +1,317 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// QUICStream::SendAsync holds no thread. It used to be SendSync run as a +// pool task: a pool thread was held until msquic accepted the buffer, which +// a slow peer stretches to its own pace, so a server streaming to a slow +// follower lost a worker of a pool the host shares, and the task reached back +// through the wrapper, which a move or a destructor in the meantime freed. +// Now the copy goes to msquic at once and its SEND_COMPLETE reports, on +// msquic's worker, exactly once. +// +// - A burst of 256 sends of 64 KiB returns with most of it still in flight +// (BytesInFlight), every send then reports true, in order, the counters +// drain to zero, and the peer gets every byte in order. +// - With every pool worker parked, a chain of 100 sends, each made from the +// previous one's onSent, completes: nothing needed the pool. +// - A connection stopped under a burst reports the rest false, each send +// exactly once, the trues before the falses. +// - Sends made just before the wrapper is destroyed still go out and +// report true. +// - Four threads sending on one stream at once all complete. +// - A send on a closed stream reports false from a pool task, not from +// inside SendAsync. + +#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; +} + +bool WaitFor(std::function done, std::chrono::milliseconds limit) { + const auto deadline = std::chrono::steady_clock::now() + limit; + while (!done()) { + if (std::chrono::steady_clock::now() >= deadline) return false; + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } + return true; +} + +// Byte `offset` of a stream, after its one-byte tag. +char Pattern(std::uint64_t offset) { return static_cast((offset * 131 + 7) & 0xff); } + +// What the server read from each stream, by the stream's first byte. +struct Received { + std::mutex mutex; + std::map> streams; + + std::optional> Of(char tag) { + std::lock_guard lock(mutex); + auto it = streams.find(tag); + if (it == streams.end()) return std::nullopt; + return it->second; + } +}; + +bool FollowsPattern(const std::vector& bytes, std::size_t expected) { + if (bytes.size() != expected + 1) return false; + for (std::size_t i = 1; i < bytes.size(); ++i) { + if (bytes[i] != Pattern(i - 1)) return false; + } + return true; +} + +// The completions of a set of sends, by index. +struct Completions { + std::mutex mutex; + std::vector calls; // how often each send's onSent fired + std::vector sent; // what it said + std::vector order; // indices, in the order they fired + + explicit Completions(std::size_t count) : calls(count, 0), sent(count, false) {} + + std::function For(std::size_t index) { + return [this, index](bool ok) { + std::lock_guard lock(mutex); + ++calls[index]; + sent[index] = ok; + order.push_back(static_cast(index)); + }; + } + std::size_t Fired() { + std::lock_guard lock(mutex); + return order.size(); + } +}; + +} // namespace + +int main() { + ThreadPool::Start(); + std::thread([] { + std::this_thread::sleep_for(std::chrono::seconds(45)); + std::println("FAIL timed out"); + std::fflush(stdout); + std::_Exit(1); + }).detach(); + + const std::string alpn = "crafter/test-send-async"; + QUICServerCredentials serverCreds; + serverCreds.selfSigned = true; + QUICClientCredentials clientCreds; + clientCreds.insecureNoServerValidation = true; + + // Leaked: readers may still be finishing when main has moved on. + Received& received = *new Received; + std::atomic readersStarted = 0; + + try { + // Each stream gets a reader on a thread of its own, not on the pool, + // which the second part parks. + ListenerQUIC listener("127.0.0.1", 0, alpn, serverCreds, [&](ClientQUIC* peer) { + peer->OnStream([&](QUICStream stream) { + auto shared = std::make_shared(std::move(stream)); + ++readersStarted; + std::thread([&received, shared] { + std::vector bytes; + try { + bytes = shared->RecieveUntilCloseSync(); + } catch (...) {} + if (bytes.empty()) return; + std::lock_guard lock(received.mutex); + received.streams[bytes.front()] = std::move(bytes); + }).detach(); + }); + }); + listener.ListenAsyncAsync(); + ClientQUIC client("127.0.0.1", listener.Port(), alpn, clientCreds); + + // ── A burst ────────────────────────────────────────────────────────── + { + constexpr std::size_t count = 256; + constexpr std::size_t size = 64 * 1024; + auto completions = std::make_shared(count + 1); + QUICStream stream = client.OpenStream(); + stream.SendAsync("A", 1, false, completions->For(count)); + std::vector chunk(size); + const auto start = std::chrono::steady_clock::now(); + for (std::size_t i = 0; i < count; ++i) { + for (std::size_t j = 0; j < size; ++j) chunk[j] = Pattern(i * size + j); + stream.SendAsync(chunk.data(), static_cast(size), i + 1 == count, completions->For(i)); + } + const auto took = std::chrono::duration_cast(std::chrono::steady_clock::now() - start); + const std::uint64_t inFlight = stream.BytesInFlight(); + Check(inFlight > 0 && stream.SendsInFlight() > 0, + std::format("256 sends of 64 KiB returned in {} ms with {} KiB still in flight: none waited for msquic", + took.count(), inFlight / 1024)); + Check(WaitFor([&] { return completions->Fired() == count + 1; }, std::chrono::seconds(20)), + "every send reported"); + { + std::lock_guard lock(completions->mutex); + const bool once = std::ranges::all_of(completions->calls, [](int calls) { return calls == 1; }); + const bool allSent = std::ranges::all_of(completions->sent, [](bool ok) { return ok; }); + std::vector data(completions->order.begin(), completions->order.end()); + std::erase(data, static_cast(count)); + Check(once && allSent && std::ranges::is_sorted(data), "each exactly once, true, and in the order sent"); + } + Check(stream.BytesInFlight() == 0 && stream.SendsInFlight() == 0, "after which nothing is in flight"); + const bool arrived = WaitFor([&] { return received.Of('A').has_value(); }, std::chrono::seconds(20)); + Check(arrived && FollowsPattern(*received.Of('A'), count * size), "and the peer read all 16 MiB, in order"); + } + + // ── No pool ────────────────────────────────────────────────────────── + { + QUICStream stream = client.OpenStream(); + const int before = readersStarted.load(); + stream.SendSync("B", 1, false); + // The server's OnStream is itself a pool task: let it start the + // reader before the pool is parked. + WaitFor([&] { return readersStarted.load() > before; }, std::chrono::seconds(10)); + + std::atomic release = false; + std::atomic parked = 0; + const unsigned workers = ThreadPool::activeThreads.load(); + for (unsigned i = 0; i < workers * 2; ++i) { + ThreadPool::Enqueue([&] { + ++parked; + while (!release.load()) std::this_thread::sleep_for(std::chrono::milliseconds(1)); + }); + } + const bool allParked = WaitFor([&] { return parked.load() >= workers; }, std::chrono::seconds(10)); + + constexpr int links = 100; + auto done = std::make_shared>(0); + auto ok = std::make_shared>(true); + // Each onSent makes the next send, on msquic's worker. + auto chain = std::make_shared>(); + QUICStream* raw = &stream; + *chain = [raw, chain, done, ok](int link) { + const char byte = Pattern(static_cast(link)); + raw->SendAsync(&byte, 1, link + 1 == links, [raw, chain, done, ok, link](bool sent) { + if (!sent) *ok = false; + ++*done; + if (link + 1 < links) (*chain)(link + 1); + }); + }; + (*chain)(0); + const bool finished = WaitFor([&] { return done->load() == links; }, std::chrono::seconds(10)); + Check(allParked && finished && ok->load(), + std::format("with all {} pool workers parked, a chain of {} sends, each from the last one's onSent, completed", + workers, done->load())); + release = true; + const bool arrived = WaitFor([&] { return received.Of('B').has_value(); }, std::chrono::seconds(10)); + Check(arrived && FollowsPattern(*received.Of('B'), links), "and the peer read them all, in order"); + *chain = nullptr; + } + + // ── The connection stopped under a burst ───────────────────────────── + { + ClientQUIC doomed("127.0.0.1", listener.Port(), alpn, clientCreds); + QUICStream stream = doomed.OpenStream(); + constexpr std::size_t count = 256; + constexpr std::size_t size = 64 * 1024; + auto completions = std::make_shared(count); + std::vector chunk(size, 'C'); + for (std::size_t i = 0; i < count; ++i) { + stream.SendAsync(chunk.data(), static_cast(size), false, completions->For(i)); + } + doomed.Stop(); + Check(WaitFor([&] { return completions->Fired() == count; }, std::chrono::seconds(20)), + "a connection stopped under a burst reports every send"); + std::lock_guard lock(completions->mutex); + const bool once = std::ranges::all_of(completions->calls, [](int calls) { return calls == 1; }); + const auto firstFalse = std::ranges::find(completions->sent, false); + const bool prefix = std::all_of(firstFalse, completions->sent.end(), [](bool ok) { return !ok; }); + Check(once && prefix && firstFalse != completions->sent.end(), + std::format("each exactly once: {} true, then {} false", + firstFalse - completions->sent.begin(), completions->sent.end() - firstFalse)); + Check(stream.BytesInFlight() == 0, "and nothing is left in flight"); + } + + // ── The wrapper destroyed with sends in flight ─────────────────────── + { + constexpr std::size_t count = 64; + constexpr std::size_t size = 64 * 1024; + auto completions = std::make_shared(count + 1); + { + QUICStream stream = client.OpenStream(); + stream.SendAsync("D", 1, false, completions->For(count)); + std::vector chunk(size); + for (std::size_t i = 0; i < count; ++i) { + for (std::size_t j = 0; j < size; ++j) chunk[j] = Pattern(i * size + j); + stream.SendAsync(chunk.data(), static_cast(size), false, completions->For(i)); + } + // ~QUICStream: Stop(), a FIN after what was sent. + } + Check(WaitFor([&] { return completions->Fired() == count + 1; }, std::chrono::seconds(20)), + "sends made just before the wrapper was destroyed all report"); + { + std::lock_guard lock(completions->mutex); + Check(std::ranges::all_of(completions->sent, [](bool ok) { return ok; }), "and report true"); + } + const bool arrived = WaitFor([&] { return received.Of('D').has_value(); }, std::chrono::seconds(20)); + Check(arrived && FollowsPattern(*received.Of('D'), count * size), "and go out, followed by the FIN"); + } + + // ── Four threads on one stream ─────────────────────────────────────── + { + constexpr int threads = 4; + constexpr int each = 50; + auto completions = std::make_shared(threads * each + 1); + QUICStream stream = client.OpenStream(); + stream.SendAsync("F", 1, false, completions->For(threads * each)); + std::vector senders; + for (int t = 0; t < threads; ++t) { + senders.emplace_back([&, t] { + const std::vector message(1024, static_cast('a' + t)); + for (int i = 0; i < each; ++i) { + stream.SendAsync(message.data(), static_cast(message.size()), false, + completions->For(static_cast(t * each + i))); + } + }); + } + for (std::thread& sender : senders) sender.join(); + stream.SendAsync(nullptr, 0, true, nullptr); + Check(WaitFor([&] { return completions->Fired() == threads * each + 1; }, std::chrono::seconds(20)), + "four threads sending on one stream at once all hear back"); + const bool arrived = WaitFor([&] { return received.Of('F').has_value(); }, std::chrono::seconds(20)); + Check(arrived && received.Of('F')->size() == 1 + threads * each * 1024, "and the peer read every byte"); + } + + // ── A closed stream ────────────────────────────────────────────────── + { + QUICStream closed; + auto inside = std::make_shared>(false); + auto fired = std::make_shared>(0); + auto sentInside = std::make_shared>(false); + auto said = std::make_shared>(true); + *inside = true; + closed.SendAsync("x", 1, false, [inside, fired, sentInside, said](bool ok) { + *sentInside = inside->load(); + *said = ok; + ++*fired; + }); + *inside = false; + Check(WaitFor([&] { return fired->load() == 1; }, std::chrono::seconds(10)) && !said->load() && !sentInside->load(), + "a send on a closed stream reports false, from a pool task rather than inside SendAsync"); + } + } catch (const std::exception& e) { + Check(false, std::format("threw: {}", e.what())); + } + + 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); +}