ParExecExample.scala

package example

import java.util.concurrent.{Executors, TimeUnit}

import cats.data.ReaderT
import cats.effect.{Concurrent, ContextShift, IO}
import cats.implicits._
import cats.kernel.Monoid
import cats.{Parallel, effect}
import example.APIClient.TotalResult
import simulacrum.typeclass

import scala.concurrent.ExecutionContext

final case class Result[A](payload: Seq[A], total: Long)

object Result {
  implicit def monoid[A]: Monoid[Result[A]] = Monoid.instance(Result[A](Seq.empty, 0), {
    case (Result(p1, t1), Result(p2, t2)) => Result(p1 ++ p2, t1 + t2)
  })
}


final case class Request[A](from: Long, size: Long)

@typeclass
sealed trait ResultGen[A] {
  def make(req: Request[A], total: Long): Result[A]
}

object ResultGen {
  def instance[A](f: => A): ResultGen[A] = new ResultGen[A] {
    override def make(req: Request[A], total: Long): Result[A] = Result(Seq.fill(total.toInt)(f), total)
  }
}

object APIClient {
  type TotalResult = Long

  def run[F[_] : Concurrent, A: ResultGen](req: Request[A]): ReaderT[F, TotalResult, Result[A]] = ReaderT { total =>
    Concurrent[F].delay {
      Thread.sleep(50)
      ResultGen[A].make(req, total)
    }
  }
}

final case class Bench(min: Long, max: Long, avg: Float, runs: Seq[Long])

object Bench {
  def apply[A](n: Long)(f: IO[A]): Unit = {
    val r = 0L.to(n).foldLeft(Bench(0, 0, 0, Seq.empty)) {
      case (stat, _) => {
        val start = System.nanoTime()
        f.unsafeRunSync()
        val end = System.nanoTime()
        val runs = stat.runs :+ (end - start)
        Bench(runs.min, runs.max, runs.sum / runs.length, runs)
      }

    }
    println(r)
  }

  def withContextShift[A](nThreads: Int, nTry:Long)(f: ContextShift[IO] => IO[A]) = {
    val es = Executors.newFixedThreadPool(nThreads)
    val s: effect.ContextShift[IO] = IO.contextShift(ExecutionContext.fromExecutorService(es))
    apply(nTry)(f(s))
    es.shutdownNow()
    es.awaitTermination(1000, TimeUnit.MILLISECONDS)
  }
}

object Main2 extends App {
  implicit val stringResultGen: ResultGen[String] = ResultGen.instance("Result")

  // 1スレッド
  Bench.withContextShift(1, 5)(implicit s => run[IO, String](10).run(500L))
  Bench.withContextShift(1, 5)(implicit s => run[IO, String](10).run(500L))

  // 5スレッド
  Bench.withContextShift(5, 5)(implicit s => run[IO, String](10).run(500L))
  Bench.withContextShift(5, 5)(implicit s => run[IO, String](10).run(500L))


  def run[F[_] : Concurrent : Parallel, A: ResultGen](size: Long): ReaderT[F, TotalResult, Result[A]] = for {
    start <- APIClient.run(Request[A](0, size))
    totalPages = start.total / size
    result <- ReaderT[F, TotalResult, Result[A]] { total =>
      2L.to(totalPages).toList
        .parTraverse(page => APIClient.run(Request[A]((page - 1) * size, size))
          .run(total))
        .map(rs => Monoid[Result[A]].combineAll(rs))
    }
  } yield result


}

MacBook Pro (13-inch, 2016, Four Thunderbolt 3 Ports)

Bench(2611349415,3190680809,2.72644864E9,List(3190680809, 2645437845, 2643156487, 2611349415, 2638400866, 2629667042))
Bench(2619746945,2654153505,2.6376215E9,List(2646086712, 2626612825, 2642719069, 2654153505, 2636410595, 2619746945))
Bench( 573457443, 590451682,  5.8321184E8,List(584264282, 573457443, 590451682, 581844766, 584933968, 584319079))
Bench( 573712735, 599906629,  5.8791386E8,List(588304737, 590146235, 578793420, 599906629, 596619243, 573712735))

threads = {1, 5, 10, 20}

Bench(2630075284,3561552907,2.7920919E9,List(3561552907, 2630075284, 2644926073, 2640361890, 2632187552, 2643447810))
Bench(2611302866,2659704498,2.63139405E9,List(2612010971, 2627109253, 2659704498, 2611302866, 2636230754, 2642006546))
Bench(581058893,598724229,5.8918989E8,List(598724229, 581058893, 582645451, 590625476, 593747410, 588337807))
Bench(573016388,598610372,5.8632198E8,List(573016388, 598610372, 583848112, 594520068, 588292169, 579644798))
Bench(316290811,329196332,3.2357232E8,List(325769662, 329196332, 316290811, 320469062, 326309433, 323398610))
Bench(317674949,328978799,3.22529792E8,List(318167103, 328978799, 322969469, 317674949, 325297628, 322090877))
Bench(215721211,229846447,2.1896744E8,List(215961961, 215721211, 229846447, 218328793, 215970937, 217975277))
Bench(213432130,220762859,2.17869584E8,List(220169043, 216496795, 220762859, 213432130, 217750237, 218606433))