Long-running loop and "Transaction failed after reaching retry limit"

[Stuart Sierra]

Hello all,
I'm trying to process about a million files. I've adapted the
parallel map function in 2 ways: 1) I don't return any values; 2) each
"worker" thread accumulates results and writes them to a file when
there are no more files to process. This works for a while, but after
a few hundred thousand iterations the worker threads throw
"java.lang.Exception: Transaction failed after reaching retry limit."

Here's an outline of my function; any help appreciated.

> (defn process-all [output-file nthreads]
> (let [pool (. Executors (newFixedThreadPool nthreads))
> todo (ref (all-files)) ; seq of files to process
> worker (fn [n] ; fn to generate worker fns
> (fn []
> (with-memory-repo ; binds vars for the worker
> (loop [] ; loop inside the dynamic env
> (let [job (sync nil
> (when @todo
> (let [item (first @todo)]
> (alter todo rest)
> item)))]
> (if job
> (do (process-one job) (recur))
> (save-results (str output-file n))))))))
> wait (fn [] (if (sync nil @todo)
> (recur)
> (. pool (shutdown))))]
> ;; Start worker threads:
> (dorun (map (fn [task] (. pool (submit task)))
> (map worker (range nthreads))))
> ;; Wait for them to finish:
> (wait)))

Thanks,
-Stuart

[John Cowan]

See http://groups.google.com/group/clojure/browse_thread/thread/a47ef556e628d526
.

--
GMail doesn't have rotating .sigs, but you can see mine at
http://www.ccil.org/~cowan/signatures

[Rich Hickey]

On Mar 10, 1:28 pm, Stuart Sierra <the.stuart.sie...@gmail.com> wrote:
> Hello all,
> I'm trying to process about a million files. I've adapted the
> parallel map function in 2 ways: 1) I don't return any values; 2) each
> "worker" thread accumulates results and writes them to a file when
> there are no more files to process. This works for a while, but after
> a few hundred thousand iterations the worker threads throw
> "java.lang.Exception: Transaction failed after reaching retry limit."
>
> Here's an outline of my function; any help appreciated.
>

Your wait function is what is called a busy-wait-loop - it is spinning/
polling and sucking away CPU cycles. You need to try a different
waiting mechanism - CountDownLatch or something.

Depending on memory usage, the problem may better suited for agents
(untested):

(defn process-all [all-files process-one nthreads]
(let [workers (map (fn [n] (agent nil)) (range nthreads))]
(map (fn [job worker] (! worker process-one job))
all-files (cycle workers))
(apply await workers)))

You don't say much about the data or the jobs (e.g. how long does it
take to run process-one once?), so it's hard to make recommendations.

Rich

[Stuart Sierra]

On Mar 10, 3:44 pm, Rich Hickey <richhic...@gmail.com> wrote:
> Your wait function is what is called a busy-wait-loop - it is spinning/
> polling and sucking away CPU cycles. You need to try a different
> waiting mechanism - CountDownLatch or something.
>
> Depending on memory usage, the problem may better suited for agents
> (untested):
>
> (defn process-all [all-files process-one nthreads]
> (let [workers (map (fn [n] (agent nil)) (range nthreads))]
> (map (fn [job worker] (! worker process-one job))
> all-files (cycle workers))
> (apply await workers)))
>
> You don't say much about the data or the jobs (e.g. how long does it
> take to run process-one once?), so it's hard to make recommendations.

Thanks, Rich. Clearly I'm still a beginner at multithreading. I
checked the running time on process-one and found it was under 100
msecs, so attempting multiple threads may have been premature evil. I
was just curious to try it.

I'm glad you mentioned agents, though, because I was also curious if I
could use agents for this task. I'll play around with it.

Thanks,
-Stuart