Stdlib: tee for emit-streams (#922)

This adds an implementation of `tee`-like functions for `emit` streams.

There is:
- `teeing` which is controlled by the consumer on the outside (i.e. if
the outside consumer stops, it stops the inside one too)
- `tee` which runs two consumers with one producer, stopping once both
are done
- `manyTee` which allows to implement `tee` for more than 2 consumers
easily

<del>Note that my initial implementation of this depended on #919. I
don't think this one does, but would probably still want to hold off on
merging this for either solving it or someone else being sufficiently
certain it doesn't.</del>

Author: marzipankaiser

examples/stdlib/stream/tee.check

@@ -0,0 +1,74 @@
+a{ ... }
+1
+2
+3
+4
+5
+Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
+a done
+b{ ... }
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10
+Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
+b done
+tee{a}{b}{ ... }
+1
+2
+3
+4
+5
+Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
+a done
+6
+7
+8
+9
+10
+Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
+b done
+tee{b}{a}{ ... }
+1
+2
+3
+4
+5
+Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
+a done
+6
+7
+8
+9
+10
+Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
+b done
+a{ teeing{b}{ ... } }
+1
+2
+3
+4
+5
+Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
+a done
+b{ teeing{a}{ ... } }
+1
+2
+3
+4
+5
+Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
+a done
+6
+7
+8
+9
+10
+Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
+b done

examples/stdlib/stream/tee.effekt

@@ -0,0 +1,39 @@
+import stream
+
+def main() = {
+  def stream() = {
+    // equivalent: with teeing[Int]{ {b} => for[Int]{b}{ v => println(v) } }
+    for[Int]{
+      [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15].each
+    }{ v => 
+      println(v)
+      do emit(v)
+    }
+  }
+  def a{ b: => Unit / emit[Int] }: Unit = {
+    println(collectList[Int]{boundary{limit[Int](5){b}}})
+    println("a done")
+  }
+  def b{ b: => Unit / emit[Int] }: Unit = {
+    println(collectList[Int]{boundary{limit[Int](10){b}}})
+    println("b done")
+  }
+  println("a{ ... }")
+  a{stream}
+  println("b{ ... }")
+  b{stream}
+  println("tee{a}{b}{ ... }")
+  tee[Int]{a}{b}{ stream() }
+  println("tee{b}{a}{ ... }")
+  tee[Int]{b}{a}{ stream() }
+  println("a{ teeing{b}{ ... } }")
+  a{
+    with teeing[Int]{b}
+    stream()
+  }
+  println("b{ teeing{a}{ ... } }")
+  b{
+    with teeing[Int]{a}
+    stream()
+  }
+}

libraries/common/stream.effekt

@@ -409,6 +409,88 @@ def each(string: String): Unit / emit[Char] =
 def collectString { stream: () => Unit / emit[Char] }: String =
   returning::collectString[Unit]{stream}.second
 
+namespace internal {
+  effect snapshot(): Unit
+}
+
+/// Use cons to handle prod, but also emit to the outside.
+/// The outside controls iteration, i.e., if cons aborts, the rest of prod will still be emitted.
+/// 
+/// Starts by executing cons. Will stop cons during execution if the outside stops consuming.
+/// 
+/// Example, printing all values consumed by the outside:
+///
+///     with teeing{ {s} => for{s}{ e => println(e) } }
+///     prod() // some producer
+///
+/// Laws-ish (hopefully):
+/// - hnd{ prd() } === hnd{ teeing{t}{ prd() } } for all hnd and t that calls its argument at most once (and has no other captures)
+def teeing[A]{ cons: { => Unit / emit[A] } => Unit }{ prod: => Unit / emit[A] }: Unit / emit[A] = region r {
+  var consDone = false
+  var k in r = box { prod() } // what still needs to run of prod after cons exits
+  try {
+    cons{
+      try {
+        prod()
+        k = box { () }
+      } with emit[A] { v =>
+        do internal::snapshot() // remember that we need to continue here if cons aborts
+        if(not(consDone)) do emit(v)
+        outer.emit(v)
+        resume(())
+      }
+      if(consDone) do stop() // cons already exited, so skip continuing there
+    } // end cons
+    consDone = true
+  }
+  with stop { () => () }
+  with outer: emit[A]{ v => resume(do emit(v)) }
+  with internal::snapshot{ () =>
+    k = box { resume(()) }
+    resume(())
+  }
+  k()
+}
+
+/// Binds a `teeing`-like function in the body, stopping the inner producer once all consumers in tees are done consuming.
+/// 
+/// Example of use, equivalent to `tee[A]{cns1}{cns2}{prd}`:
+/// 
+///     manyTee[A] { {tee} =>
+///       with tee{cns1}
+///       with tee{cns2}
+///       prd()
+///     }
+///
+def manyTee[A]{ body: { { { => Unit / emit[A] } => Unit }{ => Unit / emit[A] } => Unit / emit[A] } => Unit / emit[A] }: Unit = {
+  var running = 0
+  try {
+    body{ {hnd}{prod} =>
+      running = running + 1
+      teeing[A]{ {b} => hnd{b}; running = running - 1 }{prod}
+    }
+  } with emit[A] { _ =>
+    if(running > 0) resume(())
+  }
+}
+
+/// Streams prod to both cons1 and cons2. Stops once both cons1 and cons2 stopped.
+/// Only runs prod once, resuming at most once at each emit.
+///
+///     var sumRes = 0
+///     var productRes = 0
+///     tee{ s => sumRes = sum{s} }{ s => productRes = product{s} }{ range(0, 10) }
+///     assertEquals(sum{ range(0, 10) }, sumRes)
+///     assertEquals(product{ range(0, 10) }, productRes)
+///
+def tee[A]{ cons1: { => Unit / emit[A] } => Unit }{ cons2: { => Unit / emit[A] } => Unit }{ prod: => Unit / emit[A] }: Unit = {
+  manyTee[A]{ {tee} =>
+    with tee{cons1}
+    with tee{cons2}
+    prod()
+  }
+}
+
 namespace returning {
 
 /// Canonical handler of push streams that performs `action` for every