Merge pull request patr1ck-m#11 from patr1ck-m/improvement/data_input

Improve push to pull stream converter by setting max queue size.

Author: patr1ck-m

src/lib/queue.effekt

@@ -0,0 +1,150 @@
+module src/lib/queue
+
+import array
+
+/**
+
+Note: this is a copy of the queue.effekt from the standard library, because we need to add a size method.
+This allows us to implement more efficient implementations of stream buffers by slowing the emit down when the
+queue size exceeds a certain threshold.
+
+*/
+
+/// Mutable, automatically resizing queue.
+interface Queue[T] {
+  def empty?(): Bool
+
+  def popFront(): Option[T]
+
+  def popBack(): Option[T]
+
+  def peekFront(): Option[T]
+
+  def peekBack(): Option[T]
+
+  def pushFront(el: T): Unit
+
+  def pushBack(el: T): Unit
+
+  def size(): Int
+}
+
+
+def emptyQueue[T](): Queue[T] at global =
+  emptyQueue[T](64)
+
+def emptyQueue[T](initialCapacity: Int): Queue[T] at global = {
+
+  val contents = ref(array[Option[T]](initialCapacity, None()))
+  val head = ref(0)
+  val tail = ref(0)
+  val size = ref(0)
+  val capacity = ref(initialCapacity)
+
+  def remove(arr: Array[Option[T]], index: Int): Option[T] = {
+    with on[OutOfBounds].default { None() }
+    val value = arr.get(index)
+    arr.set(index, None())
+    value
+  }
+
+  def nonEmpty[T] { p: => Option[T] / Exception[OutOfBounds] }: Option[T] =
+    if (size.get <= 0) None() else on[OutOfBounds].default { None() } { p() }
+
+  // Exponential back-off
+  def resizeTo(requiredSize: Int): Unit =
+    if (requiredSize <= capacity.get) () else {
+      with on[OutOfBounds].ignore // should not happen
+
+      val oldSize = capacity.get
+      val newSize = capacity.get * 2
+      val oldContents = contents.get
+      val newContents = array[Option[T]](newSize, None())
+
+      if (head.get < tail.get) {
+        // The queue does not wrap around; direct copy is possible.
+        copy(oldContents, head.get, newContents, 0, size.get) // changed tail to size
+      } else if (size.get > 0) {
+        // The queue wraps around; copy in two segments.
+        copy(oldContents, head.get, newContents, 0, oldSize - head.get) // changed oldSize to oldSize - head
+        copy(oldContents, 0, newContents, oldSize - head.get, tail.get) // changed oldSize - head to oldSize - head
+      }
+
+      contents.set(newContents)
+      capacity.set(newSize)
+      head.set(0)
+      tail.set(oldSize)
+    }
+
+  def queue = new Queue[T] {
+    def empty?() = size.get <= 0
+
+    def popFront() =
+      nonEmpty {
+        val result = contents.get.remove(head.get)
+        head.set(mod(head.get + 1, capacity.get))
+        size.set(size.get - 1)
+        result
+      }
+
+    def popBack() =
+      nonEmpty {
+        tail.set(mod(tail.get - 1 + capacity.get, capacity.get))
+        val result = contents.get.remove(tail.get)
+        size.set(size.get - 1)
+        result
+      }
+
+    def peekFront() = nonEmpty { contents.get.get(head.get) }
+
+    def peekBack() = nonEmpty { contents.get.get(tail.get) }
+
+    def pushFront(el: T) = {
+      resizeTo(size.get + 1)
+      head.set(mod(head.get - 1 + capacity.get, capacity.get))
+      size.set(size.get + 1)
+      contents.get.set(head.get, Some(el))
+    }
+
+    def pushBack(el: T) = {
+      resizeTo(size.get + 1)
+      contents.get.set(tail.get, Some(el))
+      size.set(size.get + 1)
+      tail.set(mod(tail.get + 1, capacity.get))
+    }
+
+    def size() = size.get
+  }
+  box queue
+}
+
+namespace examples {
+  def main() = {
+    // queue with initial capacity 4
+    def b = emptyQueue[Int](4)
+    println(b.empty?)
+    b.pushFront(1)
+    b.pushBack(2)
+    b.pushFront(3)
+    b.pushBack(4)
+    // this will cause resizing:
+    b.pushBack(5)
+    b.pushBack(6)
+    b.pushBack(7)
+    b.pushBack(8)
+    // and again:
+    b.pushBack(9)
+
+    println(b.empty?)
+    println(b.popFront()) // Some(3)
+    println(b.popFront()) // Some(1)
+    println(b.popFront()) // Some(2)
+    println(b.popFront()) // Some(4)
+    println(b.popFront()) // Some(5)
+    println(b.popFront()) // Some(6)
+    println(b.popFront()) // Some(7)
+    println(b.popFront()) // Some(8)
+    println(b.popFront()) // Some(9)
+    println(b.popFront()) // None()
+  }
+}

src/lib/stream_input.effekt

@@ -2,18 +2,22 @@ module src/lib/stream_input
 
 import stream
 import src/lib/csv
-import queue
+import src/lib/queue
 import scanner
 import io/error
 import io/filesystem
 import io/time
 
-def pushToPullStream[T] {pushStr: () => Unit / { emit[T], stop } } {pullStr: () => Unit / { read[T] } } = {
+/// Connect a push-based stream to a pull-based stream using an internal queue which slows emits down when reaching the maximum size.
+def pushToPullStream[T](maxQueueSize: Int) {pushStr: () => Unit / { emit[T], stop } } {pullStr: () => Unit / { read[T] } } = {
   def q = emptyQueue[T]()
   var running = true
   try pushStr() with emit[T] {
     def emit(v: T) = {
       q.pushBack(v)
+      while (q.size() >= maxQueueSize) {
+        time::sleep(10)
+      }
       resume(())
     }
   } with stop {
@@ -68,11 +72,11 @@ def csvFeedStr(csvStr: String, columnName: String, delayMs: Int): Unit / { emit[
 }
 
 def csvFeedPull(path: String, columnName: String, delayMs: Int) {body: () => Unit / { read[Double], Exception[IOError], Exception[WrongFormat] } }: Unit / { Exception[IOError], Exception[WrongFormat] } = {
-  pushToPullStream[Double] { csvFeed(path, columnName, delayMs) } { body() }
+  pushToPullStream[Double](100) { csvFeed(path, columnName, delayMs) } { body() }
 }
 
 def csvFeedStrPull(csvStr: String, columnName: String, delayMs: Int) {body: () => Unit / { read[Double], Exception[WrongFormat] } }: Unit / { Exception[WrongFormat] } = {
-  pushToPullStream[Double] { csvFeedStr(csvStr, columnName, delayMs) } { body() }
+  pushToPullStream[Double](100) { csvFeedStr(csvStr, columnName, delayMs) } { body() }
 }