Add some thread tests (and fix eval) (#7992)

* keep track of who owns a mutex

* factor out and fix Deque

* use a proper Deque for thread messaging

* turns out readMessage is actually static

* try to run some threads tests

* casing is important you dirty Windows peasant

* add Deque test

* disable Deque test for Java

* allow lock timeout

* use less threads because AppVeyor chokes

* don't run thread tests for now

.gitignore

@@ -121,3 +121,4 @@ tests/benchs/export/
 tests/benchs/dump/
 tests/display/.unittest/
 tests/unit/.unittest/
+tests/threads/export/

src/macro/eval/evalDebugSocket.ml

@@ -96,8 +96,12 @@ let var_to_json name value vio env =
 		| VVector vv -> jv "Vector" (array_elems (Array.to_list vv)) (Array.length vv)
 		| VInstance vi ->
 			let class_name = EvalDebugMisc.safe_call env.env_eval EvalPrinting.value_string v in
-			let fields = instance_fields vi in
-			jv class_name (class_name) (List.length fields)
+			let num_children = match vi.ikind with
+			| IMutex _ -> 1
+			| IThread _ -> 1
+			| _ -> List.length (instance_fields vi)
+			in
+			jv class_name (class_name) num_children
 		| VPrototype proto ->
 			let fields = proto_fields proto in
 			jv "Anonymous" (s_proto_kind proto).sstring (List.length fields)
@@ -148,7 +152,7 @@ let output_threads ctx =
 	let fold id eval acc =
 		(JObject [
 			"id",JInt id;
-			"name",JString eval.thread.tname
+			"name",JString (Printf.sprintf "Thread %i" (Thread.id eval.thread.tthread));
 		]) :: acc
 	in
 	let threads = IntMap.fold fold ctx.evals [] in
@@ -266,6 +270,10 @@ let output_inner_vars v env =
 			StringHashtbl.fold (fun s (_,v) acc ->
 				(s,v) :: acc
 			) h []
+		| VInstance {ikind = IMutex mutex} ->
+			["owner",match mutex.mowner with None -> vnull | Some id -> vint id]
+		| VInstance {ikind = IThread thread} ->
+			["id",vint (Thread.id thread.tthread)]
 		| VInstance vi ->
 			let fields = instance_fields vi in
 			List.map (fun (n,v) ->

src/macro/eval/evalMain.ml

@@ -102,11 +102,9 @@ let create com api is_macro =
 	let eval = {
 		env = null_env;
 		thread = {
-			tname = "mainThread";
 			tthread = Thread.self();
-			tchannel = Event.new_channel();
-			tqueue = Queue.create ();
 			tstorage = IntMap.empty;
+			tdeque = EvalStdLib.Deque.create();
 		};
 		debug_channel = Event.new_channel ();
 		debug_state = DbgRunning;

src/macro/eval/evalStdLib.ml

@@ -54,6 +54,45 @@ let encode_i64_direct i64 =
 	let high = Int64.to_int32 (Int64.shift_right_logical i64 32) in
 	encode_i64 low high
 
+module Deque = struct
+	let create () = {
+		dvalues = [];
+		dmutex = Mutex.create();
+	}
+
+	let add this i =
+		Mutex.lock this.dmutex;
+		this.dvalues <- this.dvalues @ [i];
+		Mutex.unlock this.dmutex;
+		vnull
+
+	let pop this blocking =
+		let rec loop () =
+			Mutex.lock this.dmutex;
+			match this.dvalues with
+			| v :: vl ->
+				this.dvalues <- vl;
+				Mutex.unlock this.dmutex;
+				v
+			| [] ->
+				if not blocking then begin
+					Mutex.unlock this.dmutex;
+					vnull
+				end else begin
+					Mutex.unlock this.dmutex;
+					Thread.yield();
+					loop()
+				end
+		in
+		loop()
+
+	let push this i =
+		Mutex.lock this.dmutex;
+		this.dvalues <- i :: this.dvalues;
+		Mutex.unlock this.dmutex;
+		vnull
+end
+
 
 module StdEvalVector = struct
 	let this this = match this with
@@ -740,46 +779,20 @@ module StdDeque = struct
 		| VInstance {ikind = IDeque d} -> d
 		| _ -> unexpected_value vthis "Deque"
 
-	let lock_mutex = Mutex.create ()
-
 	let add = vifun1 (fun vthis i ->
 		let this = this vthis in
-		Mutex.lock lock_mutex;
-		this.dvalues <- this.dvalues @ [i];
-		ignore(Event.poll(Event.send this.dchannel i));
-		Mutex.unlock lock_mutex;
-		vnull;
+		Deque.add this i
 	)
 
 	let pop = vifun1 (fun vthis blocking ->
 		let this = this vthis in
-		let rec loop () =
-			Mutex.lock lock_mutex;
-			match this.dvalues with
-			| v :: vl ->
-				this.dvalues <- vl;
-				Mutex.unlock lock_mutex;
-				v
-			| [] ->
-				if blocking <> VTrue then begin
-					Mutex.unlock lock_mutex;
-					vnull
-				end else begin
-					Mutex.unlock lock_mutex;
-					ignore(Event.sync(Event.receive this.dchannel));
-					loop()
-				end
-		in
-		loop()
+		let blocking = decode_bool blocking in
+		Deque.pop this blocking
 	)
 
 	let push = vifun1 (fun vthis i ->
 		let this = this vthis in
-		Mutex.lock lock_mutex;
-		this.dvalues <- i :: this.dvalues;
-		ignore(Event.poll(Event.send this.dchannel i));
-		Mutex.unlock lock_mutex;
-		vnull;
+		Deque.push this i
 	)
 end
 
@@ -1729,19 +1742,25 @@ module StdMutex = struct
 
 	let acquire = vifun0 (fun vthis ->
 		let mutex = this vthis in
-		Mutex.lock mutex;
+		Mutex.lock mutex.mmutex;
+		mutex.mowner <- Some (Thread.id (Thread.self()));
 		vnull
 	)
 
 	let release = vifun0 (fun vthis ->
 		let mutex = this vthis in
-		Mutex.unlock mutex;
+		mutex.mowner <- None;
+		Mutex.unlock mutex.mmutex;
 		vnull
 	)
 
 	let tryAcquire = vifun0 (fun vthis ->
 		let mutex = this vthis in
-		vbool (Mutex.try_lock mutex)
+		if Mutex.try_lock mutex.mmutex then begin
+			mutex.mowner <- Some (Thread.id (Thread.self()));
+			vtrue
+		end else
+			vfalse
 	)
 end
 
@@ -2664,24 +2683,15 @@ module StdThread = struct
 		encode_instance key_eval_vm_Thread ~kind:(IThread eval.thread)
 	)
 
-	let readMessage = vifun1 (fun vthis blocking ->
-		let this = this vthis in
-		if not (Queue.is_empty this.tqueue) then
-			Queue.pop this.tqueue
-		else if blocking <> VTrue then
-			vnull
-		else begin
-			let event = Event.receive this.tchannel in
-			ignore(Event.sync event);
-			Queue.pop this.tqueue
-		end
+	let readMessage = vfun1 (fun blocking ->
+		let eval = get_eval (get_ctx()) in
+		let blocking = decode_bool blocking in
+		Deque.pop eval.thread.tdeque blocking
 	)
 
 	let sendMessage = vifun1 (fun vthis msg ->
 		let this = this vthis in
-		Queue.add msg this.tqueue;
-		ignore(Event.poll (Event.send this.tchannel msg));
-		vnull
+		Deque.push this.tdeque msg
 	)
 
 	let yield = vfun0 (fun () ->
@@ -3206,14 +3216,10 @@ let init_constructors builtins =
 						close();
 						raise exc
 				in
-				let eval = get_eval ctx in
-				let name = kind_name eval eval.env.env_info.kind in
 				let thread = {
-					tname = name;
 					tthread = Obj.magic ();
-					tchannel = Event.new_channel ();
-					tqueue = Queue.create ();
 					tstorage = IntMap.empty;
+					tdeque = Deque.create();
 				} in
 				thread.tthread <- Thread.create f thread;
 				encode_instance key_eval_vm_Thread ~kind:(IThread thread)
@@ -3221,7 +3227,11 @@ let init_constructors builtins =
 		);
 	add key_eval_vm_Mutex
 		(fun _ ->
-			encode_instance key_eval_vm_Mutex ~kind:(IMutex (Mutex.create ()))
+			let mutex = {
+				mmutex = Mutex.create();
+				mowner = None;
+			} in
+			encode_instance key_eval_vm_Mutex ~kind:(IMutex mutex)
 		);
 	add key_eval_vm_Lock
 		(fun _ ->
@@ -3240,11 +3250,7 @@ let init_constructors builtins =
 		);
 	add key_eval_vm_Deque
 		(fun _ ->
-			let deque = {
-				dvalues = [];
-				dchannel = Event.new_channel();
-			} in
-			encode_instance key_eval_vm_Deque ~kind:(IDeque deque)
+			encode_instance key_eval_vm_Deque ~kind:(IDeque (Deque.create()))
 		)
 
 let init_empty_constructors builtins =
@@ -3616,12 +3622,12 @@ let init_standard_library builtins =
 		"delay",StdThread.delay;
 		"exit",StdThread.exit;
 		"join",StdThread.join;
+		"readMessage",StdThread.readMessage;
 		"self",StdThread.self;
 		"yield",StdThread.yield;
 	] [
 		"id",StdThread.id;
 		"kill",StdThread.kill;
-		"readMessage",StdThread.readMessage;
 		"sendMessage",StdThread.sendMessage;
 	];
 	init_fields builtins (["eval";"vm"],"Tls") [] [

src/macro/eval/evalValue.ml

@@ -155,7 +155,7 @@ and vinstance_kind =
 	| IOutChannel of out_channel (* FileOutput *)
 	| ISocket of Unix.file_descr
 	| IThread of vthread
-	| IMutex of Mutex.t
+	| IMutex of vmutex
 	| ILock of vlock
 	| ITls of int
 	| IDeque of vdeque
@@ -192,16 +192,19 @@ and venum_value = {
 }
 
 and vthread = {
-	tname : string;
 	mutable tthread : Thread.t;
-	tchannel : value Event.channel;
-	tqueue : value Queue.t;
+	tdeque : vdeque;
 	mutable tstorage : value IntMap.t;
 }
 
 and vdeque = {
 	mutable dvalues : value list;
-	dchannel : value Event.channel;
+	dmutex : Mutex.t;
+}
+
+and vmutex = {
+	mmutex : Mutex.t;
+	mutable mowner : int option; (* thread ID *)
 }
 
 and vlock = {

std/eval/vm/Thread.hx

@@ -19,6 +19,7 @@
  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  * DEALINGS IN THE SOFTWARE.
  */
+
 package eval.vm;
 
 /**
@@ -33,7 +34,7 @@ extern class Thread {
 		Exceptions caused while executing `f` are printed to stderr and are not
 		propagated to the parent thread.
 	**/
-	function new(f:Void -> Void):Void;
+	function new(f:Void->Void):Void;
 
 	/**
 		Return the identifier of the given thread. A thread identifier is an integer
@@ -76,12 +77,11 @@ extern class Thread {
 		to other threads.
 	**/
 	static function yield():Void;
-
 	// neko API
-
-	function readMessage<T>(block:Bool):T;
+	static function readMessage<T>(block:Bool):T;
 	function sendMessage<T>(msg:T):Void;
-
-	static inline function create(f:Void -> Void):Thread return new Thread(f);
-	static inline function current():Thread return self();
-}
+	static inline function create(f:Void->Void):Thread
+		return new Thread(f);
+	static inline function current():Thread
+		return self();
+}

tests/runci/Config.hx

@@ -19,6 +19,7 @@ class Config {
 	static public final serverDir = cwd + "server/";
 	static public final sourcemapsDir = cwd + "sourcemaps/";
 	static public final nullSafetyDir = cwd + "nullsafety/";
+	static public final threadsDir = cwd + "threads/";
 
 	static public final ci:Null<Ci> =
 		if (Sys.getEnv("TRAVIS") == "true")

tests/runci/targets/Cpp.hx

@@ -68,6 +68,10 @@ class Cpp {
 		runCommand("haxe", ["compile-cpp.hxml"]);
 		runCpp("bin/cpp/Main-debug", []);
 
+		// changeDirectory(threadsDir);
+		// runCommand("haxe", ["build.hxml", "-cpp", "export/cpp"]);
+		// runCpp("export/cpp/Main");
+
 		// if (Sys.systemName() == "Mac")
 		// {
 		// 	changeDirectory(miscDir + "cppObjc");

tests/runci/targets/Hl.hx

@@ -44,7 +44,7 @@ class Hl {
         runCommand("cmake", [
             "--build", hlBuild
         ]);
-        
+
         addToPATH(Path.join([hlBuild, "bin"]));
         runCommand("hl", ["--version"]);
     }
@@ -54,6 +54,10 @@ class Hl {
         runCommand("haxe", ["compile-hl.hxml"].concat(args));
         runCommand("hl", ["bin/unit.hl"]);
 
+		// changeDirectory(threadsDir);
+		// runCommand("haxe", ["build.hxml", "-hl", "export/threads.hl"]);
+		// runCommand("hl", ["export/threads.hl"]);
+
         // TODO sys test
     }
 }

tests/runci/targets/Java.hx

@@ -23,6 +23,10 @@ class Java {
 		runCommand("haxe", ["compile-java.hxml"]);
 		runCommand("java", ["-jar", "bin/java/Main-Debug.jar"]);
 
+		// changeDirectory(threadsDir);
+		// runCommand("haxe", ["build.hxml", "-java", "export/java"]);
+		// runCommand("java", ["-jar", "export/java/Main.jar"]);
+
 		infoMsg("Testing java-lib extras");
 		changeDirectory('$unitDir/bin');
 		runCommand("git", ["clone", "https://github.com/waneck/java-lib-tests.git", "--depth", "1"], true);

tests/runci/targets/Macro.hx

@@ -31,5 +31,8 @@ class Macro {
 		changeDirectory(sysDir);
 		runCommand("haxe", ["compile-macro.hxml"]);
 		runCommand("haxe", ["compile-each.hxml", "--run", "Main"]);
+
+		// changeDirectory(threadsDir);
+		// runCommand("haxe", ["build.hxml", "--interp"]);
 	}
 }

tests/runci/targets/Neko.hx

@@ -12,5 +12,9 @@ class Neko {
 		changeDirectory(sysDir);
 		runCommand("haxe", ["compile-neko.hxml"]);
 		runCommand("neko", ["bin/neko/sys.n"]);
+
+		// changeDirectory(threadsDir);
+		// runCommand("haxe", ["build.hxml", "-neko", "export/threads.n"]);
+		// runCommand("neko", ["export/threads.n"]);
 	}
 }

tests/threads/build.hxml

@@ -0,0 +1,4 @@
+-cp src
+-D analyzer-optimize
+-main Main
+-lib utest

tests/threads/res/tree1.txt

@@ -0,0 +1,29 @@
+  (2)
+  (3)
+    (4)
+    (5)
+      (6)
+    (7)
+      (8)
+    (9)
+    (10)
+      (11)
+  (12)
+    (13)
+  (14)
+    (15)
+  (16)
+    (17)
+      (18)
+      (19)
+  (20)
+    (21)
+      (22)
+        (23)
+          (24)
+          (25)
+    (26)
+    (27)
+      (28)
+        (29)
+  (30)

tests/threads/src/Main.hx

@@ -0,0 +1,19 @@
+import cases.WeirdTreeSum;
+import utest.Runner;
+import utest.ui.Report;
+
+class Main {
+	static function main() {
+		var runner = new Runner();
+		runner.addCase(new cases.WeirdTreeSum());
+		#if !hl // no Lock
+		#if !java // Deque broken?
+		runner.addCase(new cases.DequeBrackets());
+		#end
+		#end
+		var report = Report.create(runner);
+		report.displayHeader = AlwaysShowHeader;
+		report.displaySuccessResults = NeverShowSuccessResults;
+		runner.run();
+	}
+}

tests/threads/src/cases/DequeBrackets.hx

@@ -0,0 +1,81 @@
+package cases;
+
+import utest.Assert;
+import haxe.ds.GenericStack;
+import utest.ITest;
+
+class DequeBrackets implements ITest {
+	public function new() {}
+
+	/**
+		Spawns a thread to insert bracket pairs into a Deque. The opening
+		one is placed in front, the closing one in the back. This is going
+		to result in something like `([{<>}])` which we check for at the end.
+	**/
+	public function test() {
+		Sys.println("Running DequeBrackets");
+		var deque = new Deque();
+		var dequeMutex = new Mutex();
+		function add(open:String, close:String) {
+			dequeMutex.acquire();
+			deque.push(open);
+			deque.add(close);
+			dequeMutex.release();
+		}
+
+		var pairs = [
+			{open: "(", close: ")"},
+			{open: "[", close: "]"},
+			{open: "{", close: "}"},
+			{open: "<", close: ">"}
+		];
+		var iterationsPerThread = 100;
+
+		var lock = new Lock();
+		var self = Thread.current();
+		Thread.create(() -> {
+			for (_ in 0...pairs.length) {
+				Assert.isTrue(lock.wait(2.));
+			}
+			self.sendMessage("done");
+		});
+		var threads = [];
+		for (pair in pairs) {
+			threads.push(Thread.create(() -> {
+				Thread.readMessage(true);
+				for (_ in 0...iterationsPerThread) {
+					add(pair.open, pair.close);
+					Sys.sleep(0.001); // sleep a bit to increase chaos
+				}
+				lock.release();
+			}));
+		}
+		for (thread in threads) {
+			thread.sendMessage("go");
+		}
+		switch (Thread.readMessage(true)) {
+			case "done":
+			case s:
+				Assert.fail("Unexpected message: " + s);
+		}
+		var stack = new GenericStack<String>();
+		function pop() {
+			return deque.pop(false);
+		}
+		for (_ in 0...pairs.length * iterationsPerThread) {
+			stack.add(pop());
+		}
+		for (elt in stack) {
+			var expected = switch (elt) {
+				case "(": ")";
+				case "<": ">";
+				case "{": "}";
+				case "[": "]";
+				case s:
+					Assert.fail("Unexpected " + s);
+					s;
+			}
+			Assert.equals(expected, pop());
+		}
+	}
+}

tests/threads/src/cases/WeirdTreeSum.hx

@@ -0,0 +1,176 @@
+package cases;
+
+import sys.io.File;
+import utest.Assert;
+
+using StringTools;
+class Ref<T> {
+	public var value:T;
+
+	public function new(value:T) {
+		this.value = value;
+	}
+
+	public function toString() {
+		return Std.string(value);
+	}
+}
+
+@:using(cases.WeirdTreeSum.TreeTools)
+
+enum Tree<T> {
+	Node(value:Ref<T>, children:Array<Tree<T>>);
+	Leaf(value:Ref<T>);
+}
+
+class TreeTools {
+	static public function toString<T>(tree:Tree<T>) {
+		var buf = new StringBuf();
+		function loop(tabs:String, tree:Tree<T>) {
+			switch (tree) {
+				case Leaf(value):
+					buf.add('$tabs($value)\n');
+				case Node(value, children):
+					buf.add('$tabs($value)\n');
+					for (child in children) {
+						loop(tabs + "  ", child);
+					}
+			}
+		}
+		loop("", tree);
+		return buf.toString();
+	}
+
+	static public function copy<T>(tree:Tree<T>) {
+		return switch (tree) {
+			case Leaf(ref): Leaf(new Ref(ref.value));
+			case Node(ref, children): Node(new Ref(ref.value), children.map(copy));
+		}
+	}
+}
+
+typedef TreeNode<T> = {
+	var indent:Int;
+	var children:Array<Tree<T>>;
+}
+
+class WeirdTreeSum implements utest.ITest {
+	public function new() {}
+
+	public function test() {
+		Sys.println("Running WeirdTreeSum");
+		var fileContent = File.getContent("res/tree1.txt");
+		var buf = new StringBuf();
+		buf.add("(1)\n");
+		for (i in 0...10) {
+			buf.add(fileContent + "\n");
+		}
+		var tree = parseTree(buf.toString().trim())[0];
+		compare(tree);
+	}
+
+	static function compare(tree:Tree<Int>) {
+		var treeSingle = tree.copy();
+		var treeMulti = tree.copy();
+		traverseSingleThreaded(treeSingle);
+		traverseMultiThreaded(treeMulti);
+		Assert.equals(treeSingle.toString(), treeMulti.toString());
+	}
+
+	static function parseTree(s:String) {
+		var split = s.split("\n");
+		var children = [];
+		var nullNode:Dynamic = null;
+		var node = {elt: {children: children, indent: -1}, next: nullNode};
+		for (offset in 0...split.length) {
+			var line = split[offset];
+			var lineIndent = line.indexOf("(");
+			var value = line.substr(lineIndent + 1, line.indexOf(")") - lineIndent - 1);
+			var ref = new Ref(Std.parseInt(value));
+			while (lineIndent <= node.elt.indent) {
+				node = node.next;
+			}
+			if (offset == split.length - 1) {
+				node.elt.children.push(Leaf(ref));
+			} else if (lineIndent >= split[offset + 1].indexOf("(")) {
+				node.elt.children.push(Leaf(ref));
+			} else {
+				var children = [];
+				node.elt.children.push(Node(ref, children));
+				node = {elt: {children: children, indent: lineIndent}, next: node};
+			}
+		}
+		return children;
+	}
+
+	static function traverseSingleThreaded(tree:Tree<Int>) {
+		return switch (tree) {
+			case Leaf(value): value.value;
+			case Node(value, children):
+				for (child in children) {
+					value.value += traverseSingleThreaded(child);
+				}
+				value.value;
+		}
+	}
+
+	/**
+		For each node, we spawn N threads where N is the number of children
+		that node has. Each thread waits for a "go" message from its parent
+		thread, and then calculates the node value recursively.
+
+		Once done, it sends a "done" message to the waiter thread, which
+		counts how many results we're still expecting. Once that number reaches
+		0, it sends "done!" to the parent thread which may then return.
+
+		This isn't meant to be fast or elegant, but tests message passing between
+		a couple thousand threads.
+	**/
+	static function traverseMultiThreaded(tree:Tree<Int>) {
+		return switch (tree) {
+			case Leaf(value): value.value;
+			case Node(value, children):
+				var self = Thread.current();
+				var todo = children.length;
+				var valueMutex = new Mutex();
+				var todoMutex = new Mutex();
+				var waiterThread = Thread.create(() -> {
+					while (true) {
+						switch (Thread.readMessage(true)) {
+							case "done":
+								todoMutex.acquire();
+								todo--;
+								todoMutex.release();
+								if (todo == 0) {
+									break;
+								}
+							case _:
+								throw "Something went wrong";
+						}
+					}
+					self.sendMessage("done!");
+				});
+				var childrenThreads = [];
+				for (child in children) {
+					var childThread = Thread.create(() -> {
+						switch (Thread.readMessage(true)) {
+							case "go":
+							case _: throw "Something went wrong";
+						}
+						valueMutex.acquire();
+						value.value += traverseMultiThreaded(child);
+						valueMutex.release();
+						waiterThread.sendMessage("done");
+					});
+					childrenThreads.push(childThread);
+				}
+				for (childThread in childrenThreads) {
+					childThread.sendMessage("go");
+				}
+				switch (Thread.readMessage(true)) {
+					case "done!": value.value;
+					case _: throw "Something went wrong";
+				}
+		}
+	}
+}

tests/threads/src/import.hx

@@ -0,0 +1,31 @@
+#if neko
+import neko.vm.Thread;
+import neko.vm.Deque;
+import neko.vm.Lock;
+import neko.vm.Tls;
+import neko.vm.Mutex;
+#elseif cpp
+import cpp.vm.Thread;
+import cpp.vm.Deque;
+import cpp.vm.Lock;
+import cpp.vm.Tls;
+import cpp.vm.Mutex;
+#elseif java
+import java.vm.Thread;
+import java.vm.Deque;
+import java.vm.Lock;
+import java.vm.Tls;
+import java.vm.Mutex;
+#elseif eval
+import eval.vm.Thread;
+import eval.vm.Deque;
+import eval.vm.Lock;
+import eval.vm.Tls;
+import eval.vm.Mutex;
+#elseif hl
+import hl.vm.Thread;
+import hl.vm.Deque;
+// import hl.vm.Lock; // TODO
+import hl.vm.Tls;
+import hl.vm.Mutex;
+#end