4

我正在尝试并行计算文件夹大小。也许这是天真的方法。我所做的是，我将每个分支节点（目录）的计算交给代理。所有叶节点的文件大小都添加到 my-size。好吧，它不起作用。:)

“扫描”工作正常，连续。'pscan' 只打印第一级的文件。

(def agents (atom []))
(def my-size (atom 0))
(def root-dir (clojure.java.io/file "/"))

(defn scan [listing]
  (doseq [f listing]
    (if (.isDirectory f)
      (scan (.listFiles f))
      (swap! my-size #(+ % (.length f))))))

(defn pscan [listing]
  (doseq [f listing]
    (if (.isDirectory f)
      (let [a (agent (.listFiles f))]
        (do (swap! agents #(conj % a))
            (send-off a pscan)
            (println (.getName f))))
    (swap! my-size #(+ %  (.length f))))))

你有什么想法，我做错了什么？

谢谢。

4

2 回答 2

15

无需使用原子保持状态。纯函数式：

(defn psize [f]
  (if (.isDirectory f)
    (apply + (pmap psize (.listFiles f)))
    (.length f)))
于 2011-05-03T16:00:12.537 回答
3

那么并行计算文件大小应该如此简单吗？

它不是 ：）

我试图更好地解决这个问题。我意识到我正在阻塞 I/O 操作，所以 pmap 不能完成这项工作。我在想也许给代理提供大量的目录（分支）来独立处理它是有意义的。看起来确实如此:) 好吧，我还没有对它进行基准测试。

它可以工作，但是在类 UNIX 系统上的符号链接可能存在一些问题。

(def user-dir (clojure.java.io/file "/home/janko/projects/"))
(def root-dir (clojure.java.io/file "/"))
(def run? (atom true))
(def *max-queue-length* 1024)
(def *max-wait-time* 1000)    ;; wait max 1 second then process anything left
(def *chunk-size* 64)
(def queue (java.util.concurrent.LinkedBlockingQueue. *max-queue-length* ))
(def agents (atom []))
(def size-total (atom 0))
(def a (agent []))

(defn branch-producer [node]
  (if @run?
    (doseq [f node]
      (when (.isDirectory f)
    (do (.put queue f)
        (branch-producer (.listFiles f)))))))

(defn producer [node]
  (future
    (branch-producer node)))

(defn node-consumer [node]
  (if (.isFile node)
    (.length node)
    0))

(defn chunk-length []
  (min (.size queue) *chunk-size*))

(defn compute-sizes [a]
  (doseq [i (map (fn [f] (.listFiles f)) a)]
    (swap! size-total #(+ % (apply + (map node-consumer i))))))

(defn consumer []
  (future
    (while @run?
      (when-let [size (if (zero? (chunk-length))
            false
            (chunk-length))] ;appropriate size of work
      (binding [a (agent [])]                    
        (dotimes [_ size]         ;give us all directories to process
          (when-let [item (.poll queue)]
            (set! a (agent (conj @a item)))))
        (swap! agents #(conj % a))
        (send-off a compute-sizes))
      (Thread/sleep *max-wait-time*)))))

您可以通过键入来启动它

    (producer (list user-dir))
    (consumer)

对于结果类型

    @size-total

你可以阻止它（有正在运行的期货 - 如果我错了，请纠正我）

    (swap! run? not)

如果您发现任何错误/错误，欢迎您分享您的想法！

于 2011-05-04T11:08:18.063 回答