ref: 03220ea343818819d6c155bd83e609ca2282a6e4
parent: 3dca3385472541f62ef9455814a5599a8e97c03a
author: iriri <[email protected]>
date: Mon Jul 23 19:19:11 EDT 2018
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")
+}