clojure - 了解 clojure 传感器的陷阱

Question

clojure 引用 contains以下关于换能器的评论，似乎在说一些关于书写和使用换能器的安全性的重要内容:

If you have a new context for applying transducers, there are a few general rules to be aware of:

• If a step function returns a reduced value, the transducible process must not supply any more inputs to the step function. The reduced value must be unwrapped with deref before completion.

• A completing process must call the completion operation on the final accumulated value exactly once.

• A transducing process must encapsulate references to the function returned by invoking a transducer - these may be stateful and unsafe for use across threads.

您能否通过一些示例来解释每个案例的含义？另外，在这种情况下，“上下文”指的是什么？

谢谢!

Answer 1

如果阶跃函数返回减小的值，则可转换过程不得向阶跃函数提供更多输入。减少的值必须在完成之前用 deref 解包。

这种情况的一个例子是 take-while传感器:

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

如您所见，它可以返回 reduced value 这意味着没有必要(实际上这将是一个错误)为这种阶跃函数提供更多的输入——我们已经知道不能再产生更多的值了。

例如在处理 (1 1 3 5 6 8 7) 时输入集合 odd?一旦我们达到值 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 数组列表中)，直到谓词函数返回不同的结果(例如，在一系列奇数之后，它会接收偶数，它将返回一串累积的奇数)。问题是如果我们到达源数据的末尾 - 将没有机会观察谓词结果值的变化，并且不会返回累积值。因此，transducible 过程必须调用 step 函数(arity 1)的完成操作，以便它可以返回其累积结果(在我们的例子中是 (6 8))。

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

当通过传递源数据和转换器实例来执行可转换过程时，它将首先调用转换器函数以生成步进函数。 The transducer is a function of the following shape :

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

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

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

也无法保证 step 函数的实现是否是线程安全的。

在这种情况下，“上下文”指的是什么？

“应用传感器的新环境”意味着实现一种新型的可转换过程。 Clojure 提供处理集合的可转导过程(例如 into 、 sequence )。 core.async 库 chan函数(其参数之一)接受转换器实例作为参数，该实例通过将转换器应用于消耗值来产生异步可转换过程，从而产生值(可以从 channel 消耗)。

例如，您可以创建一个可转换的进程来处理在套接字上接收到的数据，或者您自己的 observable 实现。

他们可以使用转换器来转换数据，因为当数据来自何处(套接字、流、集合、事件源等)时，转换器是不可知的——它只是一个用单个元素调用的函数。

他们也不关心(也不知道)应该对他们生成的结果做什么(例如，它是否应该附加到结果序列(例如 conj )？是否应该通过网络发送？插入到数据库？) - 它是通过使用由 step 函数(上面的 rf 参数)捕获的减少函数来抽象的。

因此，不要创建仅使用 conj 的阶跃函数或将元素保存到 db，我们传递一个函数，该函数具有该操作的特定实现。您的可转换过程定义了该操作是什么。