Cyclic Barrier

A re-usable synchronization primitive that allows a set of fibers to wait until they've all reached the same point.

trait CyclicBarrier[F[_]] {

  def await: F[Unit]

}

A cyclic barrier is initialized with a positive integer n and fibers which call await are semantically blocked until n of them have invoked await, at which point all of them are unblocked and the cyclic barrier is reset.

await cancelation is supported, in which case the number of fibers required to unblock the cyclic barrier is incremented again.

import cats.implicits._
import cats.effect._
import cats.effect.std.CyclicBarrier
import cats.effect.unsafe.implicits.global
import scala.concurrent.duration._

val run = (for {
  b <- CyclicBarrier[IO](2)
  f1 <- (IO.println("fast fiber before barrier") >>
      b.await >> 
      IO.println("fast fiber after barrier")
    ) .start
  f2 <- (IO.sleep(1.second) >>
      IO.println("slow fiber before barrier") >>
      IO.sleep(1.second) >>
      b.await >>
      IO.println("slow fiber after barrier")
    ).start
  _ <- (f1.join, f2.join).tupled
} yield ())
// run: IO[Unit] = FlatMap(
//   ioe = Map(
//     ioe = FlatMap(
//       ioe = Map(
//         ioe = Delay(
//           thunk = cats.effect.IO$$$Lambda$16270/0x0000000803975840@3647e6a2,
//           event = cats.effect.tracing.TracingEvent$StackTrace
//         ),
//         f = cats.effect.std.CyclicBarrier$$$Lambda$16311/0x0000000803a05040@47f4ac0f,
//         event = cats.effect.tracing.TracingEvent$StackTrace
//       ),
//       f = cats.effect.std.CyclicBarrier$$$Lambda$16312/0x0000000803a04040@6cc40162,
//       event = cats.effect.tracing.TracingEvent$StackTrace
//     ),
//     f = cats.effect.std.CyclicBarrier$$$Lambda$16313/0x0000000803a03840@2525588a,
//     event = cats.effect.tracing.TracingEvent$StackTrace
//   ),
//   f = <function1>,
//   event = cats.effect.tracing.TracingEvent$StackTrace
// )

run.unsafeRunSync()