par — common parMap/parForeach (cross-platform parallel collections PoC, scala/toolkit#31)

1-par.scala

//> using scala 3.8.3

import scala.concurrent.*
import scala.collection.BuildFrom
import scala.collection.generic.IsIterable
import java.util.concurrent.atomic.AtomicInteger

extension [Repr](xs: Repr)(using it: IsIterable[Repr])

  def parMap[B, To](parallelism: Int)(f: it.A => B)(using
      bf: BuildFrom[Repr, B, To]
  ): platform.R[To] =
    val items = it(xs).iterator.toVector
    val n     = items.size
    platform.parRun(parallelism): ec =>
      given ExecutionContext = ec
      val out  = new Array[Any](n)
      val next = AtomicInteger(0)
      def lane(): Future[Unit] = Future:
        var i = next.getAndIncrement()
        while i < n do
          out(i) = f(items(i))
          i = next.getAndIncrement()
      val lanes = Vector.fill(parallelism.min(n).max(1))(lane())
      Future.sequence(lanes).map: _ =>
        bf.fromSpecific(xs)(out.iterator.map(_.asInstanceOf[B]))

extension [A](xs: IterableOnce[A])

  def parForeach(parallelism: Int)(f: A => Unit): platform.R[Unit] =
    val items = xs.iterator.toVector
    val n     = items.size
    platform.parRun(parallelism): ec =>
      given ExecutionContext = ec
      val next = AtomicInteger(0)
      def lane(): Future[Unit] = Future:
        var i = next.getAndIncrement()
        while i < n do
          f(items(i))
          i = next.getAndIncrement()
      Future.sequence(Vector.fill(parallelism.min(n).max(1))(lane())).map(_ => ())