concurrency - 在执行之前等待另一个函数完成的 Clojure 函数

Question

我需要一个函数，当使用特定输入参数调用时执行提供的函数 g，但仅在另一个提供的函数 f 使用相同的输入参数完成执行之后。还有一个要求，当多次调用同一个参数的函数时，f只在第一次调用时执行一次，其他调用等待完成，然后直接执行g。

编辑:该解决方案在不同线程上并行运行时应该有效，并且还应该有效地使用线程。例如。阻塞应该基于每个输入而不是整个函数。

我第一次尝试的功能如下:

(defn dependent-func
  ([f g]
   (let [mem (atom {})]
     (fn [& args]
       (->> (get (locking mem
                   (swap! mem (fn [latch-map args]
                                (if (contains? latch-map args)
                                  latch-map
                                  (let [new-latch (CountDownLatch. 1)
                                        new-latch-map (assoc latch-map args new-latch)]
                                    (->> (Thread. #(do (apply f args)
                                                       (.countDown new-latch)))
                                         (.start))
                                new-latch-map))) args)) args)
            (.await))
       (apply g args)))))

这似乎满足了我的要求，并且等待 f 是基于每个输入的，所以我对此比较满意。最初我希望只使用交换!做内存更新但不幸的是交换!明确指出交换中的功能!可以多次调用(我在测试中看到过)。因此，我最终不得不在更新时锁定内存，这真的很难看。

我确信一定有一种比我更好地利用 Closure 的并发机制的更简洁的方法来执行此操作，但到目前为止我一直无法找到它。

如有任何建议，我们将不胜感激。

谢谢，

马特。

Answer 1

Clojure 对 future、promise 和 deliver 的组合非常适合启动一个进程并让多个线程等待它完成。

• Future 用于在后台启动一个线程(它可以做更多，尽管在这个例子中我不需要它)

• Promise 用于在准备就绪后立即返回一个包含答案的对象。

• Deliver 用于在准备好后提供 promise 的答案。

我还将等待部分拆分到它自己的函数中，以使代码更易于理解，这样我就可以使用内置的 memoize 函数:

这个问题很好地说明了何时使用 promise 和 deliver 而不是简单的 future。

因为我们要在运行该函数两次不安全的地方使用 memoize，我们需要注意这两个调用不会在 完全 相同的位置进入 memoize时间。所以我们只会在进入 memoize 时锁定，而不是持续时间内存功能。

hello.core> (def lock [])
#'hello.core/lock

每次调用 f 时，此函数将始终返回相同的 future 对象使用一组给定的参数，除了我们需要通过包装它来使 memoize 安全在执行锁定的函数中(您也可以为此使用代理)

hello.core> (def wait-for-function-helper             
              (memoize (fn [f args]
                         (let [answer (promise)]
                           (println "waiting for function " f " with args" args)
                           (future (deliver answer (apply f args)))
                           answer))))

#'hello.core/wait-for-function-helper
hello.core> (defn wait-for-function [& args]
              (locking lock
                (apply wait-for-function-helper args)))
#'hello.core/wait-for-function

现在我们编写使用安全内存的实际依赖函数， future 生产，等待功能功能。

hello.core> (defn dependent-func [f g & args]
              @(wait-for-function f args)
              (apply g args))
#'hello.core/dependent-func

并定义一个慢操作以查看它的实际效果:

hello.core> (defn slow-f-1 [x]
              (println "starting slow-f-1")
              (Thread/sleep 10000)
              (println "finishing slow-f-1")
              (dec x))
#'hello.core/slow-f-1

为了测试它，我们想在完全同时启动两个相同的函数。

hello.core> (do (future
                  (println "first" (dependent-func slow-f-1 inc 4)))
                (future
                  (println "second" (dependent-func slow-f-1 inc 4))))

waiting for function  
#object[clojure.core$future_call$reify__6736 0x40534083 {:status :pending, :val nil}] with args (4)
#object[hello.core$slow_f_1 0x4f9b3396 hello.core$slow_f_1@4f9b3396]
starting slow-f-1
finishing slow-f-1
second
first
5
5

如果我们再次调用它，我们会看到 slow-f-1 只运行了一次:

hello.core> (do (future
                  (println "first" (dependent-func slow-f-1 inc 4)))
                (future
                  (println "second" (dependent-func slow-f-1 inc 4))))

#object[clojure.core$future_call$reify__6736 0x3935ea29 {:status :pending, :val nil}]
first 5
second 5