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函数以创建float或doubles而不是fractions，那么在预计算概率分布时，与动态生成概率分布相比，性能会得到改善。原子中产生的分数会比原子外产生的分数慢吗？我只是做了一些简单的测试，表明情况并非如此，所以这里肯定发生了一些奇怪的事情。

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

关于转换prob分布的问题。

如果你这样重写"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"来启用/禁用并行执行。