Add callSoonThreadSafe

Author: dcarp

src/asynchronous/events.d

@@ -395,15 +395,17 @@ abstract class EventLoop
         return callback;
     }
 
-    /+
     /**
      * Like $(D_PSYMBOL call_soon()), but thread safe.
      */
-    final auto callSoonThreadsafe(alias callback, Args...)(Args args)
+    final auto callSoonThreadSafe(Dg, Args...)(Dg dg, Args args)
     {
-        return callLater!callback(0, args);
+        auto callback = new Callback!(Dg, Args)(this, dg, args);
+
+        scheduleCallbackThreadSafe(callback);
+
+        return callback;
     }
-    +/
 
     // Delayed calls
 
@@ -1180,6 +1182,8 @@ protected:
 
     void scheduleCallback(CallbackHandle callback);
 
+    void scheduleCallbackThreadSafe(CallbackHandle callback);
+
     void scheduleCallback(Duration delay, CallbackHandle callback);
 
     void scheduleCallback(SysTime when, CallbackHandle callback);

src/asynchronous/libasync/events.d

@@ -10,6 +10,7 @@ import std.string;
 import std.traits;
 import std.typecons;
 import libasync.events : EventLoop_ = EventLoop, NetworkAddress;
+import libasync.signal : AsyncSignal;
 import libasync.timer : AsyncTimer;
 import libasync.tcp : AsyncTCPConnection, AsyncTCPListener, TCPEvent;
 import libasync.threads : destroyAsyncThreads, gs_threads;
@@ -39,6 +40,9 @@ package class LibasyncEventLoop : EventLoop
     private Appender!(CallbackHandle[]) nextCallbacks2;
     private Appender!(CallbackHandle[])* currentAppender;
 
+    private Appender!(CallbackHandle[]) nextThreadSafeCallbacks;
+    private shared AsyncSignal newThreadSafeCallbacks;
+
     private LibasyncTransport[] pendingConnections;
 
     alias Listener = Tuple!(ServerImpl, "server", AsyncTCPListener, "listener");
@@ -50,6 +54,9 @@ package class LibasyncEventLoop : EventLoop
         this.eventLoop = new EventLoop_;
         this.timers = new Timers(this.eventLoop);
         this.currentAppender = &nextCallbacks1;
+
+        this.newThreadSafeCallbacks = new shared AsyncSignal(this.eventLoop);
+        this.newThreadSafeCallbacks.run(&scheduleThreadSafeCallbacks);
     }
 
     override void runForever()
@@ -97,6 +104,27 @@ package class LibasyncEventLoop : EventLoop
             currentAppender.put(callback);
     }
 
+    private void scheduleThreadSafeCallbacks()
+    {
+        synchronized (this)
+        {
+            foreach (callback; nextThreadSafeCallbacks.data)
+                scheduleCallback(callback);
+
+            nextThreadSafeCallbacks.clear;
+        }
+    }
+
+    override void scheduleCallbackThreadSafe(CallbackHandle callback)
+    {
+        synchronized (this)
+        {
+            nextThreadSafeCallbacks ~= callback;
+        }
+
+        newThreadSafeCallbacks.trigger;
+    }
+
     override void scheduleCallback(Duration delay, CallbackHandle callback)
     {
         if (delay <= Duration.zero)

test/tcp.d

@@ -3,10 +3,8 @@ import std.array;
 import std.conv;
 import std.datetime;
 import std.string;
-
 import asynchronous;
 
-
 class TestHelper
 {
     private Appender!(string[]) actualEvents_;

test/threads.d

@@ -0,0 +1,24 @@
+import core.thread;
+import std.datetime : msecs;
+import asynchronous;
+
+unittest
+{
+    auto loop = getEventLoop;
+    auto future = new Future!int;
+
+    void stopLoop()
+    {
+        future.setResult(10);
+    }
+
+    auto thread1 = new Thread({
+        Thread.sleep(2.msecs);
+        loop.callSoonThreadSafe(&stopLoop);
+    });
+    thread1.start;
+
+    loop.runUntilComplete(future);
+    assert(future.done);
+    assert(future.result == 10);
+}