使用 core.async 和非阻塞通道读取的函数如何进行记忆化？

Question

我想在一个使用core.async和<!的函数中使用memoize。

(defn foo [x]
  (go
    (<! (timeout 2000))
    (* 2 x)))

(在现实生活中，缓存服务器调用的结果可能非常有用)

我通过编写memoize的core.async版本（几乎与memoize相同的代码）来实现：

(defn memoize-async [f]
  (let [mem (atom {})]
    (fn [& args]
      (go
        (if-let [e (find @mem args)]
          (val e)
         (let [ret (<! (apply f args))]; this line differs from memoize [ret (apply f args)]
            (swap! mem assoc args ret)
            ret))))))

使用示例：

(def foo-memo (memoize-async foo))
(go (println (<! (foo-memo 3)))); delay because of (<! (timeout 2000))

(go (println (<! (foo-memo 3)))); subsequent calls are memoized => no delay

我想知道是否有更简单的方法来达到相同的结果。

**备注：我需要适用于<！>的解决方案。对于，请参阅此问题：如何备忘使用core.async和阻止通道读取的函数？**

- viebel

Answer 1

您可以使用内置的memoize函数来实现此功能。首先定义一个从通道读取并返回值的方法：

 (defn wait-for [ch]
      (<!! ch))

请注意，我们将使用<!!而不是<!，因为我们希望在所有情况下该函数都会阻塞，直到通道中有数据。当在go块内部使用时，<!仅在表单中展示此行为。

然后，您可以通过将此函数与foo组合来构建您的记忆化函数，如下所示：

(def foo-memo (memoize (comp wait-for foo)))

foo返回一个通道，因此wait-for将阻塞直到该通道有一个值（即直到foo内的操作完成）。

foo-memo可以像上面的示例一样使用，除了您不需要调用<!，因为wait-for会为您阻塞：

(go (println (foo-memo 3))

你也可以在go块之外调用它，它会表现得像你期望的那样（即阻塞调用线程，直到foo返回）。

- Jesse Rosalia

---

似乎<!!是通过在clojure端使用take!和一个promise来实现的；当take!回调触发promise时，函数才能完成。看起来有一个开源库可以将JS promises暴露给clojurescript。你可能可以使用该库来实现<!!。如果你在那之前没有解决，我今晚会尝试做些东西。 - Jesse Rosalia

尝试了将近两个小时后，我认为我无法比您的原始解决方案做得更好。我一直遇到的问题是 <! 必须直接从 go 块中调用（而不是从 go 块中调用的函数中调用），这限制了您定义一个 memo 函数来保存您想要保存的值的能力（即您不希望您的 memo 保存通道）。我之前提到的 promise 库使用回调和事件，对于这种情况没有用处。我的解决方案适用于 clojure，但我看不到在 cljs 中使其工作的方法。 - Jesse Rosalia

谢谢。我很感激你的努力。你认为我可以建议core.async团队添加memoize-async吗？ - viebel

谢谢，你绝对可以向core.async建议这种方法。我不确定具体的流程或者他们是否会接受，但是值得一试。 - Jesse Rosalia

使用像memoize-async这样的东西的一个缺点是，异步操作运行时间越长，冗余获取的机会就越高 - 没有存储任何内容来指示正在处理特定值的解析。之前曾经讨论过promise-chan - 你可以天真地记忆类似的东西，并且正确的事情会发生。如果您不想走promise-chan路线，还可以通过状态管理和通道多路复用变得疯狂。 - moe

Answer 2

这比我预期的要棘手一些。您的解决方案是不正确的，因为当您再次使用相同参数调用存储函数时，比第一次运行完其go块更早，您将再次触发它并且得到一个错误。在使用core.async处理列表时，这种情况经常发生。

下面的代码使用核心异步的发布/订阅来解决此问题（仅在CLJS中测试）。

(def lookup-sentinel  #?(:clj ::not-found :cljs (js-obj))
(def pending-sentinel #?(:clj ::pending   :cljs (js-obj))

(defn memoize-async
  [f]
  (let [>in (chan)
        pending (pub >in :args)
        mem (atom {})]
    (letfn
        [(memoized [& args]
           (go
             (let [v (get @mem args lookup-sentinel)]
               (condp identical? v
                 lookup-sentinel
                 (do
                   (swap! mem assoc args pending-sentinel)
                   (go
                     (let [ret (<! (apply f args))]
                       (swap! mem assoc args ret)
                       (put! >in {:args args :ret ret})))
                   (<! (apply memoized args)))
                 pending-sentinel
                 (let [<out (chan 1)]
                   (sub pending args <out)
                   (:ret (<! <out)))
                 v))))]
        memoized)))

注意：它可能会泄漏内存，订阅和 <out 通道未关闭

- skrat

Answer 3

我在我的一个项目中使用了这个函数来缓存HTTP调用。该函数将结果缓存一段时间，并使用障碍来防止在缓存“冷却”（由于go块内的上下文切换）时多次执行该函数。

(defn memoize-af-until
  [af ms clock]
  (let [barrier (async/chan 1)
        last-return (volatile! nil)
        last-return-ms (volatile! nil)]
    (fn [& args]
      (async/go
        (>! barrier :token)
        (let [now-ms (.now clock)]
          (when (or (not @last-return-ms) (< @last-return-ms (- now-ms ms)))
            (vreset! last-return (<! (apply af args)))
            (vreset! last-return-ms now-ms))
          (<! barrier)
          @last-return)))))

你可以通过将缓存时间设为0来测试它是否正常工作，并观察两个函数调用需要大约10秒钟。如果没有屏障，两个调用会同时完成：

(def memo (memoize-af-until #(async/timeout 5000) 0 js/Date))
(async/take! (memo) #(println "[:a] Finished"))
(async/take! (memo) #(println "[:b] Finished"))

- Milan Munzar