nbio: mpsc queue for execs on another thread

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)

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.

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() {
+}

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.

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)

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)
+}

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)