在关闭 multiples 时的竞态条件

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的问题还是使用方式的问题。不应该期望一个通道总是生成take。消费者可以通过alts或其他机制来防护“延迟tap”，也可以在“生产”方面通过策略来执行无延迟tap。

（链接：~richhickey）你能发表意见吗？

Answer 4

评论者：jreeves

当前“tap”函数有一个显式的“close?”标志，如果在被连接的通道关闭时，保证通道未被关闭，那么这个参数可能不应该存在。另外，如果我们取消自动关闭tap，是否应该从“sub”中也移除“close?”参数？

Answer 5

评论者：gshayban

不仅仅是尊重标志。与关闭行为相关，通道可以在mult过程愉快地将值分配到另一组通道（如ABA问题）的同时，无需接收任何东西进行tap和untap。也可以使关闭后分配到最后引用的一组通道的tap成为错误。这与熟悉的永久nil接收不同，但是mult已经与简单通道有所不同。

Answer 6

_由：stuart.sierra_发表的评论

我之前正在开发一个依赖于{{mult}}和{{pipeline}}的默认行为来自动关闭下游通道的系统。但有时最初的“输入”通道关闭得非常快，而通道图仍处于构建中。结果，一些输出通道被留下未关闭，一些go-loop持续运行。

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

我预期的行为是，当在具有{{close?}}参数为true({{默认设置}})的{{mult}}上调用{{tap}}时，如果mult的输入通道已经关闭，则立即关闭传递给{{tap}}的通道。

Answer 7

参考：https://clojure.atlassian.net/browse/ASYNC-64（由alexisport提交）