On Tue, 21 Jan 2025 15:13:56 GMT, Daniel Fuchs <dfu...@openjdk.org> wrote:
>> There's a subtle race condition in AggregatePublisher.AggregateSubscription.
>> The AggregatePublisher will subscribe to downstream publishers in turn,
>> subscribing to the next publisher when the previous publisher onComplete()
>> has been called.
>>
>> The request() method passed to the upstream subscriber detects that all
>> subscribers have completed if the current downstream publisher is null and
>> the queue of publisher yet to subscribe to is empty.
>>
>> The event loop is responsible for subscribing to the next publisher, and
>> does so by polling the queue and assigning the returned value to
>> this.publisher, and then subscribe to that publisher. However, when
>> subscribing to the last publisher in the queue, there is a small time window
>> where the queue can be observed to be empty, before this.publisher is
>> assigned.
>>
>> The fix adds synchronization to make sure a consistent state is observed
>> when it matters. Typically - setting `this.publisher` or taking a snapshot
>> of a publisher and its subscription, should be done within synchronized.
>
> Daniel Fuchs has updated the pull request with a new target base due to a
> merge or a rebase. The incremental webrev excludes the unrelated changes
> brought in by the merge/rebase. The pull request contains two additional
> commits since the last revision:
>
> - Merge branch 'master' into AggregatePublisher-8348108
> - 8348108: Race condition in AggregatePublisher.AggregateSubscription
test/jdk/java/net/httpclient/AggregateRequestBodyTest.java line 432:
> 430: items.addLast(item);
> 431: int available = semaphore.availablePermits();
> 432: if (semaphore.availablePermits() > Integer.MAX_VALUE - 8) {
In context of reporting the `available` value in the exception message thta
gets thrown in the next line, should this check have used the local variable
`available` or does it not matter?
-------------
PR Review Comment: https://git.openjdk.org/jdk/pull/23204#discussion_r1923934747
