Commit 5194fc57d1c206d71654b4f3e43bfcb300bf43c5 | zig

lib/std/event/future.zig+11-10

...	...	@@ -12,12 +12,13 @@ pub fn Future(comptime T: type) type {
12	12	return struct {
13	13	lock: Lock,
14	14	data: T,
	15	available: Available,
15	16
16		/// TODO make this an enum
17		/// 0 - not started
18		/// 1 - started
19		/// 2 - finished
20		available: u8,
	17	const Available = enum(u8) {
	18	NotStarted,
	19	Started,
	20	Finished,
	21	};
21	22
22	23	const Self = @This();
23	24	const Queue = std.atomic.Queue(anyframe);
...	...	@@ -34,7 +35,7 @@ pub fn Future(comptime T: type) type {
34	35	/// available.
35	36	/// Thread-safe.
36	37	pub async fn get(self: Self) T {
37		if (@atomicLoad(u8, &self.available, .SeqCst) == 2) {
	38	if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
38	39	return &self.data;
39	40	}
40	41	const held = self.lock.acquire();
...	...	@@ -46,7 +47,7 @@ pub fn Future(comptime T: type) type {
46	47	/// Gets the data without waiting for it. If it's available, a pointer is
47	48	/// returned. Otherwise, null is returned.
48	49	pub fn getOrNull(self: Self) ?T {
49		if (@atomicLoad(u8, &self.available, .SeqCst) == 2) {
	50	if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
50	51	return &self.data;
51	52	} else {
52	53	return null;
...	...	@@ -59,7 +60,7 @@ pub fn Future(comptime T: type) type {
59	60	/// It's not required to call start() before resolve() but it can be useful since
60	61	/// this method is thread-safe.
61	62	pub async fn start(self: Self) ?T {
62		const state = @cmpxchgStrong(u8, &self.available, 0, 1, .SeqCst, .SeqCst) orelse return null;
	63	const state = @cmpxchgStrong(Available, &self.available, .NotStarted, .Started, .SeqCst, .SeqCst) orelse return null;
63	64	switch (state) {
64	65	1 => {
65	66	const held = self.lock.acquire();
...	...	@@ -74,8 +75,8 @@ pub fn Future(comptime T: type) type {
74	75	/// Make the data become available. May be called only once.
75	76	/// Before calling this, modify the `data` property.
76	77	pub fn resolve(self: *Self) void {
77		const prev = @atomicRmw(u8, &self.available, .Xchg, 2, .SeqCst);
78		assert(prev == 0 or prev == 1); // resolve() called twice
	78	const prev = @atomicRmw(Available, &self.available, .Xchg, .Finished, .SeqCst);
	79	assert(prev != .Finished); // resolve() called twice
79	80	Lock.Held.release(Lock.Held{ .lock = &self.lock });
80	81	}
81	82	};

lib/std/event/rwlock.zig+16-16

...	...	@@ -13,17 +13,17 @@ const Loop = std.event.Loop;
13	13	/// When a write lock is held, it will not be released until the writer queue is empty.
14	14	/// TODO: make this API also work in blocking I/O mode
15	15	pub const RwLock = struct {
16		shared_state: u8, // TODO make this an enum
	16	shared_state: State,
17	17	writer_queue: Queue,
18	18	reader_queue: Queue,
19	19	writer_queue_empty_bit: u8, // TODO make this a bool
20	20	reader_queue_empty_bit: u8, // TODO make this a bool
21	21	reader_lock_count: usize,
22	22
23		const State = struct {
24		const Unlocked = 0;
25		const WriteLock = 1;
26		const ReadLock = 2;
	23	const State = enum(u8) {
	24	Unlocked,
	25	WriteLock,
	26	ReadLock,
27	27	};
28	28
29	29	const Queue = std.atomic.Queue(anyframe);
...	...	@@ -41,7 +41,7 @@ pub const RwLock = struct {
41	41	}
42	42
43	43	_ = @atomicRmw(u8, &self.lock.reader_queue_empty_bit, .Xchg, 1, .SeqCst);
44		if (@cmpxchgStrong(u8, &self.lock.shared_state, State.ReadLock, State.Unlocked, .SeqCst, .SeqCst) != null) {
	44	if (@cmpxchgStrong(State, &self.lock.shared_state, .ReadLock, .Unlocked, .SeqCst, .SeqCst) != null) {
45	45	// Didn't unlock. Someone else's problem.
46	46	return;
47	47	}
...	...	@@ -64,7 +64,7 @@ pub const RwLock = struct {
64	64	// We need to release the write lock. Check if any readers are waiting to grab the lock.
65	65	if (@atomicLoad(u8, &self.lock.reader_queue_empty_bit, .SeqCst) == 0) {
66	66	// Switch to a read lock.
67		_ = @atomicRmw(u8, &self.lock.shared_state, .Xchg, State.ReadLock, .SeqCst);
	67	_ = @atomicRmw(State, &self.lock.shared_state, .Xchg, .ReadLock, .SeqCst);
68	68	while (self.lock.reader_queue.get()) \|node\| {
69	69	global_event_loop.onNextTick(node);
70	70	}
...	...	@@ -72,7 +72,7 @@ pub const RwLock = struct {
72	72	}
73	73
74	74	_ = @atomicRmw(u8, &self.lock.writer_queue_empty_bit, .Xchg, 1, .SeqCst);
75		_ = @atomicRmw(u8, &self.lock.shared_state, .Xchg, State.Unlocked, .SeqCst);
	75	_ = @atomicRmw(State, &self.lock.shared_state, .Xchg, State.Unlocked, .SeqCst);
76	76
77	77	self.lock.commonPostUnlock();
78	78	}
...	...	@@ -80,7 +80,7 @@ pub const RwLock = struct {
80	80
81	81	pub fn init() RwLock {
82	82	return RwLock{
83		.shared_state = State.Unlocked,
	83	.shared_state = .Unlocked,
84	84	.writer_queue = Queue.init(),
85	85	.writer_queue_empty_bit = 1,
86	86	.reader_queue = Queue.init(),
...	...	@@ -92,7 +92,7 @@ pub const RwLock = struct {
92	92	/// Must be called when not locked. Not thread safe.
93	93	/// All calls to acquire() and release() must complete before calling deinit().
94	94	pub fn deinit(self: *RwLock) void {
95		assert(self.shared_state == State.Unlocked);
	95	assert(self.shared_state == .Unlocked);
96	96	while (self.writer_queue.get()) \|node\| resume node.data;
97	97	while (self.reader_queue.get()) \|node\| resume node.data;
98	98	}
...	...	@@ -116,7 +116,7 @@ pub const RwLock = struct {
116	116	_ = @atomicRmw(u8, &self.reader_queue_empty_bit, .Xchg, 0, .SeqCst);
117	117
118	118	// Here we don't care if we are the one to do the locking or if it was already locked for reading.
119		const have_read_lock = if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.ReadLock, .SeqCst, .SeqCst)) \|old_state\| old_state == State.ReadLock else true;
	119	const have_read_lock = if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .ReadLock, .SeqCst, .SeqCst)) \|old_state\| old_state == .ReadLock else true;
120	120	if (have_read_lock) {
121	121	// Give out all the read locks.
122	122	if (self.reader_queue.get()) \|first_node\| {
...	...	@@ -147,7 +147,7 @@ pub const RwLock = struct {
147	147	_ = @atomicRmw(u8, &self.writer_queue_empty_bit, .Xchg, 0, .SeqCst);
148	148
149	149	// Here we must be the one to acquire the write lock. It cannot already be locked.
150		if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.WriteLock, .SeqCst, .SeqCst) == null) {
	150	if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .WriteLock, .SeqCst, .SeqCst) == null) {
151	151	// We now have a write lock.
152	152	if (self.writer_queue.get()) \|node\| {
153	153	// Whether this node is us or someone else, we tail resume it.
...	...	@@ -166,7 +166,7 @@ pub const RwLock = struct {
166	166	// But if there's a writer_queue item or a reader_queue item,
167	167	// we are the actor which must loop and attempt to grab the lock again.
168	168	if (@atomicLoad(u8, &self.writer_queue_empty_bit, .SeqCst) == 0) {
169		if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.WriteLock, .SeqCst, .SeqCst) != null) {
	169	if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .WriteLock, .SeqCst, .SeqCst) != null) {
170	170	// We did not obtain the lock. Great, the queues are someone else's problem.
171	171	return;
172	172	}
...	...	@@ -177,12 +177,12 @@ pub const RwLock = struct {
177	177	}
178	178	// Release the lock again.
179	179	_ = @atomicRmw(u8, &self.writer_queue_empty_bit, .Xchg, 1, .SeqCst);
180		_ = @atomicRmw(u8, &self.shared_state, .Xchg, State.Unlocked, .SeqCst);
	180	_ = @atomicRmw(State, &self.shared_state, .Xchg, .Unlocked, .SeqCst);
181	181	continue;
182	182	}
183	183
184	184	if (@atomicLoad(u8, &self.reader_queue_empty_bit, .SeqCst) == 0) {
185		if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.ReadLock, .SeqCst, .SeqCst) != null) {
	185	if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .ReadLock, .SeqCst, .SeqCst) != null) {
186	186	// We did not obtain the lock. Great, the queues are someone else's problem.
187	187	return;
188	188	}
...	...	@@ -196,7 +196,7 @@ pub const RwLock = struct {
196	196	}
197	197	// Release the lock again.
198	198	_ = @atomicRmw(u8, &self.reader_queue_empty_bit, .Xchg, 1, .SeqCst);
199		if (@cmpxchgStrong(u8, &self.shared_state, State.ReadLock, State.Unlocked, .SeqCst, .SeqCst) != null) {
	199	if (@cmpxchgStrong(State, &self.shared_state, .ReadLock, .Unlocked, .SeqCst, .SeqCst) != null) {
200	200	// Didn't unlock. Someone else's problem.
201	201	return;
202	202	}

lib/std/lazy_init.zig+13-11

...	...	@@ -1,24 +1,26 @@
1	1	const std = @import("std.zig");
2		const builtin = @import("builtin");
3	2	const assert = std.debug.assert;
4	3	const testing = std.testing;
5		const AtomicRmwOp = builtin.AtomicRmwOp;
6		const AtomicOrder = builtin.AtomicOrder;
7	4
8	5	/// Thread-safe initialization of global data.
9	6	/// TODO use a mutex instead of a spinlock
10	7	pub fn lazyInit(comptime T: type) LazyInit(T) {
11	8	return LazyInit(T){
12	9	.data = undefined,
13		.state = 0,
14	10	};
15	11	}
16	12
17	13	fn LazyInit(comptime T: type) type {
18	14	return struct {
19		state: u8, // TODO make this an enum
	15	state: State = .NotResolved,
20	16	data: Data,
21	17
	18	const State = enum(u8) {
	19	NotResolved,
	20	Resolving,
	21	Resolved,
	22	};
	23
22	24	const Self = @This();
23	25
24	26	// TODO this isn't working for void, investigate and then remove this special case
...	...	@@ -30,14 +32,14 @@ fn LazyInit(comptime T: type) type {
30	32	/// perform the initialization and then call resolve().
31	33	pub fn get(self: *Self) ?Ptr {
32	34	while (true) {
33		var state = @cmpxchgWeak(u8, &self.state, 0, 1, AtomicOrder.SeqCst, AtomicOrder.SeqCst) orelse return null;
	35	var state = @cmpxchgWeak(State, &self.state, .NotResolved, .Resolving, .SeqCst, .SeqCst) orelse return null;
34	36	switch (state) {
35		0 => continue,
36		1 => {
	37	.NotResolved => continue,
	38	.Resolving => {
37	39	// TODO mutex instead of a spinlock
38	40	continue;
39	41	},
40		2 => {
	42	.Resolved => {
41	43	if (@sizeOf(T) == 0) {
42	44	return @as(T, undefined);
43	45	} else {
...	...	@@ -50,8 +52,8 @@ fn LazyInit(comptime T: type) type {
50	52	}
51	53
52	54	pub fn resolve(self: *Self) void {
53		const prev = @atomicRmw(u8, &self.state, AtomicRmwOp.Xchg, 2, AtomicOrder.SeqCst);
54		assert(prev == 1); // resolve() called twice
	55	const prev = @atomicRmw(State, &self.state, .Xchg, .Resolved, .SeqCst);
	56	assert(prev != .Resolved); // resolve() called twice
55	57	}
56	58	};
57	59	}

lib/std/mutex.zig+22-20

...	...	@@ -39,12 +39,14 @@ pub const Mutex = if (builtin.single_threaded)
39	39	}
40	40	else
41	41	struct {
42		state: u32, // TODO: make this an enum
	42	state: State, // TODO: make this an enum
43	43	parker: ThreadParker,
44	44
45		const Unlocked = 0;
46		const Sleeping = 1;
47		const Locked = 2;
	45	const State = enum(u32) {
	46	Unlocked,
	47	Sleeping,
	48	Locked,
	49	};
48	50
49	51	/// number of iterations to spin yielding the cpu
50	52	const SPIN_CPU = 4;
...	...	@@ -57,7 +59,7 @@ else
57	59
58	60	pub fn init() Mutex {
59	61	return Mutex{
60		.state = Unlocked,
	62	.state = .Unlocked,
61	63	.parker = ThreadParker.init(),
62	64	};
63	65	}
...	...	@@ -70,10 +72,10 @@ else
70	72	mutex: *Mutex,
71	73
72	74	pub fn release(self: Held) void {
73		switch (@atomicRmw(u32, &self.mutex.state, .Xchg, Unlocked, .Release)) {
74		Locked => {},
75		Sleeping => self.mutex.parker.unpark(&self.mutex.state),
76		Unlocked => unreachable, // unlocking an unlocked mutex
	75	switch (@atomicRmw(State, &self.mutex.state, .Xchg, .Unlocked, .Release)) {
	76	.Locked => {},
	77	.Sleeping => self.mutex.parker.unpark(@ptrCast(*const u32, &self.mutex.state)),
	78	.Unlocked => unreachable, // unlocking an unlocked mutex
77	79	else => unreachable, // should never be anything else
78	80	}
79	81	}
...	...	@@ -83,34 +85,34 @@ else
83	85	// Try and speculatively grab the lock.
84	86	// If it fails, the state is either Locked or Sleeping
85	87	// depending on if theres a thread stuck sleeping below.
86		var state = @atomicRmw(u32, &self.state, .Xchg, Locked, .Acquire);
87		if (state == Unlocked)
	88	var state = @atomicRmw(State, &self.state, .Xchg, .Locked, .Acquire);
	89	if (state == .Unlocked)
88	90	return Held{ .mutex = self };
89	91
90	92	while (true) {
91	93	// try and acquire the lock using cpu spinning on failure
92	94	var spin: usize = 0;
93	95	while (spin < SPIN_CPU) : (spin += 1) {
94		var value = @atomicLoad(u32, &self.state, .Monotonic);
95		while (value == Unlocked)
96		value = @cmpxchgWeak(u32, &self.state, Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
	96	var value = @atomicLoad(State, &self.state, .Monotonic);
	97	while (value == .Unlocked)
	98	value = @cmpxchgWeak(State, &self.state, .Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
97	99	SpinLock.yield(SPIN_CPU_COUNT);
98	100	}
99	101
100	102	// try and acquire the lock using thread rescheduling on failure
101	103	spin = 0;
102	104	while (spin < SPIN_THREAD) : (spin += 1) {
103		var value = @atomicLoad(u32, &self.state, .Monotonic);
104		while (value == Unlocked)
105		value = @cmpxchgWeak(u32, &self.state, Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
	105	var value = @atomicLoad(State, &self.state, .Monotonic);
	106	while (value == .Unlocked)
	107	value = @cmpxchgWeak(State, &self.state, .Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
106	108	std.os.sched_yield() catch std.time.sleep(1);
107	109	}
108	110
109	111	// failed to acquire the lock, go to sleep until woken up by `Held.release()`
110		if (@atomicRmw(u32, &self.state, .Xchg, Sleeping, .Acquire) == Unlocked)
	112	if (@atomicRmw(State, &self.state, .Xchg, .Sleeping, .Acquire) == .Unlocked)
111	113	return Held{ .mutex = self };
112		state = Sleeping;
113		self.parker.park(&self.state, Sleeping);
	114	state = .Sleeping;
	115	self.parker.park(@ptrCast(*const u32, &self.state), @enumToInt(State.Sleeping));
114	116	}
115	117	}
116	118	};

author	Veikka Tuominen <git@vexu.eu> 2019-11-12 17:40:21+02:00
committer	Veikka Tuominen <git@vexu.eu> 2019-11-12 17:55:54+02:00
log	5194fc57d1c206d71654b4f3e43bfcb300bf43c5
tree	02f1b897c19fbc22af9161af9f1974561c0d4b7d
parent	7e5b234b8b3a7e675a7b11eebcdaf504f2686749
signature	Commit is signed but in an unrecognized format.

use enum with atomics in std lib