1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
// SPDX-License-Identifier: MIT
// Copyright (c) 2015-2021 Zig Contributors
// This file is part of [zig](https://ziglang.org/), which is MIT licensed.
// The MIT license requires this copyright notice to be included in all copies
// and substantial portions of the software.
const std = @import("../std.zig");
const builtin = @import("builtin");
const Lock = std.event.Lock;
const testing = std.testing;
const Allocator = std.mem.Allocator;
/// ReturnType must be `void` or `E!void`
/// TODO This API was created back with the old design of async/await, when calling any
/// async function required an allocator. There is an ongoing experiment to transition
/// all uses of this API to the simpler and more resource-aware `std.event.Batch` API.
/// If the transition goes well, all usages of `Group` will be gone, and this API
/// will be deleted.
pub fn Group(comptime ReturnType: type) type {
return struct {
frame_stack: Stack,
alloc_stack: AllocStack,
lock: Lock,
allocator: *Allocator,
const Self = @This();
const Error = switch (@typeInfo(ReturnType)) {
.ErrorUnion => |payload| payload.error_set,
else => void,
};
const Stack = std.atomic.Stack(anyframe->ReturnType);
const AllocStack = std.atomic.Stack(Node);
pub const Node = struct {
bytes: []const u8 = &[0]u8{},
handle: anyframe->ReturnType,
};
pub fn init(allocator: *Allocator) Self {
return Self{
.frame_stack = Stack.init(),
.alloc_stack = AllocStack.init(),
.lock = Lock.init(),
.allocator = allocator,
};
}
/// Add a frame to the group. Thread-safe.
pub fn add(self: *Self, handle: anyframe->ReturnType) (error{OutOfMemory}!void) {
const node = try self.allocator.create(AllocStack.Node);
node.* = AllocStack.Node{
.next = undefined,
.data = Node{
.handle = handle,
},
};
self.alloc_stack.push(node);
}
/// Add a node to the group. Thread-safe. Cannot fail.
/// `node.data` should be the frame handle to add to the group.
/// The node's memory should be in the function frame of
/// the handle that is in the node, or somewhere guaranteed to live
/// at least as long.
pub fn addNode(self: *Self, node: *Stack.Node) void {
self.frame_stack.push(node);
}
/// This is equivalent to adding a frame to the group but the memory of its frame is
/// allocated by the group and freed by `wait`.
/// `func` must be async and have return type `ReturnType`.
/// Thread-safe.
pub fn call(self: *Self, comptime func: anytype, args: anytype) error{OutOfMemory}!void {
var frame = try self.allocator.create(@TypeOf(@call(.{ .modifier = .async_kw }, func, args)));
errdefer self.allocator.destroy(frame);
const node = try self.allocator.create(AllocStack.Node);
errdefer self.allocator.destroy(node);
node.* = AllocStack.Node{
.next = undefined,
.data = Node{
.handle = frame,
.bytes = std.mem.asBytes(frame),
},
};
frame.* = @call(.{ .modifier = .async_kw }, func, args);
self.alloc_stack.push(node);
}
/// Wait for all the calls and promises of the group to complete.
/// Thread-safe.
/// Safe to call any number of times.
pub fn wait(self: *Self) callconv(.Async) ReturnType {
const held = self.lock.acquire();
defer held.release();
var result: ReturnType = {};
while (self.frame_stack.pop()) |node| {
if (Error == void) {
await node.data;
} else {
(await node.data) catch |err| {
result = err;
};
}
}
while (self.alloc_stack.pop()) |node| {
const handle = node.data.handle;
if (Error == void) {
await handle;
} else {
(await handle) catch |err| {
result = err;
};
}
self.allocator.free(node.data.bytes);
self.allocator.destroy(node);
}
return result;
}
};
}
test "std.event.Group" {
// https://github.com/ziglang/zig/issues/1908
if (builtin.single_threaded) return error.SkipZigTest;
if (!std.io.is_async) return error.SkipZigTest;
// TODO this file has bit-rotted. repair it
if (true) return error.SkipZigTest;
const handle = async testGroup(std.heap.page_allocator);
}
fn testGroup(allocator: *Allocator) callconv(.Async) void {
var count: usize = 0;
var group = Group(void).init(allocator);
var sleep_a_little_frame = async sleepALittle(&count);
group.add(&sleep_a_little_frame) catch @panic("memory");
var increase_by_ten_frame = async increaseByTen(&count);
group.add(&increase_by_ten_frame) catch @panic("memory");
group.wait();
testing.expect(count == 11);
var another = Group(anyerror!void).init(allocator);
var something_else_frame = async somethingElse();
another.add(&something_else_frame) catch @panic("memory");
var something_that_fails_frame = async doSomethingThatFails();
another.add(&something_that_fails_frame) catch @panic("memory");
testing.expectError(error.ItBroke, another.wait());
}
fn sleepALittle(count: *usize) callconv(.Async) void {
std.time.sleep(1 * std.time.ns_per_ms);
_ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
}
fn increaseByTen(count: *usize) callconv(.Async) void {
var i: usize = 0;
while (i < 10) : (i += 1) {
_ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
}
}
fn doSomethingThatFails() callconv(.Async) anyerror!void {}
fn somethingElse() callconv(.Async) anyerror!void {
return error.ItBroke;
}
|
