Skip to content

Instantly share code, notes, and snippets.

@gvolpe
Last active February 15, 2024 15:29
Show Gist options
  • Star 10 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save gvolpe/44e2263f9068efe298a1f30390de6d22 to your computer and use it in GitHub Desktop.
Save gvolpe/44e2263f9068efe298a1f30390de6d22 to your computer and use it in GitHub Desktop.
parTraverse with a limit of N using a Semaphore
import cats.Traverse
import cats.effect._
import cats.effect.concurrent.Semaphore
import cats.temp.par._
import cats.syntax.all._
import scala.concurrent.duration._
object Main extends IOApp {
import ParTask._
def putStrLn[A](a: A): IO[Unit] = IO(println(a))
override def run(args: List[String]): IO[ExitCode] = {
import cats.instances.list._
parTraverseN(3, List.range(1, 10))(x => putStrLn(x) *> IO.sleep(1.second)).as(ExitCode.Success)
}
}
object ParTask {
def parTraverseN[F[_]: Concurrent: Par, G[_]: Traverse, A, B](
n: Int,
ga: G[A]
)(f: A => F[B]) =
Semaphore[F](n).flatMap { s =>
ga.parTraverse(a => s.withPermit(f(a)))
}
}
@otobrglez
Copy link

Brilliant!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment