关闭多重操作时的竞争条件

Question

当在源通道关闭的同时按压多重操作时，按压的通道可能不会被关闭。


(require '[clojure.core.async :refer (chan mult tap close!)])
(let [s (chan)
      m (mult s)
      c (chan)]
  (tap m c)
  (close! s)
  (impl/closed? c))


上述代码有时会返回 true，有时会返回 false。

*原因：* 这是由以下 {{mult}} 函数中的代码引起的


(if (nil? val)
  (doseq [[c close?] @cs]
    (when close? (close! c)))


在 cs 解引用之后的任何按压通道都不会被关闭。

*解决办法：* 解决这个问题的一个可能方法是始终关闭连接到已关闭源的多重操作通道。即


(let [s (chan)
      m (mult s)
      c (chan)]
  (close! s)
  (tap m c))  ;; 总是关闭 c


可以通过向 cs 原子添加标志来表示多重操作是打开还是关闭。如果是关闭的，任何按压的通道将自动关闭。

Answer 1

评论者：jreeves

为了参考，以下是我在多操作中使用的自定义修复

`
(defn mult [ch]
(let [state (atom [true {}])

    m (reify
        Mux
        (muxch* [_] ch)
        Mult
        (tap* [_ ch close?]
          (let [add-ch    (fn [[o? cs]] [o? (if o? (assoc cs ch close?) cs)])
                [open? _] (swap! state add-ch)]
            (when-not open? (close! ch))
            nil))
        (untap* [_ ch]
          (swap! state (fn [[open? cs]] [open? (dissoc cs ch)]))
          nil)
        (untap-all* [_]
          (swap! state (fn [[open? _]] [open? {}]))))
    dchan (chan 1)
    dctr (atom nil)
    done (fn [_] (when (zero? (swap! dctr dec))
                   (put! dchan true)))]
(go-loop []
  (let [val (<! ch)]
    (if (nil? val)
      (let [[_ cs] (swap! state (fn [[_ cs]] [false cs]))]
        (doseq [[c close?] cs]
          (when close? (close! c))))
      (let [chs (keys (second @state))]
        (reset! dctr (count chs))
        (doseq [c chs]
          (when-not (put! c val done)
            (swap! dctr dec)
            (untap* m c)))
        (when (seq chs)
          (<! dchan))
        (recur)))))
m))

`

Answer 2

评论者：dnolen

这是否在master中修复了？谢谢。

Answer 3

评论者：gshayban

我理解了场景，但老实说，我不确定这确实是 mult 或使用中的 bug。一个通道不应该总是期望产生一个 take。消费者可以通过 alt 或其他机制来防范“晚期触摸”，同时你也可以通过“生产”一端的策略强制执行无晚期触摸。

(链接：~richhickey)你能发表意见吗？

Answer 4

评论者：jreeves

“touch”函数目前有一个显式的“close？”标志，如果一个触摸通道在源通道关闭时不能保证关闭，那么这个参数可能不应该存在。另外，如果去掉了自动关闭触摸，我们应该也去掉“sub”上的“close？”参数吗？

Answer 5

评论者：gshayban

这不仅仅是尊重标志。与 close 行为相关，通道可以在接收任何内容的情况下触摸和撤销触摸，而多路复用过程则高兴地分配一个值给另一组通道（比如 ABA 问题）。也可以让在向最后 deref 集合的通道发送 close 之后再触摸成为错误。这不同于熟悉的双重空接收，但是多路复用已经与简单通道有所不同。

Answer 6

评论者：stuart.sierra

我最近正在做一个系统，该系统依赖于默认行为为 {{mult}} 和 {{pipeline}} 来自动关闭下游通道。但有时初始的“输入”通道关闭非常快，而通道图仍然在构建中。结果，一些输出通道被打开，一些 go-loops 继续运行。

我的修复方法是在处理之前尽早创建触摸，但这也让我思考了默认行为应该是什么。

我预期的行为是当在 {{mult}} 上调用 {{tap}}，并且有 {{close?}} 参数为 true（默认值），并且多路的输入通道已经关闭时，那么传递给 {{tap}} 的通道将立即关闭。

Answer 7

参考： https://clojure.atlassian.net/browse/ASYNC-64（由 alex+import 报告）