9

我在 Clojure 中编写了一些 core.async 代码，当我运行它时，它消耗了所有可用内存并因错误而失败。似乎mapcat在 core.async 管道中使用会破坏背压。（由于超出此问题范围的原因，这是不幸的。）

下面是一些代码，通过计算ing 转换器:x的进出数来演示该问题：mapcat

(ns mapcat.core
  (:require [clojure.core.async :as async]))

(defn test-backpressure [n length]
  (let [message (repeat length :x)
        input (async/chan)
        transform (async/chan 1 (mapcat seq))
        output (async/chan)
        sent (atom 0)]
    (async/pipe input transform)
    (async/pipe transform output)
    (async/go
      (dotimes [_ n]
        (async/>! input message)
        (swap! sent inc))
      (async/close! input))
    (async/go-loop [x 0]
      (when (= 0 (mod x (/ (* n length) 10)))
        (println "in:" (* @sent length) "out:" x))
      (when-let [_ (async/<! output)]
        (recur (inc x))))))

=> (test-backpressure 1000 10)
in: 10 out: 0
in: 2680 out: 1000
in: 7410 out: 2000
in: 10000 out: 3000 ; Where are the other 7000 characters?
in: 10000 out: 4000
in: 10000 out: 5000
in: 10000 out: 6000
in: 10000 out: 7000
in: 10000 out: 8000
in: 10000 out: 9000
in: 10000 out: 10000

生产者远远领先于消费者。

看来我不是第一个发现这一点的人。但是这里给出的解释似乎并没有完全涵盖它。（尽管它确实提供了一个足够的解决方法。）从概念上讲，我希望生产者领先，但只是通道中可能缓冲的少数消息的长度。

我的问题是，所有其他消息在哪里？到第四行输出 7000 :xs 下落不明。

4

1 回答 1

2

更新 2020-01-14：内存泄漏现已修复。

对“内存泄漏在哪里？”这个问题有两种可能的解释。

首先，数据保存在哪里？答案似乎在扩展变换下游的通道缓冲区中。

默认情况下，通道使用FixedBuffer( clojure.core.async.impl.buffers/FixedBuffer ) 可以判断它是否已满但不反对过满。

其次，哪段代码导致缓冲区溢出？这（如果我错了，请纠正我）似乎在（clojure.core.async.impl.channels/ManyToManyChannel）的take!方法中，其中对缓冲区的第一次调用发生在任何调用发生之前。ManyToManyChanneladd!full?

似乎take!假设它可以为它删除的每个项目添加至少一个项目到缓冲区。在长时间运行的扩展传感器的情况下，这样mapcat的假设并不总是一个安全的假设。

通过将此行更改(when (and (.hasNext iter) (not (impl/full? buf)))为 core.async 的本地副本，我可以使问题中的代码按预期运行。（注意，我对 core.async 的理解不足以保证这对于您的用例来说是一个强大的解决方案。）

更新 2016-09-17：现在有一个问题：http ://dev.clojure.org/jira/browse/ASYNC-178

更新 2020-01-14：现在已修复：https ://clojure.atlassian.net/browse/ASYNC-210 （尽管较早的票已关闭为“已拒绝”）

于 2016-06-27T15:56:02.937 回答