Scala標準のFutureと並列コレクションの実行コンテキスト

この投稿はScala Advent Calendar 2014の5日目の記事です。4日目はmeganemuraさんの Scalastyle の導入 でした。
5日に出先から戻って以降ダウンしていた影響で、公開が翌朝になっております。すいません。

記事のテーマはシンプルで、Scalaの標準ライブラリの中で勝手にうまいことやってくれる(らしい)並列・非同期処理について、実行されるスレッドがどうなっているのか調べてみるものです。

Future

どこのFutureチュートリアルでも基本的に、 import ExecutionContext.Implicits.global をとりあえず書きましょう、と但し書きしてスタートします。これがデフォルトのFuture用の実行コンテキスト。 実行コンテキストの指定をしなかった場合 Cannot find an implicit ExecutionContext と怒られます。

ExecutionContext.Implicits.global の実装は、object Implicitの中にimplicit lazy val global: ExecutionContextExecutorがあって…中略、辿っていくと最終的に、以下の部分に辿り着きます。2.11.4では scala/concurrent/impl/ExecutionContextImpl.scala 内。

    try {
      new ForkJoinPool(
        desiredParallelism,
        threadFactory,
        uncaughtExceptionHandler,
        true) // Async all the way baby
    } catch {

java.util.conncurrentにあるForkJoinPoolですね。
デフォルトのものを使った場合の挙動は以前確認してみたのですが、最大スレッド数＝CPU数としてForkJoinスレッドを扱う、無難な実装であるようです。

並列コレクション (parallel collection)

listやmap等のコレクションに対し、par を呼び出すことで簡単に並列処理になるよ！というやつです。

x: List[Int] =  // なにかListがあるとして…
scala> x.map(_ + 1)  // 普通のmap
scala> x.par.map(_ + 1)  // 並列版

基本的な説明も使い方も、公式のドキュメントにあります。設定は「並列コレクションの設定」のページ。特に実行環境を何も指定せず、おもむろに par を使っても、デフォルトの設定により実行されます。Futureと違って怒られない。

デフォルト以外の設定をする方法

Futureが要求するのはscala.concurrent.ExecutionContextExecutor で、並列コレクションが要求するのはscala.collection.parallel.TaskSupportなので微妙に違いますが、大したことはありません。

scala> import scala.concurrent._
scala> val pool = new forkjoin.ForkJoinPool(4)

pool: scala.concurrent.forkjoin.ForkJoinPool = scala.concurrent.forkjoin.ForkJoinPool@7f39e4a3[Running, parallelism = 4, size = 0, active = 0, running = 0, steals = 0, tasks = 0, submissions = 0]

並列コレクションはtasksuppoortに設定。

scala> import scala.collection.parallel
scala> val pc = parallel.mutable.ParArray(1, 2, 3)

pc: scala.collection.parallel.mutable.ParArray[Int] = ParArray(1, 2, 3)

scala> pc.tasksupport = new scala.collection.parallel.ForkJoinTaskSupport(pool)

pc.tasksupport: scala.collection.parallel.TaskSupport = scala.collection.parallel.ForkJoinTaskSupport@81df807

Futureは、object ExecutionContextを使ってなんとか。

scala> implicit val ece: ExecutionContextExecutor = scala.concurrent.ExecutionContext.fromExecutor(pool)

ece: scala.concurrent.ExecutionContextExecutor = scala.concurrent.impl.ExecutionContextImpl@2bacfa90

これでどちらも好きに設定できそうです。



この投稿はScala Advent Calendar 2014の5日目の記事でした。
6日目はkawachiさんによるscalac にもっと警告してもらう です。