Make signals rendezvous

Author: phischu

libraries/common/io/channel.effekt

@@ -1,10 +1,9 @@
 module io/channel
 
+import io
 import io/signal
 
-record Rendezvous[A, B](value: A, other: Signal[B])
-
-record Node[A, B](this: Signal[Rendezvous[A, B]], rest: Ref[Option[Node[A, B]]])
+record Node[A, B](this: Signal[A, B], rest: Ref[Option[Node[A, B]]])
 
 type Sender[A, B] = Ref[Node[A, B]]
 type Receiver[A, B] = Ref[Node[A, B]]
@@ -26,9 +25,9 @@ def send[A, B](sender: Sender[A, B], value: A): B = {
     case Some(node) =>
       sender.set(node)
   }
-  val other = signal::allocate()
-  this.unsafeSend(Rendezvous(value, other))
-  other.unsafeWait()
+  // TODO avoid yield
+  yield()
+  this.unsafeNotifyA(value)
 }
 
 def receive[A, B](receiver: Receiver[A, B], value: B): A = {
@@ -41,9 +40,9 @@ def receive[A, B](receiver: Receiver[A, B], value: B): A = {
     case Some(node) =>
       receiver.set(node)
   }
-  val Rendezvous(result, other) = this.unsafeWait()
-  other.unsafeSend(value)
-  return result
+  // TODO avoid yield
+  yield()
+  this.unsafeNotifyB(value)
 }
 
 // TODO streaming

libraries/common/io/promise.effekt

@@ -8,7 +8,7 @@ import io/signal
 
 type State[T] {
   Resolved(value: T)
-  Pending(signals: List[Signal[T]])
+  Pending(signals: List[Signal[T, Unit]])
 }
 
 extern type Promise[T]
@@ -49,11 +49,12 @@ extern {async, global} def await[T](promise: Promise[T]): T =
         val signal = signal::allocate()
         // TODO use reference.set and signal.wait
         set(reference, Pending(Cons(signal, signals)))
-        unsafeWait(signal)
+        // TODO perhaps yield to avoid stack overflow
+        unsafeNotifyB(signal, ())
     }
   }
 
-extern {io, global} def resolve[T](promise: Promise[T], value: T): Unit =
+extern {io, async, global} def resolve[T](promise: Promise[T], value: T): Unit =
   js "promise$resolve(${promise}, ${value})"
   llvm {
     val reference = toRefState(promise)
@@ -64,8 +65,9 @@ extern {io, global} def resolve[T](promise: Promise[T], value: T): Unit =
       case Pending(signals) =>
         // TODO use reference.set
         set(reference, Resolved(value))
+        // TODO perhaps yield to avoid stack overflow
         // TODO use signals.reverse.foreach
-        foreach(reverse(signals)) { signal => unsafeSend(signal, value) }
+        foreach(reverse(signals)) { signal => unsafeNotifyA(signal, value) }
     }
   }
 

libraries/common/io/signal.effekt

@@ -3,35 +3,32 @@ module io/signal
 // Signals
 // -------
 
-/// Must be sent exactly once and waited for exactly once
-extern type Signal[T]
+/// Must be notified exactly once by A and exactly once by B.
+/// Must yield before notifying or risking stack overflow.
+extern type Signal[A, B]
 
 namespace signal {
-  extern global def allocate[T](): Signal[T] =
+  extern global def allocate[A, B](): Signal[A, B] =
     llvm """
       %signal = call %Pos @c_signal_make()
       ret %Pos %signal
     """
 }
 
-extern async def unsafeSend[T](signal: Signal[T], value: T): Unit =
+extern async def unsafeNotifyA[A, B](signal: Signal[A, B], value: A): B =
   llvm """
-    call void @c_signal_send(%Pos ${signal}, %Pos ${value}, %Stack %stack)
+    call void @c_signal_notify(%Pos ${signal}, %Pos ${value}, %Stack %stack)
     ret void
   """
 
