0

Scala 2.9 / Akka2.0RC2コードの同時実行性とパフォーマンスを向上させる機会を探しています。次のコードが与えられます：

import akka.actor._

case class DataDelivery(data:Double)

class ComputeActor extends Actor {
    var buffer = scala.collection.mutable.ArrayBuffer[Double]()

    val functionsToCompute = List("f1","f2","f3","f4","f5")
    var functionMap = scala.collection.mutable.LinkedHashMap[String,(Map[String,Any]) => Double]()  
    functionMap += {"f1" -> f1}
    functionMap += {"f2" -> f2}
    functionMap += {"f3" -> f3}
    functionMap += {"f4" -> f4}
    functionMap += {"f5" -> f5}

    def updateData(data:Double):scala.collection.mutable.ArrayBuffer[Double] = {
        buffer += data
        buffer
    }

    def f1(map:Map[String,Any]):Double = {
//    println("hello from f1")
      0.0
    }

    def f2(map:Map[String,Any]):Double = {
//    println("hello from f2")
      0.0
    }

    def f3(map:Map[String,Any]):Double = {
//    println("hello from f3")
      0.0
    }

    def f4(map:Map[String,Any]):Double = {
//    println("hello from f4")
      0.0
    }

    def f5(map:Map[String,Any]):Double = {
//    println("hello from f5")
      0.0
    }

    def computeValues(immutableBuffer:IndexedSeq[Double]):Map[String,Double] = {
        var map = Map[String,Double]()
        try {
            functionsToCompute.foreach(function => {
                val value = functionMap(function)
                function match {
                    case "f1" =>
                        var v = value(Map("lookback"->10,"buffer"->immutableBuffer,"parm1"->0.0))
                        map += {function -> v}
                    case "f2" =>
                        var v = value(Map("lookback"->20,"buffer"->immutableBuffer))
                        map += {function -> v}
                    case "f3" =>
                        var v = value(Map("lookback"->30,"buffer"->immutableBuffer,"parm1"->1.0,"parm2"->false))
                        map += {function -> v}
                    case "f4" =>
                        var v = value(Map("lookback"->40,"buffer"->immutableBuffer))
                        map += {function -> v}
                    case "f5" =>
                        var v = value(Map("buffer"->immutableBuffer))
                        map += {function -> v}
                    case _ => 
                        println(this.unhandled())
                }
            })
        } catch {
            case ex: Exception =>
              ex.printStackTrace()
        }
        map
    }

    def receive = {
      case DataDelivery(data) =>
        val startTime = System.nanoTime()/1000
        val answers = computeValues(updateData(data))
        val endTime = System.nanoTime()/1000
        val elapsedTime = endTime - startTime
        println("elapsed time is " + elapsedTime)
        // reply or forward
      case msg =>
        println("msg is " + msg)
    }

}

object Test {
    def main(args:Array[String]) {
        val system = ActorSystem("actorSystem") 
        val computeActor = system.actorOf(Props(new ComputeActor),"computeActor")
        var i = 0
        while (i < 1000) {  
            computeActor ! DataDelivery(i.toDouble)
            i += 1
        }
    }
}

これを実行すると、出力（マイクロ秒に変換）は次のようになります。

elapsed time is 4898
elapsed time is 184
elapsed time is 144
    .
    .
    .
elapsed time is 109
elapsed time is 103

JVMのインクリメンタルコンパイラが起動しているのがわかります。

私は1つの迅速な勝利は変わることかもしれないと思いました

    functionsToCompute.foreach(function => {

に

    functionsToCompute.par.foreach(function => {

ただし、これにより、次の経過時間が発生します

elapsed time is 31689
elapsed time is 4874
elapsed time is 622
    .
    .
    .
elapsed time is 698
elapsed time is 2171

いくつかの情報：

1）これを2コアのMacbookProで実行しています。

2）フルバージョンでは、関数は、可変共有バッファーの一部をループする長時間実行操作です。アクターのメールボックスからのメッセージの取得がフローを制御しているため、これは問題ではないようですが、同時実行性の増加に伴う問題である可能性があります。これが、私がIndexedSeqに変換した理由です。

3）フルバージョンでは、functionsToComputeリストが異なる場合があるため、functionMap内のすべてのアイテムが必ずしも呼び出されるわけではありません（つまり、functionMap.sizeがfunctionsToCompute.sizeよりもはるかに大きい場合があります）。

4）関数は並行して計算できますが、結果のマップは戻る前に完成している必要があります

いくつかの質問：

1）並列バージョンをより高速に実行するにはどうすればよいですか？

2）非ブロッキングおよびブロッキング先物を追加することはどこに意味がありますか？

3）計算を別のアクターに転送することはどこで意味がありますか？

4）不変性/安全性を高めるためのいくつかの機会は何ですか？

ありがとう、ブルース

4

1 に答える 1

2

要求に応じて例を示します（遅延について申し訳ありません... SOの通知はありません）。

「先物の構成」に関するAkkaのドキュメントのセクションに素晴らしい例がありますが、状況に合わせてもう少し調整したものを紹介します。

さて、これを読んだ後、Akkaのウェブサイトのチュートリアルとドキュメントを読んでください。それらのドキュメントが提供する多くの重要な情報が不足しています。

import akka.dispatch.{Await, Future, ExecutionContext}
import akka.util.duration._
import java.util.concurrent.Executors

object Main {
  // This just makes the example work.  You probably have enough context
  // set up already to not need these next two lines
  val pool = Executors.newCachedThreadPool()
  implicit val ec = ExecutionContext.fromExecutorService(pool)

  // I'm simulating your function.  It just has to return a tuple, I believe
  // with a String and a Double
  def theFunction(s: String, d: Double) = (s, d)
  def main(args: Array[String]) {
    // Here we run your functions - I'm just doing a thousand of them
    // for fun.  You do what yo need to do
    val listOfFutures = (1 to 1000) map { i =>
      // Run them in parallel in the future
      Future {
        theFunction(i.toString, i.toDouble)
      }
    }
    // These lines can be composed better, but breaking them up should
    // be more illustrative.
    //
    // Turn the list of Futures (i.e. Seq[Future[(String, Double)]]) into a
    // Future with a sequence of results (i.e. Future[Seq[(String, Double)]])
    val futureOfResults = Future.sequence(listOfFutures)

    // Convert that future into another future that contains a map instead
    // instead of a sequence
    val intermediate = futureOfResults map { _.toList.toMap }

    // Wait for it complete.  Ideally you don't do this.  Continue to
    // transform the future into other forms or use pipeTo() to get it to go
    // as a result to some other Actor.  "Await" is really just evil... the
    // only place you should really use it is in silly programs like this or
    // some other special purpose app.
    val resultingMap = Await.result(intermediate, 1 second)
    println(resultingMap)

    // Again, just to make the example work
    pool.shutdown()
  }
}

これを実行するためにクラスパスに必要なのはakka-actorjarだけです。AkkaのWebサイトでは、必要なものを設定する方法を説明していますが、それは非常に簡単です。

于 2012-03-01T15:33:18.273 に答える