authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-15 11:24:26+00:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-21 13:07:04+00:00
log67eed9955005aaa02344b8d566ba140b9f2e0e18
tree1f11358b2a1fd708738a93df52ff18175a7e2c7c
parentfa7e818e144f2fd316fc54492f4699f6e1524738
signaturelock-open Commit is signed but in an unrecognized format.

std.Io.Queue: introduce closure and fix a bug

Queues can now be "closed". A closed queue cannot have more elements appended with `put`, and blocked calls to `put` will immediately unblock having failed to append some elements. Calls to `get` will continue to succeed as long as the queue buffer is non-empty, but will then never block; already-blocked calls to `get` will unblock. All queue get/put operations can now return `error.Closed` to indicate that the queue has been closed. For bulk get/put operations, they may add/receive fewer elements than the minimum requested *if* the queue was closed or the calling task was canceled. In that case, if any elements were already added/received, they are returned first, and successive calls will return `error.Closed` or `error.Canceled`. Also, fix a bug where `Queue.get` could deadlock because it incorrectly blocked until the given buffer was *filled*. Resolves: #30141

5 files changed, 370 insertions(+), 164 deletions(-)

lib/std/Io.zig+221-82
...@@ -701,7 +701,7 @@ pub const VTable = struct {...@@ -701,7 +701,7 @@ pub const VTable = struct {
701 netClose: *const fn (?*anyopaque, handle: net.Socket.Handle) void,701 netClose: *const fn (?*anyopaque, handle: net.Socket.Handle) void,
702 netInterfaceNameResolve: *const fn (?*anyopaque, *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface,702 netInterfaceNameResolve: *const fn (?*anyopaque, *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface,
703 netInterfaceName: *const fn (?*anyopaque, net.Interface) net.Interface.NameError!net.Interface.Name,703 netInterfaceName: *const fn (?*anyopaque, net.Interface) net.Interface.NameError!net.Interface.Name,
704 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) void,704 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,
705};705};
706706
707pub const Cancelable = error{707pub const Cancelable = error{
...@@ -1208,7 +1208,9 @@ pub fn Select(comptime U: type) type {...@@ -1208,7 +1208,9 @@ pub fn Select(comptime U: type) type {
1208 const args_casted: *const Args = @ptrCast(@alignCast(context));1208 const args_casted: *const Args = @ptrCast(@alignCast(context));
1209 const unerased_select: *S = @fieldParentPtr("group", group);1209 const unerased_select: *S = @fieldParentPtr("group", group);
1210 const elem = @unionInit(U, @tagName(field), @call(.auto, function, args_casted.*));1210 const elem = @unionInit(U, @tagName(field), @call(.auto, function, args_casted.*));
1211 unerased_select.queue.putOneUncancelable(unerased_select.io, elem);1211 unerased_select.queue.putOneUncancelable(unerased_select.io, elem) catch |err| switch (err) {
1212 error.Closed => unreachable,
1213 };
1212 }1214 }
1213 };1215 };
1214 _ = @atomicRmw(usize, &s.outstanding, .Add, 1, .monotonic);1216 _ = @atomicRmw(usize, &s.outstanding, .Add, 1, .monotonic);
...@@ -1222,7 +1224,10 @@ pub fn Select(comptime U: type) type {...@@ -1222,7 +1224,10 @@ pub fn Select(comptime U: type) type {
1222 /// Not threadsafe.1224 /// Not threadsafe.
1223 pub fn wait(s: *S) Cancelable!U {1225 pub fn wait(s: *S) Cancelable!U {
1224 s.outstanding -= 1;1226 s.outstanding -= 1;
1225 return s.queue.getOne(s.io);1227 return s.queue.getOne(s.io) catch |err| switch (err) {
1228 error.Canceled => |e| return e,
1229 error.Closed => unreachable,
1230 };
1226 }1231 }
12271232
1228 /// Equivalent to `wait` but requests cancellation on all remaining1233 /// Equivalent to `wait` but requests cancellation on all remaining
...@@ -1569,8 +1574,11 @@ pub const Event = enum(u32) {...@@ -1569,8 +1574,11 @@ pub const Event = enum(u32) {
1569 }1574 }
1570};1575};
15711576
1577pub const QueueClosedError = error{Closed};
1578
1572pub const TypeErasedQueue = struct {1579pub const TypeErasedQueue = struct {
1573 mutex: Mutex,1580 mutex: Mutex,
1581 closed: bool,
15741582
1575 /// Ring buffer. This data is logically *after* queued getters.1583 /// Ring buffer. This data is logically *after* queued getters.
1576 buffer: []u8,1584 buffer: []u8,
...@@ -1582,12 +1590,14 @@ pub const TypeErasedQueue = struct {...@@ -1582,12 +1590,14 @@ pub const TypeErasedQueue = struct {
15821590
1583 const Put = struct {1591 const Put = struct {
1584 remaining: []const u8,1592 remaining: []const u8,
1593 needed: usize,
1585 condition: Condition,1594 condition: Condition,
1586 node: std.DoublyLinkedList.Node,1595 node: std.DoublyLinkedList.Node,
1587 };1596 };
15881597
1589 const Get = struct {1598 const Get = struct {
1590 remaining: []u8,1599 remaining: []u8,
1600 needed: usize,
1591 condition: Condition,1601 condition: Condition,
1592 node: std.DoublyLinkedList.Node,1602 node: std.DoublyLinkedList.Node,
1593 };1603 };
...@@ -1595,6 +1605,7 @@ pub const TypeErasedQueue = struct {...@@ -1595,6 +1605,7 @@ pub const TypeErasedQueue = struct {
1595 pub fn init(buffer: []u8) TypeErasedQueue {1605 pub fn init(buffer: []u8) TypeErasedQueue {
1596 return .{1606 return .{
1597 .mutex = .init,1607 .mutex = .init,
1608 .closed = false,
1598 .buffer = buffer,1609 .buffer = buffer,
1599 .start = 0,1610 .start = 0,
1600 .len = 0,1611 .len = 0,
...@@ -1603,7 +1614,27 @@ pub const TypeErasedQueue = struct {...@@ -1603,7 +1614,27 @@ pub const TypeErasedQueue = struct {
1603 };1614 };
1604 }1615 }
16051616
1606 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) Cancelable!usize {1617 pub fn close(q: *TypeErasedQueue, io: Io) void {
1618 q.mutex.lockUncancelable(io);
1619 defer q.mutex.unlock(io);
1620 q.closed = true;
1621 {
1622 var it = q.getters.first;
1623 while (it) |node| : (it = node.next) {
1624 const getter: *Get = @alignCast(@fieldParentPtr("node", node));
1625 getter.condition.signal(io);
1626 }
1627 }
1628 {
1629 var it = q.putters.first;
1630 while (it) |node| : (it = node.next) {
1631 const putter: *Put = @alignCast(@fieldParentPtr("node", node));
1632 putter.condition.signal(io);
1633 }
1634 }
1635 }
1636
1637 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) (QueueClosedError || Cancelable)!usize {
1607 assert(elements.len >= min);1638 assert(elements.len >= min);
1608 if (elements.len == 0) return 0;1639 if (elements.len == 0) return 0;
1609 try q.mutex.lock(io);1640 try q.mutex.lock(io);
...@@ -1614,13 +1645,14 @@ pub const TypeErasedQueue = struct {...@@ -1614,13 +1645,14 @@ pub const TypeErasedQueue = struct {
1614 /// Same as `put`, except does not introduce a cancelation point.1645 /// Same as `put`, except does not introduce a cancelation point.
1615 ///1646 ///
1616 /// For a description of cancelation and cancelation points, see `Future.cancel`.1647 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1617 pub fn putUncancelable(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) usize {1648 pub fn putUncancelable(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) QueueClosedError!usize {
1618 assert(elements.len >= min);1649 assert(elements.len >= min);
1619 if (elements.len == 0) return 0;1650 if (elements.len == 0) return 0;
1620 q.mutex.lockUncancelable(io);1651 q.mutex.lockUncancelable(io);
1621 defer q.mutex.unlock(io);1652 defer q.mutex.unlock(io);
1622 return q.putLocked(io, elements, min, true) catch |err| switch (err) {1653 return q.putLocked(io, elements, min, true) catch |err| switch (err) {
1623 error.Canceled => unreachable,1654 error.Canceled => unreachable,
1655 error.Closed => |e| return e,
1624 };1656 };
1625 }1657 }
16261658
...@@ -1634,49 +1666,79 @@ pub const TypeErasedQueue = struct {...@@ -1634,49 +1666,79 @@ pub const TypeErasedQueue = struct {
1634 return if (slice.len > 0) slice else null;1666 return if (slice.len > 0) slice else null;
1635 }1667 }
16361668
1637 fn putLocked(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize, uncancelable: bool) Cancelable!usize {1669 fn putLocked(q: *TypeErasedQueue, io: Io, elements: []const u8, target: usize, uncancelable: bool) (QueueClosedError || Cancelable)!usize {
1670 // A closed queue cannot be added to, even if there is space in the buffer.
1671 if (q.closed) return error.Closed;
1672
1638 // Getters have first priority on the data, and only when the getters1673 // Getters have first priority on the data, and only when the getters
1639 // queue is empty do we start populating the buffer.1674 // queue is empty do we start populating the buffer.
16401675
1641 var remaining = elements;1676 // The number of elements we add immediately, before possibly blocking.
1677 var n: usize = 0;
1678
1642 while (q.getters.popFirst()) |getter_node| {1679 while (q.getters.popFirst()) |getter_node| {
1643 const getter: *Get = @alignCast(@fieldParentPtr("node", getter_node));1680 const getter: *Get = @alignCast(@fieldParentPtr("node", getter_node));
1644 const copy_len = @min(getter.remaining.len, remaining.len);1681 const copy_len = @min(getter.remaining.len, elements.len - n);
1645 assert(copy_len > 0);1682 assert(copy_len > 0);
1646 @memcpy(getter.remaining[0..copy_len], remaining[0..copy_len]);1683 @memcpy(getter.remaining[0..copy_len], elements[n..][0..copy_len]);
1647 remaining = remaining[copy_len..];
1648 getter.remaining = getter.remaining[copy_len..];1684 getter.remaining = getter.remaining[copy_len..];
1649 if (getter.remaining.len == 0) {1685 getter.needed -|= copy_len;
1686 n += copy_len;
1687 if (getter.needed == 0) {
1650 getter.condition.signal(io);1688 getter.condition.signal(io);
1651 if (remaining.len > 0) continue;1689 } else {
1652 } else q.getters.prepend(getter_node);1690 assert(n == elements.len); // we didn't have enough elements for the getter
1653 assert(remaining.len == 0);1691 q.getters.prepend(getter_node);
1654 return elements.len;1692 }
1693 if (n == elements.len) return elements.len;
1655 }1694 }
16561695
1657 while (q.puttableSlice()) |slice| {1696 while (q.puttableSlice()) |slice| {
1658 const copy_len = @min(slice.len, remaining.len);1697 const copy_len = @min(slice.len, elements.len - n);
1659 assert(copy_len > 0);1698 assert(copy_len > 0);
1660 @memcpy(slice[0..copy_len], remaining[0..copy_len]);1699 @memcpy(slice[0..copy_len], elements[n..][0..copy_len]);
1661 q.len += copy_len;1700 q.len += copy_len;
1662 remaining = remaining[copy_len..];1701 n += copy_len;
1663 if (remaining.len == 0) return elements.len;1702 if (n == elements.len) return elements.len;
1664 }1703 }
16651704
1666 const total_filled = elements.len - remaining.len;1705 // Don't block if we hit the target.
1667 if (total_filled >= min) return total_filled;1706 if (n >= target) return n;
16681707
1669 var pending: Put = .{ .remaining = remaining, .condition = .{}, .node = .{} };1708 var pending: Put = .{
1709 .remaining = elements[n..],
1710 .needed = target - n,
1711 .condition = .init,
1712 .node = .{},
1713 };
1670 q.putters.append(&pending.node);1714 q.putters.append(&pending.node);
1671 defer if (pending.remaining.len > 0) q.putters.remove(&pending.node);1715 defer if (pending.needed > 0) q.putters.remove(&pending.node);
1672 while (pending.remaining.len > 0) if (uncancelable)1716
1673 pending.condition.waitUncancelable(io, &q.mutex)1717 while (pending.needed > 0 and !q.closed) {
1674 else1718 if (uncancelable) {
1675 try pending.condition.wait(io, &q.mutex);1719 pending.condition.waitUncancelable(io, &q.mutex);
1676 return elements.len;1720 continue;
1721 }
1722 pending.condition.wait(io, &q.mutex) catch |err| switch (err) {
1723 error.Canceled => if (pending.remaining.len == elements.len) {
1724 // Canceled while waiting, and appended no elements.
1725 return error.Canceled;
1726 } else {
1727 // Canceled while waiting, but appended some elements, so report those first.
1728 io.recancel();
1729 return elements.len - pending.remaining.len;
1730 },
1731 };
1732 }
1733 if (pending.remaining.len == elements.len) {
1734 // The queue was closed while we were waiting. We appended no elements.
1735 assert(q.closed);
1736 return error.Closed;
1737 }
1738 return elements.len - pending.remaining.len;
1677 }1739 }
16781740
1679 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) Cancelable!usize {1741 pub fn get(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize) (QueueClosedError || Cancelable)!usize {
1680 assert(buffer.len >= min);1742 assert(buffer.len >= min);
1681 if (buffer.len == 0) return 0;1743 if (buffer.len == 0) return 0;
1682 try q.mutex.lock(io);1744 try q.mutex.lock(io);
...@@ -1687,13 +1749,14 @@ pub const TypeErasedQueue = struct {...@@ -1687,13 +1749,14 @@ pub const TypeErasedQueue = struct {
1687 /// Same as `get`, except does not introduce a cancelation point.1749 /// Same as `get`, except does not introduce a cancelation point.
1688 ///1750 ///
1689 /// For a description of cancelation and cancelation points, see `Future.cancel`.1751 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1690 pub fn getUncancelable(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize) usize {1752 pub fn getUncancelable(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize) QueueClosedError!usize {
1691 assert(buffer.len >= min);1753 assert(buffer.len >= min);
1692 if (buffer.len == 0) return 0;1754 if (buffer.len == 0) return 0;
1693 q.mutex.lockUncancelable(io);1755 q.mutex.lockUncancelable(io);
1694 defer q.mutex.unlock(io);1756 defer q.mutex.unlock(io);
1695 return q.getLocked(io, buffer, min, true) catch |err| switch (err) {1757 return q.getLocked(io, buffer, min, true) catch |err| switch (err) {
1696 error.Canceled => unreachable,1758 error.Canceled => unreachable,
1759 error.Closed => |e| return e,
1697 };1760 };
1698 }1761 }
16991762
...@@ -1703,21 +1766,23 @@ pub const TypeErasedQueue = struct {...@@ -1703,21 +1766,23 @@ pub const TypeErasedQueue = struct {
1703 return if (slice.len > 0) slice else null;1766 return if (slice.len > 0) slice else null;
1704 }1767 }
17051768
1706 fn getLocked(q: *@This(), io: Io, buffer: []u8, min: usize, uncancelable: bool) Cancelable!usize {1769 fn getLocked(q: *TypeErasedQueue, io: Io, buffer: []u8, target: usize, uncancelable: bool) (QueueClosedError || Cancelable)!usize {
1707 // The ring buffer gets first priority, then data should come from any1770 // The ring buffer gets first priority, then data should come from any
1708 // queued putters, then finally the ring buffer should be filled with1771 // queued putters, then finally the ring buffer should be filled with
1709 // data from putters so they can be resumed.1772 // data from putters so they can be resumed.
17101773
1711 var remaining = buffer;1774 // The number of elements we received immediately, before possibly blocking.
1775 var n: usize = 0;
1776
1712 while (q.gettableSlice()) |slice| {1777 while (q.gettableSlice()) |slice| {
1713 const copy_len = @min(slice.len, remaining.len);1778 const copy_len = @min(slice.len, buffer.len - n);
1714 assert(copy_len > 0);1779 assert(copy_len > 0);
1715 @memcpy(remaining[0..copy_len], slice[0..copy_len]);1780 @memcpy(buffer[n..][0..copy_len], slice[0..copy_len]);
1716 q.start += copy_len;1781 q.start += copy_len;
1717 if (q.buffer.len - q.start == 0) q.start = 0;1782 if (q.buffer.len - q.start == 0) q.start = 0;
1718 q.len -= copy_len;1783 q.len -= copy_len;
1719 remaining = remaining[copy_len..];1784 n += copy_len;
1720 if (remaining.len == 0) {1785 if (n == buffer.len) {
1721 q.fillRingBufferFromPutters(io);1786 q.fillRingBufferFromPutters(io);
1722 return buffer.len;1787 return buffer.len;
1723 }1788 }
...@@ -1726,33 +1791,64 @@ pub const TypeErasedQueue = struct {...@@ -1726,33 +1791,64 @@ pub const TypeErasedQueue = struct {
1726 // Copy directly from putters into buffer.1791 // Copy directly from putters into buffer.
1727 while (q.putters.popFirst()) |putter_node| {1792 while (q.putters.popFirst()) |putter_node| {
1728 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));1793 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));
1729 const copy_len = @min(putter.remaining.len, remaining.len);1794 const copy_len = @min(putter.remaining.len, buffer.len - n);
1730 assert(copy_len > 0);1795 assert(copy_len > 0);
1731 @memcpy(remaining[0..copy_len], putter.remaining[0..copy_len]);1796 @memcpy(buffer[n..][0..copy_len], putter.remaining[0..copy_len]);
1732 putter.remaining = putter.remaining[copy_len..];1797 putter.remaining = putter.remaining[copy_len..];
1733 remaining = remaining[copy_len..];1798 putter.needed -|= copy_len;
1734 if (putter.remaining.len == 0) {1799 n += copy_len;
1800 if (putter.needed == 0) {
1735 putter.condition.signal(io);1801 putter.condition.signal(io);
1736 if (remaining.len > 0) continue;1802 } else {
1737 } else q.putters.prepend(putter_node);1803 assert(n == buffer.len); // we didn't have enough space for the putter
1738 assert(remaining.len == 0);1804 q.putters.prepend(putter_node);
1739 q.fillRingBufferFromPutters(io);1805 }
1740 return buffer.len;1806 if (n == buffer.len) {
1807 q.fillRingBufferFromPutters(io);
1808 return buffer.len;
1809 }
1741 }1810 }
17421811
1743 // Both ring buffer and putters queue is empty.1812 // No need to call `fillRingBufferFromPutters` from this point onwards,
1744 const total_filled = buffer.len - remaining.len;1813 // because we emptied the ring buffer *and* the putter queue!
1745 if (total_filled >= min) return total_filled;
17461814
1747 var pending: Get = .{ .remaining = remaining, .condition = .{}, .node = .{} };1815 // Don't block if we hit the target or if the queue is closed. Return how
1816 // many elements we could get immediately, unless the queue was closed and
1817 // empty, in which case report `error.Closed`.
1818 if (n == 0 and q.closed) return error.Closed;
1819 if (n >= target or q.closed) return n;
1820
1821 var pending: Get = .{
1822 .remaining = buffer[n..],
1823 .needed = target - n,
1824 .condition = .init,
1825 .node = .{},
1826 };
1748 q.getters.append(&pending.node);1827 q.getters.append(&pending.node);
1749 defer if (pending.remaining.len > 0) q.getters.remove(&pending.node);1828 defer if (pending.needed > 0) q.getters.remove(&pending.node);
1750 while (pending.remaining.len > 0) if (uncancelable)1829
1751 pending.condition.waitUncancelable(io, &q.mutex)1830 while (pending.needed > 0 and !q.closed) {
1752 else1831 if (uncancelable) {
1753 try pending.condition.wait(io, &q.mutex);1832 pending.condition.waitUncancelable(io, &q.mutex);
1754 q.fillRingBufferFromPutters(io);1833 continue;
1755 return buffer.len;1834 }
1835 pending.condition.wait(io, &q.mutex) catch |err| switch (err) {
1836 error.Canceled => if (pending.remaining.len == buffer.len) {
1837 // Canceled while waiting, and received no elements.
1838 return error.Canceled;
1839 } else {
1840 // Canceled while waiting, but received some elements, so report those first.
1841 io.recancel();
1842 return buffer.len - pending.remaining.len;
1843 },
1844 };
1845 }
1846 if (pending.remaining.len == buffer.len) {
1847 // The queue was closed while we were waiting. We received no elements.
1848 assert(q.closed);
1849 return error.Closed;
1850 }
1851 return buffer.len - pending.remaining.len;
1756 }1852 }
17571853
1758 /// Called when there is nonzero space available in the ring buffer and1854 /// Called when there is nonzero space available in the ring buffer and
...@@ -1768,7 +1864,8 @@ pub const TypeErasedQueue = struct {...@@ -1768,7 +1864,8 @@ pub const TypeErasedQueue = struct {
1768 @memcpy(slice[0..copy_len], putter.remaining[0..copy_len]);1864 @memcpy(slice[0..copy_len], putter.remaining[0..copy_len]);
1769 q.len += copy_len;1865 q.len += copy_len;
1770 putter.remaining = putter.remaining[copy_len..];1866 putter.remaining = putter.remaining[copy_len..];
1771 if (putter.remaining.len == 0) {1867 putter.needed -|= copy_len;
1868 if (putter.needed == 0) {
1772 putter.condition.signal(io);1869 putter.condition.signal(io);
1773 break;1870 break;
1774 }1871 }
...@@ -1791,59 +1888,101 @@ pub fn Queue(Elem: type) type {...@@ -1791,59 +1888,101 @@ pub fn Queue(Elem: type) type {
1791 return .{ .type_erased = .init(@ptrCast(buffer)) };1888 return .{ .type_erased = .init(@ptrCast(buffer)) };
1792 }1889 }
17931890
1794 /// Appends elements to the end of the queue. The function returns when1891 pub fn close(q: *@This(), io: Io) void {
1795 /// at least `min` elements have been added to the buffer or sent1892 q.type_erased.close(io);
1796 /// directly to a consumer.1893 }
1894
1895 /// Appends elements to the end of the queue, potentially blocking if
1896 /// there is insufficient capacity. Returns when any one of the
1897 /// following conditions is satisfied:
1898 ///
1899 /// * At least `target` elements have been added to the queue
1900 /// * The queue is closed
1901 /// * The current task is canceled
1902 ///
1903 /// Returns how many of `elements` have been added to the queue, if any.
1904 /// If an error is returned, no elements have been added.
1797 ///1905 ///
1798 /// Returns how many elements have been added to the queue.1906 /// If the queue is closed or the task is canceled, but some items were
1907 /// already added before the closure or cancelation, then `put` may
1908 /// return a number lower than `target`, in which case future calls are
1909 /// guaranteed to return `error.Canceled` or `error.Closed`.
1799 ///1910 ///
1800 /// Asserts that `elements.len >= min`.1911 /// A return value of 0 is only possible if `target` is 0, in which case
1801 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) Cancelable!usize {1912 /// the call is guaranteed to queue as many of `elements` as is possible
1802 return @divExact(try q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));1913 /// *without* blocking.
1914 ///
1915 /// Asserts that `elements.len >= target`.
1916 pub fn put(q: *@This(), io: Io, elements: []const Elem, target: usize) (QueueClosedError || Cancelable)!usize {
1917 return @divExact(try q.type_erased.put(io, @ptrCast(elements), target * @sizeOf(Elem)), @sizeOf(Elem));
1803 }1918 }
18041919
1805 /// Same as `put` but blocks until all elements have been added to the queue.1920 /// Same as `put` but blocks until all elements have been added to the queue.
1806 pub fn putAll(q: *@This(), io: Io, elements: []const Elem) Cancelable!void {1921 ///
1807 assert(try q.put(io, elements, elements.len) == elements.len);1922 /// If the queue is closed or canceled, `error.Closed` or `error.Canceled`
1923 /// is returned, and it is unspecified how many, if any, of `elements` were
1924 /// added to the queue prior to cancelation or closure.
1925 pub fn putAll(q: *@This(), io: Io, elements: []const Elem) (QueueClosedError || Cancelable)!void {
1926 const n = try q.put(io, elements, elements.len);
1927 if (n != elements.len) {
1928 _ = try q.put(io, elements[n..], elements.len - n);
1929 unreachable; // partial `put` implies queue was closed or we were canceled
1930 }
1808 }1931 }
18091932
1810 /// Same as `put`, except does not introduce a cancelation point.1933 /// Same as `put`, except does not introduce a cancelation point.
1811 ///1934 ///
1812 /// For a description of cancelation and cancelation points, see `Future.cancel`.1935 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1813 pub fn putUncancelable(q: *@This(), io: Io, elements: []const Elem, min: usize) usize {1936 pub fn putUncancelable(q: *@This(), io: Io, elements: []const Elem, min: usize) QueueClosedError!usize {
1814 return @divExact(q.type_erased.putUncancelable(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));1937 return @divExact(try q.type_erased.putUncancelable(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
1815 }1938 }
18161939
1817 pub fn putOne(q: *@This(), io: Io, item: Elem) Cancelable!void {1940 /// Appends `item` to the end of the queue, blocking if the queue is full.
1941 pub fn putOne(q: *@This(), io: Io, item: Elem) (QueueClosedError || Cancelable)!void {
1818 assert(try q.put(io, &.{item}, 1) == 1);1942 assert(try q.put(io, &.{item}, 1) == 1);
1819 }1943 }
18201944
1821 /// Same as `putOne`, except does not introduce a cancelation point.1945 /// Same as `putOne`, except does not introduce a cancelation point.
1822 ///1946 ///
1823 /// For a description of cancelation and cancelation points, see `Future.cancel`.1947 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1824 pub fn putOneUncancelable(q: *@This(), io: Io, item: Elem) void {1948 pub fn putOneUncancelable(q: *@This(), io: Io, item: Elem) QueueClosedError!void {
1825 assert(q.putUncancelable(io, &.{item}, 1) == 1);1949 assert(try q.putUncancelable(io, &.{item}, 1) == 1);
1826 }1950 }
18271951
1828 /// Receives elements from the beginning of the queue. The function1952 /// Receives elements from the beginning of the queue, potentially blocking
1829 /// returns when at least `min` elements have been populated inside1953 /// if there are insufficient elements currently in the queue. Returns when
1830 /// `buffer`.1954 /// any one of the following conditions is satisfied:
1955 ///
1956 /// * At least `target` elements have been received from the queue
1957 /// * The queue is closed and contains no buffered elements
1958 /// * The current task is canceled
1959 ///
1960 /// Returns how many elements of `buffer` have been populated, if any.
1961 /// If an error is returned, no elements have been populated.
1962 ///
1963 /// If the queue is closed or the task is canceled, but some items were
1964 /// already received before the closure or cancelation, then `get` may
1965 /// return a number lower than `target`, in which case future calls are
1966 /// guaranteed to return `error.Canceled` or `error.Closed`.
1831 ///1967 ///
1832 /// Returns how many elements of `buffer` have been populated.1968 /// A return value of 0 is only possible if `target` is 0, in which case
1969 /// the call is guaranteed to fill as much of `buffer` as is possible
1970 /// *without* blocking.
1833 ///1971 ///
1834 /// Asserts that `buffer.len >= min`.1972 /// Asserts that `buffer.len >= target`.
1835 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) Cancelable!usize {1973 pub fn get(q: *@This(), io: Io, buffer: []Elem, target: usize) (QueueClosedError || Cancelable)!usize {
1836 return @divExact(try q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));1974 return @divExact(try q.type_erased.get(io, @ptrCast(buffer), target * @sizeOf(Elem)), @sizeOf(Elem));
1837 }1975 }
18381976
1839 /// Same as `get`, except does not introduce a cancelation point.1977 /// Same as `get`, except does not introduce a cancelation point.
1840 ///1978 ///
1841 /// For a description of cancelation and cancelation points, see `Future.cancel`.1979 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1842 pub fn getUncancelable(q: *@This(), io: Io, buffer: []Elem, min: usize) usize {1980 pub fn getUncancelable(q: *@This(), io: Io, buffer: []Elem, min: usize) QueueClosedError!usize {
1843 return @divExact(try q.type_erased.getUncancelable(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));1981 return @divExact(try q.type_erased.getUncancelable(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
1844 }1982 }
18451983
1846 pub fn getOne(q: *@This(), io: Io) Cancelable!Elem {1984 /// Receives one element from the beginning of the queue, blocking if the queue is empty.
1985 pub fn getOne(q: *@This(), io: Io) (QueueClosedError || Cancelable)!Elem {
1847 var buf: [1]Elem = undefined;1986 var buf: [1]Elem = undefined;
1848 assert(try q.get(io, &buf, 1) == 1);1987 assert(try q.get(io, &buf, 1) == 1);
1849 return buf[0];1988 return buf[0];
...@@ -1852,9 +1991,9 @@ pub fn Queue(Elem: type) type {...@@ -1852,9 +1991,9 @@ pub fn Queue(Elem: type) type {
1852 /// Same as `getOne`, except does not introduce a cancelation point.1991 /// Same as `getOne`, except does not introduce a cancelation point.
1853 ///1992 ///
1854 /// For a description of cancelation and cancelation points, see `Future.cancel`.1993 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1855 pub fn getOneUncancelable(q: *@This(), io: Io) Elem {1994 pub fn getOneUncancelable(q: *@This(), io: Io) QueueClosedError!Elem {
1856 var buf: [1]Elem = undefined;1995 var buf: [1]Elem = undefined;
1857 assert(q.getUncancelable(io, &buf, 1) == 1);1996 assert(try q.getUncancelable(io, &buf, 1) == 1);
1858 return buf[0];1997 return buf[0];
1859 }1998 }
18601999
lib/std/Io/Threaded.zig+19-14
...@@ -5795,11 +5795,13 @@ fn netLookup(...@@ -5795,11 +5795,13 @@ fn netLookup(
5795 host_name: HostName,5795 host_name: HostName,
5796 resolved: *Io.Queue(HostName.LookupResult),5796 resolved: *Io.Queue(HostName.LookupResult),
5797 options: HostName.LookupOptions,5797 options: HostName.LookupOptions,
5798) void {5798) net.HostName.LookupError!void {
5799 const t: *Threaded = @ptrCast(@alignCast(userdata));5799 const t: *Threaded = @ptrCast(@alignCast(userdata));
5800 const current_thread = Thread.getCurrent(t);5800 defer resolved.close(io(t));
5801 const t_io = io(t);5801 netLookupFallible(t, host_name, resolved, options) catch |err| switch (err) {
5802 resolved.putOneUncancelable(t_io, .{ .end = netLookupFallible(t, current_thread, host_name, resolved, options) });5802 error.Closed => unreachable, // `resolved` must not be closed until `netLookup` returns
5803 else => |e| return e,
5804 };
5803}5805}
58045806
5805fn netLookupUnavailable(5807fn netLookupUnavailable(
...@@ -5807,22 +5809,23 @@ fn netLookupUnavailable(...@@ -5807,22 +5809,23 @@ fn netLookupUnavailable(
5807 host_name: HostName,5809 host_name: HostName,
5808 resolved: *Io.Queue(HostName.LookupResult),5810 resolved: *Io.Queue(HostName.LookupResult),
5809 options: HostName.LookupOptions,5811 options: HostName.LookupOptions,
5810) void {5812) net.HostName.LookupError!void {
5811 _ = host_name;5813 _ = host_name;
5812 _ = options;5814 _ = options;
5813 const t: *Threaded = @ptrCast(@alignCast(userdata));5815 const t: *Threaded = @ptrCast(@alignCast(userdata));
5814 const t_io = ioBasic(t);5816 resolved.close(ioBasic(t));
5815 resolved.putOneUncancelable(t_io, .{ .end = error.NetworkDown });5817 return error.NetworkDown;
5816}5818}
58175819
5818fn netLookupFallible(5820fn netLookupFallible(
5819 t: *Threaded,5821 t: *Threaded,
5820 current_thread: *Thread,
5821 host_name: HostName,5822 host_name: HostName,
5822 resolved: *Io.Queue(HostName.LookupResult),5823 resolved: *Io.Queue(HostName.LookupResult),
5823 options: HostName.LookupOptions,5824 options: HostName.LookupOptions,
5824) !void {5825) (net.HostName.LookupError || Io.QueueClosedError)!void {
5825 if (!have_networking) return error.NetworkDown;5826 if (!have_networking) return error.NetworkDown;
5827
5828 const current_thread: *Thread = .getCurrent(t);
5826 const t_io = io(t);5829 const t_io = io(t);
5827 const name = host_name.bytes;5830 const name = host_name.bytes;
5828 assert(name.len <= HostName.max_len);5831 assert(name.len <= HostName.max_len);
...@@ -6363,7 +6366,7 @@ fn lookupDnsSearch(...@@ -6363,7 +6366,7 @@ fn lookupDnsSearch(
6363 host_name: HostName,6366 host_name: HostName,
6364 resolved: *Io.Queue(HostName.LookupResult),6367 resolved: *Io.Queue(HostName.LookupResult),
6365 options: HostName.LookupOptions,6368 options: HostName.LookupOptions,
6366) HostName.LookupError!void {6369) (HostName.LookupError || Io.QueueClosedError)!void {
6367 const t_io = io(t);6370 const t_io = io(t);
6368 const rc = HostName.ResolvConf.init(t_io) catch return error.ResolvConfParseFailed;6371 const rc = HostName.ResolvConf.init(t_io) catch return error.ResolvConfParseFailed;
63696372
...@@ -6407,7 +6410,7 @@ fn lookupDns(...@@ -6407,7 +6410,7 @@ fn lookupDns(
6407 rc: *const HostName.ResolvConf,6410 rc: *const HostName.ResolvConf,
6408 resolved: *Io.Queue(HostName.LookupResult),6411 resolved: *Io.Queue(HostName.LookupResult),
6409 options: HostName.LookupOptions,6412 options: HostName.LookupOptions,
6410) HostName.LookupError!void {6413) (HostName.LookupError || Io.QueueClosedError)!void {
6411 const t_io = io(t);6414 const t_io = io(t);
6412 const family_records: [2]struct { af: IpAddress.Family, rr: HostName.DnsRecord } = .{6415 const family_records: [2]struct { af: IpAddress.Family, rr: HostName.DnsRecord } = .{
6413 .{ .af = .ip6, .rr = .A },6416 .{ .af = .ip6, .rr = .A },
...@@ -6621,8 +6624,10 @@ fn lookupHosts(...@@ -6621,8 +6624,10 @@ fn lookupHosts(
6621 return error.DetectingNetworkConfigurationFailed;6624 return error.DetectingNetworkConfigurationFailed;
6622 },6625 },
6623 },6626 },
6624 error.Canceled => |e| return e,6627 error.Canceled,
6625 error.UnknownHostName => |e| return e,6628 error.Closed,
6629 error.UnknownHostName,
6630 => |e| return e,
6626 };6631 };
6627}6632}
66286633
...@@ -6632,7 +6637,7 @@ fn lookupHostsReader(...@@ -6632,7 +6637,7 @@ fn lookupHostsReader(
6632 resolved: *Io.Queue(HostName.LookupResult),6637 resolved: *Io.Queue(HostName.LookupResult),
6633 options: HostName.LookupOptions,6638 options: HostName.LookupOptions,
6634 reader: *Io.Reader,6639 reader: *Io.Reader,
6635) error{ ReadFailed, Canceled, UnknownHostName }!void {6640) error{ ReadFailed, Canceled, UnknownHostName, Closed }!void {
6636 const t_io = io(t);6641 const t_io = io(t);
6637 var addresses_len: usize = 0;6642 var addresses_len: usize = 0;
6638 var canonical_name: ?HostName = null;6643 var canonical_name: ?HostName = null;
lib/std/Io/net/HostName.zig+63-44
...@@ -82,19 +82,22 @@ pub const LookupError = error{...@@ -82,19 +82,22 @@ pub const LookupError = error{
82pub const LookupResult = union(enum) {82pub const LookupResult = union(enum) {
83 address: IpAddress,83 address: IpAddress,
84 canonical_name: HostName,84 canonical_name: HostName,
85 end: LookupError!void,
86};85};
8786
88/// Adds any number of `IpAddress` into resolved, exactly one canonical_name,87/// Adds any number of `LookupResult.address` into `resolved`, and exactly one
89/// and then always finishes by adding one `LookupResult.end` entry.88/// `LookupResult.canonical_name`.
90///89///
91/// Guaranteed not to block if provided queue has capacity at least 16.90/// Guaranteed not to block if provided queue has capacity at least 16.
91///
92/// Closes `resolved` before return, even on error.
93///
94/// Asserts `resolved` is not closed until this call returns.
92pub fn lookup(95pub fn lookup(
93 host_name: HostName,96 host_name: HostName,
94 io: Io,97 io: Io,
95 resolved: *Io.Queue(LookupResult),98 resolved: *Io.Queue(LookupResult),
96 options: LookupOptions,99 options: LookupOptions,
97) void {100) LookupError!void {
98 return io.vtable.netLookup(io.userdata, host_name, resolved, options);101 return io.vtable.netLookup(io.userdata, host_name, resolved, options);
99}102}
100103
...@@ -211,23 +214,25 @@ pub fn connect(...@@ -211,23 +214,25 @@ pub fn connect(
211 port: u16,214 port: u16,
212 options: IpAddress.ConnectOptions,215 options: IpAddress.ConnectOptions,
213) ConnectError!Stream {216) ConnectError!Stream {
214 var connect_many_buffer: [32]ConnectManyResult = undefined;217 var connect_many_buffer: [32]IpAddress.ConnectError!Stream = undefined;
215 var connect_many_queue: Io.Queue(ConnectManyResult) = .init(&connect_many_buffer);218 var connect_many_queue: Io.Queue(IpAddress.ConnectError!Stream) = .init(&connect_many_buffer);
216219
217 var connect_many = io.async(connectMany, .{ host_name, io, port, &connect_many_queue, options });220 var connect_many = io.async(connectMany, .{ host_name, io, port, &connect_many_queue, options });
218 var saw_end = false;
219 defer {221 defer {
220 connect_many.cancel(io);222 connect_many.cancel(io) catch {};
221 if (!saw_end) while (true) switch (connect_many_queue.getOneUncancelable(io)) {223 while (connect_many_queue.getOneUncancelable(io)) |loser| {
222 .connection => |loser| if (loser) |s| s.close(io) else |_| continue,224 if (loser) |s| s.close(io) else |_| {}
223 .end => break,225 } else |err| switch (err) {
224 };226 error.Closed => {},
227 }
225 }228 }
226229
227 var aggregate_error: ConnectError = error.UnknownHostName;230 var ip_connect_error: ?IpAddress.ConnectError = null;
228231
229 while (connect_many_queue.getOne(io)) |result| switch (result) {232 while (connect_many_queue.getOne(io)) |result| {
230 .connection => |connection| if (connection) |stream| return stream else |err| switch (err) {233 if (result) |stream| {
234 return stream;
235 } else |err| switch (err) {
231 error.SystemResources,236 error.SystemResources,
232 error.OptionUnsupported,237 error.OptionUnsupported,
233 error.ProcessFdQuotaExceeded,238 error.ProcessFdQuotaExceeded,
...@@ -237,66 +242,80 @@ pub fn connect(...@@ -237,66 +242,80 @@ pub fn connect(
237242
238 error.WouldBlock => return error.Unexpected,243 error.WouldBlock => return error.Unexpected,
239244
240 else => |e| aggregate_error = e,245 else => |e| ip_connect_error = e,
241 },246 }
242 .end => |end| {
243 saw_end = true;
244 try end;
245 return aggregate_error;
246 },
247 } else |err| switch (err) {247 } else |err| switch (err) {
248 error.Canceled => |e| return e,248 error.Canceled => |e| return e,
249 error.Closed => {
250 // There was no successful connection attempt. If there was a lookup error, return that.
251 try connect_many.await(io);
252 // Otherwise, return the error from a failed IP connection attempt.
253 return ip_connect_error orelse
254 return error.UnknownHostName;
255 },
249 }256 }
250}257}
251258
252pub const ConnectManyResult = union(enum) {
253 connection: IpAddress.ConnectError!Stream,
254 end: ConnectError!void,
255};
256
257/// Asynchronously establishes a connection to all IP addresses associated with259/// Asynchronously establishes a connection to all IP addresses associated with
258/// a host name, adding them to a results queue upon completion.260/// a host name, adding them to a results queue upon completion.
261///
262/// Closes `results` before return, even on error.
263///
264/// Asserts `results` is not closed until this call returns.
259pub fn connectMany(265pub fn connectMany(
260 host_name: HostName,266 host_name: HostName,
261 io: Io,267 io: Io,
262 port: u16,268 port: u16,
263 results: *Io.Queue(ConnectManyResult),269 results: *Io.Queue(IpAddress.ConnectError!Stream),
264 options: IpAddress.ConnectOptions,270 options: IpAddress.ConnectOptions,
265) void {271) LookupError!void {
272 defer results.close(io);
273
266 var canonical_name_buffer: [max_len]u8 = undefined;274 var canonical_name_buffer: [max_len]u8 = undefined;
267 var lookup_buffer: [32]HostName.LookupResult = undefined;275 var lookup_buffer: [32]HostName.LookupResult = undefined;
268 var lookup_queue: Io.Queue(LookupResult) = .init(&lookup_buffer);276 var lookup_queue: Io.Queue(LookupResult) = .init(&lookup_buffer);
269 var group: Io.Group = .init;277 var lookup_future = io.async(lookup, .{ host_name, io, &lookup_queue, .{
270 defer group.cancel(io);
271
272 group.async(io, lookup, .{ host_name, io, &lookup_queue, .{
273 .port = port,278 .port = port,
274 .canonical_name_buffer = &canonical_name_buffer,279 .canonical_name_buffer = &canonical_name_buffer,
275 } });280 } });
281 defer lookup_future.cancel(io) catch {};
282
283 var group: Io.Group = .init;
284 defer group.cancel(io);
276285
277 while (lookup_queue.getOne(io)) |dns_result| switch (dns_result) {286 while (lookup_queue.getOne(io)) |dns_result| switch (dns_result) {
278 .address => |address| group.async(io, enqueueConnection, .{ address, io, results, options }),287 .address => |address| group.async(io, enqueueConnection, .{ address, io, results, options }),
279 .canonical_name => continue,288 .canonical_name => continue,
280 .end => |lookup_result| {
281 group.wait(io);
282 results.putOneUncancelable(io, .{ .end = lookup_result });
283 return;
284 },
285 } else |err| switch (err) {289 } else |err| switch (err) {
286 error.Canceled => |e| {290 error.Canceled => |e| return e,
287 group.cancel(io);291 error.Closed => {
288 results.putOneUncancelable(io, .{ .end = e });292 group.wait(io);
293 return lookup_future.await(io);
289 },294 },
290 }295 }
291}296}
292
293fn enqueueConnection(297fn enqueueConnection(
294 address: IpAddress,298 address: IpAddress,
295 io: Io,299 io: Io,
296 queue: *Io.Queue(ConnectManyResult),300 queue: *Io.Queue(IpAddress.ConnectError!Stream),
297 options: IpAddress.ConnectOptions,301 options: IpAddress.ConnectOptions,
298) void {302) void {
299 queue.putOneUncancelable(io, .{ .connection = address.connect(io, options) });303 enqueueConnectionFallible(address, io, queue, options) catch |err| switch (err) {
304 error.Canceled => {},
305 };
306}
307fn enqueueConnectionFallible(
308 address: IpAddress,
309 io: Io,
310 queue: *Io.Queue(IpAddress.ConnectError!Stream),
311 options: IpAddress.ConnectOptions,
312) Io.Cancelable!void {
313 const result = address.connect(io, options);
314 errdefer if (result) |s| s.close(io) else |_| {};
315 queue.putOne(io, result) catch |err| switch (err) {
316 error.Closed => unreachable, // `queue` must not be closed
317 error.Canceled => |e| return e,
318 };
300}319}
301320
302pub const ResolvConf = struct {321pub const ResolvConf = struct {
lib/std/Io/net/test.zig+14-16
...@@ -129,7 +129,7 @@ test "resolve DNS" {...@@ -129,7 +129,7 @@ test "resolve DNS" {
129 var results_buffer: [32]net.HostName.LookupResult = undefined;129 var results_buffer: [32]net.HostName.LookupResult = undefined;
130 var results: Io.Queue(net.HostName.LookupResult) = .init(&results_buffer);130 var results: Io.Queue(net.HostName.LookupResult) = .init(&results_buffer);
131131
132 net.HostName.lookup(try .init("localhost"), io, &results, .{132 try net.HostName.lookup(try .init("localhost"), io, &results, .{
133 .port = 80,133 .port = 80,
134 .canonical_name_buffer = &canonical_name_buffer,134 .canonical_name_buffer = &canonical_name_buffer,
135 });135 });
...@@ -142,11 +142,10 @@ test "resolve DNS" {...@@ -142,11 +142,10 @@ test "resolve DNS" {
142 addresses_found += 1;142 addresses_found += 1;
143 },143 },
144 .canonical_name => |canonical_name| try testing.expectEqualStrings("localhost", canonical_name.bytes),144 .canonical_name => |canonical_name| try testing.expectEqualStrings("localhost", canonical_name.bytes),
145 .end => |end| {145 } else |err| switch (err) {
146 try end;146 error.Closed => {},
147 break;147 error.Canceled => |e| return e,
148 },148 }
149 } else |err| return err;
150149
151 try testing.expect(addresses_found != 0);150 try testing.expect(addresses_found != 0);
152 }151 }
...@@ -161,20 +160,19 @@ test "resolve DNS" {...@@ -161,20 +160,19 @@ test "resolve DNS" {
161 net.HostName.lookup(try .init("example.com"), io, &results, .{160 net.HostName.lookup(try .init("example.com"), io, &results, .{
162 .port = 80,161 .port = 80,
163 .canonical_name_buffer = &canonical_name_buffer,162 .canonical_name_buffer = &canonical_name_buffer,
164 });163 }) catch |err| switch (err) {
164 error.UnknownHostName => return error.SkipZigTest,
165 error.NameServerFailure => return error.SkipZigTest,
166 else => |e| return e,
167 };
165168
166 while (results.getOne(io)) |result| switch (result) {169 while (results.getOne(io)) |result| switch (result) {
167 .address => {},170 .address => {},
168 .canonical_name => {},171 .canonical_name => {},
169 .end => |end| {172 } else |err| switch (err) {
170 end catch |err| switch (err) {173 error.Closed => {},
171 error.UnknownHostName => return error.SkipZigTest,174 error.Canceled => |e| return e,
172 error.NameServerFailure => return error.SkipZigTest,175 }
173 else => return err,
174 };
175 break;
176 },
177 } else |err| return err;
178 }176 }
179}177}
180178
lib/std/Io/test.zig+53-8
...@@ -209,10 +209,10 @@ test "select" {...@@ -209,10 +209,10 @@ test "select" {
209 return;209 return;
210 },210 },
211 };211 };
212 defer if (get_a.cancel(io)) |_| {} else |_| @panic("fail");212 defer _ = get_a.cancel(io) catch {};
213213
214 var get_b = try io.concurrent(Io.Queue(u8).getOne, .{ &queue, io });214 var get_b = try io.concurrent(Io.Queue(u8).getOne, .{ &queue, io });
215 defer if (get_b.cancel(io)) |_| {} else |_| @panic("fail");215 defer _ = get_b.cancel(io) catch {};
216216
217 var timeout = io.async(Io.sleep, .{ io, .fromMilliseconds(1), .awake });217 var timeout = io.async(Io.sleep, .{ io, .fromMilliseconds(1), .awake });
218 defer timeout.cancel(io) catch {};218 defer timeout.cancel(io) catch {};
...@@ -225,12 +225,9 @@ test "select" {...@@ -225,12 +225,9 @@ test "select" {
225 .get_a => return error.TestFailure,225 .get_a => return error.TestFailure,
226 .get_b => return error.TestFailure,226 .get_b => return error.TestFailure,
227 .timeout => {227 .timeout => {
228 // Unblock the queues to avoid making this unit test depend on228 queue.close(io);
229 // cancellation.229 try testing.expectError(error.Closed, get_a.await(io));
230 queue.putOneUncancelable(io, 1);230 try testing.expectError(error.Closed, get_b.await(io));
231 queue.putOneUncancelable(io, 1);
232 try testing.expectEqual(1, try get_a.await(io));
233 try testing.expectEqual(1, try get_b.await(io));
234 },231 },
235 }232 }
236}233}
...@@ -256,6 +253,54 @@ test "Queue" {...@@ -256,6 +253,54 @@ test "Queue" {
256 try testQueue(5);253 try testQueue(5);
257}254}
258255
256test "Queue.close single-threaded" {
257 const io = std.testing.io;
258
259 var buf: [10]u8 = undefined;
260 var queue: Io.Queue(u8) = .init(&buf);
261
262 try queue.putAll(io, &.{ 0, 1, 2, 3, 4, 5, 6 });
263 try expectEqual(3, try queue.put(io, &.{ 7, 8, 9, 10 }, 0)); // there is capacity for 3 more items
264
265 var get_buf: [4]u8 = undefined;
266
267 // Receive some elements before closing
268 try expectEqual(4, try queue.get(io, &get_buf, 0));
269 try expectEqual(0, get_buf[0]);
270 try expectEqual(1, get_buf[1]);
271 try expectEqual(2, get_buf[2]);
272 try expectEqual(3, get_buf[3]);
273 try expectEqual(4, try queue.getOne(io));
274
275 // ...and add a couple more now there's space
276 try queue.putAll(io, &.{ 20, 21 });
277
278 queue.close(io);
279
280 // Receive more elements *after* closing
281 try expectEqual(4, try queue.get(io, &get_buf, 0));
282 try expectEqual(5, get_buf[0]);
283 try expectEqual(6, get_buf[1]);
284 try expectEqual(7, get_buf[2]);
285 try expectEqual(8, get_buf[3]);
286 try expectEqual(9, try queue.getOne(io));
287
288 // Cannot put anything while closed, even if the buffer has space
289 try expectError(error.Closed, queue.putOne(io, 100));
290 try expectError(error.Closed, queue.putAll(io, &.{ 101, 102 }));
291 try expectError(error.Closed, queue.putUncancelable(io, &.{ 103, 104 }, 0));
292
293 // Even if we ask for 3 items, the queue is closed, so we only get the last 2
294 try expectEqual(2, try queue.get(io, &get_buf, 4));
295 try expectEqual(20, get_buf[0]);
296 try expectEqual(21, get_buf[1]);
297
298 // The queue is now empty, so `get` should return `error.Closed` too
299 try expectError(error.Closed, queue.getOne(io));
300 try expectError(error.Closed, queue.get(io, &get_buf, 0));
301 try expectError(error.Closed, queue.putUncancelable(io, &get_buf, 2));
302}
303
259test "Event" {304test "Event" {
260 const global = struct {305 const global = struct {
261 fn waitAndRead(io: Io, event: *Io.Event, ptr: *const u32) Io.Cancelable!u32 {306 fn waitAndRead(io: Io, event: *Io.Event, ptr: *const u32) Io.Cancelable!u32 {