scala - 使用 Scala actors 进行阶乘计算

Question

如何使用 Scala actor 计算阶乘？

与实例相比，它会证明更省时吗

def factorial(n: Int): BigInt = (BigInt(1) to BigInt(n)).par.product

非常感谢。

Answer 1

问题

您必须将输入拆分为部分产品。然后可以并行计算此部分产品。然后将部分乘积相乘得到最终乘积。

这可以归结为更广泛的问题类别:所谓的并行前缀计算。您可以在 Wikipedia 上阅读相关信息.

简化版:当你用关联运算_ * _计算a*b*c*d时，你可以构造计算a*(b *(c*d)) 或 (a*b)*(c*d)。使用第二种方法，您可以并行计算 a*b 和 c*d，然后根据这些部分结果计算最终结果。当然，当您有更多的输入值时，您可以递归地执行此操作。

解决方案

免责声明

这听起来有点像家庭作业。所以我将提供一个具有两个属性的解决方案:

1. 它包含一个小错误
2. 它展示了如何解决一般的并行前缀，而不是直接解决问题

所以你可以看到解决方案应该如何构建，但没有人可以用它来作弊。

详细解决方案

首先我需要一些导入

导入 akka.event.Logging 导入 java.util.concurrent.TimeUnit 导入 scala.concurrent.duration.FiniteDuration 导入 akka.actor._

然后我为参与者之间的通信创建了一些辅助类

case class Calculate[T](values : Seq[T], segment : Int, parallelLimit : Int, fn : (T,T) => T)

trait CalculateResponse
case class CalculationResult[T](result : T, index : Int) extends CalculateResponse
case object Busy extends CalculateResponse

除了告诉接收者您很忙之外，Actor 还可以使用存储或实现自己的队列来获取部分结果。但在这种情况下，我认为发送方应该决定允许多少并行计算。

现在我创建 Actor :

class ParallelPrefixActor[T] extends Actor {
  val log = Logging(context.system, this)
  val subCalculation = Props(classOf[ParallelPrefixActor[BigInt]])
  val fanOut = 2
  def receive = waitForCalculation

  def waitForCalculation : Actor.Receive = {
    case c : Calculate[T] =>
      log.debug(s"Start calculation for ${c.values.length} values, segment nr. ${c.index}, from ${c.values.head} to ${c.values.last}")
      if (c.values.length < c.parallelLimit) {
        log.debug("Calculating result direct")
        val result = c.values.reduceLeft(c.fn)
        sender ! CalculationResult(result, c.index)
      }else{
        val groupSize: Int = Math.max(1, (c.values.length / fanOut) + Math.min(c.values.length % fanOut, 1))
        log.debug(s"Splitting calculation for ${c.values.length} values up to ${fanOut} children, ${groupSize} elements each, limit ${c.parallelLimit}")
        def segments=c.values.grouped(groupSize)
        log.debug("Starting children")
        segments.zipWithIndex.foreach{case (values, index) =>
          context.actorOf(subCalculation) ! c.copy(values = values, index = index)
        }
        val partialResults: Vector[T] = segments.map(_.head).to[Vector]
        log.debug(s"Waiting for ${partialResults.length} results (${partialResults.indices})")
        context.become(waitForResults(segments.length, partialResults, c, sender), discardOld = true)
      }
  }
  def waitForResults(outstandingResults : Int, partialResults : Vector[T], originalRequest : Calculate[T], originalSender : ActorRef) : Actor.Receive = {
    case c : Calculate[_] => sender ! Busy
    case r : CalculationResult[T] =>
      log.debug(s"Putting result ${r.result} on position ${r.index} in ${partialResults.length}")
      val updatedResults = partialResults.updated(r.index, r.result)
      log.debug("Killing sub-worker")
      sender ! PoisonPill
      if (outstandingResults==1) {
        log.debug("Calculating result from partial results")
        val result = updatedResults.reduceLeft(originalRequest.fn)
        originalSender ! CalculationResult(result, originalRequest.index)
        context.become(waitForCalculation, discardOld = true)
      }else{
        log.debug(s"Still waiting for ${outstandingResults-1} results")
        // For fanOut > 2 one could here already combine consecutive partial results
        context.become(waitForResults(outstandingResults-1, updatedResults, originalRequest, originalSender), discardOld = true)
      }
  }
}

