From 3a3ed48fae8c7d38f7af1fce39343b2302fd5639 Mon Sep 17 00:00:00 2001 From: laytan Date: Sat, 17 Jan 2026 21:03:25 +0100 Subject: [PATCH] nbio: mpsc queue for execs on another thread --- core/nbio/impl.odin | 21 +++++++------ core/nbio/impl_linux.odin | 5 +++ core/nbio/impl_others.odin | 3 ++ core/nbio/impl_posix.odin | 5 +++ core/nbio/impl_windows.odin | 20 +++++++----- core/nbio/mpsc.odin | 63 +++++++++++++++++++++++++++++++++++++ core/nbio/nbio.odin | 19 +++++------ 7 files changed, 108 insertions(+), 28 deletions(-) create mode 100644 core/nbio/mpsc.odin diff --git a/core/nbio/impl.odin b/core/nbio/impl.odin index 3f5191c5e..476b8ab43 100644 --- a/core/nbio/impl.odin +++ b/core/nbio/impl.odin @@ -5,10 +5,8 @@ import "base:runtime" import "base:intrinsics" import "core:container/pool" -import "core:container/queue" import "core:net" import "core:strings" -import "core:sync" import "core:time" import "core:reflect" @@ -35,7 +33,13 @@ _acquire_thread_event_loop :: proc() -> General_Error { allocator := runtime.heap_allocator() } - l.queue.data.allocator = allocator + l.allocator = allocator + + if alloc_err := mpsc_init(&l.queue, 128, l.allocator); alloc_err != nil { + l.err = .Allocation_Failed + return l.err + } + defer if l.err != nil { mpsc_destroy(&l.queue, l.allocator) } if pool_err := pool.init(&l.operation_pool, "_pool_link"); pool_err != nil { l.err = .Allocation_Failed @@ -65,7 +69,7 @@ _release_thread_event_loop :: proc() { if l.refs > 0 { l.refs -= 1 if l.refs == 0 { - queue.destroy(&l.queue) + mpsc_destroy(&l.queue, l.allocator) pool.destroy(&l.operation_pool) _destroy(l) l^ = {} @@ -85,11 +89,10 @@ _current_thread_event_loop :: #force_inline proc(loc := #caller_location) -> (^E _tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { // Receive operations queued from other threads first. - { - sync.guard(&l.queue_mu) - for op in queue.pop_front_safe(&l.queue) { - _exec(op) - } + for { + op := (^Operation)(mpsc_dequeue(&l.queue)) + if op == nil { break } + _exec(op) } return __tick(l, timeout) diff --git a/core/nbio/impl_linux.odin b/core/nbio/impl_linux.odin index cdcbeedc1..552133e67 100644 --- a/core/nbio/impl_linux.odin +++ b/core/nbio/impl_linux.odin @@ -472,6 +472,11 @@ _wake_up :: proc(l: ^Event_Loop) { assert(n == 8) } +@(private="package") +_yield :: proc() { + linux.sched_yield() +} + // Start file private. // The size of the IO Uring queues. diff --git a/core/nbio/impl_others.odin b/core/nbio/impl_others.odin index cac1f0c63..1c1e7674e 100644 --- a/core/nbio/impl_others.odin +++ b/core/nbio/impl_others.odin @@ -215,3 +215,6 @@ _associate_socket :: proc(socket: Any_Socket, l: ^Event_Loop) -> Association_Err _wake_up :: proc(l: ^Event_Loop) { } + +_yield :: proc() { +} diff --git a/core/nbio/impl_posix.odin b/core/nbio/impl_posix.odin index e003f6ea3..ed65a05b0 100644 --- a/core/nbio/impl_posix.odin +++ b/core/nbio/impl_posix.odin @@ -526,6 +526,11 @@ _wake_up :: proc(l: ^Event_Loop) { assert(n == 0) } +@(private="package") +_yield :: proc() { + posix.sched_yield() +} + // Start file private. // Max operations that can be enqueued per tick. diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index e7987671d..fbbe2c2ce 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -107,17 +107,15 @@ _Stat :: struct {} _init :: proc(l: ^Event_Loop, alloc: mem.Allocator) -> (err: General_Error) { l.allocator = alloc - mem_err: mem.Allocator_Error - if mem_err = queue.init(&l.completed, allocator = alloc); mem_err != nil { - err = .Allocation_Failed - return - } - defer if err != nil { queue.destroy(&l.completed) } + l.completed.data.allocator = l.allocator avl.init(&l.timeouts, timeouts_cmp, alloc) win.ensure_winsock_initialized() + mpsc_init(&l.completed_oob, QUEUE_SIZE, l.allocator) + defer if err != nil { mpsc_destroy(&l.completed_oob, l.allocator) } + l.iocp = win.CreateIoCompletionPort(win.INVALID_HANDLE_VALUE, nil, 0, 1) if l.iocp == nil { err = General_Error(win.GetLastError()) @@ -145,8 +143,9 @@ _init :: proc(l: ^Event_Loop, alloc: mem.Allocator) -> (err: General_Error) { @(private="package") _destroy :: proc(l: ^Event_Loop) { - queue.destroy(&l.completed) avl.destroy(&l.timeouts) + queue.destroy(&l.completed) + mpsc_destroy(&l.completed_oob, l.allocator) win.CloseHandle(l.iocp) } @@ -754,8 +753,15 @@ _wake_up :: proc(l: ^Event_Loop) { ) } +@(private="package") +_yield :: proc() { + win.SwitchToThread() +} + // Start file private. +QUEUE_SIZE :: 128 + REMOVED :: rawptr(max(uintptr)-1) INVALID_HANDLE :: Handle(win.INVALID_HANDLE) diff --git a/core/nbio/mpsc.odin b/core/nbio/mpsc.odin new file mode 100644 index 000000000..7f88829b4 --- /dev/null +++ b/core/nbio/mpsc.odin @@ -0,0 +1,63 @@ +#+private +package nbio + +import "base:runtime" + +import "core:sync" + +Multi_Producer_Single_Consumer :: struct { + count: int, + head: int, + tail: int, + buffer: []rawptr, + mask: int, +} + +mpsc_init :: proc(mpscq: ^Multi_Producer_Single_Consumer, cap: int, allocator: runtime.Allocator) -> runtime.Allocator_Error { + assert(runtime.is_power_of_two_int(cap), "cap must be a power of 2") + mpscq.buffer = make([]rawptr, cap, allocator) or_return + mpscq.mask = cap-1 + sync.atomic_thread_fence(.Release) + return nil +} + +mpsc_destroy :: proc(mpscq: ^Multi_Producer_Single_Consumer, allocator: runtime.Allocator) { + delete(mpscq.buffer, allocator) +} + +mpsc_enqueue :: proc(mpscq: ^Multi_Producer_Single_Consumer, obj: rawptr) -> bool { + count := sync.atomic_add_explicit(&mpscq.count, 1, .Acquire) + if count >= len(mpscq.buffer) { + sync.atomic_sub_explicit(&mpscq.count, 1, .Release) + return false + } + + head := sync.atomic_add_explicit(&mpscq.head, 1, .Acquire) + assert(mpscq.buffer[head & mpscq.mask] == nil) + rv := sync.atomic_exchange_explicit(&mpscq.buffer[head & mpscq.mask], obj, .Release) + assert(rv == nil) + return true +} + +mpsc_dequeue :: proc(mpscq: ^Multi_Producer_Single_Consumer) -> rawptr { + ret := sync.atomic_exchange_explicit(&mpscq.buffer[mpscq.tail], nil, .Acquire) + if ret == nil { + return nil + } + + mpscq.tail += 1 + if mpscq.tail >= len(mpscq.buffer) { + mpscq.tail = 0 + } + r := sync.atomic_sub_explicit(&mpscq.count, 1, .Release) + assert(r > 0) + return ret +} + +mpsc_count :: proc(mpscq: ^Multi_Producer_Single_Consumer) -> int { + return sync.atomic_load_explicit(&mpscq.count, .Relaxed) +} + +mpsc_cap :: proc(mpscq: ^Multi_Producer_Single_Consumer) -> int { + return len(mpscq.buffer) +} \ No newline at end of file diff --git a/core/nbio/nbio.odin b/core/nbio/nbio.odin index 274cc5291..703a2b4d7 100644 --- a/core/nbio/nbio.odin +++ b/core/nbio/nbio.odin @@ -1,11 +1,10 @@ package nbio import "base:intrinsics" +import "base:runtime" import "core:container/pool" -import "core:container/queue" import "core:net" -import "core:sync" import "core:time" /* @@ -23,14 +22,13 @@ Do not copy. */ Event_Loop :: struct /* #no_copy */ { using impl: _Event_Loop, + allocator: runtime.Allocator, err: General_Error, refs: int, now: time.Time, // Queue that is used to queue operations from another thread to be executed on this thread. - // TODO: Better data-structure. - queue: queue.Queue(^Operation), - queue_mu: sync.Mutex, + queue: Multi_Producer_Single_Consumer, operation_pool: pool.Pool(Operation), } @@ -408,13 +406,10 @@ exec :: proc(op: ^Operation, trigger_wake_up := true) { if op.l == &_tls_event_loop { _exec(op) } else { - { - // TODO: Better data-structure. - sync.guard(&op.l.queue_mu) - _, err := queue.push_back(&op.l.queue, op) - if err != nil { - panic("exec: queueing operation failed due to memory allocation failure") - } + for !mpsc_enqueue(&op.l.queue, op) { + warn("operation queue on event loop filled up") + wake_up(op.l) + _yield() } if trigger_wake_up { wake_up(op.l)