| ... | ... | @@ -1270,17 +1270,19 @@ pub const TypeErasedQueue = struct { |
| 1270 | 1270 | put_index: usize, |
| 1271 | 1271 | get_index: usize, |
| 1272 | 1272 | |
| 1273 | | putters: std.DoublyLinkedList(PutNode), |
| 1274 | | getters: std.DoublyLinkedList(GetNode), |
| 1273 | putters: std.DoublyLinkedList, |
| 1274 | getters: std.DoublyLinkedList, |
| 1275 | 1275 | |
| 1276 | | const PutNode = struct { |
| 1276 | const Put = struct { |
| 1277 | 1277 | remaining: []const u8, |
| 1278 | 1278 | condition: Condition, |
| 1279 | node: std.DoublyLinkedList.Node, |
| 1279 | 1280 | }; |
| 1280 | 1281 | |
| 1281 | | const GetNode = struct { |
| 1282 | const Get = struct { |
| 1282 | 1283 | remaining: []u8, |
| 1283 | 1284 | condition: Condition, |
| 1285 | node: std.DoublyLinkedList.Node, |
| 1284 | 1286 | }; |
| 1285 | 1287 | |
| 1286 | 1288 | pub fn init(buffer: []u8) TypeErasedQueue { |
| ... | ... | @@ -1305,16 +1307,16 @@ pub const TypeErasedQueue = struct { |
| 1305 | 1307 | |
| 1306 | 1308 | var remaining = elements; |
| 1307 | 1309 | while (true) { |
| 1308 | | const getter = q.getters.popFirst() orelse break; |
| 1309 | | const copy_len = @min(getter.data.remaining.len, remaining.len); |
| 1310 | | @memcpy(getter.data.remaining[0..copy_len], remaining[0..copy_len]); |
| 1310 | const getter: *Get = @fieldParentPtr("node", q.getters.popFirst() orelse break); |
| 1311 | const copy_len = @min(getter.remaining.len, remaining.len); |
| 1312 | @memcpy(getter.remaining[0..copy_len], remaining[0..copy_len]); |
| 1311 | 1313 | remaining = remaining[copy_len..]; |
| 1312 | | getter.data.remaining = getter.data.remaining[copy_len..]; |
| 1313 | | if (getter.data.remaining.len == 0) { |
| 1314 | | getter.data.condition.signal(io); |
| 1314 | getter.remaining = getter.remaining[copy_len..]; |
| 1315 | if (getter.remaining.len == 0) { |
| 1316 | getter.condition.signal(io); |
| 1315 | 1317 | continue; |
| 1316 | 1318 | } |
| 1317 | | q.getters.prepend(getter); |
| 1319 | q.getters.prepend(&getter.node); |
| 1318 | 1320 | assert(remaining.len == 0); |
| 1319 | 1321 | return elements.len; |
| 1320 | 1322 | } |
| ... | ... | @@ -1340,12 +1342,10 @@ pub const TypeErasedQueue = struct { |
| 1340 | 1342 | const total_filled = elements.len - remaining.len; |
| 1341 | 1343 | if (total_filled >= min) return total_filled; |
| 1342 | 1344 | |
| 1343 | | var node: std.DoublyLinkedList(PutNode).Node = .{ |
| 1344 | | .data = .{ .remaining = remaining, .condition = .{} }, |
| 1345 | | }; |
| 1346 | | q.putters.append(&node); |
| 1347 | | try node.data.condition.wait(io, &q.mutex); |
| 1348 | | remaining = node.data.remaining; |
| 1345 | var pending: Put = .{ .remaining = remaining, .condition = .{}, .node = .{} }; |
| 1346 | q.putters.append(&pending.node); |
| 1347 | try pending.condition.wait(io, &q.mutex); |
| 1348 | remaining = pending.remaining; |
| 1349 | 1349 | } |
| 1350 | 1350 | } |
| 1351 | 1351 | |
| ... | ... | @@ -1388,16 +1388,16 @@ pub const TypeErasedQueue = struct { |
| 1388 | 1388 | } |
| 1389 | 1389 | // Copy directly from putters into buffer. |
| 1390 | 1390 | while (remaining.len > 0) { |
| 1391 | | const putter = q.putters.popFirst() orelse break; |
| 1392 | | const copy_len = @min(putter.data.remaining.len, remaining.len); |
| 1393 | | @memcpy(remaining[0..copy_len], putter.data.remaining[0..copy_len]); |
| 1394 | | putter.data.remaining = putter.data.remaining[copy_len..]; |
| 1391 | const putter: *Put = @fieldParentPtr("node", q.putters.popFirst() orelse break); |
| 1392 | const copy_len = @min(putter.remaining.len, remaining.len); |
| 1393 | @memcpy(remaining[0..copy_len], putter.remaining[0..copy_len]); |
| 1394 | putter.remaining = putter.remaining[copy_len..]; |
| 1395 | 1395 | remaining = remaining[copy_len..]; |
| 1396 | | if (putter.data.remaining.len == 0) { |
| 1397 | | putter.data.condition.signal(io); |
| 1396 | if (putter.remaining.len == 0) { |
| 1397 | putter.condition.signal(io); |
| 1398 | 1398 | } else { |
| 1399 | 1399 | assert(remaining.len == 0); |
| 1400 | | q.putters.prepend(putter); |
| 1400 | q.putters.prepend(&putter.node); |
| 1401 | 1401 | return fillRingBufferFromPutters(q, io, buffer.len); |
| 1402 | 1402 | } |
| 1403 | 1403 | } |
| ... | ... | @@ -1405,12 +1405,10 @@ pub const TypeErasedQueue = struct { |
| 1405 | 1405 | const total_filled = buffer.len - remaining.len; |
| 1406 | 1406 | if (total_filled >= min) return total_filled; |
| 1407 | 1407 | |
| 1408 | | var node: std.DoublyLinkedList(GetNode).Node = .{ |
| 1409 | | .data = .{ .remaining = remaining, .condition = .{} }, |
| 1410 | | }; |
| 1411 | | q.getters.append(&node); |
| 1412 | | try node.data.condition.wait(io, &q.mutex); |
| 1413 | | remaining = node.data.remaining; |
| 1408 | var pending: Get = .{ .remaining = remaining, .condition = .{}, .node = .{} }; |
| 1409 | q.getters.append(&pending.node); |
| 1410 | try pending.condition.wait(io, &q.mutex); |
| 1411 | remaining = pending.remaining; |
| 1414 | 1412 | } |
| 1415 | 1413 | } |
| 1416 | 1414 | |
| ... | ... | @@ -1420,26 +1418,26 @@ pub const TypeErasedQueue = struct { |
| 1420 | 1418 | /// buffers been fully copied. |
| 1421 | 1419 | fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io, len: usize) usize { |
| 1422 | 1420 | while (true) { |
| 1423 | | const putter = q.putters.popFirst() orelse return len; |
| 1421 | const putter: *Put = @fieldParentPtr("node", q.putters.popFirst() orelse return len); |
| 1424 | 1422 | const available = q.buffer[q.put_index..]; |
| 1425 | | const copy_len = @min(available.len, putter.data.remaining.len); |
| 1426 | | @memcpy(available[0..copy_len], putter.data.remaining[0..copy_len]); |
| 1427 | | putter.data.remaining = putter.data.remaining[copy_len..]; |
| 1423 | const copy_len = @min(available.len, putter.remaining.len); |
| 1424 | @memcpy(available[0..copy_len], putter.remaining[0..copy_len]); |
| 1425 | putter.remaining = putter.remaining[copy_len..]; |
| 1428 | 1426 | q.put_index += copy_len; |
| 1429 | | if (putter.data.remaining.len == 0) { |
| 1430 | | putter.data.condition.signal(io); |
| 1427 | if (putter.remaining.len == 0) { |
| 1428 | putter.condition.signal(io); |
| 1431 | 1429 | continue; |
| 1432 | 1430 | } |
| 1433 | 1431 | const second_available = q.buffer[0..q.get_index]; |
| 1434 | | const second_copy_len = @min(second_available.len, putter.data.remaining.len); |
| 1435 | | @memcpy(second_available[0..second_copy_len], putter.data.remaining[0..second_copy_len]); |
| 1436 | | putter.data.remaining = putter.data.remaining[copy_len..]; |
| 1432 | const second_copy_len = @min(second_available.len, putter.remaining.len); |
| 1433 | @memcpy(second_available[0..second_copy_len], putter.remaining[0..second_copy_len]); |
| 1434 | putter.remaining = putter.remaining[copy_len..]; |
| 1437 | 1435 | q.put_index = copy_len; |
| 1438 | | if (putter.data.remaining.len == 0) { |
| 1439 | | putter.data.condition.signal(io); |
| 1436 | if (putter.remaining.len == 0) { |
| 1437 | putter.condition.signal(io); |
| 1440 | 1438 | continue; |
| 1441 | 1439 | } |
| 1442 | | q.putters.prepend(putter); |
| 1440 | q.putters.prepend(&putter.node); |
| 1443 | 1441 | return len; |
| 1444 | 1442 | } |
| 1445 | 1443 | } |