ref: 183f3d40a1dbc2254f1e01de21eb022996f40ad2
parent: e70e83ae4362b60e9eb2134fb5d388576ee90bdf
author: Ori Bernstein <ori@eigenstate.org>
date: Sat Nov 3 12:26:58 EDT 2018
First attempt at a queue API. Simple, naive, but seems to work.
--- a/lib/thread/bld.sub
+++ b/lib/thread/bld.sub
@@ -5,6 +5,7 @@
# higher level apis
future.myr
do.myr
+ queue.myr
# generic fallbacks
condvar.myr
--- /dev/null
+++ b/lib/thread/queue.myr
@@ -1,0 +1,65 @@
+use std
+use "mutex"
+use "condvar"
+
+pkg thread =
+ type queue(@a) = struct
+ hd : qnode(@a)#
+ tl : qnode(@a)#
+ mtx : mutex
+ cv : cond
+ ;;
+
+ type qnode(@a) = struct
+ v : @a
+ next : qnode(@a)#
+ ;;
+
+ generic qinit : (q : queue(@a)# -> void)
+ generic qget : (q : queue(@a)# -> @a)
+ generic qput : (q : queue(@a)#, v : @a -> void)
+;;
+
+generic Znode = (0 : qnode(@a)#)
+
+generic qinit = {q+ q.hd = Znode
+ q.tl = Znode
+ q.mtx = mkmtx()
+ q.cv = mkcond(&q.mtx)
+}
+
+
+generic qput = {q, v : @a+ var n : qnode(@a)#
+
+ n = std.mk([.next=Znode, .v=v])
+ mtxlock(&q.mtx)
+ if q.hd == Znode
+ q.hd = n
+ q.tl = n
+ else
+ q.tl.next = n
+ q.tl = n
+ ;;
+ condsignal(&q.cv)
+ mtxunlock(&q.mtx)
+}
+
+generic qget = {q+ var n, v
+
+ mtxlock(&q.mtx)
+:again
+ if q.hd == Znode
+ condwait(&q.cv)
+ goto again
+ else
+ n = q.hd
+ q.hd = q.hd.next
+ ;;
+ mtxunlock(&q.mtx)
+ v = n.v
+ std.free(n)
+ -> v
+}
--- /dev/null
+++ b/lib/thread/test/queue.myr
@@ -1,0 +1,46 @@
+use std
+use thread
+use testr
+
+const Nproduced = 100
+
+const main = {+ testr.run([
+ [.name="simple-singlethreaded", .fn=simple],
+ [.name="slow-producer", .fn={ctx; producerconsumer(ctx, 100, 100)}],+ [.name="fast-producer", .fn={ctx; producerconsumer(ctx, 10_000, 0)}],+ ][:])
+}
+
+const simple = {ctx+ var q
+
+ /* simple test */
+ thread.qinit(&q)
+ thread.qput(&q, 123)
+ thread.qput(&q, 246)
+
+ testr.check(ctx, thread.qget(&q) == 123, "expected 123")
+ testr.check(ctx, thread.qget(&q) == 246, "expected 246")
+}
+
+const producerconsumer = {ctx, count, delay+ var qp
+
+ qp = std.alloc()
+ thread.qinit(qp)
+ thread.spawn({; producer(qp, count, delay)})+
+ for var i = 0; i < count; i++
+ testr.check(ctx, thread.qget(qp) == i, "expected {}\n", i)+ ;;
+}
+
+const producer = {qp, count, delay+ for var i = 0; i < count; i++
+ thread.qput(qp, i)
+ if delay > 0
+ std.usleep(delay)
+ ;;
+ ;;
+}