-extern async def unsafeWait[T](signal: Signal[T]): T =
+extern async def unsafeNotifyB[A, B](signal: Signal[A, B], value: B): A =
   llvm """
-    call void @c_signal_wait(%Pos ${signal}, %Stack %stack)
+    call void @c_signal_notify(%Pos ${signal}, %Pos ${value}, %Stack %stack)
     ret void
   """
 
 extern llvm """
   declare %Pos @c_signal_make()
-  declare void @c_signal_send(%Pos, %Pos)
-  declare void @c_signal_wait(%Pos, %Stack)
+  declare void @c_signal_notify(%Pos, %Pos, %Stack)
 """
 
-
-
-

libraries/llvm/io.c

@@ -298,30 +298,27 @@ void c_yield(Stack stack) {
 // Signals
 // --------
 
-typedef enum { BEFORE, SENDING, WAITING, AFTER } signal_state_t;
+typedef enum { BEFORE, NOTIFIED, AFTER } signal_state_t;
 
 typedef struct {
     uint64_t rc;
     void* eraser;
     signal_state_t state;
-    union {
-        struct Pos value;
-        Stack stack;
-    } payload;
+    struct Pos value;
+    Stack stack;
 } Signal;
 
 void c_signal_erase(void *envPtr) {
     // envPtr points to a Signal _after_ the eraser, so let's adjust it to point to the beginning.
     Signal *signal = (Signal*) (envPtr - offsetof(Signal, state));
+    // TODO exploit value and stack being NULL
     signal_state_t state = signal->state;
     switch (state) {
     case BEFORE:
         break;
-    case SENDING:
-        erasePositive(signal->payload.value);
-        break;
-    case WAITING:
-        eraseStack(signal->payload.stack);
+    case NOTIFIED:
+        erasePositive(signal->value);
+        eraseStack(signal->stack);
         break;
     case AFTER:
         break;
@@ -338,30 +335,23 @@ struct Pos c_signal_make() {
     return (struct Pos) { .tag = 0, .obj = signal, };
 }
 
-void c_signal_send(struct Pos signal, struct Pos value, Stack stack) {
+void c_signal_notify(struct Pos signal, struct Pos value, Stack stack) {
     Signal* f = (Signal*)signal.obj;
     switch (f->state) {
         case BEFORE: {
-            f->state = SENDING;
-            f->payload.value = value;
+            f->state = NOTIFIED;
+            f->value = value;
+            f->stack = stack;
             erasePositive(signal);
-            // TODO avoid stack overflow without yielding
-            c_yield(stack);
             break;
         }
-        case SENDING: {
-            // TODO more graceful panic
-            fprintf(stderr, "ERROR: Signal already used for sending\n");
-            exit(1);
-            break;
-        }
-        case WAITING: {
-            Stack other = f->payload.stack;
+        case NOTIFIED: {
+            struct Pos other_value = f->value;
+            Stack other_stack = f->stack;
             f->state = AFTER;
             erasePositive(signal);
-            // TODO avoid stack overflow without yielding
-            c_yield(stack);
-            resume_Pos(other, value);
+            resume_Pos(other_stack, value);
+            resume_Pos(stack, other_value);
             break;
         }
         case AFTER: {
@@ -373,35 +363,4 @@ void c_signal_send(struct Pos signal, struct Pos value, Stack stack) {
     }
 }
 
-void c_signal_wait(struct Pos signal, Stack stack) {
-    Signal* f = (Signal*)signal.obj;
-    switch (f->state) {
-        case BEFORE: {
-            f->state = WAITING;
-            f->payload.stack = stack;
-            erasePositive(signal);
-            break;
-        }
-        case SENDING: {
-            struct Pos value = f->payload.value;
-            f->state = AFTER;
-            erasePositive(signal);
-            resume_Pos(stack, value);
-            break;
-        }
-        case WAITING: {
-            // TODO more graceful panic
-            fprintf(stderr, "ERROR: Signal already used for waiting\n");
-            exit(1);
-            break;
-        }
-        case AFTER: {
-            // TODO more graceful panic
-            fprintf(stderr, "ERROR: Waiting after signal happened\n");
-            exit(1);
-            break;
-        }
-    }
-}
-
 #endif