GPars - proper way to terminate a parallel collect

问题:

What is the best way to terminate a parallel collection (in the case of either an exception thrown by one of the threads, or an interrupt initiated by the user)?

In either case, I could easily set some flag and just check it at the top of the loop. But if I've got 10,000 items in the collection, I'd rather tell whatever feeds those into the ForkJoinPool to stop feeding them in. Let the 5 or 20 or so that already started finish, but don't start any more.

Here's a sample you could plug in to see what I mean. If I simulate the throwing of an exception by one of the threads, it appears like the collection stops processing, because the first print of thread counts (at the 'A:' message) is often very small (5 or so). However, by printing the counts outside of the GParsPool.withPool (at 'B:'), I can see that they really kept running (always all 100):

void doIt() {
    AtomicInteger threadCounter = new AtomicInteger(0)
    AtomicInteger threadCounterEnd = new AtomicInteger(0)

    try {
        GParsPool.withPool { pool ->

            try {
                (1..100).eachParallel {
                    try {
                        if (threadCounter.incrementAndGet() == 1) {
                            throw new RuntimeException('planet blew up!')
                        }

                        // Do some long work
                        Integer counter=0
                        (1..1000000).each() {counter++}

                        // Flag if we went all the way through
                        threadCounterEnd.incrementAndGet()
                    } catch (Exception exc) {
                        //pool.shutdownNow()
                        throw new RuntimeException(exc)
                    }
                }
            } catch (Exception exc) {
                println 'caught exception'
                //pool.shutdownNow()
                println "A: threads launched was ${threadCounter}, ended was ${threadCounterEnd}"
                throw new RuntimeException(exc)
            }
        }
    } catch (Exception exc) {
        exc.printStackTrace()
    }

    println "B: threads launched was ${threadCounter}, ended was ${threadCounterEnd}"
}

If I use pool.shutdown() inside the eachParallel, it has no affect. If I use pool.shutdownNow(), it ends the processing like I want, but throws a CancellationException, which obscures the real exception that I'd like to have. I could store the 'real' exception off in a variable for later access, but I have to wonder if there isn't a better way to tell the parallel collection to cleanly stop.

回答1:

There's no such mechanism ATM. Since the pool cannot distinguish between your tasks and tasks of other potential computations, it has no mechanism to stop threads from reading the task queues. If you know you are the only user of the thread pool, you may use shutdownNow() as you suggest (with the mentioned consequences).

The other option, which I feel you are also aware of, is to properly catch exceptions in the tasks directly and set a shared atomic flag to indicate to the other tasks to quit immediately.