[
https://issues.apache.org/jira/browse/GROOVY-12457?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131646#comment-18131646
]
ASF GitHub Bot commented on GROOVY-12457:
-----------------------------------------
daniellansun commented on PR #2986:
URL: https://github.com/apache/groovy/pull/2986#issuecomment-6392508175
@paulk-asert Thank you for the reproduction and for splitting the two
halves. The stall is real, and the tables are convincing: `minimumRunnable`
alone and `managedBlock` alone do not fix it, and together they do. They also
show the cost. I have taken the suggestion and removed that compensation rather
than shipping it.
`Pool.fixed(n)` is a `ForkJoinPool` of parallelism `n` again (`new
ForkJoinPool(n)`, and the same for `Pool.cpu()`). Parallel collection callbacks
on a fork-join pool are ordinary stream tasks. They are not wrapped in
`managedBlock`, so a CPU-bound body does not recruit spare workers, and the
common pool is not enlarged. On the compensated build, JDK 25, `Pool.fixed(2)`
over a CPU `forEach` grew to 6 threads; after the removal the same shape stays
at 2. Users who pick `fixed(n)` to cap CPU use keep that cap. `maximumPoolSize`
and `minimumRunnable` are the JDK defaults again, including for async
`Pool.fixed` / `Pool.cpu` callers. The JIRA description still talks about the
earlier queue-check split and does not mention this pool-constructor change;
that text needs a human update once the approach is settled. I have not edited
the ticket.
Blocking on a fork-join pool can still stall when every worker blocks with a
sibling queued. The flaky tests now do what you suggested:
`ParallelScope.withPool(parties + 1)`, with the pool created outside the scope
and closed afterwards. They assert `parallelism == threads` and `poolSize <=
threads`, and an upper bound on worker names. Twenty rounds are a concurrency
check. They are not a statistical proof at one stall in 15k–80k rounds, and
they no longer treat pool growth as expected. A separate CPU test locks the
cap. I could not reproduce the original stall on JDK 25 either. I did not
re-run your JDK 17 and 21 matrices, so those results stand.
For a blocking workload the opt-in is `Pool.virtual()`, and `Pool.io()`
where that is the virtual pool. `asForkJoinPool()` still throws for those
pools. Parallel collection methods catch that `UnsupportedOperationException`
and submit one task per element to that executor. They do not fall back to the
common pool. Any other exception from `asForkJoinPool()` propagates. On JDKs
without virtual threads, `Pool.io()` remains
`fixed(ConcurrentConfig.defaultParallelism())` and the spare-worker rule
applies. I confirmed that JDK 17 branch by reading `DefaultPool`, not by
running it: this machine is JDK 25, and the `Pool.io()` test took the
virtual-thread path. The docs no longer say that a pool of size `n` can host
`n` concurrent blockers. They say a fork-join pool's parallelism is a hard cap,
and that waiting work belongs on `virtual()` / `io()`.
Two smaller notes from the review are fixed with the same change. There is
one executor wait path, and it restores the interrupt flag on every wait; the
old path restored it only when compensation ran. A synchronous executor runs
the task on the caller and leaves a caller interrupt set. A pool thread drops a
cancellation interrupt left by the previous task, and the worker slot and the
interrupt are taken under one lock so that interrupt cannot land on the next
task that reused the thread. `grepParallel` builds the `isCase` predicate
outside the task. `collectManyParallel` applies the transform inside the stream
or flat-map task. They are no longer wrapped in different places.
Nested parallel calls stay on the current pool. A fork-join leaf does not
inherit the submitter's binding, so each callback rebinds that pool on the
worker that runs it. On a bounded executor that does not grow, a nested call
runs on the worker already executing the outer element, which keeps that worker
from waiting on a task queued behind it. `Pool.virtual()` still runs each
element as its own task.
A null element, a null reduction, and a null group key now behave the same
on a fork-join pool and on `Pool.virtual()`. `findParallel` and
`findAnyParallel` no longer finish through `Optional.of`, which threw when the
match was null. `minParallel`, `maxParallel`, and `sumParallel` fold in
encounter order and can return null; a one-element collection does not call the
comparator or operator. `groupByParallel` keeps a null key, and each group
stays in encounter order, on both paths.
I would be glad of a look at the revised contract, especially if CI was on a
JDK other than 25.
> 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
> Priority: Major
>
> 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
(v10.3.26#10030026)
