ref: 7c2f25e309226a9506ea4b37e94e560017e83994
parent: 5d5d5e417324785a9fc78ffd14a4ad5393e0fb06
author: iriri <iri@konnichiwastevenspielbergde.su>
date: Fri Jun 22 22:07:49 EDT 2018
Add semaphores and wrapper for various futexesque system calls
--- /dev/null
+++ b/bld.tags
@@ -1,0 +1,4 @@
+futex: freebsd
+futex: linux
+futex: openbsd:6.2
+futex: osx
--- a/lib/std/errno.myr
+++ b/lib/std/errno.myr
@@ -38,4 +38,5 @@
const Epipe : errno = (sys.Epipe : errno)
const Edom : errno = (sys.Edom : errno)
const Erange : errno = (sys.Erange : errno)
+ const Etimedout : errno = (sys.Etimedout : errno)
;;
--- a/lib/sys/sys+freebsd-x64.myr
+++ b/lib/sys/sys+freebsd-x64.myr
@@ -581,6 +581,13 @@
node : uint8[_Uuidnodesz];
;;
+ const Umtxabstime = 1
+ type _umtx_time = struct
+ _timeout : timespec
+ _flags : uint32
+ _clockid : uint32
+ ;;
+
/* open options */
const Ordonly : fdopt = 0x0
const Owronly : fdopt = 0x1
--- a/lib/sys/sys+osx-x64.myr
+++ b/lib/sys/sys+osx-x64.myr
@@ -18,6 +18,7 @@
type machport = int32
type signo = int32
type sigflags = int32
+ type ulockop = uint32
type fdset = struct
bits : int32[1024/4]
@@ -388,6 +389,14 @@
const Sigusr1 : signo = 30 /* user defined signal 1 */
const Sigusr2 : signo = 31 /* user defined signal 2 */
+ /* ulock ops */
+ const Ulockcompareandwait : ulockop = 0x00000001
+ const Ulockunfairlock : ulockop = 0x00000002
+ const Ulockulfwakeall : ulockop = 0x00000100
+ const Ulockulfwakethread : ulockop = 0x00000200
+ const Ulockwaitworkqdatacontention : ulockop = 0x00010000
+ const Ulocknoerrno : ulockop = 0x01000000
+
/* syscalls.
note, creat() implemented as open(path, Creat|Trunc|Wronly) */
const Syssyscall : scno = 0x2000000
@@ -737,6 +746,8 @@
const Syspid_resume : scno = 0x20001af
const Sysfileport_makeport : scno = 0x20001b0
const Sysfileport_makefd : scno = 0x20001b1
+ const Sysulock_wait : scno = 0x2000203
+ const Sysulock_wake : scno = 0x2000204
extern const syscall : (sc:scno, args:... -> int64)
@@ -823,6 +834,10 @@
new : void#, newsz : size# \
-> int)
+ /* ulock */
+ const ulock_wait : (op : ulockop, uaddr : uint64#, val : uint64, timeout : uint32 -> int)
+ const ulock_wake : (op : ulockop, uaddr : uint64#, wakeval : uint64 -> int)
+
/* filled by start code */
extern var __cenvp : byte##
;;
@@ -1079,6 +1094,14 @@
/* all args already passed through a() or ar ptrs */
-> (syscall(Sys__sysctl, \
(mib : int#), a(mib.len), old, oldsz, new, newsz) : int)
+}
+
+const ulock_wait = {op, uaddr, val, timeout+ -> (syscall(Sysulock_wait, a(op), uaddr, val, a(timeout)) : int)
+}
+
+const ulock_wake = {op, uaddr, wakeval+ -> (syscall(Sysulock_wake, a(op), uaddr, a(wakeval)) : int)
}
const waitstatus = {st--- a/lib/sys/syserrno+linux.myr
+++ b/lib/sys/syserrno+linux.myr
@@ -35,4 +35,5 @@
const Epipe : errno = -32 /* Broken pipe */
const Edom : errno = -33 /* Math argument out of domain of func */
const Erange : errno = -34 /* Math result not representable */
+ const Etimedout : errno = -110 /* Operation timed out */
;;
--- a/lib/sys/syserrno+osx.myr
+++ b/lib/sys/syserrno+osx.myr
@@ -51,4 +51,5 @@
const Eprototype : errno = -41 /* Protocol wrong type for socket */
const Enoprotoopt : errno = -42 /* Protocol not available */
const Eprotonosupport : errno = -43 /* Protocol not supported */
+ const Etimedout : errno = -60 /* Operation timed out */
;;
--- a/lib/thread/bld.sub
+++ b/lib/thread/bld.sub
@@ -1,7 +1,10 @@
lib thread =
common.myr
hookstd.myr # install thread hooks
+ mutex+futex.myr
+ sem+futex.myr
mutex.myr # fallback, for unimplemented platforms
+ sem.myr # fallback, for unimplemented platforms
#generic fallbacks
ncpu.myr
@@ -9,7 +12,7 @@
# linux impl of basic thread primitives
#condvar+linux.myr
exit+linux-x64.s
- mutex+linux.myr
+ futex+linux.myr
ncpu+linux.myr
spawn+linux.myr
@@ -16,7 +19,7 @@
# freebsd impl of thread primitives
#condvar+freebsd.myr
exit+freebsd-x64.s
- mutex+freebsd.myr
+ futex+freebsd.myr
ncpu+freebsd.myr
spawn+freebsd.myr
@@ -29,6 +32,7 @@
# osx impl of thread primitives
#condvar+osx.myr
+ futex+osx.myr
spawn+osx.myr
start+osx-x64.s
@@ -37,11 +41,12 @@
atomic-impl+plan9-x64.s
mutex+plan9.myr
ncpu+plan9.myr
+ sem+plan9.myr
spawn+plan9.myr
# openbsd impl of thread primitives
exit+openbsd-x64.s
- mutex+openbsd:6.2.myr
+ futex+openbsd:6.2.myr
ncpu+openbsd.myr
spawn+openbsd.myr
--- /dev/null
+++ b/lib/thread/futex+freebsd.myr
@@ -1,0 +1,25 @@
+use sys
+
+use "common"
+
+pkg thread =
+ const ftxwait : (uaddr : uint64#, val : uint64, timeout : sys.timespec# -> int)
+ const ftxwake : (uaddr : uint64# -> int)
+;;
+
+const ftxwait = {uaddr, val, timeout+ if timeout == Zptr
+ -> sys.umtx_op((uaddr : void#), sys.Umtxwaituintpriv, (val : uint64), Zptr, Zptr)
+ ;;
+
+ var ut : sys._umtx_time = [
+ ._timeout = timeout#
+ ._flags = sys.Umtxabstime
+ ._clockid = 1 /* CLOCK_MONOTONIC. Not exported from sys. */
+ ]
+ -> sys.umtx_op((uaddr : void#), sys.Umtxwaituintpriv, (val : uint64), (sys.sizeof(sys._umtx_time) : void#), &ut)
+}
+
+const ftxwake = {uaddr+ -> sys.umtx_op((uaddr : void#), sys.Umtxwakepriv, 1, Zptr, Zptr)
+}
--- /dev/null
+++ b/lib/thread/futex+linux.myr
@@ -1,0 +1,16 @@
+use sys
+
+use "common"
+
+pkg thread =
+ const ftxwait : (uaddr : uint64#, val : uint64, timeout : sys.timespec# -> int)
+ const ftxwake : (uaddr : uint64# -> int)
+;;
+
+const ftxwait = {uaddr, val, timeout+ -> sys.futex((uaddr : int32#), sys.Futexwait | sys.Futexpriv, val, timeout, Zptr, 0)
+}
+
+const ftxwake = {uaddr+ -> sys.futex((uaddr : int32#), sys.Futexwake | sys.Futexpriv, 1, Zptr, Zptr, 0)
+}
--- /dev/null
+++ b/lib/thread/futex+openbsd:6.2.myr
@@ -1,0 +1,16 @@
+use sys
+
+use "common"
+
+pkg thread =
+ const ftxwait : (uaddr : uint64#, val : uint64, timeout : sys.timespec# -> int)
+ const ftxwake : (uaddr : uint64# -> int)
+;;
+
+const ftxwait = {uaddr, val, timeout+ -> sys.futex((uaddr : uint32#), sys.Futexwait, val, timeout, Zptr)
+}
+
+const ftxwake = {uaddr+ -> sys.futex((uaddr : uint32#), sys.Futexwake, 1, Zptr, Zptr)
+}
--- /dev/null
+++ b/lib/thread/futex+osx.myr
@@ -1,0 +1,44 @@
+use std
+use sys
+
+use "common"
+
+pkg thread =
+ const ftxwait : (uaddr : uint64#, val : uint64, timeout : sys.timespec# -> int)
+ const ftxwake : (uaddr : uint64# -> int)
+;;
+
+/*
+ * The ulock_ functions are undocumented but the relevant source can be found at
+ * https://github.com/apple/darwin-xnu/blob/0a798f6738bc1db01281fc08ae024145e84df927/bsd/kern/sys_ulock.c
+ */
+const ftxwait = {uaddr, val, timeout+ if timeout == Zptr
+ -> sys.ulock_wait(sys.Ulockcompareandwait, uaddr, val, 0)
+ ;;
+
+ var ts
+ var err = sys.clock_gettime(`sys.Clockmonotonic, &ts)
+ std.assert(err == 0, "error: clock_gettime returned {}\n", err)+
+ var usec = 0
+ if timeout.sec > ts.sec
+ var sec = (timeout.sec - ts.sec) * 1000
+ std.assert(sec <= 0xffffffff, "error: maximum futex timeout exceeded\n")
+ usec = (sec : uint32)
+ ;;
+ if timeout.nsec > ts.nsec
+ var nsec = (timeout.nsec - ts.nsec) / 1000
+ std.assert(usec + nsec > usec, "error: maximum futex timeout exceeded\n")
+ usec += nsec
+ ;;
+
+ if usec == 0
+ -> (std.Etimedout : int)
+ ;;
+ -> sys.ulock_wait(sys.Ulockcompareandwait, uaddr, val, 0)
+}
+
+const ftxwake = {uaddr+ -> sys.ulock_wake(sys.Ulockcompareandwait, uaddr, 0)
+}
--- a/lib/thread/mutex+freebsd.myr
+++ /dev/null
@@ -1,81 +1,0 @@
-use std
-use sys
-
-use "atomic"
-use "common"
-
-pkg thread =
- type mutex = struct
- _state : uint32
- ;;
-
- const mkmtx : (-> mutex)
- const mtxlock : (mtx : mutex# -> void)
- const mtxtrylock : (mtx : mutex# -> bool)
- const mtxunlock : (mtx : mutex# -> void)
-
- pkglocal const Unlocked = 0
- pkglocal const Locked = 1
- pkglocal const Contended = 2
-;;
-
-var nspin = 10 /* FIXME: pick a sane number, based on CPU count */
-
-const mkmtx = {- -> [._state = Unlocked]
-}
-
-const mtxlock = {mtx- var c
-
- /*
- Uncontended case: we get an unlocked mutex, and we lock it.
- */
- c = Locked
- for var i = 0; i < nspin; i++
- c = xcas(&mtx._state, Unlocked, Locked)
- if c == Unlocked
- -> void
- ;;
- sys.sched_yield()
- ;;
-
- /*
- Contended case: we set the lock state to Contended. This indicates that there
- the lock is locked, and we potentially have threads waiting on it, which means
- that we will need to wake them up.
- */
- if c == Locked
- c = xchg(&mtx._state, Contended)
- ;;
-
- while c != Unlocked
- sys.umtx_op( \
- (&mtx._state : void#), \
- sys.Umtxwaituintpriv, \
- (Contended : uint64), \
- Zptr, Zptr)
- c = xchg(&mtx._state, Contended)
- ;;
-}
-
-const mtxtrylock = {mtx- -> xcas(&mtx._state, Unlocked, Locked) == Unlocked
-}
-
-const mtxunlock = {mtx- /*
- Uncontended case: If the mutex state is not contended, and we still
- are uncontended by the xchg() call, then it's safe to simply return;
- nobody was waiting for us.
- */
- if mtx._state == Contended
- mtx._state = Unlocked
- elif xchg(&mtx._state, Unlocked) == Locked
- -> void
- ;;
-
- /* wake all threads: for some reason nwake */
- sys.umtx_op((&mtx._state : void#), sys.Umtxwakepriv, 1, Zptr, Zptr)
-}
-
--- /dev/null
+++ b/lib/thread/mutex+futex.myr
@@ -1,0 +1,75 @@
+use sys
+
+use "atomic"
+use "common"
+use "futex"
+
+pkg thread =
+ type mutex = struct
+ _state : uint64
+ ;;
+
+ const mkmtx : (-> mutex)
+ const mtxlock : (mtx : mutex# -> void)
+ const mtxtrylock : (mtx : mutex# -> bool)
+ const mtxunlock : (mtx : mutex# -> void)
+
+ pkglocal const Unlocked = 0
+ pkglocal const Locked = 1
+ pkglocal const Contended = 2
+;;
+
+var nspin = 10 /* FIXME: pick a sane number, based on CPU count */
+
+const mkmtx = {+ -> [._state = Unlocked]
+}
+
+const mtxlock = {mtx+ var c
+
+ /*
+ Uncontended case: we get an unlocked mutex, and we lock it.
+ */
+ c = Locked
+ for var i = 0; i < nspin; i++
+ c = xcas(&mtx._state, Unlocked, Locked)
+ if c == Unlocked
+ -> void
+ ;;
+ ;;
+
+ /*
+ Contended case: we set the lock state to Contended. This indicates that there
+ the lock is locked, and we potentially have threads waiting on it, which means
+ that we will need to wake them up.
+ */
+ if c == Locked
+ c = xchg(&mtx._state, Contended)
+ ;;
+
+ while c != Unlocked
+ ftxwait(&mtx._state, Contended, Zptr)
+ c = xchg(&mtx._state, Contended)
+ ;;
+}
+
+const mtxtrylock = {mtx+ -> xcas(&mtx._state, Unlocked, Locked) == Unlocked
+}
+
+const mtxunlock = {mtx+ /*
+ Uncontended case: If the mutex state is not contended, and we still
+ are uncontended by the xchg() call, then it's safe to simply return;
+ nobody was waiting for us.
+ */
+ if mtx._state == Contended
+ mtx._state = Unlocked
+ elif xchg(&mtx._state, Unlocked) == Locked
+ -> void
+ ;;
+
+ /* wake one thread */
+ ftxwake(&mtx._state)
+}
--- a/lib/thread/mutex+linux.myr
+++ /dev/null
@@ -1,76 +1,0 @@
-use std
-use sys
-
-use "atomic"
-use "common"
-
-pkg thread =
- type mutex = struct
- _state : int32
- ;;
-
- const mkmtx : (-> mutex)
- const mtxlock : (mtx : mutex# -> void)
- const mtxtrylock : (mtx : mutex# -> bool)
- const mtxunlock : (mtx : mutex# -> void)
-
- pkglocal const Unlocked = 0
- pkglocal const Locked = 1
- pkglocal const Contended = 2
-;;
-
-var nspin = 10 /* FIXME: pick a sane number, based on CPU count */
-
-const mkmtx = {- -> [._state = Unlocked]
-}
-
-const mtxlock = {mtx- var c
-
- /*
- Uncontended case: we get an unlocked mutex, and we lock it.
- */
- c = Locked
- for var i = 0; i < nspin; i++
- c = xcas(&mtx._state, Unlocked, Locked)
- if c == Unlocked
- -> void
- ;;
- ;;
-
- /*
- Contended case: we set the lock state to Contended. This indicates that there
- the lock is locked, and we potentially have threads waiting on it, which means
- that we will need to wake them up.
- */
- if c == Locked
- c = xchg(&mtx._state, Contended)
- ;;
-
- while c != Unlocked
- sys.futex(&mtx._state, sys.Futexwait | sys.Futexpriv, Contended, Zptr, Zptr, 0)
- c = xchg(&mtx._state, Contended)
- ;;
-}
-
-const mtxtrylock = {mtx- -> xcas(&mtx._state, Unlocked, Locked) == Unlocked
-}
-
-const mtxunlock = {mtx- /*
- Uncontended case: If the mutex state is not contended, and we still
- are uncontended by the xchg() call, then it's safe to simply return;
- nobody was waiting for us.
- */
- if mtx._state == Contended
- mtx._state = Unlocked
- elif xchg(&mtx._state, Unlocked) == Locked
- -> void
- ;;
-
- /* wake one thread */
- sys.futex(&mtx._state, sys.Futexwake | sys.Futexpriv, 1, Zptr, Zptr, 0)
-}
-
--- a/lib/thread/mutex+openbsd:6.2.myr
+++ /dev/null
@@ -1,76 +1,0 @@
-use std
-use sys
-
-use "atomic"
-use "common"
-
-pkg thread =
- type mutex = struct
- _state : uint32
- ;;
-
- const mkmtx : (-> mutex)
- const mtxlock : (mtx : mutex# -> void)
- const mtxtrylock : (mtx : mutex# -> bool)
- const mtxunlock : (mtx : mutex# -> void)
-
- pkglocal const Unlocked : uint32 = 0
- pkglocal const Locked : uint32 = 1
- pkglocal const Contended : uint32 = 2
-;;
-
-var nspin = 10 /* FIXME: pick a sane number, based on CPU count */
-
-const mkmtx = {- -> [._state = Unlocked]
-}
-
-const mtxlock = {mtx- var c
-
- /*
- Uncontended case: we get an unlocked mutex, and we lock it.
- */
- c = Locked
- for var i = 0; i < nspin; i++
- c = xcas(&mtx._state, Unlocked, Locked)
- if c == Unlocked
- -> void
- ;;
- ;;
-
- /*
- Contended case: we set the lock state to Contended. This indicates that there
- the lock is locked, and we potentially have threads waiting on it, which means
- that we will need to wake them up.
- */
- if c == Locked
- c = xchg(&mtx._state, Contended)
- ;;
-
- while c != Unlocked
- sys.futex(&mtx._state, sys.Futexwait, (Contended : int), Zptr, Zptr)
- c = xchg(&mtx._state, Contended)
- ;;
-}
-
-const mtxtrylock = {mtx- -> xcas(&mtx._state, Unlocked, Locked) == Unlocked
-}
-
-const mtxunlock = {mtx- /*
- Uncontended case: If the mutex state is not contended, and we still
- are uncontended by the xchg() call, then it's safe to simply return;
- nobody was waiting for us.
- */
- if mtx._state == Contended
- mtx._state = Unlocked
- elif xchg(&mtx._state, Unlocked) == Locked
- -> void
- ;;
-
- /* wake one thread */
- sys.futex(&mtx._state, sys.Futexwake, 1, Zptr, Zptr)
-}
-
--- /dev/null
+++ b/lib/thread/sem+futex.myr
@@ -1,0 +1,55 @@
+use std
+use sys
+
+use "atomic"
+use "common"
+use "futex"
+
+pkg thread =
+ type sem = struct
+ _val : uint64
+ ;;
+
+ const mksem : (v : uint32 -> sem)
+ const semwait : (s : sem# -> void)
+ const semtrywait : (s : sem# -> bool)
+ const sempost : (s : sem# -> void)
+;;
+
+const mksem = {v+ -> [._val = (v : uint64)]
+}
+
+const semwait = {s+ var v = 0
+
+ for ; ;
+ while (v = s._val) > 0
+ if xcas(&s._val, v, v - 1) == v
+ -> void
+ ;;
+ ;;
+ ftxwait(&s._val, v, Zptr)
+ ;;
+ -> void /* Unreachable */
+}
+
+const semtrywait = {s+ for ; ;
+ var v = xget(&s._val)
+ if v == 0
+ -> false
+ ;;
+ if xcas(&s._val, v, v - 1) == v
+ -> true
+ ;;
+ ;;
+ -> false /* Unreachable */
+}
+
+const sempost = {s+ std.assert((xadd(&s._val, 1) : uint32) != ~0x0, "error: semaphore overflowed\n")
+
+ /* Unconditionally wake one waiter */
+ ftxwake(&s._val)
+}
--- /dev/null
+++ b/lib/thread/sem+plan9.myr
@@ -1,0 +1,54 @@
+use std
+use sys
+
+use "atomic"
+use "common"
+
+pkg thread =
+ type sem = struct
+ _user : int32
+ _kern : int32
+ ;;
+
+ const mksem : (v : uint32 -> sem)
+ const semwait : (s : sem# -> void)
+ const semtrywait : (s : sem# -> bool)
+ const sempost : (s : sem# -> void)
+;;
+
+const mksem = {v+ -> [._user = v, ._kern = 0]
+}
+
+const semwait = {s+ var u = xadd(&s._user, -1)
+ std.assert(u != 0xffffffff, "error: semaphore underflowed\n")
+
+ /* When the userspace value is negative we fall back on the kernel semaphore */
+ if u <= 0
+ while sys.semacquire((&s._kern : uint32), 1) < 0
+ /* Interrupted, retry */
+ ;;
+ ;;
+}
+
+const semtrywait = {s+ for ; ;
+ var u = xget(&s._user)
+ if u <= 0
+ -> false
+ ;;
+ if xcas(&s._user, u, u - 1) == u
+ -> true
+ ;;
+ ;;
+ -> false /* Unreachable */
+}
+
+const sempost = {s+ var u = xadd(&s._user, 1)
+ std.assert(u != 0x7fffffff, "error: semaphore overflowed\n")
+ if u < 0
+ sys.semrelease((&s._kern : uint32), 1)
+ ;;
+}
--- /dev/null
+++ b/lib/thread/sem.myr
@@ -1,0 +1,66 @@
+use std
+
+use "atomic"
+
+pkg thread =
+ type sem = struct
+ _val : uint32
+ ;;
+
+ const mksem : (v : uint32 -> sem)
+ const semwait : (s : sem# -> void)
+ const semtrywait : (s : sem# -> bool)
+ const sempost : (s : sem# -> void)
+;;
+
+const mksem = {v+ -> [._val = v]
+}
+
+const semwait = {s+ var v = 0
+
+ for var i = 0; i < 1000; i++
+ if (v = xget(&s._val)) != 0 && xcas(&s._val, v, v - 1) == v
+ -> void
+ ;;
+ ;;
+
+ for var i = 0; i < 1000; i++
+ if (v = xget(&s._val)) != 0 && xcas(&s._val, v, v - 1) == v
+ -> void
+ ;;
+ std.nanosleep(10_000)
+ ;;
+
+ for var i = 0; i < 1000; i++
+ if (v = xget(&s._val)) != 0 && xcas(&s._val, v, v - 1) == v
+ -> void
+ ;;
+ std.nanosleep(100_000)
+ ;;
+
+ for ; ;
+ if (v = xget(&s._val)) != 0 && xcas(&s._val, v, v - 1) == v
+ -> void
+ ;;
+ std.nanosleep(1_000_000)
+ ;;
+}
+
+const semtrywait = {s+ for ; ;
+ var v = xget(&s._val)
+ if v == 0
+ -> false
+ ;;
+ if xcas(&s._val, v, v - 1) == v
+ -> true
+ ;;
+ ;;
+ -> false /* Unreachable */
+}
+
+const sempost = {s+ std.assert(xadd(&s._val, 1) != ~0x0, "error: semaphore overflowed\n")
+}
--- a/lib/thread/spawn+osx.myr
+++ b/lib/thread/spawn+osx.myr
@@ -79,7 +79,6 @@
const getstk = {szvar p, m
- std.put("allocating stack {x}\n", sz)p = sys.mmap((0 : byte#), sz, sys.Mprotrw, sys.Mpriv | sys.Manon, -1, 0)
if p == sys.Mapbad
-> p
--- /dev/null
+++ b/lib/thread/test/sem.myr
@@ -1,0 +1,42 @@
+use std
+use thread
+
+const Cap = 4
+
+var write : uint32 = 0, read : uint32 = 0
+var lock
+var len
+var cap
+var buf : int[Cap]
+var done : uint32 = 0
+
+const main = {+ lock = thread.mkmtx()
+ len = thread.mksem(0)
+ cap = thread.mksem(Cap)
+
+ thread.spawn({+ for var i = 0; i < 100000; i++
+ thread.semwait(&cap)
+ thread.mtxlock(&lock)
+ buf[write++ & (Cap - 1)] = i
+ thread.mtxunlock(&lock)
+ thread.sempost(&len)
+ ;;
+ })
+
+ thread.spawn({+ for var i = 0; i < 100000; i++
+ thread.semwait(&len)
+ thread.mtxlock(&lock)
+ std.assert(i == buf[read++ & (Cap - 1)], "semaphores are broken\n")
+ thread.mtxunlock(&lock)
+ thread.sempost(&cap)
+ ;;
+ thread.xset(&done, 1)
+ })
+
+ while thread.xget(&done) == 0
+ std.nanosleep(1_000_000)
+ ;;
+}