Skip to content

Improve wake UDP API by deriving the name with the node #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leducp merged 1 commit into from
Sep 28, 2026
Merged

Improve wake UDP API by deriving the name with the node #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
12 changes: 5 additions & 7 deletions examples/hello_wait.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 <cstring>
#include <iostream>
Expand All @@ -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");
Expand Down
13 changes: 7 additions & 6 deletions include/kickmsg/Node.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
31 changes: 24 additions & 7 deletions include/kickmsg/Waker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down Expand Up @@ -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
Expand All @@ -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);

Expand All @@ -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();

Expand Down
61 changes: 31 additions & 30 deletions src/Node.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,18 +54,18 @@ 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;
}

/// 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.
Expand Down Expand Up @@ -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()));
Expand All @@ -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);
Expand Down Expand Up @@ -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<Publisher>(
make_topic_name(topic), with_leading_slash(topic),
channel::PubSub, registry::Pubsub, registry::Publisher, stamped_cfg, backend);
Expand All @@ -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<Subscriber>(
make_topic_name(topic), with_leading_slash(topic),
channel::PubSub, registry::Pubsub, registry::Subscriber, stamped_cfg);
Expand All @@ -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(
Expand All @@ -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.
Expand Down Expand Up @@ -313,15 +313,15 @@ 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);
}
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);
Expand All @@ -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<Subscriber>(
make_mailbox_name(name_.c_str(), tag),
mailbox_topic(name_.c_str(), tag),
Expand All @@ -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<Publisher>(
make_mailbox_name(owner_node, tag),
mailbox_topic(owner_node, tag),
Expand Down Expand Up @@ -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)
Expand Down
56 changes: 49 additions & 7 deletions src/Waker.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,25 @@
#include <cstring>
#include <stdexcept>
#include <string>
#include <string_view>

#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}
Expand All @@ -26,22 +40,50 @@ 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(
"UdpMulticastBackend: port_base leaves no room for PORT_SPAN");
}
// 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<uint32_t>(h & 0xFFFFu);
port_ = static_cast<uint16_t>(port_base + ((h >> 32) % PORT_SPAN));
open_sender();
Address address;
address.group = 0xEFFF0000u | static_cast<uint32_t>(hash & 0xFFFFu);
address.port = static_cast<uint16_t>(port_base + ((hash >> 32) % PORT_SPAN));
return address;
}

Waker::Waker(WakeBackend& backend)
Expand Down
Loading
Loading