[
https://issues.apache.org/jira/browse/GROOVY-12457?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133279#comment-18133279
]
ASF GitHub Bot commented on GROOVY-12457:
-----------------------------------------
paulk-asert commented on PR #2986:
URL: https://github.com/apache/groovy/pull/2986#issuecomment-6455666379
Thanks for the rework. It addresses the earlier review: `fixed(n)` and
`cpu()` are back to plain capped fork-join pools, the stall is documented
rather than hidden behind compensation, `Pool.virtual()`/`io()` no longer fall
back to the common pool, and nested calls stay on the current pool. CI is green
and the branch merges cleanly with master. I read through
`ParallelExecutorTasks` and found no correctness problems. A few things before
merge:
**1. `grepParallel`: fix now, or split it off.** There are two problems:
- **Regression:** `[1, null, 2].grepParallel(null)` now throws NPE because
`require(self, filter)` null-checks the filter. On master it returns `[null]`,
the same as sequential `grep(null)`. Null-checking only `self` fixes this.
- **Present before this PR:** it keeps an element unless `isCase` returns
exactly `Boolean.FALSE`. Sequential `grep` uses Groovy truth (`castToBoolean`).
With a filter whose `isCase` returns `x % 2`, `grep` gives `[1, 3]` but
`grepParallel` gives `[1, 2, 3, 4]`. With one whose `isCase` returns `null`,
`grep` gives `[]` but `grepParallel` returns everything. Closure, Pattern and
Class filters are unaffected because they return real booleans.
Both are one-line changes to the same method, so I'd fix them here. If you'd
rather keep this PR focused, split them into a follow-up PR together with item
2, but that follow-up needs to land before the next release so the regression
doesn't ship.
**2. `maxParallel` and nulls.** The shared min/max helper passes every
element to the comparator. Sequential `max(Comparator)` skips nulls, so `[null,
1].maxParallel(Comparator.naturalOrder())` throws where `[null,
1].max(Comparator.naturalOrder())` returns `1`. On `[null, null]`, sequential
returns `null` without calling the comparator, while the PR calls
`compare(null, null)`. Sequential `min(Comparator)` does pass nulls through, so
`minParallel` already matches. Sequential is asymmetric here, so either match
each sequential method exactly or keep the symmetric rule and document how it
differs from `max`. Either is fine; it just shouldn't diverge silently. This
goes with item 1 if you split.
**3. Per-element `Pool.withCurrent` rebinding.** Every callback inside a
`withPool`/`fixed`/`cpu` scope is now wrapped so that nested calls see the
pool. I emulated the wrapper on master: `Pool.fixed(12)`, 2M elements, a
trivial `map(i + 1)`. It goes from 2 ms to 31–32 ms on JDK 17/21 and to 14 ms
on JDK 25. A real `collectParallel { it + 1 }` on master takes 3–4 ms, so for
light callbacks the wrapper would cost more than the work. A cheaper option
would be a worker thread factory whose threads carry their owning `Pool`, with
`Pool.current()` falling back to the current worker's pool. That keeps nested
calls on the pool without rebinding per element. If you'd rather keep the
current approach, I'd like to see a JMH number for a light workload first.
**4. Interrupt semantics on the executor path.** When `find`/`any`/`every`
short-circuits, or a sibling task fails, `ParallelExecutorTasks` interrupts
callbacks that are still running. The fork-join path never does this.
Interrupting a thread blocked on an `InterruptibleChannel` (for example a
`FileChannel`) closes the channel, which users won't expect. Only internal
Javadoc mentions it; could you add a sentence to
`core-parallel-collections.adoc`?
**5. Small nit:** `groupInOrder` already inserts in encounter order. If it
used a `LinkedHashMap` instead of a `HashMap`, the keys would come out in
first-seen order, like sequential `groupBy`.
**Housekeeping:** the JIRA description still describes the dropped
compensation design and should be updated to match what the PR does now. We
also need to decide whether this goes to 6.0.x. If it does, the null-handling
changes (null matches in `find*`, null elements in `min`/`max`/`sum`, null
`groupBy` keys) need a release-notes entry.
The rebinding numbers come from emulating the wrapper on master, not from
running the PR, and the reply says so. If you'd rather not ask for the JMH
number in item 3, delete that sentence.
> 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)
