在Clojure中,我如何使用换能器执行高频率的"频率"？

Question

(问题来源:Fernando Abrao.)

我听说Clojure中传感器的性能优势,但我不知道如何使用它们.

假设我有一个qos/device-qos-range返回映射序列的函数,其中一些包含小数:samplevalue,如下所示:

[
  { :samplevalue 1.3, ... },
  { :othervalue -27.7, ... },
  { :samplevalue 7.5, ... },
  { :samplevalue 1.9, ... },
]

我想看看有多少:samplevalues落入每个整数bin,如下所示:

(frequencies
  (reduce #(if (not (nil? (:samplevalue %2)))
             (conj %1 (.intValue (:samplevalue %2))))
          []
          (qos/device-qos-range origem device qos alvo inicio fim)))

;; => {1 2, 7 1}

如何将其转换为带有传感器的快速版本,以消除中间数据结构(例如返回的数据结构reduce)？可以利用多个内核进行并行处理的代码的加分点.

Answer 1

(答案来源:Renzo Borgatti(@reborg).)

首先,让我们设置一些示例数据,稍后我们将用于性能测试.此向量包含具有相同键的500k映射.值在1/5时间重叠.

(def data 
 (mapv hash-map 
       (repeat :samplevalue) 
       (concat (range 1e5)
               (range 1e5)
               (range 1e5)
               (range 1e5)
               (range 1e5))))

现在让我们用换能器进行转换.请注意,此解决方案不是并行的.我把你缩短.intValue为正义int,这也是同样的事情.此外,:samplevalue从每个地图有条件地提取可以缩短到恰好(keep :samplevalue sequence),这相当于(remove nil? (map :samplevalue sequence)).我们将使用Criterium进行基准测试.

(require '[criterium.core :refer [quick-bench]])
(quick-bench
  (transduce
    (comp
      (keep :samplevalue)
      (map int))
    (completing #(assoc! %1 %2 (inc (get %1 %2 0))) persistent!)
    (transient {})
    data))
;; My execution time mean: 405 ms

请注意,这次我们不会将其frequencies称为外部步骤.相反,我们把它编织到了操作中.就像frequencies那样,我们已经在瞬态哈希映射上完成了操作以获得额外的性能.我们通过使用瞬态hashmap作为种子和completing最终值来调用persistent!它来实现这一点.

我们可以使这个并行.为了获得最佳性能,我们使用可变Java ConcurrentHashMap而不是不可变的Clojure数据结构.

(require '[clojure.core.reducers :as r])
(import '[java.util HashMap Collections Map]
        'java.util.concurrent.atomic.AtomicInteger
        'java.util.concurrent.ConcurrentHashMap)

(quick-bench
  (let [concurrency-level (.availableProcessors (Runtime/getRuntime))
        m (ConcurrentHashMap. (quot (count data) 2) 0.75 concurrency-level)
        combinef (fn ([] m) ([_ _]))  ; just return `m` from the combine step
        rf (fn [^Map m k]
             (let [^AtomicInteger v (or (.get m k) (.putIfAbsent m k (AtomicInteger. 1)))]
               (when v (.incrementAndGet v))
               m))
        reducef ((comp (keep :samplevalue) (map int)) rf)]
    (r/fold combinef reducef data)
    (into {} m)))
;; My execution time mean: 70 ms

这里我们使用fold从clojure.core.reducers库中实现并行性.请注意,在并行环境中,任何使用的传感器都需要无状态.另请注意,a ConcurrentHashMap不支持nil用作键或值; 幸运的是,我们不需要在这里这样做.

输出最后转换为不可变的Clojure哈希映射.您可以删除该步骤,只需使用ConcurrentHashMap实例进行额外的加速 - 在我的机器上,删除该into步骤使整个过程fold大约需要26ms.

编辑2017-11-20:用户@clojure最正确地指出,此答案的早期版本调用quick-bench了let块内部初始化并发哈希映射实例,这意味着基准测试对其所有运行使用相同的实例.我把电话移到quick-bench了let街区外面.它没有显着影响结果.