理解Clojure Transducers的陷阱

Question

Clojure参考文档中的转换器(transducers)部分包含以下有关转换器编写和使用安全性的重要说明：

如果您有一个新的上下文需要应用转换器，则需要注意以下几个通用规则：

• 如果步骤函数返回一个reduced值，则转换器过程不得再向该步骤函数提供任何输入。在完成之前，必须使用deref解包(reduced值)。

• 完成处理过程必须恰好调用一次最终累积值的completion操作。

• 转换处理过程必须封装对调用转换器所返回的函数的引用-这些可能是具有状态且跨线程使用时不安全的。

您能否解释一下每个情况的意思，可能还需要一些示例？此外，“上下文”在这里指什么？

谢谢！

- matanster

Answer 1

如果步骤函数返回了一个缩减值，可转换过程不得再提供任何输入给步骤函数。在完成之前必须使用deref解开缩减值。

这种情况的一个示例是take-while转换器：

(fn [rf]
  (fn
    ([] (rf))
    ([result] (rf result))
    ([result input]
      (if (pred input)
        (rf result input)
        (reduced result)))))

正如您所看到的，它可以返回一个“reduced”值，这意味着向此步骤函数提供更多输入是没有意义的（实际上会导致错误）- 我们已经知道不会产生更多的值。

例如，在使用“odd？”谓词处理“（1 1 3 5 6 8 7）”输入集合时，一旦我们达到值“6”，由“take-while odd？”转换器创建的步骤函数将不再返回任何值。

完成过程必须在最终累积值上恰好调用一次完成操作。

这是转换器返回有状态步骤函数的场景。一个很好的例子是“partition-by”转换器。例如，当“（partition-by odd？）”被用于可转换过程来处理“（1 3 2 4 5 2）”时，它将产生“（（1 3）（2 4）（5）（6 8））”。

(fn [rf]
  (let [a (java.util.ArrayList.)
        pv (volatile! ::none)]
    (fn
      ([] (rf))
      ([result]
         (let [result (if (.isEmpty a)
                        result
                        (let [v (vec (.toArray a))]
                          ;;clear first!
                          (.clear a)
                          (unreduced (rf result v))))]
           (rf result)))
      ([result input]
        (let [pval @pv
              val (f input)]
          (vreset! pv val)
          (if (or (identical? pval ::none)
                  (= val pval))
            (do
              (.add a input)
              result)
            (let [v (vec (.toArray a))]
              (.clear a)
              (let [ret (rf result v)]
                (when-not (reduced? ret)
                  (.add a input))
                ret))))))))

如果您查看实现，您会注意到步骤函数不会返回其累积值（存储在a数组列表中），直到谓词函数返回不同的结果（例如，在奇数序列之后，它将接收一个偶数，它将返回一系列累积的奇数）。问题是如果我们到达源数据的末尾-就没有机会观察谓词结果值的变化，并且累积值将不会被返回。因此，可转换过程必须调用步骤函数（arity 1）的完成操作，以便它可以返回其累积结果（在我们的情况下为(6 8)）。

可转换过程必须封装对通过调用变换器返回的函数的引用-这些可能是有状态的并且不安全用于跨线程使用。

当通过传递源数据和变换器实例来执行可转换过程时，它将首先调用变换器函数以生成步骤函数。变换器是以下形状的函数：

(fn [xf]
  (fn ([] ...)
      ([result] ...)
      ([result input] ...)))

因此，可转换过程将调用此顶级函数（接受 xf - 一个减少函数）以获取用于处理数据元素的实际步骤函数。问题在于，可转换过程必须保留对该步骤函数的引用，并使用相同的实例来处理特定数据源（例如，由 partition-by 转换器生成的步骤函数实例必须用于处理整个输入序列，因为它保持其内部状态，如上所示）。对于处理单个数据源使用不同的实例会产生不正确的结果。

同样，由于相同的原因，可转换过程不能重复使用步骤函数实例来处理多个数据源 - 步骤函数实例可能具有状态并保持用于处理特定数据源的内部状态。当步骤函数用于处理另一个数据源时，该状态将被破坏。

此外，并没有保证步骤函数实现是否线程安全。

在这种情况下，“上下文”是什么意思？

"应用转换器的新上下文"意味着实现一种新类型的可转换过程。Clojure提供了处理集合的可转换过程（例如into，sequence）。core.async库的chan函数（其中之一的元数）接受一个转换器实例作为参数，该实例通过将转换器应用于消耗的值来生成异步可转换过程以产生值（可以从通道中消耗），这些值可以被消费。

例如，您可以创建一个用于处理套接字接收到的数据的可转换过程，或者自己实现可观察对象。

它们可以使用转换器来转换数据，因为转换器不关心数据来自哪里（套接字、流、集合、事件源等），它只是调用单个元素的函数。

它们也不关心（也不知道）所生成的结果应该做什么（例如，应该将其附加到结果序列（例如conj）中吗？应该发送到网络吗？插入到数据库中？） - 这是通过使用由步骤函数（rf参数）捕获的减少函数进行抽象。

"

因此，我们不是创建一个仅使用conj或将元素保存到数据库的步骤函数，而是传递一个具有特定实现该操作的函数。您的可传输过程定义了该操作是什么。

- Piotrek Bzdyl

---

值得一提的是，如果我理解正确的话，可转换过程负责数据源的任何初始化和结束处理。 - matanster