Increase buffer in mapPar, add more tests, improve performance (#350)

Author: adamw

core/src/main/scala/ox/flow/FlowOps.scala

@@ -192,12 +192,16 @@ class FlowOps[+T]:
     */
   def mapPar[U](parallelism: Int)(f: T => U)(using BufferCapacity): Flow[U] = Flow.usingEmitInline: emit =>
     val s = new Semaphore(parallelism)
-    val inProgress = Channel.withCapacity[Fork[Option[U]]](parallelism)
+    // providing extra capacity in the `inProgress` channel (but still limiting it so that processing is bounded):
+    // 1. starting more forks than parallelism, so that they are read to do their work immediately after a permit becomes available
+    // 2. allowing for some slack after the mapping is completed, but its result not yet received; then, new mappings can already be started
+    val inProgress = Channel.withCapacity[Fork[Option[U]]](parallelism * 4)
     val results = BufferCapacity.newChannel[U]
 
     def forkMapping(t: T)(using OxUnsupervised): Fork[Option[U]] =
       forkUnsupervised:
         try
+          s.acquire()
           val u = f(t)
           s.release() // not in finally, as in case of an exception, no point in starting subsequent forks
           Some(u)
@@ -214,11 +218,7 @@ class FlowOps[+T]:
       // notifying only the `results` channels, as it will cause the scope to end, and any other forks to be
       // interrupted, including the inProgress-fork, which might be waiting on a join()
       forkPropagate(results):
-        last.run(
-          FlowEmit.fromInline: t =>
-            s.acquire()
-            inProgress.sendOrClosed(forkMapping(t)).discard
-        )
+        last.run(FlowEmit.fromInline(t => inProgress.sendOrClosed(forkMapping(t)).discard))
         inProgress.doneOrClosed().discard
 
       // a fork in which we wait for the created forks to finish (in sequence), and forward the mapped values to `results`

core/src/test/scala/ox/flow/FlowOpsMapParTest.scala

@@ -112,4 +112,125 @@ class FlowOpsMapParTest extends AnyFlatSpec with Matchers with Eventually:
     // checking if the forks aren't left running
     sleep(200.millis)
     trail.get shouldBe Vector("done", "done", "exception") // TODO: 3 isn't cancelled because it's already taken off the queue
+
+  // Edge Cases
+  it should "handle empty flow" in supervised:
+    // given
+    val flow = Flow.fromIterable(List.empty[Int])
+    val processedCount = new AtomicInteger(0)
+
+    // when
+    val result = flow.mapPar(5): i =>
+      processedCount.incrementAndGet()
+      i * 2
+
+    // then
+    result.runToList() shouldBe List.empty
+    processedCount.get() shouldBe 0
+
+  it should "handle flow with exactly parallelism number of elements" in supervised:
+    // given
+    val parallelism = 3
+    val flow = Flow.fromIterable(1 to parallelism)
+    val running = new AtomicInteger(0)
+    val maxRunning = new AtomicInteger(0)
+
+    def f(i: Int) =
+      val current = running.incrementAndGet()
+      maxRunning.updateAndGet(current.max)
+      try
+        sleep(100.millis)
+        i * 2
+      finally running.decrementAndGet().discard
+      end try
+    end f
+
+    // when
+    val result = flow.mapPar(parallelism)(f).runToList()
+
+    // then
+    result shouldBe List(2, 4, 6)
+    maxRunning.get() shouldBe parallelism
+
+  it should "handle flow with less than parallelism number of elements" in supervised:
+    // given
+    val flow = Flow.fromIterable(1 to 2)
+    val running = new AtomicInteger(0)
+    val maxRunning = new AtomicInteger(0)
+
+    def f(i: Int) =
+      val current = running.incrementAndGet()
+      maxRunning.updateAndGet(current.max)
+      try
+        sleep(100.millis)
+        i * 2
+      finally running.decrementAndGet().discard
+      end try
+    end f
+
+    // when
+    val result = flow.mapPar(5)(f).runToList()
+
+    // then
+    result shouldBe List(2, 4)
+    maxRunning.get() shouldBe 2 // should never exceed actual number of elements
+
+  // Order Preservation Tests
+  it should "preserve order even with varying processing times" in supervised:
+    // given
+    val flow = Flow.fromIterable(1 to 10)
+
+    def f(i: Int) =
+      // Later elements finish faster to test order preservation
+      val delay = if i <= 5 then (6 - i) * 50 else 50
+      sleep(delay.millis)
+      i * 2
+
+    // when
+    val result = flow.mapPar(3)(f).runToList()
+
+    // then
+    result shouldBe List(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)
+
+  it should "preserve order with random processing times" in supervised:
+    // given
+    val elements = 1 to 20
+    val flow = Flow.fromIterable(elements)
+
+    def f(i: Int) =
+      // Random delay to test order preservation
+      val delay = scala.util.Random.nextInt(100) + 10
+      sleep(delay.millis)
+      i
+
+    // when
+    val result = flow.mapPar(5)(f).runToList()
+
+    // then
+    result shouldBe elements.toList
+
+  // Other
+  it should "work with very high parallelism values" in supervised:
+    // given
+    val flow = Flow.fromIterable(1 to 5)
+    val running = new AtomicInteger(0)
+    val maxRunning = new AtomicInteger(0)
+
+    def f(i: Int) =
+      val current = running.incrementAndGet()
+      maxRunning.updateAndGet(current.max)
+      try
+        sleep(50.millis)
+        i * 2
+      finally running.decrementAndGet().discard
+      end try
+    end f
+
+    // when
+    val result = flow.mapPar(1000)(f).runToList()
+
+    // then
+    result shouldBe List(2, 4, 6, 8, 10)
+    maxRunning.get() shouldBe 5 // Should not exceed actual number of elements
+
 end FlowOpsMapParTest