3

如何使用 Scala 演员计算阶乘？

与例如相比，它是否会证明更省时

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

非常感谢。

4

1 回答 1

4

问题

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

这可以归结为更广泛的问题：所谓的并行前缀计算。您可以在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

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

现在我创建演员：

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 更快）。因此，对数字进行洗牌可能是个好主意，这样大数字就会与小数字混合在一起。这在这种情况下有效，因为我们使用了交换操作。并行前缀计算通常只需要一个关联操作即可工作。1 * ... * 1090 * ... * 100

表现

理论上

对于少量数据，参与者解决方案的性能比“幼稚”解决方案（使用并行集合）更差。当您在专用硬件（例如显卡或 FPGA）或多台机器上进行复杂计算或分发计算时，actor 解决方案将大放异彩。有了演员，你可以控制，谁做哪个计算，你甚至可以重新启动“悬空计算”。这可以给一个很大的加速。

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

一些测量

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

首先我设置了演员系统：

// 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))

我得到以下结果

> “par product”的时间：134.38289ms
  res0: scala.math.BigInt = 284625968091705451890641321211986889014805140170279923
  079417999427441134000376444377299078675778477581588406214231752883004233994015
  351873905242116138271617481982419982759241828925978789812425312059465996259867
  065601615720360323979263287367170557419759620994797203461536981198970926112775
  004841988454104755446424421365733030767036288258035489674611170973695786036701
  910715127305872810411586405612811653853259684258259955846881464304255898366493
  170592517172042765974074461334000541940524623034368691540594040662278282483715
  120383221786446271838229238996389928272218797024593876938030946273322925705554
  596900278752822425443480211275590191694254290289169072190970836905398737474524
  833728995218023632827412170402680867692104515558405671725553720158521328290342
  799898184493136...

  “演员产品”时间：1310.217247ms
  res2: Any = CalculationResult(28462596809170545189064132121198688901480514017027
  992307941799942744113400037644437729907867577847758158840621423175288300423399
  401535187390524211613827161748198241998275924182892597878981242531205946599625
  986706560161572036032397926328736717055741975962099479720346153698119897092611
  277500484198845410475544642442136573303076703628825803548967461117097369578603
  670191071512730587281041158640561281165385325968425825995584688146430425589836
  649317059251717204276597407446133400054194052462303436869154059404066227828248
  371512038322178644627183822923899638992827221879702459387693803094627332292570
  555459690027875282242544348021127559019169425429028916907219097083690539873747
  452483372899521802363282741217040268086769210451555840567172555372015852132829
  034279989818449...

> “par sum”的时间：6.488620999999999ms
  res3: scala.math.BigInt = 50005000

>“演员总和”时间：657.752832ms
  res5: Any = CalculationResult(50005000,0)

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

于 2014-02-07T14:17:51.790 回答