js,checker: fix some modules build for v -b js self, fix or block check in ast.CallExpr (#12231)

2023-08-10 21:13:21 +03:00 · 2021-10-19 12:11:54 +03:00
parent 39c3817ce4
commit c1aa782a6c
11 changed files with 314 additions and 245 deletions
--- a/vlib/sync/channels.c.v
+++ b/vlib/sync/channels.c.v
@@ -0,0 +1,731 @@
+module sync
+
+import time
+import rand
+
+$if windows {
+	#flag -I @VEXEROOT/thirdparty/stdatomic/win
+} $else {
+	#flag -I @VEXEROOT/thirdparty/stdatomic/nix
+}
+
+$if linux {
+	$if tinyc {
+		$if amd64 {
+			// most Linux distributions have /usr/lib/libatomic.so, but Ubuntu uses gcc version specific dir
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/6
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/7
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/8
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/9
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/10
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/11
+			#flag -L/usr/lib/gcc/x86_64-linux-gnu/12
+		} $else $if arm64 {
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/6
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/7
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/8
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/9
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/10
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/11
+			#flag -L/usr/lib/gcc/aarch64-linux-gnu/12
+		}
+		#flag -latomic
+	}
+}
+
+#include <atomic.h>
+// The following functions are actually generic in C
+fn C.atomic_load_ptr(voidptr) voidptr
+fn C.atomic_store_ptr(voidptr, voidptr)
+fn C.atomic_compare_exchange_weak_ptr(voidptr, voidptr, voidptr) bool
+fn C.atomic_compare_exchange_strong_ptr(voidptr, voidptr, voidptr) bool
+fn C.atomic_exchange_ptr(voidptr, voidptr) voidptr
+fn C.atomic_fetch_add_ptr(voidptr, voidptr) voidptr
+fn C.atomic_fetch_sub_ptr(voidptr, voidptr) voidptr
+
+fn C.atomic_load_u16(voidptr) u16
+fn C.atomic_store_u16(voidptr, u16)
+fn C.atomic_compare_exchange_weak_u16(voidptr, voidptr, u16) bool
+fn C.atomic_compare_exchange_strong_u16(voidptr, voidptr, u16) bool
+fn C.atomic_exchange_u16(voidptr, u16) u16
+fn C.atomic_fetch_add_u16(voidptr, u16) u16
+fn C.atomic_fetch_sub_u16(voidptr, u16) u16
+
+fn C.atomic_load_u32(voidptr) u32
+fn C.atomic_store_u32(voidptr, u32)
+fn C.atomic_compare_exchange_weak_u32(voidptr, voidptr, u32) bool
+fn C.atomic_compare_exchange_strong_u32(voidptr, voidptr, u32) bool
+fn C.atomic_exchange_u32(voidptr, u32) u32
+fn C.atomic_fetch_add_u32(voidptr, u32) u32
+fn C.atomic_fetch_sub_u32(voidptr, u32) u32
+
+fn C.atomic_load_u64(voidptr) u64
+fn C.atomic_store_u64(voidptr, u64)
+fn C.atomic_compare_exchange_weak_u64(voidptr, voidptr, u64) bool
+fn C.atomic_compare_exchange_strong_u64(voidptr, voidptr, u64) bool
+fn C.atomic_exchange_u64(voidptr, u64) u64
+fn C.atomic_fetch_add_u64(voidptr, u64) u64
+fn C.atomic_fetch_sub_u64(voidptr, u64) u64
+
+const (
+	// how often to try to get data without blocking before to wait for semaphore
+	spinloops     = 750
+	spinloops_sem = 4000
+)
+
+enum BufferElemStat {
+	unused = 0
+	writing
+	written
+	reading
+}
+
+struct Subscription {
+mut:
+	sem  &Semaphore
+	prev &&Subscription
+	nxt  &Subscription
+}
+
+enum Direction {
+	pop
+	push
+}
+
+struct Channel {
+	ringbuf   &byte // queue for buffered channels
+	statusbuf &byte // flags to synchronize write/read in ringbuf
+	objsize   u32
+mut: // atomic
+	writesem           Semaphore // to wake thread that wanted to write, but buffer was full
+	readsem            Semaphore // to wake thread that wanted to read, but buffer was empty
+	writesem_im        Semaphore
+	readsem_im         Semaphore
+	write_adr          C.atomic_uintptr_t // if != NULL the next obj can be written here without wait
+	read_adr           C.atomic_uintptr_t // if != NULL an obj can be read from here without wait
+	adr_read           C.atomic_uintptr_t // used to identify origin of writesem
+	adr_written        C.atomic_uintptr_t // used to identify origin of readsem
+	write_free         u32 // for queue state
+	read_avail         u32
+	buf_elem_write_idx u32
+	buf_elem_read_idx  u32
+	// for select
+	write_subscriber &Subscription
+	read_subscriber  &Subscription
+	write_sub_mtx    u16
+	read_sub_mtx     u16
+	closed           u16
+pub:
+	cap u32 // queue length in #objects
+}
+
+pub fn new_channel<T>(n u32) &Channel {
+	st := sizeof(T)
+	if isreftype(T) {
+		return new_channel_st(n, st)
+	} else {
+		return new_channel_st_noscan(n, st)
+	}
+}
+
+fn new_channel_st(n u32, st u32) &Channel {
+	wsem := if n > 0 { n } else { 1 }
+	rsem := if n > 0 { u32(0) } else { 1 }
+	rbuf := if n > 0 { unsafe { malloc(int(n * st)) } } else { &byte(0) }
+	sbuf := if n > 0 { vcalloc_noscan(int(n * 2)) } else { &byte(0) }
+	mut ch := Channel{
+		objsize: st
+		cap: n
+		write_free: n
+		read_avail: 0
+		ringbuf: rbuf
+		statusbuf: sbuf
+		write_subscriber: 0
+		read_subscriber: 0
+	}
+	ch.writesem.init(wsem)
+	ch.readsem.init(rsem)
+	ch.writesem_im.init(0)
+	ch.readsem_im.init(0)
+	return &ch
+}
+
+fn new_channel_st_noscan(n u32, st u32) &Channel {
+	$if gcboehm_opt ? {
+		wsem := if n > 0 { n } else { 1 }
+		rsem := if n > 0 { u32(0) } else { 1 }
+		rbuf := if n > 0 { unsafe { malloc_noscan(int(n * st)) } } else { &byte(0) }
+		sbuf := if n > 0 { vcalloc_noscan(int(n * 2)) } else { &byte(0) }
+		mut ch := Channel{
+			objsize: st
+			cap: n
+			write_free: n
+			read_avail: 0
+			ringbuf: rbuf
+			statusbuf: sbuf
+			write_subscriber: 0
+			read_subscriber: 0
+		}
+		ch.writesem.init(wsem)
+		ch.readsem.init(rsem)
+		ch.writesem_im.init(0)
+		ch.readsem_im.init(0)
+		return &ch
+	} $else {
+		return new_channel_st(n, st)
+	}
+}
+
+pub fn (ch &Channel) auto_str(typename string) string {
+	return 'chan $typename{cap: $ch.cap, closed: $ch.closed}'
+}
+
+pub fn (mut ch Channel) close() {
+	open_val := u16(0)
+	if !C.atomic_compare_exchange_strong_u16(&ch.closed, &open_val, 1) {
+		return
+	}
+	mut nulladr := voidptr(0)
+	for !C.atomic_compare_exchange_weak_ptr(unsafe { &voidptr(&ch.adr_written) }, &nulladr,
+		voidptr(-1)) {
+		nulladr = voidptr(0)
+	}
+	ch.readsem_im.post()
+	ch.readsem.post()
+	mut null16 := u16(0)
+	for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
+		null16 = u16(0)
+	}
+	if ch.read_subscriber != voidptr(0) {
+		ch.read_subscriber.sem.post()
+	}
+	C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
+	null16 = u16(0)
+	for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
+		null16 = u16(0)
+	}
+	if ch.write_subscriber != voidptr(0) {
+		ch.write_subscriber.sem.post()
+	}
+	C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
+	ch.writesem.post()
+	if ch.cap == 0 {
+		C.atomic_store_ptr(unsafe { &voidptr(&ch.read_adr) }, voidptr(0))
+	}
+	ch.writesem_im.post()
+}
+
+[inline]
+pub fn (mut ch Channel) len() int {
+	return int(C.atomic_load_u32(&ch.read_avail))
+}
+
+[inline]
+pub fn (mut ch Channel) closed() bool {
+	return C.atomic_load_u16(&ch.closed) != 0
+}
+
+[inline]
+pub fn (mut ch Channel) push(src voidptr) {
+	if ch.try_push_priv(src, false) == .closed {
+		panic('push on closed channel')
+	}
+}
+
+[inline]
+pub fn (mut ch Channel) try_push(src voidptr) ChanState {
+	return ch.try_push_priv(src, true)
+}
+
+fn (mut ch Channel) try_push_priv(src voidptr, no_block bool) ChanState {
+	if C.atomic_load_u16(&ch.closed) != 0 {
+		return .closed
+	}
+	spinloops_sem_, spinloops_ := if no_block { 1, 1 } else { sync.spinloops, sync.spinloops_sem }
+	mut have_swapped := false
+	for {
+		mut got_sem := false
+		mut wradr := C.atomic_load_ptr(unsafe { &voidptr(&ch.write_adr) })
+		for wradr != C.NULL {
+			if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.write_adr) },
+				&wradr, voidptr(0))
+			{
+				// there is a reader waiting for us
+				unsafe { C.memcpy(wradr, src, ch.objsize) }
+				mut nulladr := voidptr(0)
+				for !C.atomic_compare_exchange_weak_ptr(unsafe { &voidptr(&ch.adr_written) },
+					&nulladr, wradr) {
+					nulladr = voidptr(0)
+				}
+				ch.readsem_im.post()
+				return .success
+			}
+		}
+		if no_block && ch.cap == 0 {
+			return .not_ready
+		}
+		// get token to read
+		for _ in 0 .. spinloops_sem_ {
+			if got_sem {
+				break
+			}
+			got_sem = ch.writesem.try_wait()
+		}
+		if !got_sem {
+			if no_block {
+				return .not_ready
+			}
+			ch.writesem.wait()
+		}
+		if C.atomic_load_u16(&ch.closed) != 0 {
+			ch.writesem.post()
+			return .closed
+		}
+		if ch.cap == 0 {
+			// try to advertise current object as readable
+			mut read_in_progress := false
+			C.atomic_store_ptr(unsafe { &voidptr(&ch.read_adr) }, src)
+			wradr = C.atomic_load_ptr(unsafe { &voidptr(&ch.write_adr) })
+			if wradr != C.NULL {
+				mut src2 := src
+				if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.read_adr) },
+					&src2, voidptr(0))
+				{
+					ch.writesem.post()
+					continue
+				} else {
+					read_in_progress = true
+				}
+			}
+			if !read_in_progress {
+				mut null16 := u16(0)
+				for !C.atomic_compare_exchange_weak_u16(voidptr(&ch.read_sub_mtx), &null16,
+					u16(1)) {
+					null16 = u16(0)
+				}
+				if ch.read_subscriber != voidptr(0) {
+					ch.read_subscriber.sem.post()
+				}
+				C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
+			}
+			mut src2 := src
+			for sp := u32(0); sp < spinloops_ || read_in_progress; sp++ {
+				if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.adr_read) },
+					&src2, voidptr(0))
+				{
+					have_swapped = true
+					read_in_progress = true
+					break
+				}
+				src2 = src
+			}
+			mut got_im_sem := false
+			for sp := u32(0); sp < spinloops_sem_ || read_in_progress; sp++ {
+				got_im_sem = ch.writesem_im.try_wait()
+				if got_im_sem {
+					break
+				}
+			}
+			for {
+				if got_im_sem {
+					got_im_sem = false
+				} else {
+					ch.writesem_im.wait()
+				}
+				if C.atomic_load_u16(&ch.closed) != 0 {
+					if have_swapped
+						|| C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.adr_read) }, &src2, voidptr(0)) {
+						ch.writesem.post()
+						return .success
+					} else {
+						return .closed
+					}
+				}
+				if have_swapped
+					|| C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.adr_read) }, &src2, voidptr(0)) {
+					ch.writesem.post()
+					break
+				} else {
+					// this semaphore was not for us - repost in
+					ch.writesem_im.post()
+					if src2 == voidptr(-1) {
+						ch.readsem.post()
+						return .closed
+					}
+					src2 = src
+				}
+			}
+			return .success
+		} else {
+			// buffered channel
+			mut space_in_queue := false
+			mut wr_free := C.atomic_load_u32(&ch.write_free)
+			for wr_free > 0 {
+				space_in_queue = C.atomic_compare_exchange_weak_u32(&ch.write_free, &wr_free,
+					wr_free - 1)
+				if space_in_queue {
+					break
+				}
+			}
+			if space_in_queue {
+				mut wr_idx := C.atomic_load_u32(&ch.buf_elem_write_idx)
+				for {
+					mut new_wr_idx := wr_idx + 1
+					for new_wr_idx >= ch.cap {
+						new_wr_idx -= ch.cap
+					}
+					if C.atomic_compare_exchange_strong_u32(&ch.buf_elem_write_idx, &wr_idx,
+						new_wr_idx)
+					{
+						break
+					}
+				}
+				mut wr_ptr := ch.ringbuf
+				mut status_adr := ch.statusbuf
+				unsafe {
+					wr_ptr += wr_idx * ch.objsize
+					status_adr += wr_idx * sizeof(u16)
+				}
+				mut expected_status := u16(BufferElemStat.unused)
+				for !C.atomic_compare_exchange_weak_u16(status_adr, &expected_status,
+					u16(BufferElemStat.writing)) {
+					expected_status = u16(BufferElemStat.unused)
+				}
+				unsafe {
+					C.memcpy(wr_ptr, src, ch.objsize)
+				}
+				C.atomic_store_u16(unsafe { &u16(status_adr) }, u16(BufferElemStat.written))
+				C.atomic_fetch_add_u32(&ch.read_avail, 1)
+				ch.readsem.post()
+				mut null16 := u16(0)
+				for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
+					null16 = u16(0)
+				}
+				if ch.read_subscriber != voidptr(0) {
+					ch.read_subscriber.sem.post()
+				}
+				C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
+				return .success
+			} else {
+				if no_block {
+					return .not_ready
+				}
+				ch.writesem.post()
+			}
+		}
+	}
+	// we should not get here but the V compiler want's to see a return statement
+	assert false
+	return .success
+}
+
+[inline]
+pub fn (mut ch Channel) pop(dest voidptr) bool {
+	return ch.try_pop_priv(dest, false) == .success
+}
+
+[inline]
+pub fn (mut ch Channel) try_pop(dest voidptr) ChanState {
+	return ch.try_pop_priv(dest, true)
+}
+
+fn (mut ch Channel) try_pop_priv(dest voidptr, no_block bool) ChanState {
+	spinloops_sem_, spinloops_ := if no_block { 1, 1 } else { sync.spinloops, sync.spinloops_sem }
+	mut have_swapped := false
+	mut write_in_progress := false
+	for {
+		mut got_sem := false
+		if ch.cap == 0 {
+			// unbuffered channel - first see if a `push()` has adversized
+			mut rdadr := C.atomic_load_ptr(unsafe { &voidptr(&ch.read_adr) })
+			for rdadr != C.NULL {
+				if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.read_adr) },
+					&rdadr, voidptr(0))
+				{
+					// there is a writer waiting for us
+					unsafe { C.memcpy(dest, rdadr, ch.objsize) }
+					mut nulladr := voidptr(0)
+					for !C.atomic_compare_exchange_weak_ptr(unsafe { &voidptr(&ch.adr_read) },
+						&nulladr, rdadr) {
+						nulladr = voidptr(0)
+					}
+					ch.writesem_im.post()
+					return .success
+				}
+			}
+			if no_block {
+				if C.atomic_load_u16(&ch.closed) == 0 {
+					return .not_ready
+				} else {
+					return .closed
+				}
+			}
+		}
+		// get token to read
+		for _ in 0 .. spinloops_sem_ {
+			if got_sem {
+				break
+			}
+			got_sem = ch.readsem.try_wait()
+		}
+		if !got_sem {
+			if no_block {
+				if C.atomic_load_u16(&ch.closed) == 0 {
+					return .not_ready
+				} else {
+					return .closed
+				}
+			}
+			ch.readsem.wait()
+		}
+		if ch.cap > 0 {
+			// try to get buffer token
+			mut obj_in_queue := false
+			mut rd_avail := C.atomic_load_u32(&ch.read_avail)
+			for rd_avail > 0 {
+				obj_in_queue = C.atomic_compare_exchange_weak_u32(&ch.read_avail, &rd_avail,
+					rd_avail - 1)
+				if obj_in_queue {
+					break
+				}
+			}
+			if obj_in_queue {
+				mut rd_idx := C.atomic_load_u32(&ch.buf_elem_read_idx)
+				for {
+					mut new_rd_idx := rd_idx + 1
+					for new_rd_idx >= ch.cap {
+						new_rd_idx -= ch.cap
+					}
+					if C.atomic_compare_exchange_weak_u32(&ch.buf_elem_read_idx, &rd_idx,
+						new_rd_idx)
+					{
+						break
+					}
+				}
+				mut rd_ptr := ch.ringbuf
+				mut status_adr := ch.statusbuf
+				unsafe {
+					rd_ptr += rd_idx * ch.objsize
+					status_adr += rd_idx * sizeof(u16)
+				}
+				mut expected_status := u16(BufferElemStat.written)
+				for !C.atomic_compare_exchange_weak_u16(status_adr, &expected_status,
+					u16(BufferElemStat.reading)) {
+					expected_status = u16(BufferElemStat.written)
+				}
+				unsafe {
+					C.memcpy(dest, rd_ptr, ch.objsize)
+				}
+				C.atomic_store_u16(unsafe { &u16(status_adr) }, u16(BufferElemStat.unused))
+				C.atomic_fetch_add_u32(&ch.write_free, 1)
+				ch.writesem.post()
+				mut null16 := u16(0)
+				for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
+					null16 = u16(0)
+				}
+				if ch.write_subscriber != voidptr(0) {
+					ch.write_subscriber.sem.post()
+				}
+				C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
+				return .success
+			}
+		}
+		// try to advertise `dest` as writable
+		C.atomic_store_ptr(unsafe { &voidptr(&ch.write_adr) }, dest)
+		if ch.cap == 0 {
+			mut rdadr := C.atomic_load_ptr(unsafe { &voidptr(&ch.read_adr) })
+			if rdadr != C.NULL {
+				mut dest2 := dest
+				if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.write_adr) },
+					&dest2, voidptr(0))
+				{
+					ch.readsem.post()
+					continue
+				} else {
+					write_in_progress = true
+				}
+			}
+		}
+		if ch.cap == 0 && !write_in_progress {
+			mut null16 := u16(0)
+			for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
+				null16 = u16(0)
+			}
+			if ch.write_subscriber != voidptr(0) {
+				ch.write_subscriber.sem.post()
+			}
+			C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
+		}
+		mut dest2 := dest
+		for sp := u32(0); sp < spinloops_ || write_in_progress; sp++ {
+			if C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.adr_written) },
+				&dest2, voidptr(0))
+			{
+				have_swapped = true
+				break
+			} else if dest2 == voidptr(-1) {
+				ch.readsem.post()
+				return .closed
+			}
+			dest2 = dest
+		}
+		mut got_im_sem := false
+		for sp := u32(0); sp < spinloops_sem_ || write_in_progress; sp++ {
+			got_im_sem = ch.readsem_im.try_wait()
+			if got_im_sem {
+				break
+			}
+		}
+		for {
+			if got_im_sem {
+				got_im_sem = false
+			} else {
+				ch.readsem_im.wait()
+			}
+			if have_swapped
+				|| C.atomic_compare_exchange_strong_ptr(unsafe { &voidptr(&ch.adr_written) }, &dest2, voidptr(0)) {
+				ch.readsem.post()
+				break
+			} else {
+				// this semaphore was not for us - repost in
+				ch.readsem_im.post()
+				if dest2 == voidptr(-1) {
+					ch.readsem.post()
+					return .closed
+				}
+				dest2 = dest
+			}
+		}
+		break
+	}
+	return .success
+}
+
+// Wait `timeout` on any of `channels[i]` until one of them can push (`is_push[i] = true`) or pop (`is_push[i] = false`)
+// object referenced by `objrefs[i]`. `timeout = time.infinite` means wait unlimited time. `timeout <= 0` means return
+// immediately if no transaction can be performed without waiting.
+// return value: the index of the channel on which a transaction has taken place
+//               -1 if waiting for a transaction has exceeded timeout
+//               -2 if all channels are closed
+
+pub fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []voidptr, timeout time.Duration) int {
+	assert channels.len == dir.len
+	assert dir.len == objrefs.len
+	mut subscr := []Subscription{len: channels.len}
+	mut sem := unsafe { Semaphore{} }
+	sem.init(0)
+	for i, ch in channels {
+		subscr[i].sem = unsafe { &sem }
+		if dir[i] == .push {
+			mut null16 := u16(0)
+			for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
+				null16 = u16(0)
+			}
+			subscr[i].prev = unsafe { &ch.write_subscriber }
+			unsafe {
+				subscr[i].nxt = &Subscription(C.atomic_exchange_ptr(&voidptr(&ch.write_subscriber),
+					&subscr[i]))
+			}
+			if voidptr(subscr[i].nxt) != voidptr(0) {
+				subscr[i].nxt.prev = unsafe { &subscr[i].nxt }
+			}
+			C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
+		} else {
+			mut null16 := u16(0)
+			for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
+				null16 = u16(0)
+			}
+			subscr[i].prev = unsafe { &ch.read_subscriber }
+			unsafe {
+				subscr[i].nxt = &Subscription(C.atomic_exchange_ptr(&voidptr(&ch.read_subscriber),
+					&subscr[i]))
+			}
+			if voidptr(subscr[i].nxt) != voidptr(0) {
+				subscr[i].nxt.prev = unsafe { &subscr[i].nxt }
+			}
+			C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
+		}
+	}
+	stopwatch := if timeout == time.infinite || timeout <= 0 {
+		time.StopWatch{}
+	} else {
+		time.new_stopwatch()
+	}
+	mut event_idx := -1 // negative index means `timed out`
+
+	outer: for {
+		rnd := rand.u32_in_range(0, u32(channels.len))
+		mut num_closed := 0
+		for j, _ in channels {
+			mut i := j + int(rnd)
+			if i >= channels.len {
+				i -= channels.len
+			}
+			if dir[i] == .push {
+				stat := channels[i].try_push_priv(objrefs[i], true)
+				if stat == .success {
+					event_idx = i
+					break outer
+				} else if stat == .closed {
+					num_closed++
+				}
+			} else {
+				stat := channels[i].try_pop_priv(objrefs[i], true)
+				if stat == .success {
+					event_idx = i
+					break outer
+				} else if stat == .closed {
+					num_closed++
+				}
+			}
+		}
+		if num_closed == channels.len {
+			event_idx = -2
+			break outer
+		}
+		if timeout <= 0 {
+			break outer
+		}
+		if timeout != time.infinite {
+			remaining := timeout - stopwatch.elapsed()
+			if !sem.timed_wait(remaining) {
+				break outer
+			}
+		} else {
+			sem.wait()
+		}
+	}
+	// reset subscribers
+	for i, ch in channels {
+		if dir[i] == .push {
+			mut null16 := u16(0)
+			for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
+				null16 = u16(0)
+			}
+			unsafe {
+				*subscr[i].prev = subscr[i].nxt
+			}
+			if subscr[i].nxt != 0 {
+				subscr[i].nxt.prev = subscr[i].prev
+				// just in case we have missed a semaphore during restore
+				subscr[i].nxt.sem.post()
+			}
+			C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
+		} else {
+			mut null16 := u16(0)
+			for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
+				null16 = u16(0)
+			}
+			unsafe {
+				*subscr[i].prev = subscr[i].nxt
+			}
+			if subscr[i].nxt != 0 {
+				subscr[i].nxt.prev = subscr[i].prev
+				subscr[i].nxt.sem.post()
+			}
+			C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
+		}
+	}
+	sem.destroy()
+	return event_idx
+}