| ... | ... | @@ -2,8 +2,6 @@ const std = @import("../std.zig"); |
| 2 | 2 | const builtin = @import("builtin"); |
| 3 | 3 | const assert = std.debug.assert; |
| 4 | 4 | const testing = std.testing; |
| 5 | | const AtomicRmwOp = builtin.AtomicRmwOp; |
| 6 | | const AtomicOrder = builtin.AtomicOrder; |
| 7 | 5 | const Loop = std.event.Loop; |
| 8 | 6 | |
| 9 | 7 | /// many producer, many consumer, thread-safe, runtime configurable buffer size |
| ... | ... | @@ -98,18 +96,18 @@ pub fn Channel(comptime T: type) type { |
| 98 | 96 | |
| 99 | 97 | // TODO test canceling a put() |
| 100 | 98 | errdefer { |
| 101 | | _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 99 | _ = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst); |
| 102 | 100 | const need_dispatch = !self.putters.remove(&queue_node); |
| 103 | 101 | self.loop.cancelOnNextTick(&my_tick_node); |
| 104 | 102 | if (need_dispatch) { |
| 105 | 103 | // oops we made the put_count incorrect for a period of time. fix by dispatching. |
| 106 | | _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 104 | _ = @atomicRmw(usize, &self.put_count, .Add, 1, .SeqCst); |
| 107 | 105 | self.dispatch(); |
| 108 | 106 | } |
| 109 | 107 | } |
| 110 | 108 | suspend { |
| 111 | 109 | self.putters.put(&queue_node); |
| 112 | | _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 110 | _ = @atomicRmw(usize, &self.put_count, .Add, 1, .SeqCst); |
| 113 | 111 | |
| 114 | 112 | self.dispatch(); |
| 115 | 113 | } |
| ... | ... | @@ -118,8 +116,7 @@ pub fn Channel(comptime T: type) type { |
| 118 | 116 | /// await this function to get an item from the channel. If the buffer is empty, the frame will |
| 119 | 117 | /// complete when the next item is put in the channel. |
| 120 | 118 | pub async fn get(self: *SelfChannel) T { |
| 121 | | // TODO integrate this function with named return values |
| 122 | | // so we can get rid of this extra result copy |
| 119 | // TODO https://github.com/ziglang/zig/issues/2765 |
| 123 | 120 | var result: T = undefined; |
| 124 | 121 | var my_tick_node = Loop.NextTickNode.init(@frame()); |
| 125 | 122 | var queue_node = std.atomic.Queue(GetNode).Node.init(GetNode{ |
| ... | ... | @@ -131,19 +128,19 @@ pub fn Channel(comptime T: type) type { |
| 131 | 128 | |
| 132 | 129 | // TODO test canceling a get() |
| 133 | 130 | errdefer { |
| 134 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 131 | _ = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst); |
| 135 | 132 | const need_dispatch = !self.getters.remove(&queue_node); |
| 136 | 133 | self.loop.cancelOnNextTick(&my_tick_node); |
| 137 | 134 | if (need_dispatch) { |
| 138 | 135 | // oops we made the get_count incorrect for a period of time. fix by dispatching. |
| 139 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 136 | _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst); |
| 140 | 137 | self.dispatch(); |
| 141 | 138 | } |
| 142 | 139 | } |
| 143 | 140 | |
| 144 | 141 | suspend { |
| 145 | 142 | self.getters.put(&queue_node); |
| 146 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 143 | _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst); |
| 147 | 144 | |
| 148 | 145 | self.dispatch(); |
| 149 | 146 | } |
| ... | ... | @@ -183,19 +180,19 @@ pub fn Channel(comptime T: type) type { |
| 183 | 180 | // TODO test canceling getOrNull |
| 184 | 181 | errdefer { |
| 185 | 182 | _ = self.or_null_queue.remove(&or_null_node); |
| 186 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 183 | _ = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst); |
| 187 | 184 | const need_dispatch = !self.getters.remove(&queue_node); |
| 188 | 185 | self.loop.cancelOnNextTick(&my_tick_node); |
| 189 | 186 | if (need_dispatch) { |
| 190 | 187 | // oops we made the get_count incorrect for a period of time. fix by dispatching. |
| 191 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 188 | _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst); |
| 192 | 189 | self.dispatch(); |
| 193 | 190 | } |
| 194 | 191 | } |
| 195 | 192 | |
| 196 | 193 | suspend { |
| 197 | 194 | self.getters.put(&queue_node); |
| 198 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 195 | _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst); |
| 199 | 196 | self.or_null_queue.put(&or_null_node); |
| 200 | 197 | |
| 201 | 198 | self.dispatch(); |
| ... | ... | @@ -205,21 +202,21 @@ pub fn Channel(comptime T: type) type { |
| 205 | 202 | |
| 206 | 203 | fn dispatch(self: *SelfChannel) void { |
| 207 | 204 | // set the "need dispatch" flag |
| 208 | | _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst); |
| 205 | _ = @atomicRmw(u8, &self.need_dispatch, .Xchg, 1, .SeqCst); |
| 209 | 206 | |
| 210 | 207 | lock: while (true) { |
| 211 | 208 | // set the lock flag |
| 212 | | const prev_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst); |
| 209 | const prev_lock = @atomicRmw(u8, &self.dispatch_lock, .Xchg, 1, .SeqCst); |
| 213 | 210 | if (prev_lock != 0) return; |
| 214 | 211 | |
| 215 | 212 | // clear the need_dispatch flag since we're about to do it |
| 216 | | _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst); |
| 213 | _ = @atomicRmw(u8, &self.need_dispatch, .Xchg, 0, .SeqCst); |
| 217 | 214 | |
| 218 | 215 | while (true) { |
| 219 | 216 | one_dispatch: { |
| 220 | 217 | // later we correct these extra subtractions |
| 221 | | var get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 222 | | var put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 218 | var get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst); |
| 219 | var put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst); |
| 223 | 220 | |
| 224 | 221 | // transfer self.buffer to self.getters |
| 225 | 222 | while (self.buffer_len != 0) { |
| ... | ... | @@ -238,7 +235,7 @@ pub fn Channel(comptime T: type) type { |
| 238 | 235 | self.loop.onNextTick(get_node.tick_node); |
| 239 | 236 | self.buffer_len -= 1; |
| 240 | 237 | |
| 241 | | get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 238 | get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst); |
| 242 | 239 | } |
| 243 | 240 | |
| 244 | 241 | // direct transfer self.putters to self.getters |
| ... | ... | @@ -258,8 +255,8 @@ pub fn Channel(comptime T: type) type { |
| 258 | 255 | self.loop.onNextTick(get_node.tick_node); |
| 259 | 256 | self.loop.onNextTick(put_node.tick_node); |
| 260 | 257 | |
| 261 | | get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 262 | | put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 258 | get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst); |
| 259 | put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst); |
| 263 | 260 | } |
| 264 | 261 | |
| 265 | 262 | // transfer self.putters to self.buffer |
| ... | ... | @@ -271,13 +268,13 @@ pub fn Channel(comptime T: type) type { |
| 271 | 268 | self.buffer_index +%= 1; |
| 272 | 269 | self.buffer_len += 1; |
| 273 | 270 | |
| 274 | | put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst); |
| 271 | put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst); |
| 275 | 272 | } |
| 276 | 273 | } |
| 277 | 274 | |
| 278 | 275 | // undo the extra subtractions |
| 279 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 280 | | _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst); |
| 276 | _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst); |
| 277 | _ = @atomicRmw(usize, &self.put_count, .Add, 1, .SeqCst); |
| 281 | 278 | |
| 282 | 279 | // All the "get or null" functions should resume now. |
| 283 | 280 | var remove_count: usize = 0; |
| ... | ... | @@ -286,18 +283,18 @@ pub fn Channel(comptime T: type) type { |
| 286 | 283 | self.loop.onNextTick(or_null_node.data.data.tick_node); |
| 287 | 284 | } |
| 288 | 285 | if (remove_count != 0) { |
| 289 | | _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, remove_count, AtomicOrder.SeqCst); |
| 286 | _ = @atomicRmw(usize, &self.get_count, .Sub, remove_count, .SeqCst); |
| 290 | 287 | } |
| 291 | 288 | |
| 292 | 289 | // clear need-dispatch flag |
| 293 | | const need_dispatch = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst); |
| 290 | const need_dispatch = @atomicRmw(u8, &self.need_dispatch, .Xchg, 0, .SeqCst); |
| 294 | 291 | if (need_dispatch != 0) continue; |
| 295 | 292 | |
| 296 | | const my_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst); |
| 293 | const my_lock = @atomicRmw(u8, &self.dispatch_lock, .Xchg, 0, .SeqCst); |
| 297 | 294 | assert(my_lock != 0); |
| 298 | 295 | |
| 299 | 296 | // we have to check again now that we unlocked |
| 300 | | if (@atomicLoad(u8, &self.need_dispatch, AtomicOrder.SeqCst) != 0) continue :lock; |
| 297 | if (@atomicLoad(u8, &self.need_dispatch, .SeqCst) != 0) continue :lock; |
| 301 | 298 | |
| 302 | 299 | return; |
| 303 | 300 | } |
| ... | ... | @@ -327,16 +324,13 @@ test "std.event.Channel" { |
| 327 | 324 | } |
| 328 | 325 | |
| 329 | 326 | async fn testChannelGetter(loop: *Loop, channel: *Channel(i32)) void { |
| 330 | | const value1_promise = async channel.get(); |
| 331 | | const value1 = await value1_promise; |
| 327 | const value1 = channel.get(); |
| 332 | 328 | testing.expect(value1 == 1234); |
| 333 | 329 | |
| 334 | | const value2_promise = async channel.get(); |
| 335 | | const value2 = await value2_promise; |
| 330 | const value2 = channel.get(); |
| 336 | 331 | testing.expect(value2 == 4567); |
| 337 | 332 | |
| 338 | | const value3_promise = async channel.getOrNull(); |
| 339 | | const value3 = await value3_promise; |
| 333 | const value3 = channel.getOrNull(); |
| 340 | 334 | testing.expect(value3 == null); |
| 341 | 335 | |
| 342 | 336 | const last_put = async testPut(channel, 4444); |