mirrors/zig
1
0
Fork 0
mirror of https://codeberg.org/ziglang/zig.git synced 2025-12-06 13:54:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Io: update for new linked list API

Browse source
This commit is contained in:
Andrew Kelley 2025-04-04 15:19:55 -07:00
parent 4b657d2de5
commit e1cbcecf89
2 changed files with 43 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

82
lib/std/Io.zig
View file

@ -917,17 +917,19 @@ pub const TypeErasedQueue = struct {
put_index: usize, put_index: usize,
get_index: usize, get_index: usize,
putters: std.DoublyLinkedList(PutNode), putters: std.DoublyLinkedList,
getters: std.DoublyLinkedList(GetNode), getters: std.DoublyLinkedList,
const PutNode = struct { const Put = struct {
remaining: []const u8, remaining: []const u8,
condition: Condition, condition: Condition,
node: std.DoublyLinkedList.Node,
}; };
const GetNode = struct { const Get = struct {
remaining: []u8, remaining: []u8,
condition: Condition, condition: Condition,
node: std.DoublyLinkedList.Node,
}; };
pub fn init(buffer: []u8) TypeErasedQueue { pub fn init(buffer: []u8) TypeErasedQueue {
@ -952,16 +954,16 @@ pub const TypeErasedQueue = struct {
var remaining = elements; var remaining = elements;
while (true) { while (true) {
const getter = q.getters.popFirst() orelse break; const getter: *Get = @fieldParentPtr("node", q.getters.popFirst() orelse break);
const copy_len = @min(getter.data.remaining.len, remaining.len); const copy_len = @min(getter.remaining.len, remaining.len);
@memcpy(getter.data.remaining[0..copy_len], remaining[0..copy_len]); @memcpy(getter.remaining[0..copy_len], remaining[0..copy_len]);
remaining = remaining[copy_len..]; remaining = remaining[copy_len..];
getter.data.remaining = getter.data.remaining[copy_len..]; getter.remaining = getter.remaining[copy_len..];
if (getter.data.remaining.len == 0) { if (getter.remaining.len == 0) {
getter.data.condition.signal(io); getter.condition.signal(io);
continue; continue;
} }
q.getters.prepend(getter); q.getters.prepend(&getter.node);
assert(remaining.len == 0); assert(remaining.len == 0);
return elements.len; return elements.len;
} }
@ -987,12 +989,10 @@ pub const TypeErasedQueue = struct {
const total_filled = elements.len - remaining.len; const total_filled = elements.len - remaining.len;
if (total_filled >= min) return total_filled; if (total_filled >= min) return total_filled;
var node: std.DoublyLinkedList(PutNode).Node = .{ var pending: Put = .{ .remaining = remaining, .condition = .{}, .node = .{} };
.data = .{ .remaining = remaining, .condition = .{} }, q.putters.append(&pending.node);
}; try pending.condition.wait(io, &q.mutex);
q.putters.append(&node); remaining = pending.remaining;
try node.data.condition.wait(io, &q.mutex);
remaining = node.data.remaining;
} }
} }
@ -1035,16 +1035,16 @@ pub const TypeErasedQueue = struct {
} }
// Copy directly from putters into buffer. // Copy directly from putters into buffer.
while (remaining.len > 0) { while (remaining.len > 0) {
const putter = q.putters.popFirst() orelse break; const putter: *Put = @fieldParentPtr("node", q.putters.popFirst() orelse break);
const copy_len = @min(putter.data.remaining.len, remaining.len); const copy_len = @min(putter.remaining.len, remaining.len);
@memcpy(remaining[0..copy_len], putter.data.remaining[0..copy_len]); @memcpy(remaining[0..copy_len], putter.remaining[0..copy_len]);
putter.data.remaining = putter.data.remaining[copy_len..]; putter.remaining = putter.remaining[copy_len..];
remaining = remaining[copy_len..]; remaining = remaining[copy_len..];
if (putter.data.remaining.len == 0) { if (putter.remaining.len == 0) {
putter.data.condition.signal(io); putter.condition.signal(io);
} else { } else {
assert(remaining.len == 0); assert(remaining.len == 0);
q.putters.prepend(putter); q.putters.prepend(&putter.node);
return fillRingBufferFromPutters(q, io, buffer.len); return fillRingBufferFromPutters(q, io, buffer.len);
} }
} }
@ -1052,12 +1052,10 @@ pub const TypeErasedQueue = struct {
const total_filled = buffer.len - remaining.len; const total_filled = buffer.len - remaining.len;
if (total_filled >= min) return total_filled; if (total_filled >= min) return total_filled;
var node: std.DoublyLinkedList(GetNode).Node = .{ var pending: Get = .{ .remaining = remaining, .condition = .{}, .node = .{} };
.data = .{ .remaining = remaining, .condition = .{} }, q.getters.append(&pending.node);
}; try pending.condition.wait(io, &q.mutex);
q.getters.append(&node); remaining = pending.remaining;
try node.data.condition.wait(io, &q.mutex);
remaining = node.data.remaining;
} }
} }
@ -1067,26 +1065,26 @@ pub const TypeErasedQueue = struct {
/// buffers been fully copied. /// buffers been fully copied.
fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io, len: usize) usize { fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io, len: usize) usize {
while (true) { while (true) {
const putter = q.putters.popFirst() orelse return len; const putter: *Put = @fieldParentPtr("node", q.putters.popFirst() orelse return len);
const available = q.buffer[q.put_index..]; const available = q.buffer[q.put_index..];
const copy_len = @min(available.len, putter.data.remaining.len); const copy_len = @min(available.len, putter.remaining.len);
@memcpy(available[0..copy_len], putter.data.remaining[0..copy_len]); @memcpy(available[0..copy_len], putter.remaining[0..copy_len]);
putter.data.remaining = putter.data.remaining[copy_len..]; putter.remaining = putter.remaining[copy_len..];
q.put_index += copy_len; q.put_index += copy_len;
if (putter.data.remaining.len == 0) { if (putter.remaining.len == 0) {
putter.data.condition.signal(io); putter.condition.signal(io);
continue; continue;
} }
const second_available = q.buffer[0..q.get_index]; const second_available = q.buffer[0..q.get_index];
const second_copy_len = @min(second_available.len, putter.data.remaining.len); const second_copy_len = @min(second_available.len, putter.remaining.len);
@memcpy(second_available[0..second_copy_len], putter.data.remaining[0..second_copy_len]); @memcpy(second_available[0..second_copy_len], putter.remaining[0..second_copy_len]);
putter.data.remaining = putter.data.remaining[copy_len..]; putter.remaining = putter.remaining[copy_len..];
q.put_index = copy_len; q.put_index = copy_len;
if (putter.data.remaining.len == 0) { if (putter.remaining.len == 0) {
putter.data.condition.signal(io); putter.condition.signal(io);
continue; continue;
} }
q.putters.prepend(putter); q.putters.prepend(&putter.node);
return len; return len;
} }
} }

