From bf93f22debb9249f8d9f0d402963575ab454aa32 Mon Sep 17 00:00:00 2001 From: Aoto Date: Fri, 2 Oct 2026 12:55:20 +0200 Subject: [PATCH] let one refused datagram fail without stalling the batch --- README.md | 3 +- docs/batches.md | 53 ++++++++---- docs/testing.md | 2 +- examples/echo.cpp | 10 ++- src/dgram/batch.cppm | 122 +++++++++++++++++++++------- tests/concurrency_tests.cpp | 2 +- tests/demux_tests.cpp | 10 +-- tests/doc_examples.cpp | 5 +- tests/loopback_tests.cpp | 156 ++++++++++++++++++++++++++++++++---- tests/metadata_tests.cpp | 30 +++---- tests/offload_tests.cpp | 8 +- tests/pacing_tests.cpp | 3 +- 12 files changed, 311 insertions(+), 93 deletions(-) diff --git a/README.md b/README.md index 476c265..8868bc6 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,8 @@ for (const auto& d : in->datagrams() | std::views::filter(dgram::is_intact)) { (void)out->stage(piece, d.from()); } } -(void)out->flush(*sock); // discarding the result keeps a failed send staged +(void)out->flush(*sock); // a refused datagram is dropped; the rest still go +out->discard(); // whatever a stalled flush kept still points into `in` ``` `then`, `map`, `recover` and `tap` compose a `result` left to right, in the order diff --git a/docs/batches.md b/docs/batches.md index 14126e8..b24911c 100644 --- a/docs/batches.md +++ b/docs/batches.md @@ -105,14 +105,16 @@ bool stage(std::span payload, const endpoint& to); bool stage(std::span payload, const endpoint& to, const control&); bool stage_copy(std::span payload, const endpoint& to, const control& = {}); // SlotBytes > 0 -result flush(const socket& sock, int flags = 0); +flushed flush(const socket& sock, int flags = 0); +void discard(); bool full() const; std::size_t staged() const; ``` `stage` references the caller's bytes rather than copying, so bouncing a datagram straight out of a receive slot costs nothing. **The referenced bytes must stay -valid until `flush` returns.** It yields `false` when the batch is full. +valid until a flush sends them or `discard()` drops them.** It yields `false` +when the batch is full. `stage_copy` copies into the batch's own slot, for bytes that will not outlive the flush. It requires `SlotBytes > 0` and yields `false` if the payload exceeds @@ -122,24 +124,45 @@ The `control` overload attaches ancillary data to that one datagram; see [metadata](metadata.md). `msg_controllen` is set from what was actually built, zero included, so nothing carries over between datagrams staged in the same slot. -`flush` sends everything staged and clears the batch, returning how many were -taken. A short send is not an error and the remainder is dropped: datagram -delivery is unreliable by definition, so requeueing buys nothing a protocol layer -cannot do better. - -**A failed send leaves the batch staged.** That case is not a short send: nothing -went out at all, so clearing would hand the caller an empty batch and no way to -tell what it lost. The argument for dropping a short send's tail is that the -protocol layer can do better, which is only true while the protocol still has the -bytes. `EAGAIN` on a non-blocking socket is the ordinary way to get here; flush -again once the socket is writable, or call `discard()` to drop it. +`flush` sends everything staged and reports what happened: + +| `flushed` member | Description | +|------------------|-------------| +| `sent` | Datagrams the kernel took. | +| `rejected` | Entries dropped because the kernel refused that datagram. | +| `last_rejection` | Why the last of those was refused. | +| `stalled` | Why the flush stopped early. Set means entries are still staged. | +| `drained()` | `stalled` is empty: every entry was sent or rejected. | + +`sendmmsg` stops at the first datagram it cannot send, so `flush` decides who the +error belongs to. + +- **The datagram's own** (`EMSGSIZE`, `EHOSTUNREACH`, `EPERM` from a firewall, + `EACCES` on broadcast, ...): that entry is dropped and the rest are still sent. + One bad destination cannot hold up everyone else in the batch, and a + per-destination error that repeats cannot wedge it. +- **The socket's** (`EAGAIN`, `ENOBUFS`, `ENOMEM`, `EINTR`, `EBADF`, `ENOTSOCK`, + `EOPNOTSUPP`): the next entry would hit it too, so `flush` stops. Everything + not yet sent stays staged, in order, including the tail of a send that got + partway. Flush again once the socket is writable, or `discard()` it. + +**Entries a stalled flush keeps still reference the caller's bytes.** If those +are receive slots, the next `receive` overwrites them, and a later flush would +send new datagrams' bytes to old datagrams' peers. Flush again before receiving, +or discard: ```cpp -if (const auto sent{tx->flush(*sock)}; !sent && sent.error() == dgram::would_block) { - tx->discard(); // or leave it staged and flush again when writable +const auto sent{tx->flush(*sock)}; +if (sent.stalled == dgram::would_block) { + wait_until_writable(*sock); // then flush again, before the next receive +} else if (sent.stalled) { + tx->discard(); } ``` +Stalled entries keep their slots until they go out, so `full()` can be true with +fewer than `Capacity` staged. + ## Retaining bytes past the batch There is no retention mechanism. A protocol that reassembles or retransmits, as diff --git a/docs/testing.md b/docs/testing.md index 0919fda..131e299 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -18,7 +18,7 @@ checkout. |-------|--------| | `doc_examples.cpp` | Every snippet printed in `README.md` and `docs/`, compiled and run. A snippet in a document is untested code that looks authoritative; this is what stops one rotting. | | `compile_checks.cpp` | Pure `static_assert`. Feature concepts, control-buffer sizing, batch geometry, which resources may be carved from, which metadata is readable, and SipHash against its published vectors, split at every word boundary. Compiling is the test; it is never run. | -| `loopback_tests.cpp` | Batch reuse, truncation reporting (including a caller-supplied `MSG_TRUNC`), nonblocking receive, staging limits, a failed flush leaving the batch staged, the error combinators. | +| `loopback_tests.cpp` | Batch reuse, truncation reporting (including a caller-supplied `MSG_TRUNC`), nonblocking receive, staging limits, the error combinators. Flushing: a refused datagram dropped while the rest still go, socket-level errors stalling with everything kept staged in order, and `send_all` against scripted `sendmmsg` results, since loopback never returns `EAGAIN` on send. | | `metadata_tests.cpp` | Destination address and ECN over loopback, both families, including a feature that was never enabled and an undersized control buffer. DSCP and ECN together in both families, and a per-datagram marking replacing the socket's DSCP. Dual-stack sockets: ECN from and to IPv4 peers, and a reply from the arrival address. | | `malformed_control_tests.cpp` | Control buffers the kernel would never write. Every case here was found by the fuzzer first. | | `offload_tests.cpp` | GRO and GSO over loopback, and the segmentation walk on its own: short tail, exact multiple, stride of zero, stride past the end, empty payload. | diff --git a/examples/echo.cpp b/examples/echo.cpp index ec65a0b..b5387da 100644 --- a/examples/echo.cpp +++ b/examples/echo.cpp @@ -103,8 +103,14 @@ int main(const int argc, const char* const* argv) { } } - if (const auto sent{tx->flush(*sock)}; !sent) { - std::println(stderr, "flush: {}", dgram::describe(sent.error())); + const auto sent{tx->flush(*sock)}; + if (sent.last_rejection) { + std::println(stderr, "flush: {} refused, last: {}", sent.rejected, dgram::describe(*sent.last_rejection)); + } + if (sent.stalled) { + // What is still staged points into the receive slots the next receive overwrites. + std::println(stderr, "flush stalled: {}", dgram::describe(*sent.stalled)); + tx->discard(); } } } diff --git a/src/dgram/batch.cppm b/src/dgram/batch.cppm index f81fd3a..12e2170 100644 --- a/src/dgram/batch.cppm +++ b/src/dgram/batch.cppm @@ -327,13 +327,69 @@ private: * Transmit * ============================================================================ */ +/** @brief What one `flush` did. */ +export struct flushed { + std::size_t sent{}; ///< datagrams the kernel took + std::size_t rejected{}; ///< entries dropped because the kernel refused that datagram + std::optional last_rejection{}; ///< why the last of those was refused + std::optional stalled{}; ///< why it stopped early; every unsent entry is still staged + + /** @brief Nothing is left staged: every entry was sent or rejected. */ + [[nodiscard]] constexpr bool drained() const noexcept { return !stalled.has_value(); } +}; + +namespace detail { + +/** + * @brief Errors that belong to the socket or the moment, not to one datagram. + * + * Retrying the next entry would hit them again, so they stop a flush. Anything + * else is the datagram's own fault and drops only that entry. + */ +inline constexpr std::array socket_errors{would_block, interrupted, out_of_memory, errc{ENOBUFS}, errc{EBADF}, errc{ENOTSOCK}, errc{EOPNOTSUPP}}; + +/** + * @brief Send `pending` front to back, through `send` standing in for one `sendmmsg`. + * + * `sendmmsg` stops at the first datagram it cannot send and, if it sent any + * before it, reports only the count. The next call starts at that datagram and + * gets its error. Exactly `sent + rejected` entries from the front are consumed. + */ +export template + requires std::is_invocable_r_v, Send&, std::span<::mmsghdr>> +[[nodiscard]] flushed send_all(std::span<::mmsghdr> pending, Send&& send) noexcept { + flushed out{}; + while (!pending.empty()) { + const auto n{std::invoke(send, pending)}; + if (n.has_value() && *n > 0) { + const auto taken{std::min(static_cast(*n), pending.size())}; + out.sent += taken; + pending = pending.subspan(taken); + continue; + } + // A zero count, which a non-empty vector never gets, stalls rather than spins. + const auto e{n.error_or(would_block)}; + if (std::ranges::contains(socket_errors, e)) { + out.stalled = e; + return out; + } + ++out.rejected; + out.last_rejection = e; + pending = pending.subspan(1); + } + return out; +} + +} // namespace detail + /** * @brief Fixed-size `sendmmsg` argument block over caller-provided memory. * * `stage` references the caller's bytes instead of copying them, so echoing * straight out of a `receive_batch` slot costs nothing. The referenced bytes - * must stay put until `flush` returns. `SlotBytes` of zero carves no payload - * memory at all and leaves only the referencing path available. + * must stay put until a flush sends them or `discard()` drops them, which a + * stalled flush postpones. `SlotBytes` of zero carves no payload memory at all + * and leaves only the referencing path available. */ export template class transmit_batch { static_assert(Capacity > 0 && Capacity <= IOV_MAX, "capacity must fit IOV_MAX"); @@ -368,9 +424,16 @@ public: return b; } - /** @brief Whether another datagram fits before a flush is required. */ - [[nodiscard]] constexpr bool full() const noexcept { return staged_ == Capacity; } - [[nodiscard]] constexpr std::size_t staged() const noexcept { return staged_; } + /** + * @brief Whether another datagram fits before a flush is required. + * + * Entries a stalled flush left behind keep their slots, so this can be + * true with fewer than `Capacity` staged. + */ + [[nodiscard]] constexpr bool full() const noexcept { return end_ == Capacity; } + + /** @brief Entries waiting to be sent. */ + [[nodiscard]] constexpr std::size_t staged() const noexcept { return end_ - head_; } /** * @brief Queue `payload` for `to`, referencing the bytes in place. @@ -396,7 +459,7 @@ public: if (full()) [[unlikely]] { return false; } - const auto i{staged_++}; + const auto i{end_++}; addrs_[i] = to; // sendmmsg only reads through iov_base, but the field is not const-qualified. iovs_[i].iov_base = const_cast(payload.data()); @@ -417,38 +480,38 @@ public: if (full() || payload.size() > SlotBytes) [[unlikely]] { return false; } - const auto dest{payload_.subspan(staged_ * SlotBytes, payload.size())}; + const auto dest{payload_.subspan(end_ * SlotBytes, payload.size())}; std::ranges::copy(payload, dest.begin()); return stage(dest, to, ancillary); } /** - * @brief Send everything staged, then clear the batch. - * - * A short send is not an error: `sendmmsg` reports how many it took and the - * remainder is dropped. Datagram delivery is unreliable by definition, so - * requeueing the tail buys nothing a protocol layer cannot do better. + * @brief Send everything staged, dropping only what the kernel refuses. * - * **A failed send is different and the batch is left staged.** Nothing went - * out, so clearing would hand the caller an empty batch and no way to know - * what it lost, and "a protocol layer can do better" is only true while the - * protocol still has the bytes. `EAGAIN` on a non-blocking socket is the - * common case: flush again once it is writable, or `discard()` to drop it. + * An error that belongs to one datagram (`EMSGSIZE`, `EHOSTUNREACH`, + * `EPERM`, ...) drops that entry, counts it in `rejected`, and the rest are + * still sent, so one bad destination cannot hold up the batch. An error that + * belongs to the socket (`EAGAIN`, `ENOBUFS`, `EINTR`, `EBADF`, ...) stops + * the flush and sets `stalled`. Everything not yet sent stays staged and + * still references the caller's bytes: flush again once the socket is + * writable, or `discard()` it before those bytes change. */ - [[nodiscard]] result flush(const socket& sock, const int flags = 0) noexcept { - if (staged_ == 0) { - return std::size_t{0}; - } - const int n{::sendmmsg(sock.native(), msgs_.data(), static_cast(staged_), flags)}; - if (n < 0) [[unlikely]] { - return fail(); + [[nodiscard]] flushed flush(const socket& sock, const int flags = 0) noexcept { + const auto out{detail::send_all(msgs_.subspan(head_, end_ - head_), [&sock, flags](const std::span<::mmsghdr> pending) noexcept { + return invoke_syscall(::sendmmsg, sock.native(), pending.data(), static_cast(pending.size()), flags); + })}; + head_ += out.sent + out.rejected; + if (head_ == end_) { + discard(); } - staged_ = 0; - return static_cast(n); + return out; } - /** @brief Drop everything staged without sending it, after a failed `flush`. */ - void discard() noexcept { staged_ = 0; } + /** @brief Drop everything staged without sending it. */ + void discard() noexcept { + head_ = 0; + end_ = 0; + } private: transmit_batch() noexcept = default; @@ -481,7 +544,8 @@ private: std::span addrs_{}; std::span control_{}; std::span payload_{}; - std::size_t staged_{}; + std::size_t head_{}; ///< first entry not yet sent; a stalled flush leaves it past zero + std::size_t end_{}; ///< one past the last staged entry, where `stage` writes next }; } // namespace dgram diff --git a/tests/concurrency_tests.cpp b/tests/concurrency_tests.cpp index e570f1f..b5459ce 100644 --- a/tests/concurrency_tests.cpp +++ b/tests/concurrency_tests.cpp @@ -71,7 +71,7 @@ TEST(Concurrency, WorkersShareNothing) { if (!tx->stage_copy(bytes_of(text), target, marked)) { continue; } - ASSERT_TRUE(tx->flush(*peer).has_value()); + ASSERT_TRUE(tx->flush(*peer).drained()); // Nonblocking, so poll briefly rather than parking a thread. for (int attempt{}; attempt < 2000; ++attempt) { diff --git a/tests/demux_tests.cpp b/tests/demux_tests.cpp index 92bdfa8..1f0deb8 100644 --- a/tests/demux_tests.cpp +++ b/tests/demux_tests.cpp @@ -382,7 +382,7 @@ TEST(Route, DeliversToTheMatchingFlowAndReportsTheRest) { const auto say = [&](dgram::socket& s, std::string_view text) { ASSERT_TRUE(tx->stage_copy({reinterpret_cast(text.data()), text.size()}, net.target)); - ASSERT_TRUE(tx->flush(s).has_value()); + ASSERT_TRUE(tx->flush(s).drained()); }; say(*known, "from-known"); say(*stranger, "from-stranger"); @@ -443,7 +443,7 @@ TEST(Route, CoalescedSlotRoutesPerDatagramNotPerSlot) { dgram::control ancillary{}; ancillary.set(mtu); ASSERT_TRUE(tx->stage(payload, net.target, ancillary)); - ASSERT_TRUE(tx->flush(*sender).has_value()); + ASSERT_TRUE(tx->flush(*sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u) << "GRO should coalesce into one slot"; { @@ -475,7 +475,7 @@ TEST(Route, TruncatedDatagramsAreCountedNotDelivered) { constexpr std::string_view text{"far longer than eight bytes"}; ASSERT_TRUE(tx->stage_copy({reinterpret_cast(text.data()), text.size()}, net.target)); - ASSERT_TRUE(tx->flush(*sender).has_value()); + ASSERT_TRUE(tx->flush(*sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto counts{dgram::route(*rx, dgram::by_peer{}, *table, 0ns)}; @@ -507,7 +507,7 @@ struct routing_fixture { template dgram::routed exchange(rx_batch& rx, tx_batch& tx, dgram::socket& s, std::string_view text, Table& table, Unmatched&& unmatched, Context&&... context) { EXPECT_TRUE(tx.stage_copy({reinterpret_cast(text.data()), text.size()}, net.target)); - EXPECT_TRUE(tx.flush(s).has_value()); + EXPECT_TRUE(tx.flush(s).drained()); for (int attempt{}; attempt < 100; ++attempt) { if (const auto got{rx.receive(net.receiver)}; got && *got > 0) { @@ -674,7 +674,7 @@ TEST(Route, ReportsHowManySlotsTheDatagramsCameIn) { dgram::control ancillary{}; (void)ancillary.set(stride); ASSERT_TRUE(tx->stage_copy(payload, net.target, ancillary)); - ASSERT_TRUE(tx->flush(*sender).has_value()); + ASSERT_TRUE(tx->flush(*sender).drained()); dgram::routed counts{}; for (int attempt{}; attempt < 100 && counts.handled() < 4; ++attempt) { diff --git a/tests/doc_examples.cpp b/tests/doc_examples.cpp index b06f583..0397972 100644 --- a/tests/doc_examples.cpp +++ b/tests/doc_examples.cpp @@ -96,7 +96,10 @@ TEST(DocExamples, AtAGlance) { } } } - ASSERT_TRUE(out->flush(*sock).has_value()); + const auto sent{out->flush(*sock)}; + out->discard(); + EXPECT_TRUE(sent.drained()); + EXPECT_EQ(sent.sent, 1u); EXPECT_EQ(echoed, 1u); } diff --git a/tests/loopback_tests.cpp b/tests/loopback_tests.cpp index d70a1b6..42559db 100644 --- a/tests/loopback_tests.cpp +++ b/tests/loopback_tests.cpp @@ -2,6 +2,7 @@ // SPDX-License-Identifier: MIT #include +#include #include import std; @@ -99,8 +100,8 @@ TEST(Loopback, DeliversAWholeBatch) { ASSERT_TRUE(tx->stage_copy(bytes_of(p), net.target)); } const auto sent{tx->flush(net.sender)}; - ASSERT_TRUE(sent.has_value()); - EXPECT_EQ(*sent, sent_count); + ASSERT_TRUE(sent.drained()); + EXPECT_EQ(sent.sent, sent_count); const auto got{rx->receive(net.receiver)}; ASSERT_TRUE(got.has_value()); @@ -125,7 +126,7 @@ TEST(Loopback, SurvivesRepeatedReceivesOnTheSameBatch) { for (int round{}; round < 4; ++round) { const auto text{std::format("round-{}", round)}; ASSERT_TRUE(tx->stage_copy(bytes_of(text), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); const auto got{rx->receive(net.receiver)}; ASSERT_TRUE(got.has_value()); @@ -150,7 +151,7 @@ TEST(Loopback, ReportsTruncationRatherThanCorrupting) { ASSERT_TRUE(tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("far longer than eight bytes"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); const auto got{rx->receive(net.receiver)}; ASSERT_TRUE(got.has_value()); @@ -174,7 +175,7 @@ TEST(Loopback, CallerMsgTruncDoesNotStretchThePayloadPastTheSlot) { ASSERT_TRUE(tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("far longer than eight bytes"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver, MSG_TRUNC), 1u); const auto d{*rx->datagrams().begin()}; @@ -205,7 +206,7 @@ TEST(Loopback, NonblockingReceiveReportsWouldBlockThenRecovers) { constexpr std::string_view text{"after the empty poll"}; ASSERT_TRUE(tx->stage_copy(bytes_of(text), target)); - ASSERT_TRUE(tx->flush(sender).has_value()); + ASSERT_TRUE(tx->flush(sender).drained()); // Poll until the loopback datagram lands; the point is what arrives, not when. dgram::result got{std::unexpected{dgram::would_block}}; @@ -264,29 +265,150 @@ TEST(Error, PipeCombinatorsChainLeftToRight) { EXPECT_EQ(*recovered, 7); } -/* A failed send is not a short send: nothing left, so the caller must still be - holding everything it staged. Clearing here would lose datagrams silently for - any caller that stages by reference and then drops its own copy. */ -TEST(TransmitBatch, AFailedFlushLeavesTheBatchStaged) { +/* ============================================================================ + * Flushing: who an error belongs to + * ============================================================================ */ + +/** Stands in for sendmmsg: replays scripted results and records each call's vector length. */ +struct scripted_send { + std::vector> script; + std::vector lengths{}; + + dgram::result operator()(const std::span<::mmsghdr> pending) { + lengths.push_back(pending.size()); + const auto next{script.front()}; + script.erase(script.begin()); + return next; + } +}; + +/* sendmmsg reports a datagram's error only on the call that starts at it. */ +TEST(SendAll, ARejectedDatagramIsSkippedAndTheRestStillGo) { + std::array<::mmsghdr, 6> msgs{}; + scripted_send send{{2, std::unexpected{dgram::message_too_long}, 3}}; + + const auto out{dgram::detail::send_all(msgs, send)}; + EXPECT_EQ(out.sent, 5u); + EXPECT_EQ(out.rejected, 1u); + EXPECT_EQ(out.last_rejection, dgram::message_too_long); + EXPECT_TRUE(out.drained()); + EXPECT_EQ(send.lengths, (std::vector{6, 4, 3})) << "each call starts where the last one stopped"; +} + +TEST(SendAll, ASocketErrorStopsAfterWhatWasAlreadySent) { + std::array<::mmsghdr, 6> msgs{}; + scripted_send send{{2, std::unexpected{dgram::errc{EHOSTUNREACH}}, std::unexpected{dgram::would_block}}}; + + const auto out{dgram::detail::send_all(msgs, send)}; + EXPECT_EQ(out.sent, 2u); + EXPECT_EQ(out.rejected, 1u); + EXPECT_EQ(out.stalled, dgram::would_block); + EXPECT_FALSE(out.drained()); + EXPECT_EQ(send.lengths, (std::vector{6, 4, 3})) << "three entries consumed, three left"; +} + +TEST(SendAll, EverySocketLevelErrorStalls) { + for (const auto e : {dgram::would_block, dgram::interrupted, dgram::out_of_memory, dgram::errc{ENOBUFS}, dgram::errc{EBADF}, dgram::errc{ENOTSOCK}, + dgram::errc{EOPNOTSUPP}}) { + std::array<::mmsghdr, 2> msgs{}; + scripted_send send{{std::unexpected{e}}}; + const auto out{dgram::detail::send_all(msgs, send)}; + EXPECT_EQ(out.stalled, e) << dgram::describe(e); + EXPECT_EQ(out.sent + out.rejected, 0u); + } +} + +TEST(SendAll, AZeroCountStallsRatherThanSpins) { + std::array<::mmsghdr, 2> msgs{}; + scripted_send send{{0}}; + const auto out{dgram::detail::send_all(msgs, send)}; + EXPECT_EQ(out.stalled, dgram::would_block); + EXPECT_EQ(send.lengths.size(), 1u); +} + +/* The bug this guards: an error on the first entry used to leave the whole batch + staged, so one destination the kernel always refuses stopped every reply. */ +TEST(TransmitBatch, ARefusedDatagramIsDroppedAndTheRestStillGo) { + loopback net{}; + auto rx{rx_batch::carve(net.arena)}; + auto tx{tx_batch::carve(net.arena)}; + ASSERT_TRUE(rx.has_value() && tx.has_value()); + + const std::vector oversized(70'000); // past the 65507-byte UDP limit: EMSGSIZE + ASSERT_TRUE(tx->stage(oversized, net.target)); + ASSERT_TRUE(tx->stage_copy(bytes_of("first"), net.target)); + ASSERT_TRUE(tx->stage(oversized, net.target)); + ASSERT_TRUE(tx->stage_copy(bytes_of("second"), net.target)); + + const auto sent{tx->flush(net.sender)}; + EXPECT_EQ(sent.sent, 2u); + EXPECT_EQ(sent.rejected, 2u); + EXPECT_EQ(sent.last_rejection, dgram::message_too_long); + EXPECT_TRUE(sent.drained()); + EXPECT_EQ(tx->staged(), 0u); + + ASSERT_EQ(*rx->receive(net.receiver), 2u); + const auto texts{rx->datagrams() | std::views::transform([](const auto& d) { + return std::string{reinterpret_cast(d.payload().data()), d.payload().size()}; + }) | std::ranges::to()}; + EXPECT_EQ(texts, (std::vector{"first", "second"})); +} + +TEST(TransmitBatch, ADestinationTheSocketCannotReachIsRejectedNotKept) { libmem::arena arena{dgram::transmit_batch<4, 64>::footprint()}; auto tx{dgram::transmit_batch<4, 64>::carve(arena)}; ASSERT_TRUE(tx.has_value()); - - // A socket that cannot send: never bound, and pointed at nothing reachable. auto sender{dgram::socket::open(dgram::family::inet4)}; ASSERT_TRUE(sender.has_value()); - const auto unreachable{*dgram::endpoint::parse(dgram::family::inet6, "::1", 9)}; const std::array payload{}; - ASSERT_TRUE(tx->stage_copy(payload, unreachable)); - ASSERT_EQ(tx->staged(), 1u); + ASSERT_TRUE(tx->stage_copy(payload, *dgram::endpoint::parse(dgram::family::inet6, "::1", 9))); const auto sent{tx->flush(*sender)}; - ASSERT_FALSE(sent.has_value()) << "a v6 destination on a v4 socket cannot be sent"; - EXPECT_EQ(tx->staged(), 1u) << "the caller has to be able to see what did not go"; + EXPECT_EQ(sent.rejected, 1u) << "a v6 destination on a v4 socket is that datagram's problem"; + EXPECT_TRUE(sent.last_rejection.has_value()); + EXPECT_EQ(tx->staged(), 0u); +} + +/* A socket-level error keeps everything staged, in order, and the batch keeps + accepting entries behind it. MSG_OOB is refused for UDP before any datagram. */ +TEST(TransmitBatch, AStalledFlushKeepsTheBatchStagedForTheNextOne) { + loopback net{}; + auto rx{rx_batch::carve(net.arena)}; + auto tx{tx_batch::carve(net.arena)}; + ASSERT_TRUE(rx.has_value() && tx.has_value()); + + ASSERT_TRUE(tx->stage_copy(bytes_of("one"), net.target)); + ASSERT_TRUE(tx->stage_copy(bytes_of("two"), net.target)); + + const auto stalled{tx->flush(net.sender, MSG_OOB)}; + EXPECT_EQ(stalled.stalled, dgram::errc{EOPNOTSUPP}); + EXPECT_EQ(stalled.sent + stalled.rejected, 0u); + EXPECT_EQ(tx->staged(), 2u) << "the caller has to be able to see what did not go"; + + ASSERT_TRUE(tx->stage_copy(bytes_of("three"), net.target)); + const auto sent{tx->flush(net.sender)}; + EXPECT_TRUE(sent.drained()); + EXPECT_EQ(sent.sent, 3u); + EXPECT_EQ(tx->staged(), 0u); + + ASSERT_EQ(*rx->receive(net.receiver), 3u); + const auto texts{rx->datagrams() | std::views::transform([](const auto& d) { + return std::string{reinterpret_cast(d.payload().data()), d.payload().size()}; + }) | std::ranges::to()}; + EXPECT_EQ(texts, (std::vector{"one", "two", "three"})); +} + +TEST(TransmitBatch, DiscardDropsAStalledBatch) { + loopback net{}; + auto tx{tx_batch::carve(net.arena)}; + ASSERT_TRUE(tx.has_value()); + ASSERT_TRUE(tx->stage_copy(bytes_of("dropped"), net.target)); + ASSERT_FALSE(tx->flush(net.sender, MSG_OOB).drained()); tx->discard(); EXPECT_EQ(tx->staged(), 0u); + EXPECT_FALSE(tx->full()); } } // namespace diff --git a/tests/metadata_tests.cpp b/tests/metadata_tests.cpp index 862e08a..9517887 100644 --- a/tests/metadata_tests.cpp +++ b/tests/metadata_tests.cpp @@ -58,7 +58,7 @@ TEST(Metadata, ReportsDestinationAddressOverIPv4) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("where did i land"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -79,7 +79,7 @@ TEST(Metadata, ReportsDestinationAddressOverIPv6) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("v6 please"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -103,7 +103,7 @@ TEST_P(EcnRoundTrip, MarkedCodepointSurvives) { dgram::control ancillary{}; ancillary.set({.ecn = marked}); ASSERT_TRUE(tx->stage_copy(bytes_of("marked"), net.target, ancillary)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -124,7 +124,7 @@ TEST(Metadata, UnmarkedDatagramReadsAsNotEct) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("plain"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -177,7 +177,7 @@ TEST_P(TrafficClass, DscpAndEcnArriveTogether) { dgram::control ancillary{}; ancillary.set(sent); ASSERT_TRUE(tx->stage_copy(bytes_of("marked"), net.target, ancillary)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto meta{(*rx->datagrams().begin()).meta()}; @@ -198,14 +198,14 @@ TEST_P(TrafficClass, PerDatagramMarkingReplacesTheSocketDscp) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("socket default"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); EXPECT_EQ((*rx->datagrams().begin()).meta().get()->dscp, dgram::dscp::ef); dgram::control ancillary{}; ancillary.set({.ecn = dgram::ecn_codepoint::ect0}); ASSERT_TRUE(tx->stage_copy(bytes_of("own marking"), net.target, ancillary)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto got{*(*rx->datagrams().begin()).meta().get()}; EXPECT_EQ(got.dscp, dgram::dscp::df) << "the datagram said df, and that wins over the socket"; @@ -248,7 +248,7 @@ TEST(DualStack, ReportsEcnForIPv4Arrivals) { dgram::control ancillary{}; ancillary.set({.ecn = marked}); ASSERT_TRUE(tx->stage_copy(bytes_of("from v4"), target, ancillary)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(receiver), 1u); const auto d{*rx->datagrams().begin()}; @@ -277,7 +277,7 @@ TEST(DualStack, MarksEcnOnDatagramsToV4MappedPeers) { ancillary.set({.ecn = marked}); ASSERT_TRUE(tx->stage_copy(bytes_of("to v4-mapped"), target, ancillary)); const auto sent{tx->flush(sender)}; - ASSERT_TRUE(sent.has_value()) << dgram::describe(sent.error()); + ASSERT_EQ(sent.sent, 1u) << sent.last_rejection.transform(dgram::describe).value_or("stalled"); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto& got{(*rx->datagrams().begin()).meta().get()}; @@ -306,7 +306,7 @@ TEST(DualStack, RepliesToAnIPv4PeerFromTheArrivalAddress) { dgram::control request{}; request.set({.ecn = dgram::ecn_codepoint::ect1}); ASSERT_TRUE(client_tx->stage_copy(bytes_of("ping"), server_target, request)); - ASSERT_TRUE(client_tx->flush(client.receiver).has_value()); + ASSERT_TRUE(client_tx->flush(client.receiver).drained()); ASSERT_EQ(*server_rx->receive(server), 1u); const auto arrival{*server_rx->datagrams().begin()}; @@ -318,7 +318,7 @@ TEST(DualStack, RepliesToAnIPv4PeerFromTheArrivalAddress) { reply.set({.ecn = *meta.get()}); ASSERT_TRUE(server_tx->stage_copy(bytes_of("pong"), arrival.from(), reply)); const auto sent{server_tx->flush(server)}; - ASSERT_TRUE(sent.has_value()) << dgram::describe(sent.error()); + ASSERT_EQ(sent.sent, 1u) << sent.last_rejection.transform(dgram::describe).value_or("stalled"); ASSERT_EQ(*client_rx->receive(client.receiver), 1u); const auto back{*client_rx->datagrams().begin()}; @@ -352,7 +352,7 @@ TEST(Metadata, FeatureNeverEnabledReadsAsAbsent) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("only ecn"), target)); - ASSERT_TRUE(tx->flush(sender).has_value()); + ASSERT_TRUE(tx->flush(sender).drained()); ASSERT_EQ(*rx->receive(receiver), 1u); const auto view{rx->datagrams()}; @@ -379,7 +379,7 @@ TEST(Metadata, UndersizedControlBufferTruncatesWithoutOverreading) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("too much metadata"), target)); - ASSERT_TRUE(tx->flush(sender).has_value()); + ASSERT_TRUE(tx->flush(sender).drained()); ASSERT_EQ(*rx->receive(receiver), 1u); const auto view{rx->datagrams()}; @@ -408,7 +408,7 @@ TEST(Metadata, StaleControlLengthDoesNotLeakBetweenDatagrams) { marked.set({.ecn = dgram::ecn_codepoint::ce}); ASSERT_TRUE(tx->stage_copy(bytes_of("marked"), net.target, marked)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); { const auto view{rx->datagrams()}; @@ -417,7 +417,7 @@ TEST(Metadata, StaleControlLengthDoesNotLeakBetweenDatagrams) { // Same slot, now with nothing attached. ASSERT_TRUE(tx->stage_copy(bytes_of("plain"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); { const auto view{rx->datagrams()}; diff --git a/tests/offload_tests.cpp b/tests/offload_tests.cpp index 5581f02..0f1d439 100644 --- a/tests/offload_tests.cpp +++ b/tests/offload_tests.cpp @@ -127,7 +127,7 @@ TEST(Offload, GsoSendCoalescesIntoOneGroSlot) { ancillary.set(mtu); ASSERT_TRUE(tx->stage(payload, net.target, ancillary)); const auto sent{tx->flush(net.sender)}; - ASSERT_TRUE(sent.has_value()) << dgram::describe(sent.error()); + ASSERT_EQ(sent.sent, 1u) << sent.last_rejection.transform(dgram::describe).value_or("stalled"); ASSERT_EQ(*rx->receive(net.receiver), 1u) << "GRO should deliver the lot in one slot"; @@ -165,7 +165,7 @@ TEST(Offload, UncoalescedDatagramIsOneSegment) { ASSERT_TRUE(rx.has_value() && tx.has_value()); ASSERT_TRUE(tx->stage_copy(bytes_of("a single small datagram"), net.target)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -190,7 +190,7 @@ TEST(Offload, SegmentAndEcnTravelTogether) { ancillary.set({.ecn = dgram::ecn_codepoint::ect0}); ASSERT_TRUE(tx->stage(payload, net.target, ancillary)); - ASSERT_TRUE(tx->flush(net.sender).has_value()); + ASSERT_TRUE(tx->flush(net.sender).drained()); ASSERT_EQ(*rx->receive(net.receiver), 1u); const auto view{rx->datagrams()}; @@ -218,7 +218,7 @@ TEST(Offload, UndersizedSlotReportsTruncation) { dgram::control ancillary{}; ancillary.set(mtu); ASSERT_TRUE(tx->stage(payload, target, ancillary)); - ASSERT_TRUE(tx->flush(sender).has_value()); + ASSERT_TRUE(tx->flush(sender).drained()); ASSERT_EQ(*rx->receive(receiver), 1u); const auto view{rx->datagrams()}; diff --git a/tests/pacing_tests.cpp b/tests/pacing_tests.cpp index fb75eca..6dc3ce4 100644 --- a/tests/pacing_tests.cpp +++ b/tests/pacing_tests.cpp @@ -196,8 +196,7 @@ TEST(TransmitTime, PacedDatagramStillArrives) { ASSERT_TRUE(tx->stage_copy(bytes_of("paced"), target, ancillary)); const auto sent{tx->flush(*sender)}; - ASSERT_TRUE(sent.has_value()) << dgram::describe(sent.error()); - EXPECT_EQ(*sent, 1u); + EXPECT_EQ(sent.sent, 1u) << sent.last_rejection.transform(dgram::describe).value_or("stalled"); // Without fq on the interface the kernel ignores the departure time and // sends immediately, so this arrives either way. What is asserted is that