Implement blocking loops instead of busy ones (#106)

* try a different approach for blocking loops

* maybe like this

* remove unneeded code

* add wakeUp function

* try waking up via an async for luv schedulers

* try to be more accurate

* disable issue 37 to avoid distractions

* avoid race condition for now

* track wokenUp state explicitly

* don't try to wake up C++'s luv scheduler like that

* try with a sanity mutex

* Revert "disable issue 37 to avoid distractions"

This reverts commit a367b16.

* Reapply "disable issue 37 to avoid distractions"

This reverts commit a50a2a0.

* try with ThreadPoolDispatcher again

* loop at end of entrypoint tests to avoid dangling events

* how's this?

* it just doesn't work the way I want it to

* use semaphore to ensure onCompletion is done before returning

* add IShutDown to handle dispatcher shutdown

* add runMode argument to loop

* neko works now

* try with a simple semaphore

Author: Simn

src/hxcoro/dispatchers/IShutDown.hx

@@ -0,0 +1,5 @@
+package hxcoro.dispatchers;
+
+interface IShutDown {
+	function shutDown():Void;
+}

src/hxcoro/dispatchers/LuvDispatcher.cpp.hx

@@ -8,7 +8,7 @@ import haxe.coro.schedulers.IScheduler;
 import haxe.ds.Option;
 import hxcoro.schedulers.LuvScheduler;
 
-class LuvDispatcher extends Dispatcher
+class LuvDispatcher extends Dispatcher implements IShutDown
 {
 	final loop:LuvLoop;
 	final workQueue:AsyncDeque<()->Void>;

src/hxcoro/run/LoopRun.hx

@@ -2,7 +2,6 @@ package hxcoro.run;
 
 import haxe.Timer;
 import haxe.coro.context.Context;
-import hxcoro.exceptions.TimeoutException;
 import hxcoro.schedulers.ILoop;
 import hxcoro.task.CoroTask;
 import hxcoro.task.ICoroTask;
@@ -40,9 +39,21 @@ class LoopRun {
 		that are already running.
 	**/
 	static function awaitTaskCompletion<T>(loop:ILoop, task:ICoroTask<T>) {
+		#if target.threaded
+		final semaphore = new sys.thread.Semaphore(0);
+		task.onCompletion((_, _) -> {
+			loop.wakeUp();
+			semaphore.release();
+		});
+		#end
+
 		while (task.isActive()) {
-			loop.loop();
+			loop.loop(Once);
 		}
+
+		#if target.threaded
+		semaphore.acquire();
+		#end
 	}
 
 	/**

src/hxcoro/run/Setup.hx

@@ -1,5 +1,7 @@
 package hxcoro.run;
 
+import haxe.coro.schedulers.IScheduler;
+import hxcoro.dispatchers.IShutDown;
 import haxe.coro.BaseContinuation;
 import haxe.coro.context.Context;
 import haxe.coro.dispatchers.Dispatcher;
@@ -57,19 +59,25 @@ class Setup {
 
 	#if (cpp && hxcpp_luv_io)
 
-	static public function createLuv() {
+	static public function createLuvGen(createDispatcher:(cpp.luv.Luv.LuvLoop, IScheduler) -> Dispatcher) {
 		final loop = cpp.luv.Luv.allocLoop();
 		final scheduler = new hxcoro.schedulers.LuvScheduler(loop);
-		final dispatcher = new hxcoro.dispatchers.LuvDispatcher(loop, scheduler);
+		final dispatcher = createDispatcher(loop, scheduler);
 		function finalize() {
-			dispatcher.shutDown();
+			scheduler.shutDown();
+			if (dispatcher is IShutDown) {
+				(cast dispatcher : IShutDown).shutDown();
+			}
 			cpp.luv.Luv.stopLoop(loop);
-			cpp.luv.Luv.shutdownLoop(loop);
 			cpp.luv.Luv.freeLoop(loop);
 		}
 		return new LoopSetup(scheduler, dispatcher, finalize);
 	}
 
+	static public function createLuv() {
+		return createLuvGen((uvLoop, loop) -> new hxcoro.dispatchers.LuvDispatcher(uvLoop, loop));
+	}
+
 	#elseif interp
 
 	static public function createLuv() {

src/hxcoro/schedulers/EventLoopScheduler.hx

@@ -1,5 +1,7 @@
 package hxcoro.schedulers;
 
+import hxcoro.concurrent.BackOff;
+import hxcoro.schedulers.ILoop;
 import haxe.coro.IContinuation;
 import haxe.Timer;
 import haxe.Int64;
@@ -40,8 +42,9 @@ class HeapScheduler implements IScheduler {
 }
 
 class EventLoopScheduler extends HeapScheduler implements ILoop {
-	public function loop() {
+	function loopOnce() {
 		final currentTime = now();
+		var didSomething = false;
 		while (true) {
 			futureMutex.acquire();
 			var minimum = heap.minimum();
@@ -52,12 +55,27 @@ class EventLoopScheduler extends HeapScheduler implements ILoop {
 			heap.extract();
 			futureMutex.release();
 
+			didSomething = true;
 			minimum.dispatch();
 		}
 
 		futureMutex.release();
+		return didSomething;
 	}
 
+	public function loop(runMode:RunMode) {
+		switch (runMode) {
+			case NoWait:
+				loopOnce();
+			case Once:
+				while (!loopOnce()) {
+					BackOff.backOff();
+				}
+		}
+	}
+
+	public function wakeUp() {}
+
 	public function toString() {
 		return '[EventLoopScheduler]';
 	}

src/hxcoro/schedulers/ILoop.hx

@@ -1,5 +1,11 @@
 package hxcoro.schedulers;
 
+enum abstract RunMode(Int) {
+	final Once = 1;
+	final NoWait = 2;
+}
+
 interface ILoop {
-	function loop():Void;
+	function loop(runMode:RunMode):Void;
+	function wakeUp():Void;
 }

src/hxcoro/schedulers/LuvScheduler.cpp.hx

@@ -162,8 +162,12 @@ class LuvScheduler implements IScheduler implements ILoop {
 		consumeDeque(closeQueue, event -> event.stop());
 	}
 
-	public function loop() {
-		cpp.luv.Luv.runLoop(uvLoop, NoWait);
+	public function loop(runMode:RunMode) {
+		cpp.luv.Luv.runLoop(uvLoop, cast runMode);
+	}
+
+	public function wakeUp() {
+		@:privateAccess eventQueue.async.send();
 	}
 
 	public function shutDown() {

src/hxcoro/schedulers/LuvScheduler.eval.hx

@@ -143,8 +143,12 @@ class LuvScheduler implements IScheduler implements ILoop {
 		return uvLoop.now().toInt64();
 	}
 
-	public function loop() {
-		uvLoop.run(NOWAIT);
+	public function loop(runMode:ILoop.RunMode) {
+		uvLoop.run(cast runMode);
+	}
+
+	public function wakeUp() {
+		@:privateAccess eventQueue.async.send();
 	}
 
 	inline function consumeDeque<T>(deque:AsyncDeque<T>, f:T->Void) {

src/hxcoro/schedulers/ThreadAwareScheduler.hx

@@ -1,16 +1,18 @@
 package hxcoro.schedulers;
 
-import haxe.coro.IContinuation;
-import haxe.Exception;
-import sys.thread.Thread;
-import hxcoro.ds.CircularVector;
-import haxe.Timer;
 import haxe.Int64;
-import sys.thread.Tls;
-import sys.thread.Deque;
-import haxe.exceptions.ArgumentException;
+import haxe.Timer;
+import haxe.coro.IContinuation;
 import haxe.coro.schedulers.IScheduler;
 import haxe.coro.schedulers.ISchedulerHandle;
+import haxe.exceptions.ArgumentException;
+import hxcoro.concurrent.AtomicInt;
+import hxcoro.ds.CircularVector;
+import hxcoro.schedulers.ILoop;
+import sys.thread.Deque;
+import sys.thread.Semaphore;
+import sys.thread.Thread;
+import sys.thread.Tls;
 
 private class CircularQueueData {
 	public var read:Int;
@@ -82,13 +84,15 @@ class ThreadAwareScheduler implements IScheduler implements ILoop {
 	final queueTls:Tls<TlsQueue>;
 	final queueDeque:Deque<TlsQueueEvent>;
 	final rootEvent:ScheduledEvent;
+	final semaphore:Semaphore;
 	var firstQueue:Null<TlsQueue>;
 
 	public function new() {
 		heap = new MinimumHeap();
 		queueTls = new Tls();
 		queueDeque = new Deque();
 		rootEvent = new ScheduledEvent(null, 0);
+		semaphore = new Semaphore(0);
 	}
 
 	function getTlsQueue():TlsQueue {
@@ -114,9 +118,8 @@ class ThreadAwareScheduler implements IScheduler implements ILoop {
 		}
 
 		final event = new ScheduledEvent(cont, now() + ms);
-
 		getTlsQueue().add(event);
-
+		semaphore.release();
 		return event;
     }
 
@@ -151,7 +154,7 @@ class ThreadAwareScheduler implements IScheduler implements ILoop {
 		}
 	}
 
-	public function loop() {
+	function loopNoWait() {
 		final currentTime = now();
 
 		// First we consume the coordination deque so we know all queues.
@@ -193,15 +196,41 @@ class ThreadAwareScheduler implements IScheduler implements ILoop {
 			current = current.next;
 		}
 		var event:Null<ScheduledEvent> = rootEvent;
+		var didSomething = false;
 		while (true) {
 			event = event.next;
 			if (event == null) {
-				break;
+				return didSomething;
 			}
+			didSomething = true;
 			event.dispatch();
 		}
 	}
 
+	public function loop(runMode:RunMode) {
+		var wasWokenUp = false;
+		while(true) {
+			if (loopNoWait() || wasWokenUp || runMode == NoWait) {
+				// If we did something we're fine.
+				return;
+			}
+			final minimum = heap.minimum();
+			if (minimum != null) {
+				// If there's a scheduled event, its due time is our max timeout time.
+				final timeout:Float = (Int64.toInt(minimum.runTime - now())) / 1000;
+				semaphore.tryAcquire(timeout);
+			} else {
+				// Otherwise we wait until something happens.
+				semaphore.acquire();
+			}
+			wasWokenUp = true;
+		}
+	}
+
+	public function wakeUp() {
+		semaphore.release();
+	}
+
 	public function dump() {
 		Sys.println("ThreadAwareScheduler");
 		Sys.println('\theap minimum: ${heap.minimum()}');

src/hxcoro/schedulers/VirtualTimeScheduler.hx

@@ -1,5 +1,6 @@
 package hxcoro.schedulers;
 
+import hxcoro.schedulers.ILoop;
 import haxe.Int64;
 import haxe.exceptions.ArgumentException;
 
@@ -55,7 +56,9 @@ class VirtualTimeScheduler extends EventLoopScheduler.HeapScheduler implements I
 		return hasMoreEvents;
 	}
 
-	public function loop() {
-		while (advanceBy(1)) {}
+	public function loop(runMode:RunMode) {
+		advanceBy(1);
 	}
+
+	public function wakeUp() {}
 }