Basic streaming (#527)

Effects for push and pulls streams and handlers for them.

---------

Co-authored-by: Jonathan Brachthäuser <[email protected]>

effekt/jvm/src/test/scala/effekt/ChezSchemeTests.scala

@@ -21,13 +21,13 @@ abstract class ChezSchemeTests extends EffektTests {
 
     examplesDir / "llvm",
 
-    // bidirectional handlers
-    examplesDir / "pos" / "maps.effekt",
-
     // bidirectional effects are not yet supported in our Chez backend
+    examplesDir / "pos" / "maps.effekt",
     examplesDir / "pos" / "bidirectional",
     examplesDir / "pos" / "object",
     examplesDir / "pos" / "type_omission_op.effekt",
+    examplesDir / "benchmarks" / "input_output" / "word_count_ascii.effekt",
+    examplesDir / "benchmarks" / "input_output" / "word_count_utf8.effekt",
 
     // unsafe continuations are not yet supported in our Chez backend
     examplesDir / "pos" / "unsafe_cont.effekt",

effekt/jvm/src/test/scala/effekt/LLVMTests.scala

@@ -36,6 +36,8 @@ class LLVMTests extends EffektTests {
     examplesDir / "benchmarks" / "are_we_fast_yet" / "towers.effekt",
     examplesDir / "benchmarks" / "are_we_fast_yet" / "permute.effekt",
     examplesDir / "benchmarks" / "are_we_fast_yet" / "queens.effekt",
+    examplesDir / "benchmarks" / "input_output" / "word_count_ascii.effekt",
+    examplesDir / "benchmarks" / "input_output" / "word_count_utf8.effekt",
   )
 
   /**

effekt/shared/src/main/scala/effekt/core/PolymorphismBoxing.scala

@@ -489,6 +489,7 @@ object PolymorphismBoxing extends Phase[CoreTransformed, CoreTransformed] {
         case core.Type.TDouble  => Pure.Literal(13.37, core.Type.TDouble)
         // Do strings need to be boxed? Really?
         case core.Type.TString  => Pure.Literal("<?nothing>", core.Type.TString)
+        case core.Type.TByte    => Pure.Literal(1337, core.Type.TByte)
         case t if box.isDefinedAt(t) => sys error s"No default value defined for ${t}"
         case _ => sys error s"Trying to unbox Nothing to ${t}"
       }

effekt/shared/src/main/scala/effekt/core/Transformer.scala

@@ -640,7 +640,7 @@ object Transformer extends Phase[Typechecked, CoreTransformed] {
                 val resumeFun: core.BlockLit = core.BlockLit(Nil, List(resumeArgCapture), Nil, List(resumeArgParam),
                   core.Stmt.Resume(resumeVar, core.Stmt.App(resumeArgVar, Nil, Nil, bvars)))
 
-                core.Operation(op.definition, tps, Nil, vps, bparams,
+                core.Operation(op.definition, tps, cps, vps, bparams,
                   core.Shift(prompt, core.BlockLit(Nil, List(resumeCapture), Nil, resumeParam :: Nil,
                     core.Scope(List(core.Definition.Def(resumeSymbol, resumeFun)),
                       transform(body)))))

examples/benchmarks/input_output/word_count_ascii.check

@@ -0,0 +1 @@
+6170

examples/benchmarks/input_output/word_count_ascii.effekt

@@ -0,0 +1,75 @@
+import examples/benchmarks/runner
+
+import io/error
+import io/filesystem
+import stream
+
+record Output(chars: Int, words: Int, lines: Int)
+
+def formatWith(output: Output, filename: String): String =
+  output.lines.show ++ " " ++ output.words.show ++ " " ++ output.chars.show ++ " " ++ filename
+
+def isSpace(b: Byte) = b.toInt match {
+  case 32 => true // ' '
+  case 9  => true // '\t'
+  case 10 => true // '\n'
+  case 11 => true // '\v'
+  case 12 => true // '\f'
+  case 13 => true // '\r'
+  case _ => false
+}
+
+def isNewline(b: Byte) = b.toInt == 10 // \n
+
+def countWords(): Output / read[Byte] = {
+
+  var chars = 0
+  var words = 0
+  var lines = 0
+  var wasSpace = true
+
+  exhaustively { do read[Byte]() } { c =>
+
+    chars = chars + 1
+
+    val currentIsSpace = isSpace(c)
+
+    if (wasSpace && not(currentIsSpace)) {
+      words = words + 1
+    }
+    wasSpace = currentIsSpace
+
+    if (isNewline(c)) {
+      lines = lines + 1
+    }
+  }
+
+  Output(chars, words, lines)
+}
+
+def run(n: Int) = {
+  with on[IOError].panic;
+
+  val filename = "/tmp/word_count_ascii.txt"
+
+  val _ = {
+    with writeFile(filename)
+    with repeat(n)
+    for[Int] { range(32, 127) } { c =>
+      repeat(10) {
+        do emit(c.toByte)
+      }
+      do emit(10.toByte)
+    }
+  }
+
+  val output = {
+    with readFile(filename)
+    countWords()
+  }
+
+  return output.chars + output.words + output.lines
+}
+
+def main() = benchmark(5){run}
+

examples/benchmarks/input_output/word_count_utf8.check

@@ -0,0 +1 @@
+650

examples/benchmarks/input_output/word_count_utf8.effekt

@@ -0,0 +1,66 @@
+import examples/benchmarks/runner
+
+import io/error
+import io/filesystem
+import char
+import stream
+
+record Output(chars: Int, words: Int, lines: Int)
+
+def formatWith(output: Output, filename: String): String =
+  output.lines.show ++ " " ++ output.words.show ++ " " ++ output.chars.show ++ " " ++ filename
+
+def countWords(): Output / read[Char] = {
+
+  var chars = 0
+  var words = 0
+  var lines = 0
+  var wasSpace = true
+
+  exhaustively[Char] { do read() } { c =>
+
+    chars = chars + 1
+
+    val currentIsSpace = isWhitespace(c)
+
+    if (wasSpace && not(currentIsSpace)) {
+      words = words + 1
+    }
+    wasSpace = currentIsSpace
+
+    if (c == '\n') {
+      lines = lines + 1
+    }
+  }
+
+  Output(chars, words, lines)
+}
+
+def run(n: Int) = {
+  with on[IOError].panic;
+
+  val filename = "/tmp/word_count_utf8.txt"
+
+  val _ = {
+    with writeFile(filename)
+    with encodeUTF8
+    with repeat(n)
+    for[Int] { range(128512, 128522) } { c =>
+      repeat(10) {
+        do emit(c.toChar)
+      }
+      do emit(10.toChar)
+    }
+  }
+
+  val output = {
+    with readFile(filename)
+    with decodeUTF8
+    countWords()
+  }
+
+  return output.chars + output.words + output.lines
+}
+
+def main() = benchmark(5){run}
+

examples/stdlib/bytearray/bytearray.effekt

@@ -11,5 +11,5 @@ def main() = {
 
   b.unsafeSet(1, 102.toByte)
 
-  println(b.toUTF8) // hfllo
+  println(b.toString) // hfllo
 }

libraries/common/array.effekt

@@ -77,30 +77,7 @@ extern global def unsafeSet[T](arr: Array[T], index: Int, value: T): Unit =
     ret %Pos %z
   """
 
-/// Creates a copy of `arr`
-def copy[T](arr: Array[T]): Array[T] = {
-  with on[OutOfBounds].default { <> }; // should not happen
-  val len = arr.size;
-  val newArray = allocate[T](len);
-  copy[T](arr, 0, newArray, 0, len);
-  newArray
-}
 
-/// Copies `length`-many elements from `from` to `to`
-/// starting at `start` (in `from`) and `offset` (in `to`)
-def copy[T](from: Array[T], start: Int, to: Array[T], offset: Int, length: Int): Unit / Exception[OutOfBounds] = {
-  val startValid  = start >= 0 && start + length <= from.size
-  val offsetValid = offset >= 0 && offset + length <= to.size
-
-  def go(i: Int, j: Int, length: Int): Unit =
-    if (length > 0) {
-      to.unsafeSet(j, from.unsafeGet(i))
-      go(i + 1, j + 1, length - 1)
-    }
-
-  if (startValid && offsetValid) go(start, offset, length)
-  else do raise(OutOfBounds(), "Array index out of bounds, when copying")
-}
 
 // Derived operations:
 
@@ -126,6 +103,44 @@ def build[T](size: Int) { index: Int => T }: Array[T] = {
   arr
 }
 
+def resize[T](source: Array[T], size: Int): Array[T] = {
+  val target = allocate(size)
+  val n = min(source.size, target.size)
+  def go(i: Int): Array[T] =
+    if (i < n) {
+      target.unsafeSet(i, source.unsafeGet(i))
+      go(i + 1)
+    } else {
+      target
+    }
+  go(0)
+}
+
+/**
+ * Creates a copy of `arr`
+ */
+def copy[T](array: Array[T]): Array[T] =
+  array.resize(array.size)
+
+/**
+ * Copies `length`-many elements from `from` to `to`
+ * starting at `start` (in `from`) and `offset` (in `to`)
+ */
+def copy[T](from: Array[T], start: Int, to: Array[T], offset: Int, length: Int): Unit / Exception[OutOfBounds] = {
+  val startValid  = start >= 0 && start + length <= from.size
+  val offsetValid = offset >= 0 && offset + length <= to.size
+
+  def go(i: Int, j: Int, length: Int): Unit =
+    if (length > 0) {
+      to.unsafeSet(j, from.unsafeGet(i))
+      go(i + 1, j + 1, length - 1)
+    }
+
+  if (startValid && offsetValid) go(start, offset, length)
+  else do raise(OutOfBounds(), "Array index out of bounds, when copying")
+}
+
+
 // Utility functions:
 
 def toList[T](arr: Array[T]): List[T] = {
@@ -140,26 +155,22 @@ def toList[T](arr: Array[T]): List[T] = {
 
 def foreach[T](arr: Array[T]){ action: T => Unit }: Unit =
   each(0, arr.size) { i =>
-    val x: T = arr.unsafeGet(i)
-    action(x)
+    action(arr.unsafeGet(i))
   }
 
 def foreach[T](arr: Array[T]){ action: (T) {Control} => Unit }: Unit =
   each(0, arr.size) { (i) {label} =>
-    val x: T = arr.unsafeGet(i)
-    action(x) {label}
+    action(arr.unsafeGet(i)) {label}
   }
 
 def foreachIndex[T](arr: Array[T]){ action: (Int, T) => Unit }: Unit =
   each(0, arr.size) { i =>
-    val x: T = arr.unsafeGet(i)
-    action(i, x)
+    action(i, arr.unsafeGet(i))
   }
 
 def foreachIndex[T](arr: Array[T]){ action: (Int, T) {Control} => Unit }: Unit =
   each(0, arr.size) { (i) {label} =>
-    val x: T = arr.unsafeGet(i)
-    action(i, x) {label}
+    action(i, arr.unsafeGet(i)) {label}
   }
 
 def sum(list: Array[Int]): Int = {

libraries/common/bytearray.effekt

@@ -50,13 +50,33 @@ def resize(source: ByteArray, size: Int): ByteArray = {
   go(0)
 }
 
-extern pure def fromUTF8(str: String): ByteArray =
+def foreach(arr: ByteArray){ action: Byte => Unit }: Unit =
+  each(0, arr.size) { i =>
+    action(arr.unsafeGet(i))
+  }
+
+def foreach(arr: ByteArray){ action: (Byte) {Control} => Unit }: Unit =
+  each(0, arr.size) { (i) {label} =>
+    action(arr.unsafeGet(i)) {label}
+  }
+
+def foreachIndex(arr: ByteArray){ action: (Int, Byte) => Unit }: Unit =
+  each(0, arr.size) { i =>
+    action(i, arr.unsafeGet(i))
+  }
+
+def foreachIndex(arr: ByteArray){ action: (Int, Byte) {Control} => Unit }: Unit =
+  each(0, arr.size) { (i) {label} =>
+    action(i, arr.unsafeGet(i)) {label}
+  }
+
+extern pure def fromString(str: String): ByteArray =
   js "(new TextEncoder().encode(${str}))"
   llvm """
     ret %Pos ${str}
   """
 
-extern pure def toUTF8(arr: ByteArray): String =
+extern pure def toString(arr: ByteArray): String =
   js "(new TextDecoder('utf-8').decode(${arr}))"
   // assuming the buffer is already in UTF-8
   llvm """

libraries/common/effekt.effekt

@@ -566,6 +566,11 @@ extern pure def bitwiseAnd(x: Int, y: Int): Int =
   chez "(logand ${x} ${y})"
   llvm "%z = and %Int ${x}, ${y} ret %Int %z"
 
+extern pure def bitwiseOr(x: Int, y: Int): Int =
+  js "(${x} | ${y})"
+  chez "(logior ${x} ${y})"
+  llvm "%z = or %Int ${x}, ${y} ret %Int %z"
+
 extern pure def bitwiseXor(x: Int, y: Int): Int =
   js "(${x} ^ ${y})"
   chez "(logxor ${x} ${y})"

libraries/common/io/filesystem.effekt

@@ -44,7 +44,7 @@ def readFile(path: String): String / Exception[IOError] = {
   def go(): String = {
     read(file, buffer, offset, chunkSize, -1) match {
       case 0 =>
-        buffer.resize(offset).toUTF8
+        buffer.resize(offset).toString
       case n =>
         offset = offset + n
         if (offset + chunkSize > buffer.size) {
@@ -63,7 +63,7 @@ def writeFile(path: String, contents: String): Unit / Exception[IOError] = {
   with on[IOError].finalize { close(file) }
 
   val chunkSize = 1048576 // 1MB
-  val buffer = contents.fromUTF8
+  val buffer = contents.fromString
   var offset = 0;
 
   def go(): Unit = {