Add 'stream::fix' fixpoint operator (#867)

Added together with a test.

Not sure if the semantics are what we desire wrt stack, cc @b-studios.

Author: jiribenes

effekt/jvm/src/test/scala/effekt/core/VMTests.scala

@@ -803,6 +803,22 @@ class VMTests extends munit.FunSuite {
       shifts = 0,
       resumes = 0
     )),
+
+    examplesDir / "stdlib" / "stream" / "fix.effekt" -> Some(Summary(
+      staticDispatches = 38,
+      dynamicDispatches = 52,
+      patternMatches = 37,
+      branches = 0,
+      pushedFrames = 34,
+      poppedFrames = 34,
+      allocations = 37,
+      closures = 37,
+      variableReads = 10,
+      variableWrites = 9,
+      resets = 0,
+      shifts = 0,
+      resumes = 0
+    )),
   )
 
   val testFiles: Seq[(File, Option[Summary])] =

examples/stdlib/stream/fix.check

@@ -0,0 +1,10 @@
+1
+2
+3
+4
+0
+100
+-60
+42
+-50
+sum: 42

examples/stdlib/stream/fix.effekt

@@ -0,0 +1,35 @@
+import stream
+
+type Message {
+  Primitive(n: Int)
+  Complex(msgs: List[Message])
+}
+
+def solveOne(m: Message): Unit / { emit[Int], emit[Message] } = m match {
+  case Primitive(n) => do emit[Int](n)
+  case Complex(msgs) => msgs.each
+}
+
+def main() = {
+  def solved() = fix[Message] { msg => solveOne(msg) } {
+    do emit(
+      Complex([
+        Primitive(1),
+        Complex([Primitive(2), Primitive(3), Primitive(4)]),
+        Complex([]),
+        Complex([Complex([Primitive(0)]), Primitive(100)])
+      ])
+    )
+    do emit(Primitive(-60))
+    do emit(Complex([]))
+    do emit(Complex([Complex([]), Primitive(42)]))
+    do emit(Primitive(-50))
+  }
+
+  var sum = 0
+  for[Int] {solved} { n =>
+    sum = n + sum
+    println(n)
+  }
+  println("sum: " ++ show(sum))
+}

libraries/common/stream.effekt

@@ -30,6 +30,17 @@ effect stop(): Nothing
 def for[A] { stream: () => Unit / emit[A] } { action: A => Unit }: Unit =
   returning::for[A, Unit]{stream}{action}
 
+/// Runs the stream as a message queue,
+/// handling each of its elements that can produce new elements ("messages"),
+/// until a fixed point is reached.
+def fix[T] { one: T => Unit / emit[T] } { stream: => Unit / emit[T] }: Unit =
+  try {
+    stream()
+  } with emit[T] { value =>
+    fix[T] {one} { one(value) }
+    resume(())
+  }
+
 /// Turns a `list` into a producer of a push stream
 /// by emitting each contained value left-to-right.
 def each[A](list: List[A]): Unit / emit[A] =