Clojure：并行计算产生的数据访问速度似乎要慢得多

Question

我有一个函数可以同时计算文本文件中某些特征的频率并整理数据。该函数的输出是存储在持久映射中的数千个频率分布。举个简单的例子：

{"dogs" {"great dane" 2, "poodle" 4}, "cats" {"siamese" 1 "tom" 3}}

以及产生这个的代码：

(defn do-the-thing-1 [lines species_list]
  ;; we know the full list of species beforehand so to avoid thread contention
  ;; for a single resource, make an atom for each species
  (let [resultdump      (reduce #(assoc %1 %2 (atom {})) {} species_list)
        line-processor  (fn [line]
                          (fn [] ; return a function that will do the work when invoked
                            (doseq [[species breed] (extract-pairs line)]
                              (swap! ; increase the count for this species-breed pair
                                (resultdump species)
                                update-in [breed] #(+ 1 (or % 0))))))
        pool            (Executors/newFixedThreadPool 4)]
    ;; queue up the tasks
    (doseq [future (.invokeAll pool (map line-processor lines))]
      (.get future))
    (.shutdown pool)
    (deref-vals result)))

(defn deref-vals [species_map]
  (into {} (for [[species fdist] species_map] [species @fdist]))

这很好用。问题是我需要先将它们转换为概率分布，然后才能使用它们。例如

{"dogs" {"great dane" 1/3, "poodle" 2/3}, "cats" {"siamese" 1/4, "tom" 3/4}}

这是执行此操作的函数：

(defn freq->prob 
  "Converts a frequency distribution into a probability distribution"
  [fdist]
  (let [sum (apply + (vals fdist))]
    (persistent!
      (reduce
        (fn [dist [key val]] (assoc! dist key (/ val sum)))
        (transient fdist)
        (seq fdist)))))

在处理流水线中的下一步消耗分布时即时执行此转换可提供合理的速度，但也有相当数量的冗余转换，因为某些分布被多次使用。当我修改函数以在返回结果之前并行执行转换时，后期处理的速度会急剧下降。

这是修改后的函数：

(defn do-the-thing-2 [lines species_list]
  ;; we know the full list of species beforehand so to avoid thread contention
  ;; for a single resource, make an atom for each species
  (let [resultdump      (reduce #(assoc %1 %2 (atom {})) {} species_list)
        line-processor  (fn [line]
                          (fn [] ; return a function that will do the work when invoked
                            (doseq [[species breed] (extract-pairs line)]
                              (swap! ; increase the count for this species-breed pair
                                (resultdump species)
                                update-in [breed] #(+ 1 (or % 0))))))
        pool            (Executors/newFixedThreadPool 4)]
    ;; queue up the tasks
    (doseq [future (.invokeAll pool (map line-processor lines))]
      (.get future))

    ;; this is the only bit that has been added
    (doseq [future (.invokeAll pool (map
                                      (fn [fdist_atom]
                                        #(reset! fdist_atom (freq->prob @fdist_atom)))
                                      (vals resultdump)))]
      (.get future))

    (.shutdown pool)
    (deref-vals result)))

是的，这使得之后的一切都比在每次访问结果地图时简单地调用 freq->prob 时慢大约 10 倍，尽管返回的数据是相同的。任何人都可以就为什么会这样或我能做些什么提出原因吗？

编辑：我现在怀疑它与 Clojure 的分数有关。如果我修改 freq->prob 函数以创建浮点数或双精度数而不是分数，则在预先计算概率分布而不是动态生成概率分布时性能会有所提高。是不是在原子中产生的分数比在原子外产生的分数运行得慢？我刚刚进行了一些简单的测试，表明情况并非如此，所以这里肯定发生了一些奇怪的事情。

Answer 1

我不是 100% 确定我遵循了你的逻辑，但你的地图功能在这里：

(map
    (fn [fdist_atom]
        #(reset! fdist_atom (freq->prob @fdist_atom)))
    (vals resultdump))

看起来不对。如果要根据旧值更新原子，swap! 比 reset! 更适合应用于原子的取消引用值的函数。这样看起来会更好：

(map
    (fn [fdist_atom] (swap! fdist_atom freq->prob))
    (vals resultdump))

Answer 2

关于转换概率分布的问题。

如果你像这样重写'freq-prob'：

(defn cnv-freq [m]
 (let [t (apply + (vals m))]
  (into {} (map (fn [[k v]] [k (/ v t)]) m))))
(defn freq-prob [m]
 (into {} (pmap (fn [[k v]] [k (cnv-freq v)]) m)))

您可以通过将“pmap”更改为“map”来启用/禁用并行执行。