asynchronous - 当mapcat破坏core.async中的背压时，内存泄漏在哪里？

Question

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

以下是一些代码，它通过将: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

生产者的竞争远远领先于消费者。

看来我不是第一个发现此问题的人。
但是 here给出的解释似乎并没有涵盖它。 (尽管它确实提供了适当的解决方法。)
从概念上讲，我希望生产者处于领先地位，但仅取决于 channel 中可能缓冲的少量消息的长度。

我的问题是，所有其他消息在哪里？到输出的第四行，无法说明 :x 7000。

Answer 1

UPDATE 2020-01-14:现在修复了内存泄漏。

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

首先，数据存放在哪里？答案似乎在扩展转换下游的 channel 缓冲区中。

默认情况下， channel 使用FixedBuffer(clojure.core.async.impl.buffers/FixedBuffer)，可以告诉它是否已满，但不反对过度满。

其次，哪段代码会导致缓冲区过满？这(如果我错了，请纠正我)似乎出现在take!(the ManyToManyChannel method)的clojure.core.async.impl.channels/ManyToManyChannel中，其中在缓冲区中的first call to add! 发生在发生任何calls to full? 之前。

似乎take!假定它可以为它删除的每个项目将至少一个项目添加到缓冲区中。对于长时间运行的扩展传感器(例如mapcat)，这并不总是一个安全的假设。

通过在core.async的本地副本中将this line更改为(when (and (.hasNext iter) (not (impl/full? buf)))，我可以使问题中的代码按预期运行。 (注:我对core.async的理解不足以保证它是您使用案例的可靠解决方案。)

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

2020年1月14日更新:此问题自https://clojure.atlassian.net/browse/ASYNC-210起已修复(尽管之前的凭单已被关闭为“拒绝”)