在同步多线程memoization的功能参数 [英] Synchronizing on function parameter for multithreaded memoization

问题描述

我的核心问题是：我怎么能实现同步的方法创建的对象的组合和方法参数

My core question is: how can I implement synchronization in a method on the combination of the object instance and the method parameter?

下面是我的情况的细节。我用下面的code实现记忆化，改编自这个答案：

Here are the details of my situation. I'm using the following code to implement memoization, adapted from this answer:

/**
 * Memoizes a unary function
 * @param f the function to memoize
 * @tparam T the argument type
 * @tparam R the result type
 */
class Memoized[-T, +R](f: T => R) extends (T => R) {

  import scala.collection.mutable

  private[this] val cache = mutable.Map.empty[T, R]

  def apply(x: T): R = cache.getOrElse(x, {
    val y = f(x)
    cache += ((x, y))
    y
  })
}

在我的项目，我memoizing 未来 s到重复数据删除异步API调用。使用时，此工作的罚款为...产量来映射在产生期货，与标准创建的 ExcecutionContext ，但当我升级到斯卡拉异步这些期货更好的操控性。但是，我意识到，多线程该库使用允许多个线程进入适用，击败记忆化，因为异步块所有并联执行，进入缓存之前，否则容易的thunk 可以用新的未来更新。

In my project, I'm memoizing Futures to deduplicate asynchronous API calls. This worked fine when using for...yield to map over the resulting futures, created with the standard ExcecutionContext, but when I upgraded to Scala Async for nicer handling of these futures. However, I realized that the multithreading that library uses allowed multiple threads to enter apply, defeating memoization, because the async blocks all executed in parallel, entering the "orElse" thunk before cache could be updated with a new Future.

要解决这个问题，我把主要的应用功能在 this.synchronized 块：

To work around this, I put the main apply function in a this.synchronized block:

def apply(x: T): R = this.synchronized {
  cache.getOrElse(x, {
    val y = f(x)
    cache += ((x, y))
    y
  })
}

这恢复了memoized行为。其缺点是，这将阻止具有不同PARAMS呼叫，至少直到创建预订。我不知道是否有建立更细粒度的同步上的 Memoized 实例的组合和值 X 参数适用。通过这种方式，只要求将重复数据删除将被阻止。

This restored the memoized behavior. The drawback is that this will block calls with different params, at least until the Future is created. I'm wondering if there is a way to set up finer grained synchronization on the combination of the Memoized instance and the value of the x parameter to apply. That way, only calls that would be deduplicated will be blocked.

作为一个方面说明，我不知道这是真正的性能至关重要，因为synchronized块会释放一次未来创建并返回（我想？） 。但是，如果有这个是我没有想到的任何问题，我也想知道。

As a side note, I'm not sure this is truly performance critical, because the synchronized block will release once the Future is created and returned (I think?). But if there are any concerns with this that I'm not thinking of, I would also like to know.

推荐答案

阿卡演员与期货相结合提供了一个强大的方式来包装了可变状态而不阻塞。下面是如何使用记忆化的演员一个简单的例子：

Akka actors combined with futures provide a powerful way to wrap over mutable state without blocking. Here is a simple example of how to use an Actor for memoization:

import akka.actor._
import akka.util.Timeout
import akka.pattern.ask
import scala.concurrent._
import scala.concurrent.duration._

class Memoize(system: ActorSystem) {
  class CacheActor(f: Any => Future[Any]) extends Actor {
    private[this] val cache = scala.collection.mutable.Map.empty[Any, Future[Any]]

    def receive = {
      case x => sender ! cache.getOrElseUpdate(x, f(x))
    }
  }

  def apply[K, V](f: K => Future[V]): K => Future[V] = {
    val fCast = f.asInstanceOf[Any => Future[Any]]
    val actorRef = system.actorOf(Props(new CacheActor(fCast)))
    implicit val timeout = Timeout(5.seconds)
    import system.dispatcher
    x => actorRef.ask(x).asInstanceOf[Future[Future[V]]].flatMap(identity)
  }
}

我们可以用它喜欢的：

val system = ActorSystem()
val memoize = new Memoize(system)
val f = memoize { x: Int =>
  println("Computing for " + x)
  scala.concurrent.Future.successful {
    Thread.sleep(1000)
    x + 1
  }
}
import system.dispatcher
f(5).foreach(println)
f(5).foreach(println)

和计算5将只打印一个单一的时间，但6将打印两次。

And "Computing for 5" will only print a single time, but "6" will print twice.

有一些可怕看 asInstanceOf 呼叫，但它是完全类型安全的。

There are some scary looking asInstanceOf calls, but it is perfectly type-safe.

这篇关于在同步多线程memoization的功能参数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！