优化

使用并行前缀计算并不是最优的。计算较大数字乘积的参与者比计算较小数字乘积的参与者做更多的工作(例如，在计算 1 * ... * 100 时，计算 更快code>1 * ... * 10 比 90 * ... * 100)。所以打乱数字可能是个好主意，这样大数字就会和小数字混合在一起。这在这种情况下有效，因为我们使用了交换操作。并行前缀计算一般只需要一个关联运算即可。

性能

理论上

对于少量数据，参与者解决方案的性能比“朴素”解决方案(使用并行集合)差。当您进行复杂计算或将计算分布在专用硬件(例如图形卡或 FPGA)或多台机器上时，Actor 解决方案将会大放异彩。有了你可以控制的 Actor ，谁做哪个计算，你甚至可以重新启动“挂起的计算”。这可以给出 big speed up .

在单台机器上，当您的内存架构不一致时，actor 解决方案可能会有所帮助。然后，您可以通过将内存固定到特定处理器的方式来组织参与者。

一些测量

我在 IntelliJ IDEA 中使用 Scala 工作表进行了一些实际性能测量。

首先我设置了 Actor 系统:

// Setup the actor system
val system = ActorSystem("root")
// Start one calculation actor
val calculationStart = Props(classOf[ParallelPrefixActor[BigInt]])


val calcolon = system.actorOf(calculationStart, "Calcolon-BigInt")

val inbox = Inbox.create(system)

然后我定义了一个辅助方法来测量时间:

// Helper function to measure time
def time[A] (id : String)(f: => A) = {
  val start = System.nanoTime()
  val result = f
  val stop = System.nanoTime()
  println(s"""Time for "${id}": ${(stop-start)*1e-6d}ms""")
  result
}

然后我做了一些性能测量:

// Test code
val limit = 10000
def testRange = (1 to limit).map(BigInt(_))

time("par product")(testRange.par.product)
val timeOut = FiniteDuration(240, TimeUnit.SECONDS)
inbox.send(calcolon, Calculate[BigInt]((1 to limit).map(BigInt(_)), 0, 10, _ * _))
time("actor product")(inbox.receive(timeOut))

time("par sum")(testRange.par.sum)
inbox.send(calcolon, Calculate[BigInt](testRange, 0, 5, _ + _))
time("actor sum")(inbox.receive(timeOut))

我得到了以下结果

> Time for "par product": 134.38289ms  res0: scala.math.BigInt = 284625968091705451890641321211986889014805140170279923  079417999427441134000376444377299078675778477581588406214231752883004233994015  351873905242116138271617481982419982759241828925978789812425312059465996259867  065601615720360323979263287367170557419759620994797203461536981198970926112775  004841988454104755446424421365733030767036288258035489674611170973695786036701  910715127305872810411586405612811653853259684258259955846881464304255898366493  170592517172042765974074461334000541940524623034368691540594040662278282483715  120383221786446271838229238996389928272218797024593876938030946273322925705554  596900278752822425443480211275590191694254290289169072190970836905398737474524  833728995218023632827412170402680867692104515558405671725553720158521328290342  799898184493136...  Time for "actor product": 1310.217247ms  res2: Any = CalculationResult(28462596809170545189064132121198688901480514017027  992307941799942744113400037644437729907867577847758158840621423175288300423399  401535187390524211613827161748198241998275924182892597878981242531205946599625  986706560161572036032397926328736717055741975962099479720346153698119897092611  277500484198845410475544642442136573303076703628825803548967461117097369578603  670191071512730587281041158640561281165385325968425825995584688146430425589836  649317059251717204276597407446133400054194052462303436869154059404066227828248  371512038322178644627183822923899638992827221879702459387693803094627332292570  555459690027875282242544348021127559019169425429028916907219097083690539873747  452483372899521802363282741217040268086769210451555840567172555372015852132829  034279989818449...> Time for "par sum": 6.488620999999999ms  res3: scala.math.BigInt = 50005000> Time for "actor sum": 657.752832ms  res5: Any = CalculationResult(50005000,0)

你可以很容易地看到 actor 版本比使用并行集合要慢得多。