shithub: mc

Info  •  Files  •  Log  •  Branches

Add ftxwakeall and waitgrps to libthread.

A waitgrp provides more or less the same functionality as a pthread_barrier_t or a sync.WaitGroup from Go. Since "barrier" sounds similar to a memory barrier/fence, I went with the latter name.

--- a/lib/thread/bld.sub

+++ b/lib/thread/bld.sub

@@ -3,8 +3,10 @@

 	hookstd.myr	# install thread hooks

 	mutex+futex.myr

 	sem+futex.myr

+	waitgrp+futex.myr

 	mutex.myr	# fallback, for unimplemented platforms

 	sem.myr		# fallback, for unimplemented platforms

+	waitgrp.myr	# fallback, for unimplemented platforms

 	#generic fallbacks

 	ncpu.myr

--- a/lib/thread/futex+freebsd.myr

+++ b/lib/thread/futex+freebsd.myr

@@ -9,6 +9,7 @@

 	const ftxwait : (uaddr : ftxtag#, val : ftxtag, timeout : sys.timespec# -> int)

 	const ftxwake : (uaddr : ftxtag# -> int)

+	const ftxwakeall : (uaddr : ftxtag# -> int)

 ;;

 const ftxwait = {uaddr, val, timeout

@@ -30,6 +31,10 @@

 const ftxwake = {uaddr

 	-> sys.umtx_op((uaddr : void#), sys.Umtxwakepriv, 1, Zptr, Zptr)

+}

+

+const ftxwakeall = {uaddr

+	-> sys.umtx_op((uaddr : void#), sys.Umtxwakepriv, 0x7fffffff, Zptr, Zptr)

 }

 impl atomic ftxtag =

--- a/lib/thread/futex+linux.myr

+++ b/lib/thread/futex+linux.myr

@@ -9,6 +9,7 @@

 	const ftxwait : (uaddr : ftxtag#, val : ftxtag, timeout : sys.timespec# -> int)

 	const ftxwake : (uaddr : ftxtag# -> int)

+	const ftxwakeall : (uaddr : ftxtag# -> int)

 ;;

 const ftxwait = {uaddr, val, timeout

@@ -22,6 +23,10 @@

 const ftxwake = {uaddr

 	-> (sys.futex((uaddr : int32#), sys.Futexwake | sys.Futexpriv, 1, Zptr, Zptr, 0) : int)

+}

+

+const ftxwakeall = {uaddr

+	-> (sys.futex((uaddr : int32#), sys.Futexwake | sys.Futexpriv, 0x7fffffff, Zptr, Zptr, 0) : int)

 }

 impl atomic ftxtag =

--- a/lib/thread/futex+openbsd:6.2.myr

+++ b/lib/thread/futex+openbsd:6.2.myr

@@ -9,6 +9,7 @@

 	const ftxwait : (uaddr : ftxtag#, val : ftxtag, timeout : sys.timespec# -> int)

 	const ftxwake : (uaddr : ftxtag# -> int)

+	const ftxwakeall : (uaddr : ftxtag# -> int)

 ;;

 const ftxwait = {uaddr, val, timeout

@@ -17,6 +18,10 @@

 const ftxwake = {uaddr

 	-> sys.futex((uaddr : uint32#), sys.Futexwake, 1, Zptr, Zptr)

+}

+

+const ftxwakeall = {uaddr

+	-> sys.futex((uaddr : uint32#), sys.Futexwake, 0x7fffffff, Zptr, Zptr)

 }

 impl atomic ftxtag =

--- a/lib/thread/futex+osx.myr

+++ b/lib/thread/futex+osx.myr

@@ -10,6 +10,7 @@

 	const ftxwait : (uaddr : ftxtag#, val : ftxtag, timeout : sys.timespec# -> int)

 	const ftxwake : (uaddr : ftxtag# -> int)

+	const ftxwakeall : (uaddr : ftxtag# -> int)

 ;;

 /*

@@ -46,6 +47,10 @@

 const ftxwake = {uaddr

 	-> sys.ulock_wake(sys.Ulockcompareandwait, (uaddr : uint64#), 0)

+}

+

+const ftxwakeall = {uaddr

+	-> sys.ulock_wake(sys.Ulockcompareandwait | sys.Ulockulfwakeall, (uaddr : uint64#), 0)

 }

 impl atomic ftxtag =

--- a/lib/thread/mutex+futex.myr

+++ b/lib/thread/mutex+futex.myr

@@ -1,5 +1,3 @@

-use sys

-

 use "atomic"

 use "common"

 use "futex"

--- a/lib/thread/sem+futex.myr

+++ b/lib/thread/sem+futex.myr

@@ -1,5 +1,4 @@

 use std

-use sys

 use "atomic"

 use "common"

@@ -31,7 +30,6 @@

 		;;

 		ftxwait(&s._val, v, Zptr)

 	;;

-	-> void /* Unreachable */

 }

 const semtrywait = {s

--- /dev/null

+++ b/lib/thread/test/waitgrp.myr

@@ -1,0 +1,19 @@

+use std

+use thread

+

+const Threadc = 8

+

+var wg

+

+const main = {

+	wg = thread.mkwg(Threadc)

+

+	for var i = 0; i < Threadc; i++

+		thread.spawn({

+			std.usleep(3_000_000)

+			thread.wgpost(&wg)

+			thread.wgwait(&wg)

+		})

+	;;

+	thread.wgwait(&wg)

+}

--- /dev/null

+++ b/lib/thread/waitgrp+futex.myr

@@ -1,0 +1,37 @@

+use std

+

+use "atomic"

+use "common"

+use "futex"

+

+pkg thread =

+	type waitgrp = struct

+		_val : ftxtag

+	;;

+

+	const mkwg : (v : uint32 -> waitgrp)

+	const wgwait : (w : waitgrp# -> void)

+	const wgpost : (w : waitgrp# -> void)

+;;

+

+const mkwg = {v

+	-> [._val = (v : ftxtag)]

+}

+

+const wgwait = {w

+	var v = 0

+

+	while (v = xget(&w._val)) != 0

+		ftxwait(&w._val, v, Zptr)

+	;;

+}

+

+const wgpost = {w

+	var v = xadd(&w._val, -1)

+

+	if v == 1

+		ftxwakeall(&w._val)

+		-> void

+	;;

+	std.assert(v > 0, "error: waitgroup underflowed\n")

+}

--- /dev/null

+++ b/lib/thread/waitgrp.myr

@@ -1,0 +1,50 @@

+use std

+

+use "atomic"

+

+pkg thread =

+	type waitgrp = struct

+		_val : uint32

+	;;

+

+	const mkwg : (v : uint32 -> waitgrp)

+	const wgwait : (w : waitgrp# -> void)

+	const wgpost : (w : waitgrp# -> void)

+;;

+

+const mkwg = {v

+	-> [._val = v]

+}

+

+const wgwait = {w

+	for var i = 0; i < 1000; i++

+		if xget(&w._val) == 0

+			-> void

+		;;

+	;;

+

+	for var i = 0; i < 1000; i++

+		if xget(&w._val) == 0

+			-> void

+		;;

+		std.nanosleep(10_000)

+	;;

+

+	for var i = 0; i < 1000; i++

+		if xget(&w._val) == 0

+			-> void

+		;;

+		std.nanosleep(100_000)

+	;;

+

+	for ; ;

+		if xget(&w._val) == 0

+			-> void

+		;;

+		std.nanosleep(1_000_000)

+	;;

+}

+

+const wgpost = {w

+	std.assert(xadd(&w._val, -1) > 0, "error: waitgroup underflowed\n")

+}