Part 4. Cores talking to cores · lesson 30 of 42
30. Full buffers¶
A sender must not overrun a receiver that is busy, and Komorebi does it differently for each
kind of message: in a dynamic run it gives each core a quota, a number of sends that it
may have alive, and send returns false when the quota is used up. This lesson relies on
lesson 28, A routed message and lesson 29, Forward and consume.
Why a limit¶
A receiver takes messages at its own speed. If senders may send as many as they like, the messages pile up and something overflows: a buffer in L1 with no more room, or a queue in DRAM that wraps round over messages not yet read. A program that loses messages is wrong in ways that depend on timing, which is the worst kind. So every way of sending has a rule that makes overflow impossible:
| Kind | What stops a sender | What the sender sees |
|---|---|---|
| port (lesson 25, A port) | each sender has a slot of its own in the receiver's arrival buffer, and exclusive=True keeps two commands' tiles apart |
nothing: it can always write |
| broadcast (lesson 27, Broadcast) | a window of blocks in flight: a slot is rewritten only when every receiver has finished with it | reserve waits |
| routed message (lesson 28, A routed message) | a quota of messages alive for each core | send returns false |
In a dynamic run every core has a quota to start with: max_live divided by the number
of cores that take messages. send takes one unit. consume gives one unit back to the core
where the message ends. A forward changes nothing. With max_live messages at most alive
in the whole run, and an inbox of at least max_live messages for each core, no inbox can
overflow, even when every message is addressed to one core. The price is that quota stays
with the messages: a core that only sends runs dry, and a core that only consumes collects
more than it started with.
The general form¶
if (!messages.send(DESTINATION, message)) {
... // no quota left: do something else
}
const uint32_t left = messages.quota();
messages.sendreturnstruewhen the message was sent andfalsewhen this core has no quota. When it returns false, nothing was sent.messages.quota()is the sends this core has left.LIVEismax_live, the most messages alive at once on the card. Each core starts withLIVE / 118: 120 cores less 2 counter tiles.LIVEmust be at least 118.QUEUEisqueue_messages, the room for messages in each core's inbox, a power of two. It may not be smaller thanLIVE: Komorebi refuses such a program before it opens the card.
One line at a time¶
program = km.Program(
"full_buffers",
kernels={
"reader": RUNTIME / "messages_reader.cpp",
"compute": KERNELS / "compute.cpp",
"writer": RUNTIME / "messages_writer.cpp",
},
handlers=KERNELS / "handlers.h",
).dynamic("flood", message_bytes=64, max_live=MAX_LIVE, queue_messages=2048)
max_live is 1180: 10 units of quota for each of the 118 cores. The inboxes hold 2048.
compute.run([&](const km_command_t& command) {
if (km::handler(command) != kHandlerFlood) {
return;
}
const uint32_t notes_address = command.a0;
km::Messages messages(compute, command);
uint32_t message[kMessageWords] = {};
uint32_t accepted = 0;
uint32_t received = 0;
if (messages.seeding()) {
if (messages.rank() == 0) {
const uint32_t attempts = 2 * messages.quota();
for (uint32_t attempt = 0; attempt < attempts; attempt++) {
message[kValue] = attempt;
accepted += messages.send(attempt % messages.count(), message);
}
}
messages.seeded();
}
while (km::Packet packet = messages.next()) {
for (uint32_t index = 0; index < packet.count; index++) {
received++;
messages.consume();
}
messages.done(packet);
}
const uint32_t quota_left = messages.quota();
UNPACK(({
volatile tt_l1_ptr uint32_t* notes = km::l1(notes_address);
notes[0] = accepted;
notes[1] = received;
notes[2] = quota_left;
}));
});
Rank 0 tries twice as many sends as its quota, counting the ones that were accepted: the first 10 are, the other 10 return false. The messages are spread round the 6 cores. Every core consumes what it gets and reads its quota at the end.
The whole program¶
"""Full buffers: a core may have only so many messages alive; send says no past that."""
from pathlib import Path
import numpy as np
import komorebi as km
HERE = Path(__file__).resolve().parent
KERNELS = HERE / "kernels"
RUNTIME = HERE.parents[1] / "src" / "komorebi_native" / "kernels"
GROUP = 8 # 6 cores take messages; the last 2 cores of a dynamic group keep counters
RANKS = GROUP - 2
QUOTA = 10 # sends each core may have alive
MAX_LIVE = 118 * QUOTA # in the whole card: 120 cores less the 2 counter tiles
with km.Session() as session:
notes = session.tensor(
np.zeros((120, 4), np.int32), "int32", shard_grid=session.grid
)
program = km.Program(
"full_buffers",
kernels={
"reader": RUNTIME / "messages_reader.cpp",
"compute": KERNELS / "compute.cpp",
"writer": RUNTIME / "messages_writer.cpp",
},
handlers=KERNELS / "handlers.h",
).dynamic("flood", message_bytes=64, max_live=MAX_LIVE, queue_messages=2048)
run = session.load(program)
handle = run.submit("flood", group=(0, GROUP), args=[notes.address], exclusive=True)
run.wait(handle, timeout=10.0)
run.stop()
kept = notes.read()[:RANKS, :3]
for rank, (accepted, received, quota_left) in enumerate(kept.tolist()):
print(
f"rank {rank}: accepted {accepted} sends, received {received}, "
f"quota left {quota_left}"
)
assert kept[:, 0].tolist() == [QUOTA] + [0] * (RANKS - 1)
assert kept[:, 1].sum() == QUOTA
assert (kept[:, 2] == QUOTA - kept[:, 0] + kept[:, 1]).all()
print(f"rank 0 asked {2 * QUOTA} times and {QUOTA} sends were accepted")
// The compute kernel of a dynamic handler: rank 0 sends messages until send says no
// more, and every core consumes what it receives and notes how much quota it has left.
#include "api/compute/common.h"
#include "komorebi/device/km_messages.hpp"
#include "komorebi/device/km_worker.hpp"
#include "handlers.h"
void kernel_main() {
km::Compute compute;
compute.run([&](const km_command_t& command) {
if (km::handler(command) != kHandlerFlood) {
return;
}
const uint32_t notes_address = command.a0;
km::Messages messages(compute, command);
uint32_t message[kMessageWords] = {};
uint32_t accepted = 0;
uint32_t received = 0;
if (messages.seeding()) {
if (messages.rank() == 0) {
const uint32_t attempts = 2 * messages.quota();
for (uint32_t attempt = 0; attempt < attempts; attempt++) {
message[kValue] = attempt;
accepted += messages.send(attempt % messages.count(), message);
}
}
messages.seeded();
}
while (km::Packet packet = messages.next()) {
for (uint32_t index = 0; index < packet.count; index++) {
received++;
messages.consume();
}
messages.done(packet);
}
const uint32_t quota_left = messages.quota();
UNPACK(({
volatile tt_l1_ptr uint32_t* notes = km::l1(notes_address);
notes[0] = accepted;
notes[1] = received;
notes[2] = quota_left;
}));
});
}
rank 0: accepted 10 sends, received 2, quota left 2
rank 1: accepted 0 sends, received 2, quota left 12
rank 2: accepted 0 sends, received 2, quota left 12
rank 3: accepted 0 sends, received 2, quota left 12
rank 4: accepted 0 sends, received 1, quota left 11
rank 5: accepted 0 sends, received 1, quota left 11
rank 0 asked 20 times and 10 sends were accepted
Rank 0 had 10 units, sent 10 messages and was refused the other 10 sends. Its 10 messages went to ranks 0 to 5 (ranks 0 to 3 got two each, ranks 4 and 5 one), and each message that ended at a core gave that core a unit back. The core's quota at the end is its 10, less what it sent, plus what it received: 2 for rank 0, 12 for ranks 1 to 3, 11 for ranks 4 and 5. The script checks each of these.
Common mistakes¶
- Ignoring the result of
send. A refused send sends nothing; a program that does not look has lost that message. - A seed that sends more messages than the quota. The seeding sends count too.
- A
max_livethat is less than the number of cores: "no quota" is the error, before the card is opened. - Moving the quota between cores. Komorebi does not do it: a program in which some cores only send and others only consume runs the senders dry.
Next: lesson 31, Streams and graphs describes a whole chain of steps once, instead of core by core.