mirrors/Zag
1
0
Fork 0
mirror of https://github.com/rdmsr/zag.git synced 2026-10-27 13:58:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

ke/log_ring: cleanup

Browse Source
This commit is contained in:
Loïc Grégoire
2026-08-09 19:23:03 -04:00
parent 5cd981070e
commit f78d10b62e
1 changed files with 178 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/kern/log_ring.zig
+178 -67
View File
@@ -1,25 +1,32 @@
//! Lock-free multi-producer multi-consumer ring buffer designed for kernel logging.
//! Lock-free multi-producer multi-consumer ring buffer designed
//! for kernel logging.
//!
//! # Overview
//!
//! `RingBuffer` is a concurrent ring buffer that allows multiple writers and readers
//! to operate simultaneously without locks, making it safe to use from interrupt and
//! NMI context. It is modelled after Linux's `printk_ringbuffer`, but is simpler (no "committed" state).
//! `RingBuffer` is a concurrent ring buffer that allows multiple writers
//! and readers to operate concurrently without locks, and also supports re-
//! entrancy, making it safe to use from interrupt context.
//! It is modelled after Linux's `printk_ringbuffer`, but is simplified,
//! there is no support for pr_cont or the "committed" state.
//! See linux's printk_ringbuffer.c for more details.
//!
//! The ring buffer is composed of three parallel arrays:
//!
//! - `descs` - descriptor ring: tracks the state and data location of each record
//! - `data` - data ring: byte storage for the actual messages
//! - `infos` - array containing metadata: stores sequence number, timestamp, and length per record
//! - `descs`: tracks the state and data location of each record
//! - `data`: byte storage for the actual messages
//! - `infos`: metadata: sequence number, timestamp, and length per record
//!
//! `infos` is indexed by descriptor IDs, so it doesn't *really* count as a ring.
//! `head_id` is the ID of the newest descriptor, the last slot claimed by a writer.
//! `tail_id` is the ID of the oldest descriptor, the oldest slot which is the next to be evicted.
//! `infos` is indexed by descriptor ID, so it doesn't *really* count as a ring.
//! `head_id` is the ID of the newest descriptor, the last slot claimed by a
//! writer.
//! `tail_id` is the ID of the oldest descriptor, the oldest slot which is the
//! next to be evicted.
//! `tail_id` must always point to a free or published descriptor.
//! The number of live descriptors at any moment is `head_id - tail_id`,
//! when this reaches the maximum descriptor count (`desc_count`), the ring is full and `tail_id` must be advanced.
//! `head_lpos` and `tail_lpos` are the logical positions of the next data block to be written and the oldest data block, respectively.
//! when this reaches the maximum descriptor count (`desc_count`),
//! the ring is full and `tail_id` must be advanced.
//! `head_lpos` and `tail_lpos` are the logical positions of the next data block
//! to be written and the oldest data block, respectively.
//!
//! # Descriptor states
//!
@@ -30,9 +37,9 @@
//! `Reserved` A writer has claimed this slot and is actively writing.
//! `Published` The record is complete and visible to readers.
//! `Free` The slot has been recycled and is available for reuse.
//! `Miss` Pseudo-state: the slot has been recycled to a newer generation.
//! Returned when the ID embedded in a descriptor state does not match
//! the expected ID, indicating the record is gone.
//! `Miss` Pseudo-state: the slot has been reused in a newer generation.
//! Returned when the ID embedded in a descriptor state does not
//! match the expected ID, indicating the record is gone.
//!
//! # IDs and logical positions
//!
@@ -47,7 +54,8 @@
//! to the same physical slot can always be distinguished by comparing the full
//! integer, which prevents ABA bugs.
//! The `State` atomic packs both the descriptor ID and its state into a single
//! `usize`, allowing both to be read and updated atomically via CAS (a CAS is used to prevent ABA where the generation could overflow):
//! `usize`, allowing both to be read and updated atomically via CAS
//! (a CAS is used to prevent ABA where the generation could overflow):
//!
//! [ id (upper 62 bits) | state (lower 2 bits) ] (on 64-bit platforms)
//!
@@ -57,6 +65,8 @@
//! If an allocation would cross the boundary, the block is placed at offset 0 of
//! the next generation and a stub (containing only the descriptor ID) is left at
//! the original position so the data ring can be walked linearly during eviction.
//!
const std = @import("std");
const r = @import("root");
const rtl = @import("rtl");
@@ -76,7 +86,6 @@ const DescState = enum(u2) {
};
const usize_bits = @bitSizeOf(usize);
// Id is a monotonically increasing counter. `id & desc_mask` maps to the physical array index.
const Id = @Int(.unsigned, usize_bits - 2);
const IdSigned = @Int(.signed, usize_bits - 2);
@@ -118,7 +127,8 @@ pub const ReserveError = error{ NoSpace, ReservationFailed, ZeroSize };
pub const ReadError = error{NotYetAvailable};
// Logical positions are always aligned to `@sizeOf(usize)` (even numbers).
// Thus, 1 (an odd number) is guaranteed to never collide with a real data block position.
// Thus, 1 (an odd number) is guaranteed to never collide with a real data
// block position.
const lpos_no_data = 1;
pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
@@ -146,7 +156,8 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
const Self = @This();
// Return the exact Id of the descriptor that occupied this physical slot one generation ago.
// Return the exact Id of the descriptor that occupied this physical slot
// one generation ago.
fn prev_wrap(id: Id) Id {
return id -% desc_count;
}
@@ -180,15 +191,19 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
}
fn check_size(size: usize) bool {
// If the size is larger than half of the data buffer, it can never fit in the ring buffer.
// If the size is larger than half of the data buffer,
// it can never fit in the ring buffer.
return size <= data_size / 2;
}
// Return true if `lpos_current` has not yet reached `lpos_next`.
// Uses wrapping subtraction so that:
// - if lpos_current > lpos_next (another CPU already advanced past the target), returns false.
// - if lpos_current == lpos_next (already at target), the -% 1 wraps to usize max, returns false.
// - if the distance exceeds data_size (should never happen), returns false.
// - if current > next (another CPU already advanced past the target),
// returns false.
// - if current == next (already at target), the -1 wraps to usize max,
// returns false.
// - if the distance exceeds data_size (should never happen),
// returns false.
fn need_more_space(lpos_current: usize, lpos_next: usize) bool {
return lpos_next -% lpos_current -% 1 < data_size;
}
@@ -229,11 +244,18 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
return result;
}
fn read_desc_finalized(self: *Self, id: Id, seq: u64, desc_out: *Desc) !void {
fn read_desc_finalized(
self: *Self,
id: Id,
seq: u64,
desc_out: *Desc,
) !void {
const res = self.read_desc(id);
desc_out.* = res.desc;
if (res.state == .Miss or res.state == .Reserved or (res.seq != seq)) {
if (res.state == .Miss or res.state == .Reserved or
(res.seq != seq))
{
return error.Invalid;
}
@@ -243,10 +265,21 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
fn free_desc(self: *Self, id: Id) void {
var desc = self.to_desc(id);
const expected_state: State = .{ .state = .Published, .id = id };
const new_state: State = .{ .state = .Free, .id = id };
const expected_state: State = .{
.state = .Published,
.id = id,
};
const new_state: State = .{
.state = .Free,
.id = id,
};
_ = desc.state.cmpxchgStrong(expected_state, new_state, .release, .monotonic);
_ = desc.state.cmpxchgStrong(
expected_state,
new_state,
.release,
.monotonic,
);
}
fn free_data(self: *Self, begin: usize, end: usize) ?usize {
@@ -256,8 +289,15 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
while (need_more_space(cur_begin, end)) {
const blk = self.to_block(cur_begin);
// Loading the ID here is purposely racey because we check if the descriptor is valid anyway.
const id: Id = @truncate(@atomicLoad(usize, @as(*usize, @ptrCast(&blk.id)), .monotonic));
// Loading the ID here is purposely racey because
// we check if the descriptor is valid anyway.
const id: Id = @truncate(
@atomicLoad(
usize,
@as(*usize, @ptrCast(&blk.id)),
.monotonic,
),
);
const res = self.read_desc(id);
desc = res.desc;
@@ -268,14 +308,16 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
},
.Published => {
if (desc.blk_pos.begin != cur_begin) {
// The descriptor does not point back to the data block, we must've lost a race.
// The descriptor does not point back to
// the data block, we must've lost a race.
return null;
}
self.free_desc(id);
},
.Free => {
if (desc.blk_pos.begin != cur_begin) {
// The descriptor does not point back to the data block, we must've lost a race.
// The descriptor does not point back to
// the data block, we must've lost a race.
return null;
}
},
@@ -301,14 +343,20 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
// Invalidate all data blocks before the new tail position.
if (self.free_data(cur_tail, new_tail_lpos)) |new_tail| {
// Successfully freed some blocks, try advancing the tail.
if (self.tail_lpos.cmpxchgWeak(cur_tail, new_tail, .release, .monotonic)) |real_cur_tail| {
if (self.tail_lpos.cmpxchgWeak(
cur_tail,
new_tail,
.release,
.monotonic,
)) |real_cur_tail| {
cur_tail = real_cur_tail;
std.atomic.spinLoopHint();
} else {
break;
}
} else {
// Failed to free the data, which means we either lost a race or the descriptor is still in use.
// Failed to free the data, which means we either lost a race
// or the descriptor is still in use.
// Reload the tail position to try again.
const new_tail = self.tail_lpos.load(.acquire);
if (new_tail == cur_tail) {
@@ -334,12 +382,14 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
self.free_desc(tail_id);
},
.Reserved => {
// The descriptor is still reserved, we can't do anything but fail the reservation.
// The descriptor is still reserved,
// we can't do anything but fail the reservation.
return error.ReservationFailed;
},
.Miss => {
if (tail_desc.state.load(.monotonic).id == prev_wrap(tail_id)) {
// This must mean that the descriptor is currently getting reserved by another writer.
// This must mean that the descriptor is currently
// getting reserved by another writer.
return error.ReservationFailed;
}
@@ -347,23 +397,32 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
return;
},
.Free => {
// The descriptor is already free, we can just advance the tail.
// The descriptor is already free, we can just advance the
// tail.
},
}
// Invalidate data blocks
try self.data_advance_tail(tail_desc.blk_pos.end);
// Read the next descriptor after tail_id, because the tail must always be in a finalized or free state.
// Read the next descriptor after tail_id, because the tail must
// always be in a finalized or free state.
const next_res = self.read_desc(tail_id +% 1);
if (next_res.state == .Published or next_res.state == .Free) {
// Advance the tail to the next descriptor.
// release: push all previous state changes before the tail update, guaranteeing the new tail is visible after all evictions.
_ = self.tail_id.cmpxchgStrong(@as(usize, tail_id), @as(usize, tail_id +% 1), .release, .monotonic);
// release: push all previous state changes before the tail update,
// guaranteeing the new tail is visible after all evictions.
_ = self.tail_id.cmpxchgStrong(
@as(usize, tail_id),
@as(usize, tail_id +% 1),
.release,
.monotonic,
);
} else {
// The next descriptor is not in the desired state.
// One possibility is that the tail was pushed by another CPU, in which case we are good.
// One possibility is that the tail was pushed by another CPU,
// in which case we are good.
const tail_id_cur = self.tail_id.load(.acquire);
if (tail_id_cur != tail_id) {
// Another CPU pushed the tail, we are good.
@@ -388,12 +447,19 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
const tail_id = self.tail_id.load(.acquire);
if (prev_id == tail_id) {
// Wrapped back to the tail, need to make space by advancing it.
// Wrapped back to the tail, need to make space by
// advancing it.
try self.advance_tail(prev_id);
}
// Ordering: acq_rel to claim head and publish, acquire on failure to retry.
if (self.head_id.cmpxchgWeak(head_id, new_id, .acq_rel, .acquire)) |val| {
// Ordering: acq_rel to claim head and publish,
// acquire on failure to retry.
if (self.head_id.cmpxchgWeak(
head_id,
new_id,
.acq_rel,
.acquire,
)) |val| {
head_id = @truncate(val);
std.atomic.spinLoopHint();
} else {
@@ -403,16 +469,28 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
const desc = self.to_desc(new_id);
// Ordering: monotonic is safe here because acq_rel on head_id published our claim.
// Ordering: monotonic is safe here because acq_rel on head_id
// published our claim.
const prev_state = desc.state.load(.monotonic);
const raw: usize = @bitCast(prev_state);
if (raw != 0 and get_desc_state(prev_id, prev_state) != .Free) {
if (raw != 0 and
get_desc_state(prev_id, prev_state) != .Free)
{
return error.ReservationFailed;
}
// Ordering: release ensures all setup is published before making the descriptor reserved.
if (desc.state.cmpxchgStrong(prev_state, .{ .state = .Reserved, .id = new_id }, .release, .monotonic) != null) {
// Ordering: release ensures all setup is published before making
// the descriptor reserved.
if (desc.state.cmpxchgStrong(
prev_state,
.{
.state = .Reserved,
.id = new_id,
},
.release,
.monotonic,
) != null) {
return error.ReservationFailed;
}
@@ -421,7 +499,8 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
// Return true if the range [begin, end) crosses the buffer boundary.
// Compares the generation counters (upper bits) of begin and end-1.
// The -% 1 ensures a block that fits exactly at the end does not count as wrapping.
// The -% 1 ensures a block that fits exactly at the end does not count
// as wrapping.
fn data_wraps(begin: usize, end: usize) bool {
return (begin >> data_size_bits) != ((end -% 1) >> data_size_bits);
}
@@ -431,8 +510,9 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
const next_lpos = lpos +% size;
// We never physically split a payload across the ring boundary.
// If it would cross the boundary, we advance the `lpos` to the next generation boundary
// so the data can sit contiguously at physical offset 0.
// If it would cross the boundary, we advance the `lpos` to the next
// generation boundary so the data can sit contiguously at
// physical offset 0.
if (data_wraps(begin_lpos, next_lpos)) {
return ((next_lpos >> data_size_bits) << data_size_bits) + size;
}
@@ -462,7 +542,12 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
};
// Try to claim the new head position.
if (self.head_lpos.cmpxchgWeak(head_lpos, new_head_lpos, .acq_rel, .acquire)) |real_head_lpos| {
if (self.head_lpos.cmpxchgWeak(
head_lpos,
new_head_lpos,
.acq_rel,
.acquire,
)) |real_head_lpos| {
head_lpos = real_head_lpos;
std.atomic.spinLoopHint();
} else {
@@ -517,25 +602,33 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
/// Initialize and return a new ring buffer instance.
pub fn init() Self {
// Start near usize max so that the first wrap of the ID/lpos space happens
// almost immediately, testing overflow handling early rather than after
// billions of records.
// Start near usize max so that the first wrap of the ID/lpos space
// happens almost immediately, testing overflow handling early
// rather than after billions of records.
const dummy_id: Id = 0 -% @as(Id, desc_count + 1);
const blk0: usize = -%@as(usize, data_size);
var self: Self = std.mem.zeroes(Self);
// Place a dummy descriptor in the last slot. head_id and tail_id both start
// here so that the first reservation (head_id + 1) lands in slot 0, giving
// the first real record sequence number 0.
self.descs[desc_count - 1].state.store(.{ .state = .Free, .id = dummy_id }, .monotonic);
self.descs[desc_count - 1].blk_pos = .{ .begin = lpos_no_data, .end = lpos_no_data };
// Place a dummy descriptor in the last slot. head_id and tail_id
// both start here so that the first reservation (head_id + 1) lands
// in slot 0, giving the first real record sequence number 0.
self.descs[desc_count - 1].state.store(.{
.state = .Free,
.id = dummy_id,
}, .monotonic);
self.descs[desc_count - 1].blk_pos = .{
.begin = lpos_no_data,
.end = lpos_no_data,
};
self.head_id.store(dummy_id, .monotonic);
self.tail_id.store(dummy_id, .monotonic);
// Make it -(desc_count) so that when the slot is first reserved and the sequence is advanced by desc_count,
// the first record gets sequence number 0.
// Make it -(desc_count) so that when the slot is first reserved and
// the sequence is advanced by desc_count, the first record gets
// sequence number 0.
self.infos[0].sequence = -%(@as(u64, desc_count));
self.head_lpos.store(blk0, .monotonic);
@@ -565,17 +658,24 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
if (seq == 0 and (id & desc_mask) != 0) {
// First time this slot has been used.
// Slot 0 never hits this branch, its seq is set to -(desc_count)
// in init() so that the else branch below produces seq=0 on its first use.
// in init() so that the else branch below produces seq=0 on
// its first use.
info.sequence = @as(u64, id & desc_mask);
} else {
// Slot is being recycled (or is slot 0 on first use).
// Adding desc_count advances the sequence by one full ring wrap,
// keeping sequence numbers monotonically increasing across generations.
// keeping sequence numbers monotonically increasing across
// generations.
info.sequence = seq +% desc_count;
}
reservation.buf = self.alloc_data(size, id, &desc.blk_pos) catch |e| {
// On failure, publish the descriptor as-is to clean up the reservation.
reservation.buf = self.alloc_data(
size,
id,
&desc.blk_pos,
) catch |e| {
// On failure, publish the descriptor as-is to clean up
// the reservation.
self.publish(reservation);
return e;
};
@@ -589,7 +689,18 @@ pub fn RingBuffer(data_size_bits: usize, avg_msg_bits: usize) type {
pub fn publish(self: *Self, res: Reservation) void {
const desc = self.to_desc(res.id);
_ = desc.state.cmpxchgStrong(.{ .state = .Reserved, .id = res.id }, .{ .state = .Published, .id = res.id }, .release, .monotonic);
_ = desc.state.cmpxchgStrong(
.{
.state = .Reserved,
.id = res.id,
},
.{
.state = .Published,
.id = res.id,
},
.release,
.monotonic,
);
// Restore IRQ state after publishing the new entry.
ki.impl.restore_interrupts(res.irq_state);