From cd6b4e89604ae31ff61f617cdb74bf289128ec33 Mon Sep 17 00:00:00 2001 From: Philippe Leduc Date: Mon, 28 Sep 2026 10:27:23 +0200 Subject: [PATCH] Improve wake UDP API by deriving the name with the node --- ARCHITECTURE.md | 16 +++---- examples/hello_wait.cc | 12 ++--- include/kickmsg/Node.h | 13 +++--- include/kickmsg/Waker.h | 31 +++++++++--- src/Node.cc | 61 ++++++++++++------------ src/Waker.cc | 56 +++++++++++++++++++--- src/os/posix/WakeBackends.cc | 3 +- tests/unit/node-t.cc | 16 +++++++ tests/unit/wait_fd-t.cc | 91 +++++++++++++++++++++++++++++++++++- 9 files changed, 232 insertions(+), 67 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index deb69f7..1c1a8ec 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -455,14 +455,14 @@ is that a descriptor **composes** -- with a socket, a timer, another channel -- which a futex word cannot. On a single channel with nothing else to watch, `receive()` is both simpler and faster. -Give each channel its own instance. The **port** is what separates them: -a socket bound to `INADDR_ANY` receives every datagram arriving on its -port, including groups it never joined, because the membership decides -whether the *host* accepts the packet rather than which socket gets it. -The **group** is what fans one channel's wake out to all its subscribers. -Constructed from a channel name, both ends derive the same group and port -without coordinating; a constructor taking an explicit group and port is -there for a caller that would rather pin them. +Give each channel its own instance. POSIX receivers bind to the multicast +group, so unrelated traffic on the same port does not wake them. Windows +receivers bind to `INADDR_ANY` and may wake spuriously on a port collision. + +Both ends derive the group and port from the same name, or from the channel +identity stamped by `Node` via `for_topic`, `for_broadcast`, or `for_mailbox`. +That identity includes the channel kind and namespace; mailboxes also include +the owner. Different identities can still hash to the same UDP address. The park protocol is the same Dekker pair as the futex path, with one extra constraint. `arm_wait()` samples `write_pos` **before** deciding the diff --git a/examples/hello_wait.cc b/examples/hello_wait.cc index c44af6e..fc5d2c1 100644 --- a/examples/hello_wait.cc +++ b/examples/hello_wait.cc @@ -5,8 +5,7 @@ /// ends a wake backend instead and the subscriber hands out a descriptor, so one loop can /// serve it alongside a socket, a timer, or anything else the caller already polls. /// -/// Single-process for simplicity; in production the publisher and the subscriber are two -/// processes, each building its own backend from the same name. +/// Single-process for simplicity; each process normally builds its own backend. #include #include @@ -28,12 +27,11 @@ int main() { kickmsg::SharedMemory::unlink("/demo_wait_temperature"); - // Both ends are given the same backend. Nothing about it is negotiated through shared - // memory: each side derives the same address from the name. - UdpMulticastBackend backend("demo/temperature"); - Node sensor("sensor", "demo_wait"); - auto pub = sensor.advertise("temperature", {}, &backend); + + // Each node in the namespace derives the same wake address for this topic. + auto backend = UdpMulticastBackend::for_topic(sensor, "temperature"); + auto pub = sensor.advertise("temperature", {}, &backend); Node display("display", "demo_wait"); auto sub = display.subscribe("temperature"); diff --git a/include/kickmsg/Node.h b/include/kickmsg/Node.h index 623d261..3d58e71 100644 --- a/include/kickmsg/Node.h +++ b/include/kickmsg/Node.h @@ -157,17 +157,18 @@ namespace kickmsg std::string const& name() const { return name_; } std::string const& kmsg_namespace() const { return namespace_; } + /// Identity a channel's region is stamped with. Hashes raw channel names, + /// but sanitizes mailbox owners like Node names. The same hashes detect + /// shm-name collisions and select wake addresses. + uint64_t topic_identity(char const* topic) const; + uint64_t broadcast_identity(char const* channel) const; + uint64_t mailbox_identity(char const* owner, char const* tag) const; + private: std::string make_topic_name(char const* topic) const; std::string make_broadcast_name(char const* channel) const; std::string make_mailbox_name(char const* owner, char const* tag) const; - // Identity hashes over the raw (pre-sanitization) coordinates plus - // a kind tag; shm-name collisions are then detected at open. - uint64_t make_topic_identity(char const* topic) const; - uint64_t make_broadcast_identity(char const* channel) const; - uint64_t make_mailbox_identity(char const* owner, char const* tag) const; - // unordered_map guarantees reference stability for elements // (only iterators are invalidated on rehash), so pointers // returned here remain valid across subsequent insertions. diff --git a/include/kickmsg/Waker.h b/include/kickmsg/Waker.h index 46163a1..362e30e 100644 --- a/include/kickmsg/Waker.h +++ b/include/kickmsg/Waker.h @@ -8,6 +8,8 @@ namespace kickmsg { + class Node; + /// A pollable cross-process wake: the subscriber gets a descriptor for its own /// poll/epoll/kqueue loop, and the publisher makes it readable. /// @@ -55,10 +57,8 @@ namespace kickmsg /// datagram to every joined socket. Use it when the descriptor has to sit in an event /// loop next to other sources. For a single channel, receive() is faster. /// - /// Give each channel its own instance. The port is what keeps channels apart: a - /// socket bound to INADDR_ANY gets every datagram on its port, even for groups it - /// never joined, because the membership only decides whether the host accepts the - /// packet. The group is what delivers one wake to every subscriber. + /// Give each channel its own instance. POSIX receivers filter by group; Windows + /// receivers may wake spuriously when channels share a port. /// /// This is a hint that something arrived, not an authenticated channel. Any local /// process can join the group and send to it, which keeps descriptors readable and @@ -72,11 +72,17 @@ namespace kickmsg static constexpr uint16_t DEFAULT_PORT_BASE = 27182; static constexpr uint16_t PORT_SPAN = 512; - /// Derives a group in 239.255.0.0/16 and a port in [port_base, +PORT_SPAN) from - /// `name`, so both ends agree without coordinating. Two names colliding onto one - /// port wake each other spuriously but stay correct. + /// Maps `name` to a group in 239.255.0.0/16 and a port in + /// [port_base, port_base + PORT_SPAN). Collisions cause spurious wakes. UdpMulticastBackend(char const* name, uint16_t port_base = DEFAULT_PORT_BASE); + /// Derive the address from the Node's channel identity. Pass the same raw name + /// used for the channel: "/fault" and "fault" differ. + static UdpMulticastBackend for_topic(Node const& node, char const* topic, uint16_t port_base = DEFAULT_PORT_BASE); + static UdpMulticastBackend for_broadcast(Node const& node, char const* channel, uint16_t port_base = DEFAULT_PORT_BASE); + /// `owner_node` defaults to `node`, the side that created the mailbox. + static UdpMulticastBackend for_mailbox(Node const& node, char const* tag, char const* owner_node = nullptr, uint16_t port_base = DEFAULT_PORT_BASE); + /// Exact group and port, for a caller that would rather pin them than derive them. UdpMulticastBackend(uint32_t group, uint16_t port); @@ -94,6 +100,17 @@ namespace kickmsg uint16_t port() const { return port_; } private: + struct Address + { + uint32_t group; + uint16_t port; + }; + + /// Throws std::invalid_argument when port_base leaves no room for PORT_SPAN. + static Address derive(uint64_t hash, uint16_t port_base); + + UdpMulticastBackend(Address address); + /// Throws std::runtime_error when the host cannot carry multicast. void open_sender(); diff --git a/src/Node.cc b/src/Node.cc index 8e75ec4..578a97b 100644 --- a/src/Node.cc +++ b/src/Node.cc @@ -54,7 +54,7 @@ namespace kickmsg std::string mailbox_topic(char const* owner, char const* tag) { std::string out = "/"; - out += owner; + out += sanitize_shm_component(owner, "mailbox owner"); out += '/'; out += tag; return out; @@ -62,10 +62,10 @@ namespace kickmsg /// Chain one identity component: bytes, then length as a separator /// so ("ab","c") and ("a","bc") never hash alike. - uint64_t identity_chain(std::string_view s, uint64_t h) + uint64_t identity_chain(std::string_view component, uint64_t identity) { - h = hash::fnv1a_64(s, h); - return hash::fnv1a_64(s.size(), h); + identity = hash::fnv1a_64(component, identity); + return hash::fnv1a_64(component.size(), identity); } /// Cache hits must validate identity and config because sanitized names can collide. @@ -179,14 +179,14 @@ namespace kickmsg // Reuse cached regions; create() would replace the shared object. if (auto* r = find_region(shm_name)) { - check_cached_identity(*r, shm_name, make_topic_identity(topic)); + check_cached_identity(*r, shm_name, topic_identity(topic)); check_cached_config(*r, shm_name, channel::PubSub, cfg); touch_registry(shm_name, topic_path, channel::PubSub, registry::Pubsub, registry::Publisher); return Publisher(*r, backend); } channel::Config stamped_cfg = cfg; - stamped_cfg.identity = make_topic_identity(topic); + stamped_cfg.identity = topic_identity(topic); auto [it, _] = regions_.emplace( shm_name, SharedRegion::create(shm_name.c_str(), channel::PubSub, stamped_cfg, name_.c_str())); @@ -201,14 +201,14 @@ namespace kickmsg auto topic_path = with_leading_slash(topic); if (auto* r = find_region(shm_name)) { - check_cached_identity(*r, shm_name, make_topic_identity(topic)); + check_cached_identity(*r, shm_name, topic_identity(topic)); touch_registry(shm_name, topic_path, channel::PubSub, registry::Pubsub, registry::Subscriber); return Subscriber(*r); } auto [it, _] = regions_.emplace( shm_name, - SharedRegion::open(shm_name.c_str(), make_topic_identity(topic))); + SharedRegion::open(shm_name.c_str(), topic_identity(topic))); touch_registry(shm_name, topic_path, channel::PubSub, registry::Pubsub, registry::Subscriber); return Subscriber(it->second); @@ -242,7 +242,7 @@ namespace kickmsg WakeBackend* backend) { channel::Config stamped_cfg = cfg; - stamped_cfg.identity = make_topic_identity(topic); + stamped_cfg.identity = topic_identity(topic); return create_or_open_handle( make_topic_name(topic), with_leading_slash(topic), channel::PubSub, registry::Pubsub, registry::Publisher, stamped_cfg, backend); @@ -251,7 +251,7 @@ namespace kickmsg Subscriber Node::subscribe_or_create(char const* topic, channel::Config const& cfg) { channel::Config stamped_cfg = cfg; - stamped_cfg.identity = make_topic_identity(topic); + stamped_cfg.identity = topic_identity(topic); return create_or_open_handle( make_topic_name(topic), with_leading_slash(topic), channel::PubSub, registry::Pubsub, registry::Subscriber, stamped_cfg); @@ -264,14 +264,14 @@ namespace kickmsg auto topic_path = with_leading_slash(channel); if (auto* r = find_region(shm_name)) { - check_cached_identity(*r, shm_name, make_broadcast_identity(channel)); + check_cached_identity(*r, shm_name, broadcast_identity(channel)); check_cached_config(*r, shm_name, channel::Broadcast, cfg); touch_registry(shm_name, topic_path, channel::Broadcast, registry::Broadcast, registry::Both); return BroadcastHandle{Publisher{*r, backend}, Subscriber{*r}}; } channel::Config stamped_cfg = cfg; - stamped_cfg.identity = make_broadcast_identity(channel); + stamped_cfg.identity = broadcast_identity(channel); auto [it, _] = regions_.emplace( shm_name, SharedRegion::create_or_open( @@ -285,7 +285,7 @@ namespace kickmsg { channel::Config mbx_cfg = cfg; mbx_cfg.max_subscribers = 1; - mbx_cfg.identity = make_mailbox_identity(name_.c_str(), tag); + mbx_cfg.identity = mailbox_identity(name_.c_str(), tag); auto shm_name = make_mailbox_name(name_.c_str(), tag); auto topic_path = mailbox_topic(name_.c_str(), tag); // Reuse the region so a duplicate subscriber fails without replacing it. @@ -313,7 +313,7 @@ namespace kickmsg auto topic_path = mailbox_topic(owner_node, tag); if (auto* r = find_region(shm_name)) { - check_cached_identity(*r, shm_name, make_mailbox_identity(owner_node, tag)); + check_cached_identity(*r, shm_name, mailbox_identity(owner_node, tag)); touch_registry(shm_name, topic_path, channel::PubSub, registry::Mailbox, registry::Publisher); return Publisher(*r, backend); @@ -321,7 +321,7 @@ namespace kickmsg auto [it, _] = regions_.emplace( shm_name, SharedRegion::open(shm_name.c_str(), - make_mailbox_identity(owner_node, tag))); + mailbox_identity(owner_node, tag))); // Mailbox sender is the Publisher side. touch_registry(shm_name, topic_path, channel::PubSub, registry::Mailbox, registry::Publisher); @@ -333,7 +333,7 @@ namespace kickmsg { channel::Config mbx_cfg = cfg; mbx_cfg.max_subscribers = 1; - mbx_cfg.identity = make_mailbox_identity(name_.c_str(), tag); + mbx_cfg.identity = mailbox_identity(name_.c_str(), tag); return create_or_open_handle( make_mailbox_name(name_.c_str(), tag), mailbox_topic(name_.c_str(), tag), @@ -346,7 +346,7 @@ namespace kickmsg { channel::Config mbx_cfg = cfg; mbx_cfg.max_subscribers = 1; - mbx_cfg.identity = make_mailbox_identity(owner_node, tag); + mbx_cfg.identity = mailbox_identity(owner_node, tag); return create_or_open_handle( make_mailbox_name(owner_node, tag), mailbox_topic(owner_node, tag), @@ -462,26 +462,27 @@ namespace kickmsg // Raw per-call components disambiguate colliding sanitized names; the // leading kind tag keeps the three channel kinds in disjoint domains. - uint64_t Node::make_topic_identity(char const* topic) const + uint64_t Node::topic_identity(char const* topic) const { - uint64_t h = identity_chain("topic", hash::FNV1A_64_OFFSET_BASIS); - h = identity_chain(namespace_, h); - return identity_chain(topic, h); + uint64_t identity = identity_chain("topic", hash::FNV1A_64_OFFSET_BASIS); + identity = identity_chain(namespace_, identity); + return identity_chain(topic, identity); } - uint64_t Node::make_broadcast_identity(char const* channel) const + uint64_t Node::broadcast_identity(char const* channel) const { - uint64_t h = identity_chain("broadcast", hash::FNV1A_64_OFFSET_BASIS); - h = identity_chain(namespace_, h); - return identity_chain(channel, h); + uint64_t identity = identity_chain("broadcast", hash::FNV1A_64_OFFSET_BASIS); + identity = identity_chain(namespace_, identity); + return identity_chain(channel, identity); } - uint64_t Node::make_mailbox_identity(char const* owner, char const* tag) const + uint64_t Node::mailbox_identity(char const* owner, char const* tag) const { - uint64_t h = identity_chain("mailbox", hash::FNV1A_64_OFFSET_BASIS); - h = identity_chain(namespace_, h); - h = identity_chain(owner, h); - return identity_chain(tag, h); + // Match the owner's sanitized Node name even when the sender spells it differently. + uint64_t identity = identity_chain("mailbox", hash::FNV1A_64_OFFSET_BASIS); + identity = identity_chain(namespace_, identity); + identity = identity_chain(sanitize_shm_component(owner, "mailbox owner"), identity); + return identity_chain(tag, identity); } SharedRegion* Node::find_region(std::string const& shm_name) diff --git a/src/Waker.cc b/src/Waker.cc index be34f3e..70adc18 100644 --- a/src/Waker.cc +++ b/src/Waker.cc @@ -1,11 +1,25 @@ -#include #include +#include +#include #include "kickmsg/Hash.h" +#include "kickmsg/Node.h" #include "kickmsg/Waker.h" namespace kickmsg { + namespace + { + char const* require_name(char const* name, char const* what) + { + if (name == nullptr) + { + throw std::invalid_argument(std::string{"UdpMulticastBackend: "} + what + " must not be null"); + } + return name; + } + } + UdpMulticastBackend::UdpMulticastBackend(uint32_t group, uint16_t port) : group_{group} , port_{port} @@ -26,11 +40,39 @@ namespace kickmsg } UdpMulticastBackend::UdpMulticastBackend(char const* name, uint16_t port_base) + : UdpMulticastBackend(derive(hash::fnv1a_64(std::string_view{require_name(name, "name")}), port_base)) + { + } + + UdpMulticastBackend UdpMulticastBackend::for_topic(Node const& node, char const* topic, uint16_t port_base) + { + return UdpMulticastBackend(derive(node.topic_identity(require_name(topic, "topic")), port_base)); + } + + UdpMulticastBackend UdpMulticastBackend::for_broadcast(Node const& node, char const* channel, uint16_t port_base) { - if (name == nullptr) + return UdpMulticastBackend(derive(node.broadcast_identity(require_name(channel, "channel")), port_base)); + } + + UdpMulticastBackend UdpMulticastBackend::for_mailbox(Node const& node, char const* tag, char const* owner_node, uint16_t port_base) + { + char const* owner = owner_node; + if (owner == nullptr) { - throw std::invalid_argument("UdpMulticastBackend: name must not be null"); + owner = node.name().c_str(); } + return UdpMulticastBackend(derive(node.mailbox_identity(owner, require_name(tag, "tag")), port_base)); + } + + UdpMulticastBackend::UdpMulticastBackend(Address address) + : group_{address.group} + , port_{address.port} + { + open_sender(); + } + + UdpMulticastBackend::Address UdpMulticastBackend::derive(uint64_t hash, uint16_t port_base) + { if (port_base == 0 or port_base > UINT16_MAX - PORT_SPAN + 1) { throw std::invalid_argument( @@ -38,10 +80,10 @@ namespace kickmsg } // 239.255.0.0/16 is administratively scoped: routers never forward it. The port // takes the other half of the hash, so the two collisions stay independent. - uint64_t const h = hash::fnv1a_64(name, std::strlen(name)); - group_ = 0xEFFF0000u | static_cast(h & 0xFFFFu); - port_ = static_cast(port_base + ((h >> 32) % PORT_SPAN)); - open_sender(); + Address address; + address.group = 0xEFFF0000u | static_cast(hash & 0xFFFFu); + address.port = static_cast(port_base + ((hash >> 32) % PORT_SPAN)); + return address; } Waker::Waker(WakeBackend& backend) diff --git a/src/os/posix/WakeBackends.cc b/src/os/posix/WakeBackends.cc index 70ab5fc..c0f7e04 100644 --- a/src/os/posix/WakeBackends.cc +++ b/src/os/posix/WakeBackends.cc @@ -103,9 +103,10 @@ namespace kickmsg // Best effort: SO_REUSEADDR already covers multicast rebinding. (void) ::setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(on)); #endif + // Binding to the group filters other traffic on the same port. sockaddr_in addr{}; addr.sin_family = AF_INET; - addr.sin_addr.s_addr = htonl(INADDR_ANY); + addr.sin_addr.s_addr = htonl(group_); addr.sin_port = htons(port_); if (::bind(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { diff --git a/tests/unit/node-t.cc b/tests/unit/node-t.cc index 2311c62..cd63112 100644 --- a/tests/unit/node-t.cc +++ b/tests/unit/node-t.cc @@ -158,6 +158,22 @@ TEST_F(NodeTest, MailboxPattern) EXPECT_EQ(std::string(static_cast(msg->data()), msg->len()), reply); } +TEST_F(NodeTest, MailboxOwnerMayBeNamedAsGivenToItsNode) +{ + track("test", "robot.arm_mbx_inbox"); + + kickmsg::Node owner("/robot/arm", "test"); + auto inbox = owner.create_mailbox("inbox", small_cfg()); + + kickmsg::Node sender("sender", "test"); + auto pub = sender.open_mailbox("/robot/arm", "inbox"); + EXPECT_EQ(owner.mailbox_identity(owner.name().c_str(), "inbox"), sender.mailbox_identity("/robot/arm", "inbox")); + + uint32_t val = 7; + ASSERT_GE(pub.send(&val, sizeof(val)), 0); + EXPECT_TRUE(inbox.try_receive().has_value()); +} + // --- Schema descriptor through the Node API ------------------------------- namespace diff --git a/tests/unit/wait_fd-t.cc b/tests/unit/wait_fd-t.cc index 2a403eb..a91f8fd 100644 --- a/tests/unit/wait_fd-t.cc +++ b/tests/unit/wait_fd-t.cc @@ -12,6 +12,12 @@ #include "kickmsg/WaitSet.h" #include "kickmsg/Subscriber.h" +#ifndef _WIN32 + #include + #include + #include +#endif + using namespace kickmsg; class WaitFdTest : public ::testing::Test @@ -89,11 +95,94 @@ TEST_F(WaitFdTest, TwoBackendsForOneNameAgreeWithoutCoordinating) EXPECT_LT(publisher_side.port(), UdpMulticastBackend::DEFAULT_PORT_BASE + UdpMulticastBackend::PORT_SPAN); - // A different channel lands on a different port, which is what isolates them. UdpMulticastBackend elsewhere{SHM_OTHER}; EXPECT_NE(publisher_side.port(), elsewhere.port()); } +TEST_F(WaitFdTest, NodeScopedBackendsAgreeAcrossNodesOfOneNamespace) +{ + Node publisher_node("publisher", "robot/left"); + Node subscriber_node("subscriber", "robot.left"); + + auto topic_pub = UdpMulticastBackend::for_topic(publisher_node, "fault"); + auto topic_sub = UdpMulticastBackend::for_topic(subscriber_node, "fault"); + EXPECT_EQ(topic_pub.group(), topic_sub.group()); + EXPECT_EQ(topic_pub.port(), topic_sub.port()); + + auto bcast_a = UdpMulticastBackend::for_broadcast(publisher_node, "fault"); + auto bcast_b = UdpMulticastBackend::for_broadcast(subscriber_node, "fault"); + EXPECT_EQ(bcast_a.group(), bcast_b.group()); + EXPECT_EQ(bcast_a.port(), bcast_b.port()); + + Node owner_node("robot/arm", "robot.left"); + auto mailbox_owner = UdpMulticastBackend::for_mailbox(owner_node, "cmd"); + auto mailbox_sender = UdpMulticastBackend::for_mailbox(publisher_node, "cmd", "/robot/arm"); + EXPECT_EQ(mailbox_owner.group(), mailbox_sender.group()); + EXPECT_EQ(mailbox_owner.port(), mailbox_sender.port()); +} + +TEST_F(WaitFdTest, NodeScopedBackendsUseKindAndNamespace) +{ + Node left("node", "robot.left"); + Node right("node", "robot.right"); + + auto topic = UdpMulticastBackend::for_topic(left, "fault"); + EXPECT_NE(topic.port(), UdpMulticastBackend::for_topic(right, "fault").port()); + EXPECT_NE(topic.port(), UdpMulticastBackend::for_topic(left, "status").port()); + EXPECT_NE(topic.port(), UdpMulticastBackend::for_broadcast(left, "fault").port()); + EXPECT_NE(topic.port(), UdpMulticastBackend::for_mailbox(left, "fault").port()); + + auto own_mailbox = UdpMulticastBackend::for_mailbox(left, "fault"); + EXPECT_NE(own_mailbox.port(), UdpMulticastBackend::for_mailbox(left, "fault", "other").port()); +} + +TEST_F(WaitFdTest, NodeScopedBackendsRejectANullName) +{ + Node node("node", "robot.left"); + EXPECT_THROW(UdpMulticastBackend::for_topic(node, nullptr), std::invalid_argument); + EXPECT_THROW(UdpMulticastBackend::for_broadcast(node, nullptr), std::invalid_argument); + EXPECT_THROW(UdpMulticastBackend::for_mailbox(node, nullptr), std::invalid_argument); + EXPECT_THROW(UdpMulticastBackend(nullptr), std::invalid_argument); +} + +#ifndef _WIN32 +TEST_F(WaitFdTest, DifferentGroupsOnOnePortDoNotWakeEachOther) +{ + constexpr uint16_t shared_port = 31000; + UdpMulticastBackend first(0xEFFF1234u, shared_port); + UdpMulticastBackend second(0xEFFF5678u, shared_port); + Waker first_waker(first); + Waker second_waker(second); + ASSERT_TRUE(first_waker.valid()); + ASSERT_TRUE(second_waker.valid()); + + first.signal(); + EXPECT_TRUE(readable(wait_descriptor(first_waker), 500ms)); + EXPECT_FALSE(readable(wait_descriptor(second_waker), 50ms)); + first_waker.drain(); +} + +TEST_F(WaitFdTest, AUnicastToTheWakePortDoesNotWake) +{ + constexpr uint16_t port = 31001; + UdpMulticastBackend carrier(0xEFFF1234u, port); + Waker waker(carrier); + ASSERT_TRUE(waker.valid()); + + int stray = ::socket(AF_INET, SOCK_DGRAM, 0); + ASSERT_GE(stray, 0); + sockaddr_in to{}; + to.sin_family = AF_INET; + to.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + to.sin_port = htons(port); + char const datagram = 1; + ASSERT_EQ(1, ::sendto(stray, &datagram, 1, 0, reinterpret_cast(&to), sizeof(to))); + ::close(stray); + + EXPECT_FALSE(readable(wait_descriptor(waker), 50ms)); +} +#endif + TEST_F(WaitFdTest, TheUdpBackendRejectsAnAddressItCannotDeliverTo) { // Port 0 binds an ephemeral port but is a literal sendto destination.