轻量级的 Scala fork join 语法

Question

尽管即将推出 java 7 标准 fork/join 框架，但我正在构建一些语法轻量级的辅助方法，以便客户端并行运行代码。 这是一个可运行的 main 方法来说明这个想法。

import actors.Futures

object ForkTest2 {



  def main(args: Array[String]) {
    test1
    test2
  }



  def test1 {
    val (a, b, c) =fork({
      Thread.sleep(500)
      println("inside fx1 ",+System.currentTimeMillis)
      true
    }, {
      Thread.sleep(1000)
      println("inside fx2 ",+System.currentTimeMillis)
      "stringResult"
    }, {
      Thread.sleep(1500)
      println("inside fx3 ",+System.currentTimeMillis)
      1
    })

    println(b, a, c)
    true
  }

  def test2 {
    val results = forkAll({
      () =>
              Thread.sleep(500)
              println("inside fx1 ",+System.currentTimeMillis)
              true
    }, {
      () =>
              Thread.sleep(1000)
              println("inside fx2 ",+System.currentTimeMillis)
              "stringResult"
    }, {
      () =>
              Thread.sleep(1500)
              println("inside fx3 ",+System.currentTimeMillis)
              1
    }, {
      () =>
              Thread.sleep(2000)
              println("inside fx4 ",+System.currentTimeMillis)
              1.023
    })

    println(results)
    true
  }

  val tenMinutes = 1000 * 60 * 10

  def fork[A, B, C](
          fx1: => A,
          fx2: => B,
          fx3: => C
          ) = {
    val re1 = Futures.future(fx1)
    val re2 = Futures.future(fx2)
    val re3 = Futures.future(fx3)
    //default wait 10 minutes
    val result = Futures.awaitAll(tenMinutes, re1, re2, re3)
    (
            result(0).asInstanceOf[Option[A]],
            result(1).asInstanceOf[Option[B]],
            result(2).asInstanceOf[Option[C]]

            )
  }

  type fxAny = () => Any

  def forkAll(
          fx1: fxAny*
          ): List[Any] = {
    val results = fx1.toList.map {fx: fxAny => Futures.future(fx())}
    Futures.awaitAll(tenMinutes, results: _*)
  }
}

示例输出是

(inside fx1 ,1263804802301)
(inside fx2 ,1263804802801)
(inside fx3 ,1263804803301)
(Some(stringResult),Some(true),Some(1))
(inside fx1 ,1263804803818)
(inside fx2 ,1263804804318)
(inside fx3 ,1263804804818)
(inside fx4 ,1263804805318)
List(Some(true), Some(stringResult), Some(1), Some(1.023))

test 1 说明类型安全返回类型

test 2 说明任意输入参数

我希望将这两种测试方法结合起来，以便客户端代码可以与类型安全返回类型并行运行任意函数。

关于任意函数参数的另一点是：

我认为该行

  type fxAny = () => Any

确实应该编码为

  type fxAny =  => Any

，但 scala 编译器不允许我这样做。

任何帮助都是值得赞赏的。

Answer 1

Eric Torreborre 在 @retronym 提供的链接中写道：

trait LazyParameters { 
  /** transform a value to a zero-arg function returning that value */ 
  implicit def toLazyParameter[T](value: =>T) = new LazyParameter(() => value) 
  /** class holding a value to be evaluated lazily */ 
  class LazyParameter[T](value: ()=>T) { 
    lazy val v = value() 
    def apply() = v 
  } 
} 

这是您的测试的 LazyParameter 版本：

object ForkTest2 extends LazyParameters {

...

def forkAll(fx1: LazyParameter[Any]*): List[Any] = {
  val results = fx1.toList.map {
    fx: LazyParameter[Any] => Futures.future(fx.apply())}
  Futures.awaitAll(tenMinutes, results: _*)
}

编辑：正如您所注意到的，隐式评估按名称参数，并且不会继续下去评估延迟。为什么不直接使用“未来”这个词呢？我个人认为这使代码更具可读性。

import actors.Futures
import actors.Futures.future
import actors.Future

……

def test2 {
  val results = forkAll(
    future {
      Thread.sleep(500)
      println("inside fx1 ",+System.currentTimeMillis)
      true
    },
    future {
      Thread.sleep(1000)
      println("inside fx2 ",+System.currentTimeMillis)
      "stringResult"
    },
    future {
      Thread.sleep(1500)
      println("inside fx3 ",+System.currentTimeMillis)
      1
    },
    future {
      Thread.sleep(2000)
      println("inside fx4 ",+System.currentTimeMillis)
      1.023
    })

  println(results)
  true
}

​

def forkAll(futures: Future[Any]*): List[Any] = {
  println("forkAll")
  Futures.awaitAll(tenMinutes, futures: _*)
}

Answer 2

由于以下错误，您不能使用按名称调用类型作为重复参数（也称为可变参数）： https://lampsvn.epfl.ch/trac/scala/ticket/237

请参阅此处最近的讨论：http://old.nabble.com/Lazy-varargs-td27169264.html

Answer 3

任何感兴趣的人的示例实现，

对于带有几个函数参数的用例，可能只是定义具有不同数量参数的 fork 函数列表（遵循元组定义中的想法）
并享受类型安全返回类型

object ForkTest4 extends LazyParameters {
  def main(args: Array[String]) {
    test4
  }

  def test4 {
    println("Begin test 4")
    //Without the explicit call to type conversion would cause early evaluation
    //    val result4 = forkAll({
    val result4 = forkAll(l {
      Thread.sleep(500)
      println("inside fx4 ", +System.currentTimeMillis)
      true
    })
    println(result4)
    true
  }

  val tenMinutes = 1000 * 60 * 10

  def forkAll(fx1: (() => Any)*): List[Any] = {

    val results = fx1.toList.map {

      fx: (() => Any) => {

        val result = Futures.future(fx.apply())

        result
      }
    }
    Futures.awaitAll(tenMinutes, results: _*)
  }

  def l[T](t: => T): (() => T) = () => t

  implicit def implicitLazy[T](t: => T): (() => T) = () => t
}