Benchmark IO.async vs manual Deferred

DeferredAsyncPerf.scala

import cats.effect.std.Dispatcher
import cats.effect.{IO, IOApp}
import cats.syntax.all._

import java.util.concurrent.atomic.AtomicInteger

object DeferredAsyncPerf extends IOApp.Simple {

  trait Model {
    def addListener(onResponse: Either[Throwable, Int] => Unit): Unit

    def addListenerInfallible(onResponse: Int => Unit): Unit

    def removeListener(onResponse: Either[Throwable, Int] => Unit): Unit
  }

  val model = new Model {
    private val counter = new AtomicInteger(0)

    def addListener(onResponse: Either[Throwable, Int] => Unit): Unit = onResponse(Right(1))

    def addListenerInfallible(onResponse: Int => Unit): Unit = onResponse(1)

    def removeListener(onResponse: Either[Throwable, Int] => Unit): Unit = ()
  }

  val asyncViaAsync: IO[Int] = IO.async { cb =>
    IO {
      model.addListener(cb)
      Some(IO(model.removeListener(cb)))
    }
  }

  def asyncViaDeferred(dispatcher: Dispatcher[IO]): IO[Int] =
    IO.deferred[Either[Throwable, Int]].flatMap { promise =>
      val cb: Either[Throwable, Int] => Unit = result => dispatcher.unsafeRunAndForget(promise.complete(result))
      model.addListener(cb)
      promise.get.rethrow.onCancel(IO(model.removeListener(cb)))
    }

  def asyncViaDeferredInfallible(dispatcher: Dispatcher[IO]): IO[Int] = {
    IO.deferred[Int].flatMap { promise =>
      model.addListenerInfallible(n => dispatcher.unsafeRunAndForget(promise.complete(n)))
      promise.get
    }
  }


  def run: IO[Unit] = {
    val n = 500_000

    def benchmark(dispatcher: Dispatcher[IO]): IO[Unit] = {
      val viaAsync = asyncViaAsync.replicateA_(n).timed.map(_._1.toMillis)
      val viaDeferred = asyncViaDeferred(dispatcher).replicateA_(n).timed.map(_._1.toMillis)
      val viaDeferredInfallible = asyncViaDeferredInfallible(dispatcher).replicateA_(n).timed.map(_._1.toMillis)

      viaAsync >> IO.print("Benchmark viaAsync: ") >> viaAsync.flatTap(IO.println).void >>
        viaDeferred >> IO.print("Benchmark viaDeferred: ") >> viaDeferred.flatTap(IO.println).void >>
        viaDeferredInfallible >> IO.print("Benchmark viaDeferredInfallible: ") >> viaDeferredInfallible.flatTap(IO.println).void
    }

    (1 to 3).toList.traverse_ { n =>
      IO.println(s"Run #$n") >>
        IO.println("Sequential dispatcher") >>
        Dispatcher.sequential[IO].use(benchmark) >>
        IO.println("\nParallel dispatcher") >>
        Dispatcher.parallel[IO].use(benchmark) >>
        IO.println("*********************************")
    }
  }
}

Results:

Sequential dispatcher
Benchmark viaAsync: 57
Benchmark viaDeferred: 3039
Benchmark viaDeferredInfallible: 4008

Parallel dispatcher
Benchmark viaAsync: 55
Benchmark viaDeferred: 7891
Benchmark viaDeferredInfallible: 8090