From 402b1d64c39f8b70166db810fad4e97d7a0905e3 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Fri, 8 Aug 2025 10:52:18 +0200 Subject: [PATCH 1/7] Add parFlatTraverseN and parFlatSequenceN Add equivalents of the Parallel methods that allow flattening the result of a limited parallel traversal. These are helpful in case like big database lookups, where a set of IDs is batched, then each batch is queried and produces a collection of results that needs to be flattened into a larger result. --- .../src/main/scala/cats/effect/IO.scala | 13 +++++ .../cats/effect/kernel/GenConcurrent.scala | 28 +++++++++-- .../kernel/syntax/GenConcurrentSyntax.scala | 20 +++++++- .../cats/effect/kernel/SyntaxSuite.scala | 10 ++++ .../src/test/scala/cats/effect/IOSuite.scala | 48 +++++++++++++++++++ 5 files changed, 114 insertions(+), 5 deletions(-) diff --git a/core/shared/src/main/scala/cats/effect/IO.scala b/core/shared/src/main/scala/cats/effect/IO.scala index 41fb79db63..8df5621453 100644 --- a/core/shared/src/main/scala/cats/effect/IO.scala +++ b/core/shared/src/main/scala/cats/effect/IO.scala @@ -1599,6 +1599,13 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara def parTraverseN_[T[_]: Foldable, A, B](n: Int)(ta: T[A])(f: A => IO[B]): IO[Unit] = _asyncForIO.parTraverseN_(n)(ta)(f) + /** + * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. + */ + def parFlatTraverseN[T[_]: Traverse: cats.FlatMap, A, B](n: Int)(ta: T[A])( + f: A => IO[T[B]]): IO[T[B]] = + _asyncForIO.parFlatTraverseN(n)(ta)(f) + /** * Like `Parallel.parSequence` */ @@ -1623,6 +1630,12 @@ object IO extends IOCompanionPlatform with IOLowPriorityImplicits with TuplePara def parSequenceN_[T[_]: Foldable, A](n: Int)(tma: T[IO[A]]): IO[Unit] = _asyncForIO.parSequenceN_(n)(tma) + /** + * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. + */ + def parFlatSequenceN[T[_]: Traverse: cats.FlatMap, A](n: Int)(tmta: T[IO[T[A]]]): IO[T[A]] = + _asyncForIO.parFlatSequenceN(n)(tmta) + /** * Like `Parallel.parReplicateA`, but limits the degree of parallelism. */ diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index 024f640197..87b3d473dc 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -16,11 +16,11 @@ package cats.effect.kernel -import cats.{Foldable, Monoid, Semigroup, Traverse} +import cats.{FlatMap, Foldable, Monoid, Semigroup, Traverse} import cats.data.{EitherT, IorT, Kleisli, OptionT, WriterT} -import cats.effect.kernel.instances.spawn._ -import cats.effect.kernel.syntax.all._ -import cats.syntax.all._ +import cats.effect.kernel.instances.spawn.* +import cats.effect.kernel.syntax.all.* +import cats.syntax.all.* trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { @@ -155,6 +155,26 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { MiniSemaphore[F](n).flatMap { sem => ta.parTraverse_ { a => sem.withPermit(f(a)) } } } + /** + * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. + */ + def parFlatSequenceN[T[_]: Traverse: FlatMap, A](n: Int)(tma: T[F[T[A]]]): F[T[A]] = + parFlatTraverseN(n)(tma)(identity) + + /** + * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. Note that the + * semantics of this operation aim to maximise fairness: when a spot to execute becomes + * available, every task has a chance to claim it, and not only the next `n` tasks in `ta` + */ + def parFlatTraverseN[T[_]: Traverse: FlatMap, A, B](n: Int)(ta: T[A])( + f: A => F[T[B]]): F[T[B]] = { + require(n >= 1, s"Concurrency limit should be at least 1, was: $n") + + implicit val F: GenConcurrent[F, E] = this + + MiniSemaphore[F](n).flatMap { sem => ta.parFlatTraverse { a => sem.withPermit(f(a)) } } + } + override def racePair[A, B](fa: F[A], fb: F[B]) : F[Either[(Outcome[F, E, A], Fiber[F, E, B]), (Fiber[F, E, A], Outcome[F, E, B])]] = { implicit val F: GenConcurrent[F, E] = this diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/syntax/GenConcurrentSyntax.scala b/kernel/shared/src/main/scala/cats/effect/kernel/syntax/GenConcurrentSyntax.scala index 35b7604aac..164d83822b 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/syntax/GenConcurrentSyntax.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/syntax/GenConcurrentSyntax.scala @@ -16,7 +16,7 @@ package cats.effect.kernel.syntax -import cats.{Foldable, Traverse} +import cats.{FlatMap, Foldable, Traverse} import cats.effect.kernel.GenConcurrent trait GenConcurrentSyntax { @@ -34,6 +34,11 @@ trait GenConcurrentSyntax { ): ConcurrentParSequenceNOps[T, F, A] = new ConcurrentParSequenceNOps(wrapped) + implicit def concurrentParFlatSequenceOps[T[_], F[_], A]( + wrapped: T[F[T[A]]] + ): ConcurrentParFlatSequenceNOps[T, F, A] = + new ConcurrentParFlatSequenceNOps(wrapped) + } final class GenConcurrentOps_[F[_], A] private[syntax] (private val wrapped: F[A]) @@ -57,6 +62,11 @@ final class ConcurrentParTraverseNOps[T[_], A] private[syntax] ( f: A => F[B] )(implicit T: Foldable[T], F: GenConcurrent[F, ?]): F[Unit] = F.parTraverseN_(n)(wrapped)(f) + + def parFlatTraverseN[F[_], B](n: Int)( + f: A => F[T[B]] + )(implicit T: Traverse[T], FM: FlatMap[T], F: GenConcurrent[F, ?]): F[T[B]] = + F.parFlatTraverseN(n)(wrapped)(f) } final class ConcurrentParSequenceNOps[T[_], F[_], A] private[syntax] ( @@ -68,3 +78,11 @@ final class ConcurrentParSequenceNOps[T[_], F[_], A] private[syntax] ( def parSequenceN_(n: Int)(implicit T: Foldable[T], F: GenConcurrent[F, ?]): F[Unit] = F.parSequenceN_(n)(wrapped) } + +final class ConcurrentParFlatSequenceNOps[T[_], F[_], A] private[syntax] ( + private val wrapped: T[F[T[A]]] +) extends AnyVal { + def parFlatSequenceN( + n: Int)(implicit T: Traverse[T], FM: FlatMap[T], F: GenConcurrent[F, ?]): F[T[A]] = + F.parFlatSequenceN(n)(wrapped) +} diff --git a/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala b/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala index 6f920fa5ea..475ab259ff 100644 --- a/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala +++ b/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala @@ -57,6 +57,16 @@ class SyntaxSuite { result: F[Unit] } + { + val result = List(target).parFlatTraverseN(3)(t => F.pure(List(t))) + result: F[List[A]] + } + + { + val result = List(target.map(List(_))).parFlatSequenceN(3) + result: F[List[A]] + } + { val result = target.parReplicateAN(3)(5) result: F[List[A]] diff --git a/tests/shared/src/test/scala/cats/effect/IOSuite.scala b/tests/shared/src/test/scala/cats/effect/IOSuite.scala index 73dedf6129..3bb9c06d5c 100644 --- a/tests/shared/src/test/scala/cats/effect/IOSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/IOSuite.scala @@ -1702,6 +1702,54 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu assertCompleteAs(p, true) } + real("parFlatTraverseN - throw when n < 1") { + IO.defer { + List.empty[Int].parFlatTraverseN(0)(List(_).pure[IO]) + }.mustFailWith[IllegalArgumentException] + } + + real("parFlatTraverseN - propagate errors") { + List(1, 2, 3) + .parFlatTraverseN(2) { (n: Int) => + if (n == 2) IO.raiseError(new RuntimeException) else List(n).pure[IO] + } + .mustFailWith[RuntimeException] + } + + ticked("parFlatTraverseN - be cancelable") { implicit ticker => + val p = for { + f <- List(1, 2, 3).parFlatTraverseN(2)(_ => IO.never[List[Int]]).start + _ <- IO.sleep(100.millis) + _ <- f.cancel + } yield true + + assertCompleteAs(p, true) + } + + real("parFlatSequenceN - throw when n < 1") { + IO.defer { + List.empty[IO[List[Int]]].parFlatSequenceN(0) + }.mustFailWith[IllegalArgumentException] + } + + real("parFlatSequenceN - propagate errors") { + List(1, 2, 3).map { (n: Int) => + if (n == 2) IO.raiseError(new RuntimeException) else List(n).pure[IO] + } + .parFlatSequenceN(2) + .mustFailWith[RuntimeException] + } + + ticked("parFlatSequenceN - be cancelable") { implicit ticker => + val p = for { + f <- List(1, 2, 3).map(_ => IO.never[List[IO[Int]]]).parFlatSequenceN(2).start + _ <- IO.sleep(100.millis) + _ <- f.cancel + } yield true + + assertCompleteAs(p, true) + } + real("parallel - run parallel actually in parallel") { val x = IO.sleep(2.seconds) >> IO.pure(1) val y = IO.sleep(2.seconds) >> IO.pure(2) From 976ac27b33456fc7f44e6d256cf0e10b88321813 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Fri, 8 Aug 2025 11:00:21 +0200 Subject: [PATCH 2/7] Fix SyntaxSuite and formatting --- .../src/main/scala/cats/effect/kernel/GenConcurrent.scala | 6 +++--- .../src/test/scala/cats/effect/kernel/SyntaxSuite.scala | 2 +- tests/shared/src/test/scala/cats/effect/IOSuite.scala | 5 ++--- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index 87b3d473dc..bddd3cbaf9 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -18,9 +18,9 @@ package cats.effect.kernel import cats.{FlatMap, Foldable, Monoid, Semigroup, Traverse} import cats.data.{EitherT, IorT, Kleisli, OptionT, WriterT} -import cats.effect.kernel.instances.spawn.* -import cats.effect.kernel.syntax.all.* -import cats.syntax.all.* +import cats.effect.kernel.instances.spawn._ +import cats.effect.kernel.syntax.all._ +import cats.syntax.all._ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { diff --git a/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala b/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala index 475ab259ff..38b5ec8db6 100644 --- a/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala +++ b/kernel/shared/src/test/scala/cats/effect/kernel/SyntaxSuite.scala @@ -58,7 +58,7 @@ class SyntaxSuite { } { - val result = List(target).parFlatTraverseN(3)(t => F.pure(List(t))) + val result = List(target).parFlatTraverseN(3)(t => t.map(List(_))) result: F[List[A]] } diff --git a/tests/shared/src/test/scala/cats/effect/IOSuite.scala b/tests/shared/src/test/scala/cats/effect/IOSuite.scala index 3bb9c06d5c..abea7b43a6 100644 --- a/tests/shared/src/test/scala/cats/effect/IOSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/IOSuite.scala @@ -1733,9 +1733,8 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu } real("parFlatSequenceN - propagate errors") { - List(1, 2, 3).map { (n: Int) => - if (n == 2) IO.raiseError(new RuntimeException) else List(n).pure[IO] - } + List(1, 2, 3) + .map { (n: Int) => if (n == 2) IO.raiseError(new RuntimeException) else List(n).pure[IO] } .parFlatSequenceN(2) .mustFailWith[RuntimeException] } From 4500e205419241d449bd8ef0559d1b8f8dd390b6 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Thu, 10 Sep 2026 16:28:55 +0200 Subject: [PATCH 3/7] Adjust parFlatTraverseN implementation With the rewrite of parTraverseN's internals for performance, the MiniSemaphore approach is no longer optimal. This extracts the parTraverseN rewrite into a private helper that takes a function that converts the result into its final state, either via `sequence` for the parTraverseN case or via `flatSequence` for `parFlatTraverseN`. --- .../cats/effect/kernel/GenConcurrent.scala | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index 50634022d8..aa5915a34e 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -143,6 +143,16 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { * limit. */ def parTraverseN[T[_]: Traverse, A, B](n: Int)(ta: T[A])(f: A => F[B]): F[T[B]] = { + implicit val F: GenConcurrent[F, E] = this + parTraverseNImpl[T, A, B, B](n)(ta)(f)(_.sequence[F, B]) + } + + /** + * Shared core implementation for both [[parTraverseN()]] and [[parFlatTraverseN()]], taking a function that decides + * on how to sequence the result. + * @tparam B the intermediate result type of the function `f`, needs to be sequenceable to `C` + */ + private def parTraverseNImpl[T[_]: Traverse, A, B, C](n: Int)(ta: T[A])(f: A => F[B])(seq: T[F[B]] => F[T[C]]): F[T[C]] = { require(n >= 1, s"Concurrency limit should be at least 1, was: $n") implicit val F: GenConcurrent[F, E] = this @@ -242,7 +252,7 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { } } - results.flatMap(_.sequence).onCancel(cancelAllAndJoin) + results.flatMap(seq).onCancel(cancelAllAndJoin) } } } @@ -349,11 +359,8 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { */ def parFlatTraverseN[T[_]: Traverse: FlatMap, A, B](n: Int)(ta: T[A])( f: A => F[T[B]]): F[T[B]] = { - require(n >= 1, s"Concurrency limit should be at least 1, was: $n") - implicit val F: GenConcurrent[F, E] = this - - MiniSemaphore[F](n).flatMap { sem => ta.parFlatTraverse { a => sem.withPermit(f(a)) } } + parTraverseNImpl[T, A, T[B], B](n)(ta)(f)(_.flatSequence) } override def racePair[A, B](fa: F[A], fb: F[B]) From af04b9bba108a2b9a0702d23df6a2b482ed53478 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Thu, 10 Sep 2026 16:34:36 +0200 Subject: [PATCH 4/7] Scalafmt GenConcurrent --- .../main/scala/cats/effect/kernel/GenConcurrent.scala | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index aa5915a34e..b4a72aa8f5 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -148,11 +148,13 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { } /** - * Shared core implementation for both [[parTraverseN()]] and [[parFlatTraverseN()]], taking a function that decides - * on how to sequence the result. - * @tparam B the intermediate result type of the function `f`, needs to be sequenceable to `C` + * Shared core implementation for both [[parTraverseN()]] and [[parFlatTraverseN()]], taking a + * function that decides on how to sequence the result. + * @tparam B + * the intermediate result type of the function `f`, needs to be sequenceable to `C` */ - private def parTraverseNImpl[T[_]: Traverse, A, B, C](n: Int)(ta: T[A])(f: A => F[B])(seq: T[F[B]] => F[T[C]]): F[T[C]] = { + private def parTraverseNImpl[T[_]: Traverse, A, B, C](n: Int)(ta: T[A])(f: A => F[B])( + seq: T[F[B]] => F[T[C]]): F[T[C]] = { require(n >= 1, s"Concurrency limit should be at least 1, was: $n") implicit val F: GenConcurrent[F, E] = this From 15d0223197febba51c780fff09a8f10b22f343e9 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Mon, 14 Sep 2026 12:57:39 +0200 Subject: [PATCH 5/7] Add success test cases Ensure parFlatTraverseN and parFlatSequenceN yield the correct results and run in the expected time, ensuring parallelism. --- .../src/test/scala/cats/effect/IOSuite.scala | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/shared/src/test/scala/cats/effect/IOSuite.scala b/tests/shared/src/test/scala/cats/effect/IOSuite.scala index ddac7aff4a..d9a2b276f4 100644 --- a/tests/shared/src/test/scala/cats/effect/IOSuite.scala +++ b/tests/shared/src/test/scala/cats/effect/IOSuite.scala @@ -2268,6 +2268,15 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu assertCompleteAs(test.attempt.void, ()) } + ticked("parFlatTraverseN - run tasks in parallel and flatten the results (ticked)") { + implicit ticker => + val p = List(1, 2, 3) + .parFlatTraverseN(2) { (n: Int) => IO.sleep(2.seconds) >> IO.pure(List.fill(n)(())) } + .timeoutTo(5.seconds, IO(fail("parFlatSequenceN took too long"))) + + assertCompleteAs(p, List.fill(1 + 2 + 3)(())) + } + real("parFlatTraverseN - throw when n < 1") { IO.defer { List.empty[Int].parFlatTraverseN(0)(List(_).pure[IO]) @@ -2292,6 +2301,16 @@ class IOSuite extends BaseScalaCheckSuite with DisciplineSuite with IOPlatformSu assertCompleteAs(p, true) } + ticked("parFlatSequenceN - run tasks in parallel and flatten the results (ticked)") { + implicit ticker => + val p = List(1, 2, 3) + .map { (n: Int) => IO.sleep(2.seconds) >> IO.pure(List.fill(n)(())) } + .parFlatSequenceN(2) + .timeoutTo(5.seconds, IO(fail("parFlatSequenceN took too long"))) + + assertCompleteAs(p, List.fill(1 + 2 + 3)(())) + } + real("parFlatSequenceN - throw when n < 1") { IO.defer { List.empty[IO[List[Int]]].parFlatSequenceN(0) From f5f4d3cc72b220c5f41b52915fe9f634b0676de9 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Mon, 14 Sep 2026 13:01:39 +0200 Subject: [PATCH 6/7] Adjust docs in regard to fairness --- .../src/main/scala/cats/effect/kernel/GenConcurrent.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index b4a72aa8f5..b01f7b545e 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -349,15 +349,15 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { } /** - * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. + * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. See + * [[parSequenceN()]] for fairness considerations, the same semantics apply here. */ def parFlatSequenceN[T[_]: Traverse: FlatMap, A](n: Int)(tma: T[F[T[A]]]): F[T[A]] = parFlatTraverseN(n)(tma)(identity) /** - * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. Note that the - * semantics of this operation aim to maximise fairness: when a spot to execute becomes - * available, every task has a chance to claim it, and not only the next `n` tasks in `ta` + * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. See + * [[parTraverseN()]] for fairness considerations, the same semantics apply here. */ def parFlatTraverseN[T[_]: Traverse: FlatMap, A, B](n: Int)(ta: T[A])( f: A => F[T[B]]): F[T[B]] = { From 9e3dc7d8e44222da6c7737a8764413c156b168d7 Mon Sep 17 00:00:00 2001 From: Yannick Heiber Date: Mon, 14 Sep 2026 13:44:48 +0200 Subject: [PATCH 7/7] Fix scaladoc references to other methods --- .../main/scala/cats/effect/kernel/GenConcurrent.scala | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala index b01f7b545e..da74552d65 100644 --- a/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala +++ b/kernel/shared/src/main/scala/cats/effect/kernel/GenConcurrent.scala @@ -148,7 +148,7 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { } /** - * Shared core implementation for both [[parTraverseN()]] and [[parFlatTraverseN()]], taking a + * Shared core implementation for both [[parTraverseN]] and [[parFlatTraverseN]], taking a * function that decides on how to sequence the result. * @tparam B * the intermediate result type of the function `f`, needs to be sequenceable to `C` @@ -349,15 +349,15 @@ trait GenConcurrent[F[_], E] extends GenSpawn[F, E] { } /** - * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. See - * [[parSequenceN()]] for fairness considerations, the same semantics apply here. + * Like `Parallel.parFlatSequence`, but limits the degree of parallelism. See [[parSequenceN]] + * for fairness considerations, the same semantics apply here. */ def parFlatSequenceN[T[_]: Traverse: FlatMap, A](n: Int)(tma: T[F[T[A]]]): F[T[A]] = parFlatTraverseN(n)(tma)(identity) /** - * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. See - * [[parTraverseN()]] for fairness considerations, the same semantics apply here. + * Like `Parallel.parFlatTraverse`, but limits the degree of parallelism. See [[parTraverseN]] + * for fairness considerations, the same semantics apply here. */ def parFlatTraverseN[T[_]: Traverse: FlatMap, A, B](n: Int)(ta: T[A])( f: A => F[T[B]]): F[T[B]] = {