authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-11-03 21:05:18-05:00
committergravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-11-04 04:24:32-05:00
logea76946d2af588d135e79a97f19d128acfeb8915
treef4d800b03a8497afa6fcb87aa080e4462a41369f
parentc603d27f900540b52586ee2638060a4b2ac7c557

Io.Queue: fix empty and full states being indistinguishable.

2 files changed, 64 insertions(+), 64 deletions(-)

lib/std/Io.zig+43-64
...@@ -1238,8 +1238,8 @@ pub const TypeErasedQueue = struct {...@@ -1238,8 +1238,8 @@ pub const TypeErasedQueue = struct {
12381238
1239 /// Ring buffer. This data is logically *after* queued getters.1239 /// Ring buffer. This data is logically *after* queued getters.
1240 buffer: []u8,1240 buffer: []u8,
1241 put_index: usize,1241 start: usize,
1242 get_index: usize,1242 len: usize,
12431243
1244 putters: std.DoublyLinkedList,1244 putters: std.DoublyLinkedList,
1245 getters: std.DoublyLinkedList,1245 getters: std.DoublyLinkedList,
...@@ -1260,8 +1260,8 @@ pub const TypeErasedQueue = struct {...@@ -1260,8 +1260,8 @@ pub const TypeErasedQueue = struct {
1260 return .{1260 return .{
1261 .mutex = .init,1261 .mutex = .init,
1262 .buffer = buffer,1262 .buffer = buffer,
1263 .put_index = 0,1263 .start = 0,
1264 .get_index = 0,1264 .len = 0,
1265 .putters = .{},1265 .putters = .{},
1266 .getters = .{},1266 .getters = .{},
1267 };1267 };
...@@ -1286,6 +1286,16 @@ pub const TypeErasedQueue = struct {...@@ -1286,6 +1286,16 @@ pub const TypeErasedQueue = struct {
1286 };1286 };
1287 }1287 }
12881288
1289 fn puttableSlice(q: *const TypeErasedQueue) ?[]u8 {
1290 const unwrapped_index = q.start + q.len;
1291 const wrapped_index, const overflow = @subWithOverflow(unwrapped_index, q.buffer.len);
1292 const slice = switch (overflow) {
1293 1 => q.buffer[unwrapped_index..],
1294 0 => q.buffer[wrapped_index..q.start],
1295 };
1296 return if (slice.len > 0) slice else null;
1297 }
1298
1289 fn putLocked(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize, uncancelable: bool) Cancelable!usize {1299 fn putLocked(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize, uncancelable: bool) Cancelable!usize {
1290 // Getters have first priority on the data, and only when the getters1300 // Getters have first priority on the data, and only when the getters
1291 // queue is empty do we start populating the buffer.1301 // queue is empty do we start populating the buffer.
...@@ -1306,20 +1316,12 @@ pub const TypeErasedQueue = struct {...@@ -1306,20 +1316,12 @@ pub const TypeErasedQueue = struct {
1306 return elements.len;1316 return elements.len;
1307 }1317 }
13081318
1309 {1319 while (q.puttableSlice()) |slice| {
1310 const available = q.buffer[q.put_index..];1320 const copy_len = @min(slice.len, remaining.len);
1311 const copy_len = @min(available.len, remaining.len);1321 assert(copy_len > 0);
1312 @memcpy(available[0..copy_len], remaining[0..copy_len]);1322 @memcpy(slice[0..copy_len], remaining[0..copy_len]);
1313 remaining = remaining[copy_len..];1323 q.len += copy_len;
1314 q.put_index += copy_len;
1315 if (remaining.len == 0) return elements.len;
1316 }
1317 {
1318 const available = q.buffer[0..q.get_index];
1319 const copy_len = @min(available.len, remaining.len);
1320 @memcpy(available[0..copy_len], remaining[0..copy_len]);
1321 remaining = remaining[copy_len..];1324 remaining = remaining[copy_len..];
1322 q.put_index = copy_len;
1323 if (remaining.len == 0) return elements.len;1325 if (remaining.len == 0) return elements.len;
1324 }1326 }
13251327
...@@ -1354,46 +1356,32 @@ pub const TypeErasedQueue = struct {...@@ -1354,46 +1356,32 @@ pub const TypeErasedQueue = struct {
1354 };1356 };
1355 }1357 }
13561358
1359 fn gettableSlice(q: *const TypeErasedQueue) ?[]const u8 {
1360 const overlong_slice = q.buffer[q.start..];
1361 const slice = overlong_slice[0..@min(overlong_slice.len, q.len)];
1362 return if (slice.len > 0) slice else null;
1363 }
1364
1357 fn getLocked(q: *@This(), io: Io, buffer: []u8, min: usize, uncancelable: bool) Cancelable!usize {1365 fn getLocked(q: *@This(), io: Io, buffer: []u8, min: usize, uncancelable: bool) Cancelable!usize {
1358 // The ring buffer gets first priority, then data should come from any1366 // The ring buffer gets first priority, then data should come from any
1359 // queued putters, then finally the ring buffer should be filled with1367 // queued putters, then finally the ring buffer should be filled with
1360 // data from putters so they can be resumed.1368 // data from putters so they can be resumed.
13611369
1362 var remaining = buffer;1370 var remaining = buffer;
1363 if (q.get_index <= q.put_index) {1371 while (q.gettableSlice()) |slice| {
1364 const available = q.buffer[q.get_index..q.put_index];1372 const copy_len = @min(slice.len, remaining.len);
1365 const copy_len = @min(available.len, remaining.len);1373 assert(copy_len > 0);
1366 @memcpy(remaining[0..copy_len], available[0..copy_len]);1374 @memcpy(remaining[0..copy_len], slice[0..copy_len]);
1367 q.get_index += copy_len;1375 q.start += copy_len;
1376 if (q.buffer.len - q.start == 0) q.start = 0;
1377 q.len -= copy_len;
1368 remaining = remaining[copy_len..];1378 remaining = remaining[copy_len..];
1369 if (remaining.len == 0) {1379 if (remaining.len == 0) {
1370 q.fillRingBufferFromPutters(io);1380 q.fillRingBufferFromPutters(io);
1371 return buffer.len;1381 return buffer.len;
1372 }1382 }
1373 } else {
1374 {
1375 const available = q.buffer[q.get_index..];
1376 const copy_len = @min(available.len, remaining.len);
1377 @memcpy(remaining[0..copy_len], available[0..copy_len]);
1378 q.get_index += copy_len;
1379 remaining = remaining[copy_len..];
1380 if (remaining.len == 0) {
1381 q.fillRingBufferFromPutters(io);
1382 return buffer.len;
1383 }
1384 }
1385 {
1386 const available = q.buffer[0..q.put_index];
1387 const copy_len = @min(available.len, remaining.len);
1388 @memcpy(remaining[0..copy_len], available[0..copy_len]);
1389 q.get_index = copy_len;
1390 remaining = remaining[copy_len..];
1391 if (remaining.len == 0) {
1392 q.fillRingBufferFromPutters(io);
1393 return buffer.len;
1394 }
1395 }
1396 }1383 }
1384
1397 // Copy directly from putters into buffer.1385 // Copy directly from putters into buffer.
1398 while (q.putters.popFirst()) |putter_node| {1386 while (q.putters.popFirst()) |putter_node| {
1399 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));1387 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));
...@@ -1410,6 +1398,7 @@ pub const TypeErasedQueue = struct {...@@ -1410,6 +1398,7 @@ pub const TypeErasedQueue = struct {
1410 q.fillRingBufferFromPutters(io);1398 q.fillRingBufferFromPutters(io);
1411 return buffer.len;1399 return buffer.len;
1412 }1400 }
1401
1413 // Both ring buffer and putters queue is empty.1402 // Both ring buffer and putters queue is empty.
1414 const total_filled = buffer.len - remaining.len;1403 const total_filled = buffer.len - remaining.len;
1415 if (total_filled >= min) return total_filled;1404 if (total_filled >= min) return total_filled;
...@@ -1432,30 +1421,20 @@ pub const TypeErasedQueue = struct {...@@ -1432,30 +1421,20 @@ pub const TypeErasedQueue = struct {
1432 fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io) void {1421 fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io) void {
1433 while (q.putters.popFirst()) |putter_node| {1422 while (q.putters.popFirst()) |putter_node| {
1434 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));1423 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));
1435 {1424 while (q.puttableSlice()) |slice| {
1436 const available = q.buffer[q.put_index..];1425 const copy_len = @min(slice.len, putter.remaining.len);
1437 const copy_len = @min(available.len, putter.remaining.len);1426 assert(copy_len > 0);
1438 @memcpy(available[0..copy_len], putter.remaining[0..copy_len]);1427 @memcpy(slice[0..copy_len], putter.remaining[0..copy_len]);
1428 q.len += copy_len;
1439 putter.remaining = putter.remaining[copy_len..];1429 putter.remaining = putter.remaining[copy_len..];
1440 q.put_index += copy_len;
1441 if (putter.remaining.len == 0) {1430 if (putter.remaining.len == 0) {
1442 putter.condition.signal(io);1431 putter.condition.signal(io);
1443 continue;1432 break;
1444 }
1445 }
1446 {
1447 const available = q.buffer[0..q.get_index];
1448 const copy_len = @min(available.len, putter.remaining.len);
1449 @memcpy(available[0..copy_len], putter.remaining[0..copy_len]);
1450 putter.remaining = putter.remaining[copy_len..];
1451 q.put_index = copy_len;
1452 if (putter.remaining.len == 0) {
1453 putter.condition.signal(io);
1454 continue;
1455 }1433 }
1434 } else {
1435 q.putters.prepend(putter_node);
1436 break;
1456 }1437 }
1457 q.putters.prepend(putter_node);
1458 break;
1459 }1438 }
1460 }1439 }
1461};1440};
lib/std/Io/test.zig+21
...@@ -208,3 +208,24 @@ test "select" {...@@ -208,3 +208,24 @@ test "select" {
208 },208 },
209 }209 }
210}210}
211
212fn testQueue(comptime len: usize) !void {
213 const io = testing.io;
214 var buf: [len]usize = undefined;
215 var queue: Io.Queue(usize) = .init(&buf);
216 var begin: usize = 0;
217 for (1..len + 1) |n| {
218 const end = begin + n;
219 for (begin..end) |i| try queue.putOne(io, i);
220 for (begin..end) |i| try expect(try queue.getOne(io) == i);
221 begin = end;
222 }
223}
224
225test "Queue" {
226 try testQueue(1);
227 try testQueue(2);
228 try testQueue(3);
229 try testQueue(4);
230 try testQueue(5);
231}