| ... | @@ -248,8 +248,16 @@ fn findReadyFiber(el: *EventLoop, thread: *Thread) ?*Fiber { | ... | @@ -248,8 +248,16 @@ fn findReadyFiber(el: *EventLoop, thread: *Thread) ?*Fiber { |
| 248 | if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0; | 248 | if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0; |
| 249 | const steal_ready_search_thread = &el.threads.allocated[0..active_threads][thread.steal_ready_search_index]; | 249 | const steal_ready_search_thread = &el.threads.allocated[0..active_threads][thread.steal_ready_search_index]; |
| 250 | if (steal_ready_search_thread == thread) continue; | 250 | if (steal_ready_search_thread == thread) continue; |
| 251 | const ready_fiber = @atomicRmw(?*Fiber, &steal_ready_search_thread.ready_queue, .And, Fiber.finished, .acquire) orelse continue; | 251 | const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue; |
| 252 | if (ready_fiber == Fiber.finished) continue; | 252 | if (ready_fiber == Fiber.finished) continue; |
| | 253 | if (@cmpxchgWeak( |
| | 254 | ?*Fiber, |
| | 255 | &steal_ready_search_thread.ready_queue, |
| | 256 | ready_fiber, |
| | 257 | null, |
| | 258 | .acquire, |
| | 259 | .monotonic, |
| | 260 | )) |_| continue; |
| 253 | @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release); | 261 | @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release); |
| 254 | ready_fiber.queue_next = null; | 262 | ready_fiber.queue_next = null; |
| 255 | return ready_fiber; | 263 | return ready_fiber; |
| ... | @@ -297,7 +305,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void { | ... | @@ -297,7 +305,7 @@ fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void { |
| 297 | &idle_search_thread.ready_queue, | 305 | &idle_search_thread.ready_queue, |
| 298 | null, | 306 | null, |
| 299 | ready_queue.head, | 307 | ready_queue.head, |
| 300 | .acq_rel, | 308 | .release, |
| 301 | .monotonic, | 309 | .monotonic, |
| 302 | )) |_| continue; | 310 | )) |_| continue; |
| 303 | getSqe(&thread.io_uring).* = .{ | 311 | getSqe(&thread.io_uring).* = .{ |
| ... | @@ -1268,9 +1276,9 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I | ... | @@ -1268,9 +1276,9 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I |
| 1268 | const fiber = thread.currentFiber(); | 1276 | const fiber = thread.currentFiber(); |
| 1269 | const prev = @atomicRmw(?*Fiber, cond_state, .Xchg, fiber, .acquire); | 1277 | const prev = @atomicRmw(?*Fiber, cond_state, .Xchg, fiber, .acquire); |
| 1270 | assert(prev == null); // More than one wait on same Condition is illegal. | 1278 | assert(prev == null); // More than one wait on same Condition is illegal. |
| 1271 | mutex.unlock(io(el)); | 1279 | mutex.unlock(el.io()); |
| 1272 | el.yield(null, .nothing); | 1280 | el.yield(null, .nothing); |
| 1273 | try mutex.lock(io(el)); | 1281 | try mutex.lock(el.io()); |
| 1274 | } | 1282 | } |
| 1275 | | 1283 | |
| 1276 | fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void { | 1284 | fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void { |