7
lib/std/Io/EventLoop.zig
View file

@ -13,7 +13,7 @@ main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fib
threads: Thread.List, threads: Thread.List,
detached: struct { detached: struct {
mutex: std.Io.Mutex, mutex: std.Io.Mutex,
list: std.DoublyLinkedList(void), list: std.DoublyLinkedList,
}, },
/// Empirically saw >128KB being used by the self-hosted backend to panic. /// Empirically saw >128KB being used by the self-hosted backend to panic.
@ -226,7 +226,6 @@ pub fn deinit(el: *EventLoop) void {
detached.detached_queue_node = .{ detached.detached_queue_node = .{
.prev = &detached.detached_queue_node, .prev = &detached.detached_queue_node,
.next = &detached.detached_queue_node, .next = &detached.detached_queue_node,
.data = {},
}; };
break :detached_future @ptrCast(detached.fiber); break :detached_future @ptrCast(detached.fiber);
}, &.{}, .@"1"); }, &.{}, .@"1");
@ -753,7 +752,7 @@ const DetachedClosure = struct {
event_loop: *EventLoop, event_loop: *EventLoop,
fiber: *Fiber, fiber: *Fiber,
start: *const fn (context: *const anyopaque) void, start: *const fn (context: *const anyopaque) void,
detached_queue_node: std.DoublyLinkedList(void).Node, detached_queue_node: std.DoublyLinkedList.Node,
fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 { fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure)); return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
@ -818,7 +817,7 @@ fn go(
.event_loop = event_loop, .event_loop = event_loop,
.fiber = fiber, .fiber = fiber,
.start = start, .start = start,
.detached_queue_node = .{ .data = {} }, .detached_queue_node = .{},
}; };
{ {
event_loop.detached.mutex.lock(event_loop.io()) catch |err| switch (err) { event_loop.detached.mutex.lock(event_loop.io()) catch |err| switch (err) {