Daniel Sun created GROOVY-12457:
-----------------------------------
Summary: Blocking parallel-collection callbacks can fail to run a
queued sibling task
Key: GROOVY-12457
URL: https://issues.apache.org/jira/browse/GROOVY-12457
Project: Groovy
Issue Type: Bug
Reporter: Daniel Sun
h2. Context
{{{}eachParallel{}}}, {{{}collectParallel{}}}, and the other parallel
collection methods run their callback on a {{{}ForkJoinPool{}}}: the pool bound
by {{{}ParallelScope.withPool{}}}, or {{ForkJoinPool.commonPool()}} when no
pool is bound. {{@Parallel}} {{for-in}} loops are rewritten to call
{{{}eachParallel{}}}. The methods arrived with GROOVY-11952
([GEP-18|https://groovy.apache.org/wiki/GEP-18.html]).
A pool of size {{n}} is the intended way to run {{n}} callbacks that block at
the same time.
h2. Problem
This rendezvous fails intermittently, including when the pool size equals the
number of parties:
{code:groovy}
import groovy.concurrent.ParallelScope
import java.util.concurrent.CyclicBarrier
import java.util.concurrent.TimeUnit
int parties = 2
ParallelScope.withPool(parties) { scope ->
def barrier = new CyclicBarrier(parties)
(1..parties).eachParallel {
barrier.await(5, TimeUnit.SECONDS)
}
}
{code}
{{CyclicBarrier.await}} throws {{TimeoutException}} because the second element
never starts. The failure is reported as a nested
{{{}UndeclaredThrowableException{}}}:
||Layer||What the caller sees||
|Closure|{{TimeoutException}} from {{CyclicBarrier.await}}|
|{{Consumer}} proxy|{{{}UndeclaredThrowableException{}}}, because {{accept}}
does not declare checked exceptions|
|{{ForkJoinTask.getThrowableException}}|A second
{{{}UndeclaredThrowableException{}}}, rebuilt on the joining thread|
The same shape fails for {{@Parallel}} {{{}for (item in 1..n){}}}. It shows up
more often once the pool's workers have already run and parked. A freshly
created pool can fail as well.
h2. Cause
{{parallelStream()}} forks the sibling leaf onto the current worker's queue,
then runs one leaf on that same worker. {{CyclicBarrier.await}} is not a
{{{}ForkJoinPool.ManagedBlocker{}}}, so the pool does not release or replace
the blocked worker. If the other worker has already scanned and parked, the
fork signal is not always delivered again. The sibling stays on the queue until
the barrier times out.
Sizing the pool to exactly {{n}} threads does not close the race. The idle
thread is there; it is parked.
h2. Change
The proposed change runs the user callback inside {{ForkJoinPool.managedBlock}}
when the current worker still has queued tasks. That releases a parked worker,
or creates a spare, before the callback can block. A leaf whose queue is empty
is unchanged.
* JDK 20 and later read {{{}ForkJoinWorkerThread.getQueuedTaskCount(){}}}.
* JDK 17 does not have that method, so the check uses the pool-wide
queued-task estimate.
The wrapper covers every parallel-collection callback, including
{{eachWithIndexParallel}} and the {{eachParallel}} call emitted for
{{{}@Parallel{}}}.
Compensation runs only when a sibling is already queued. A callback that blocks
before any sibling has been forked can still stall, and so can a pool that
refuses another thread. Work that spends most of its time waiting still fits
{{Pool.io()}} or {{Pool.virtual()}} better than a fixed {{{}ForkJoinPool{}}}.
h2. Tests
* {{ParallelCollectionTest}} and {{ParallelAnnotationTest}} repeat the
two-party barrier on one pool, so later rounds run against workers parked by
the first round. The barrier must trip.
* The thread-count assertion allows a spare ({{{}size() >= parties{}}}),
because {{managedBlock}} may add one while a worker is blocked.
* The full two test classes passed on JDK 25. The three barrier tests also
passed on JDK 17, which exercises the queue-length fallback.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
