Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343//! Shared decoded-block cache for subscriber cold replay.//!//! Immutable sealed blocks are keyed by segment checksum, block index, and a//! per-segment invalidation generation. Concurrent misses for one key decode//! exactly once; unrelated keys decode concurrently. Handles pin an item//! across LRU eviction, and four memo lanes let the subscribe layer share v1,//! v2, and their dictionary-compressed wire frames across readers. Memo growth//! is charged to the same byte budget that owns the decoded block.
const std = @import("std");const segment = @import("segment.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;
pub const default_max_bytes: usize = 64 * 1024 * 1024;
pub const Key = struct { segment_index: u64, checksum: u64, block_index: u64, generation: u64,};
pub const MemoLane = enum(u2) { v1, v2, zstd_v1, zstd_v2 };
const Memo = struct { done: bool = false, body: ?[]u8 = null, err: ?anyerror = null,};
const Item = struct { key: Key, block: segment.Block, memos: []Memo, memo_mu: Io.Mutex = .init, bytes: usize, refs: usize = 0, resident: bool = false, prev: ?*Item = null, next: ?*Item = null,
fn deinit(self: *Item, allocator: Allocator) void { for (self.memos) |memo| if (memo.body) |body| allocator.free(body); allocator.free(self.memos); self.block.deinit(allocator); allocator.destroy(self); }};
const Inflight = struct { done: Io.Condition = .init, complete: bool = false, participants: usize = 1, item: ?*Item = null, err: ?anyerror = null,};
pub const Handle = struct { cache: *Cache, item: *Item,
pub fn events(self: *const Handle) []const segment.Event { return self.item.block.events; }
/// Compute one immutable memo at most once. `compute` must allocate a /// returned body with the cache allocator; the cache owns it thereafter. pub fn memoize( self: *Handle, event_index: usize, lane: MemoLane, ctx: anytype, comptime compute: fn (@TypeOf(ctx), Allocator, segment.Event) anyerror!?[]u8, ) !?[]const u8 { if (event_index >= self.item.block.events.len) return error.BadEventIndex; const at = event_index * 4 + @backingInt(lane); self.item.memo_mu.lockUncancelable(self.cache.io); defer self.item.memo_mu.unlock(self.cache.io); const memo = &self.item.memos[at]; if (!memo.done) { memo.body = compute(ctx, self.cache.allocator, self.item.block.events[event_index]) catch |err| { memo.err = err; memo.done = true; return err; }; memo.done = true; if (memo.body) |body| self.cache.grow(self.item, body.len); } if (memo.err) |err| return err; return memo.body; }
pub fn deinit(self: *Handle) void { const cache = self.cache; cache.mu.lockUncancelable(cache.io); std.debug.assert(self.item.refs > 0); self.item.refs -= 1; const destroy = self.item.refs == 0 and !self.item.resident; cache.mu.unlock(cache.io); if (destroy) self.item.deinit(cache.allocator); self.* = undefined; }};
pub const Cache = struct { allocator: Allocator, io: Io, max_bytes: usize, mu: Io.Mutex = .init, items: std.AutoHashMapUnmanaged(Key, *Item) = .empty, inflight: std.AutoHashMapUnmanaged(Key, *Inflight) = .empty, generations: std.AutoHashMapUnmanaged(u64, u64) = .empty, head: ?*Item = null, tail: ?*Item = null, cur_bytes: usize = 0, decode_count: std.atomic.Value(u64) = .init(0),
pub fn init(allocator: Allocator, io: Io, max_bytes: usize) Cache { return .{ .allocator = allocator, .io = io, .max_bytes = if (max_bytes == 0) default_max_bytes else max_bytes, }; }
pub fn deinit(self: *Cache) void { self.mu.lockUncancelable(self.io); std.debug.assert(self.inflight.count() == 0); var it = self.items.valueIterator(); while (it.next()) |item_ptr| { const item = item_ptr.*; std.debug.assert(item.refs == 0); item.resident = false; item.deinit(self.allocator); } self.items.deinit(self.allocator); self.inflight.deinit(self.allocator); self.generations.deinit(self.allocator); self.head = null; self.tail = null; self.cur_bytes = 0; self.mu.unlock(self.io); self.* = undefined; }
pub fn key(self: *Cache, segment_index: u64, checksum: u64, block_index: usize) Key { self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); return .{ .segment_index = segment_index, .checksum = checksum, .block_index = block_index, .generation = self.generations.get(segment_index) orelse 0, }; }
/// Return a pinned decoded block. The first caller for a key executes the /// real decoder outside the cache lock; concurrent callers wait for that /// result while other keys remain independent. pub fn acquire( self: *Cache, key_value: Key, ctx: anytype, comptime decode: fn (@TypeOf(ctx), Allocator) anyerror!segment.Block, ) !Handle { self.mu.lockUncancelable(self.io); if (self.items.get(key_value)) |item| { item.refs += 1; self.touchLocked(item); self.mu.unlock(self.io); return .{ .cache = self, .item = item }; } if (self.inflight.get(key_value)) |flight| { flight.participants += 1; while (!flight.complete) flight.done.waitUncancelable(self.io, &self.mu); const item = flight.item; const err = flight.err; self.finishFlightLocked(flight); self.mu.unlock(self.io); if (err) |value| return value; return .{ .cache = self, .item = item.? }; }
const flight = self.allocator.create(Inflight) catch |err| { self.mu.unlock(self.io); return err; }; flight.* = .{}; self.inflight.put(self.allocator, key_value, flight) catch |err| { self.allocator.destroy(flight); self.mu.unlock(self.io); return err; }; self.mu.unlock(self.io);
_ = self.decode_count.fetchAdd(1, .monotonic); var decoded = decode(ctx, self.allocator) catch |err| { self.completeFailure(key_value, flight, err); return err; }; var item = self.createItem(key_value, decoded) catch |err| { decoded.deinit(self.allocator); self.completeFailure(key_value, flight, err); return err; };
self.mu.lockUncancelable(self.io); item.refs = flight.participants; const current_generation = self.generations.get(key_value.segment_index) orelse 0; if (current_generation == key_value.generation) { self.items.put(self.allocator, key_value, item) catch |err| { self.finishCompletionLocked(key_value, flight, null, err); self.mu.unlock(self.io); item.refs = 0; item.deinit(self.allocator); return err; }; item.resident = true; self.linkFrontLocked(item); self.cur_bytes += item.bytes; self.evictLocked(); } self.finishCompletionLocked(key_value, flight, item, null); self.mu.unlock(self.io); return .{ .cache = self, .item = item }; }
fn createItem(self: *Cache, key_value: Key, block: segment.Block) !*Item { const item = try self.allocator.create(Item); errdefer self.allocator.destroy(item); const memos = try self.allocator.alloc(Memo, block.events.len * 4); @memset(memos, .{}); item.* = .{ .key = key_value, .block = block, .memos = memos, .bytes = blockBytes(block.events), }; return item; }
/// Match upstream's decoded-byte estimate: borrowed event fields plus a /// fixed per-entry allowance for the Event/Entry/pointer structures. fn blockBytes(events: []const segment.Event) usize { var total: usize = 0; for (events) |event| { total += event.payload.len + event.did.len + event.collection.len + event.rkey.len + event.rev.len + 128; } return total; }
fn completeFailure(self: *Cache, key_value: Key, flight: *Inflight, err: anyerror) void { self.mu.lockUncancelable(self.io); self.finishCompletionLocked(key_value, flight, null, err); self.mu.unlock(self.io); }
fn finishCompletionLocked(self: *Cache, key_value: Key, flight: *Inflight, item: ?*Item, err: ?anyerror) void { _ = self.inflight.remove(key_value); flight.item = item; flight.err = err; flight.complete = true; flight.done.broadcast(self.io); self.finishFlightLocked(flight); }
fn finishFlightLocked(self: *Cache, flight: *Inflight) void { std.debug.assert(flight.participants > 0); flight.participants -= 1; if (flight.participants == 0) self.allocator.destroy(flight); }
pub fn invalidateSegment(self: *Cache, segment_index: u64) !void { self.mu.lockUncancelable(self.io); errdefer self.mu.unlock(self.io); const generation = self.generations.get(segment_index) orelse 0; try self.generations.put(self.allocator, segment_index, generation +% 1); var current = self.head; while (current) |item| { const next = item.next; if (item.key.segment_index == segment_index) self.removeResidentLocked(item); current = next; } self.mu.unlock(self.io); }
pub fn bytes(self: *Cache) usize { self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); return self.cur_bytes; }
fn grow(self: *Cache, item: *Item, delta: usize) void { if (delta == 0) return; self.mu.lockUncancelable(self.io); if (item.resident) { item.bytes += delta; self.cur_bytes += delta; self.evictLocked(); } self.mu.unlock(self.io); }
fn touchLocked(self: *Cache, item: *Item) void { if (self.head == item) return; self.unlinkLocked(item); self.linkFrontLocked(item); }
fn linkFrontLocked(self: *Cache, item: *Item) void { item.prev = null; item.next = self.head; if (self.head) |head| head.prev = item else self.tail = item; self.head = item; }
fn unlinkLocked(self: *Cache, item: *Item) void { if (item.prev) |prev| prev.next = item.next else self.head = item.next; if (item.next) |next| next.prev = item.prev else self.tail = item.prev; item.prev = null; item.next = null; }
fn removeResidentLocked(self: *Cache, item: *Item) void { if (!item.resident) return; _ = self.items.remove(item.key); self.unlinkLocked(item); self.cur_bytes -= item.bytes; item.resident = false; if (item.refs == 0) item.deinit(self.allocator); }
fn evictLocked(self: *Cache) void { while (self.cur_bytes > self.max_bytes) { const item = self.tail orelse break; self.removeResidentLocked(item); } }};