代码之家 › 专栏 › 技术社区 › Carcigenicate

为什么在另一个线程中循环处理大量数据会导致GC过度活动,并阻止释放某些数据?

garbage-collection clojure concurrency multithreading

Carcigenicate · 技术社区 · 6 年前

我正在编写代码,它接受由 pmap ,并将它们绘制到 BufferedImage . 三天来,我一直在想,为什么画突然开始冻结,最后停了大约三分之一的路。

我终于把范围缩小到在另一个线程中循环处理大量数据。

这是我想到的最好的MCVE:

(ns mandelbrot-redo.irrelevant.write-image-mcve
  (:import [java.awt.image BufferedImage]
           (java.util.concurrent Executors Executor)))

(defn lazy-producer [width height]
  (for [y (range height)
        x (range width)]
    [x y (+ x y)]))

; This works fine; finishing after about 5 seconds when width=5000
(defn sync-consumer [results width height]
  (time
    (doseq [[i result] (map vector (range) results)]
      (when (zero? (rem i 1e6))
        (println (str (double (/ i (* width height) 0.01)) "%")))

      ((fn boop [x] x) result)))) ; Data gets consumed here

; This gets to ~30%, then begins being interupted by 1-4 second lags
(defn async-consumer [results width height]
  (doto
    (Thread. ^Runnable
             (fn []
               (sync-consumer results width height)
               (println "Done...")))
    (.start)))

(defn -main []
  (let [width 5000
        height (int (* width 2/3))]
    (-> (lazy-producer width height)
        (async-consumer width height))))

什么时候? -main 与 sync-consumer ,几秒钟后结束。与 async-consumer 然而,它达到了25%左右,然后开始缓慢爬行到最后打印的百分比是30%。如果我离开它,我会得到一个OOME。

如果我使用显式 Thread. 异步使用者 ,它会挂起并崩溃。如果我用 future 但是,它完成得很好,就像 同步消费者 .

我得到的唯一提示是,当我在VisualVM中运行这个函数时,我看到了 Long 使用异步版本时:

同步版本显示 长 相比之下,一次大约是45mb。

CPU使用率也大不相同:

有大量的GC峰值,但看起来不像 长 正在处理。

我能够使用 未来 为了这个,但我已经被它的异常吞咽行为咬了那么多次,我很犹豫。

为什么会这样?为什么在一个新线程中运行它会导致GC变得疯狂,而同时又没有释放数字呢?

有人能解释这种行为吗?

3 回复 | 直到 6 年前

Alex Miller 6 年前

sync版本似乎正在处理16M+的结果,但由于本地清除,sync版本不会保留结果序列的头部。这意味着,在运行时,将创建、处理和GC'ed值。

异步的结束了 results 在fn和will中保持头部,将所有16M+值保存在内存中,可能导致GC抖动?

实际上,我无法重现您所描述的内容—同步和异步对我来说所用的时间与上面所述的时间差不多。(Clojure 1.9,Java 1.8)。

Alan Thompson 6 年前

我简化了你的例子,得到了不一致的结果。我怀疑手册 Thread 对象以某种方式被视为(有时)作为守护进程线程,因此JVM有时在完成之前退出。

(def N 5e3)
(def total-count (* N N))
(def report-fact (int (/ total-count 20)))

(defn lazy-producer []
  (for [y (range N)
        x (range N)]
    [x y (+ x y)]))

(defn sync-consumer [results]
  (println "sync-consumer:  start")
  (time
    (doseq [[i result] (map vector (range) results)]
      (when (zero? (rem i report-fact))
        (println (str (Math/round (/ (* 100 i) total-count)) " %")))))
  (println "sync-consumer:  stop"))

 (defn async-consumer [results]
  ; (spyx (count results))
  (spyx (nth results 99))
  (let [thread (Thread. (fn []
                          (println "thread  start")
                          (sync-consumer results)
                          (println "thread  done")
                          (flush)))]
    ; (.setDaemon thread false)
    (.start thread)
    (println "daemon?   " (.isDaemon thread))
    thread))

(dotest
  (println "test - start")
  (let [thread (async-consumer
                 (lazy-producer))]
    (when true
      (println "test - sleeping")
      (Thread/sleep 5000))
 ;  (.join thread)
  )
  (println "test - end"))

结果如下:

----------------------------------
   Clojure 1.9.0    Java 10.0.1
----------------------------------

lein test tst.demo.core
test - start
(nth results 99) => [99 0 99]
daemon?    false
test - sleeping
thread  start
sync-consumer:  start
0 %
5 %
10 %
15 %
20 %
25 %
30 %
35 %
40 %
45 %
50 %
55 %
test - end

Ran 2 tests containing 0 assertions.
0 failures, 0 errors.
60 %
lein test  54.58s user 1.37s system 372% cpu 15.028 total

如果我们取消注释 (.join thread) 线,我们得到一个完整的运行:

~/expr/demo > lein test

----------------------------------
   Clojure 1.9.0    Java 10.0.1
----------------------------------

lein test tst.demo.core
test - start
(nth results 99) => [99 0 99]
daemon?    false
test - sleeping
thread  start
sync-consumer:  start
0 %
5 %
10 %
15 %
20 %
25 %
30 %
35 %
40 %
45 %
50 %
55 %
60 %
65 %
70 %
75 %
80 %
85 %
90 %
95 %
"Elapsed time: 9388.313828 msecs"
sync-consumer:  stop
thread  done
test - end

Ran 2 tests containing 0 assertions.
0 failures, 0 errors.
lein test  72.52s user 1.69s system 374% cpu 19.823 total

似乎克洛伊丘早就离开了手册。 螺纹 反对。

也许你发现了一个(间歇性的)虫子。

Carcigenicate 6 年前

多亏了阿玛洛和亚历克斯,我才成功。

我在评论中实现了@amalloy的建议,这两个变体在这里和我的实际案例中都有效:

; Brittle since "once" is apparently more of an implementation detail of the language
(defn async-consumer [results width height]
  (doto
    (Thread. ^Runnable
             (^:once fn* []
               (sync-consumer results width height)
               (println "Done...")))
    (.start)))

; Arguably less brittle under the assumption that if they replace "once" with another mechanism,
;  they'll update "delay".
(defn async-consumer [results width height]
  (let [d (delay (sync-consumer results width height))]
    (doto
      (Thread. ^Runnable
               (fn []
                 @d
                 (println "Done...")))
      (.start))))

我还尝试更新到1.9.0。我想也许能解决这个问题,因为@Alex说他在1.9.0上,不能复制这个,而且 there's also this bug fix that seems related. 不幸的是,我没注意到有什么不同。

如果有一个实际的,可靠的机制,那就好了。 ^:once 看起来不错,但我不想用如果只是为了让它以后有可能破裂,和使用 delay 似乎是为了利用它的内在 (^:once fn* ...) .

哦,好吧,至少现在有用了。谢谢你们。