archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-08-27 23:04:50 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

8338383: Implement JEP 491: Synchronize Virtual Threads without Pinning

Browse source 
Co-authored-by: Patricio Chilano Mateo <pchilanomate@openjdk.org>
Co-authored-by: Alan Bateman <alanb@openjdk.org>
Co-authored-by: Andrew Haley <aph@openjdk.org>
Co-authored-by: Fei Yang <fyang@openjdk.org>
Co-authored-by: Coleen Phillimore <coleenp@openjdk.org>
Co-authored-by: Richard Reingruber <rrich@openjdk.org>
Co-authored-by: Martin Doerr <mdoerr@openjdk.org>
Reviewed-by: aboldtch, dholmes, coleenp, fbredberg, dlong, sspitsyn
This commit is contained in:
Patricio Chilano Mateo 2024-11-12 15:23:48 +00:00
parent 8a2a75e56d
commit 78b80150e0
246 changed files with 8295 additions and 2755 deletions
Download patch file Download diff file Expand all files Collapse all files

80
src/java.base/macosx/classes/sun/nio/ch/KQueueSelectorImpl.java
View file

@ -35,7 +35,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import jdk.internal.misc.Blocker;
import static sun.nio.ch.KQueue.EVFILT_READ;
import static sun.nio.ch.KQueue.EVFILT_WRITE;
@ -109,36 +108,71 @@ class KQueueSelectorImpl extends SelectorImpl {
int numEntries;
processUpdateQueue();
processDeregisterQueue();
try {
begin(blocking);
do {
long startTime = timedPoll ? System.nanoTime() : 0;
boolean attempted = Blocker.begin(blocking);
try {
if (Thread.currentThread().isVirtual()) {
numEntries = (timedPoll)
? timedPoll(TimeUnit.MILLISECONDS.toNanos(to))
: untimedPoll(blocking);
} else {
try {
begin(blocking);
do {
long startTime = timedPoll ? System.nanoTime() : 0;
numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, to);
} finally {
Blocker.end(attempted);
}
if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
// timed poll interrupted so need to adjust timeout
long adjust = System.nanoTime() - startTime;
to -= TimeUnit.NANOSECONDS.toMillis(adjust);
if (to <= 0) {
// timeout expired so no retry
numEntries = 0;
if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
// timed poll interrupted so need to adjust timeout
long adjust = System.nanoTime() - startTime;
to -= TimeUnit.NANOSECONDS.toMillis(adjust);
if (to <= 0) {
// timeout expired so no retry
numEntries = 0;
}
}
}
} while (numEntries == IOStatus.INTERRUPTED);
assert IOStatus.check(numEntries);
} finally {
end(blocking);
} while (numEntries == IOStatus.INTERRUPTED);
} finally {
end(blocking);
}
}
assert IOStatus.check(numEntries);
processDeregisterQueue();
return processEvents(numEntries, action);
}
/**
* If blocking, parks the current virtual thread until a file descriptor is polled
* or the thread is interrupted.
*/
private int untimedPoll(boolean block) throws IOException {
int numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
if (block) {
while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
Poller.pollSelector(kqfd, 0);
numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
}
}
return numEntries;
}
/**
* Parks the current virtual thread until a file descriptor is polled, or the thread
* is interrupted, for up to the specified waiting time.
*/
private int timedPoll(long nanos) throws IOException {
long startNanos = System.nanoTime();
int numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
long remainingNanos = nanos - (System.nanoTime() - startNanos);
if (remainingNanos <= 0) {
// timeout
break;
}
Poller.pollSelector(kqfd, remainingNanos);
numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
}
return numEntries;
}
/**
* Process changes to the interest ops.
*/