Skip to content

Commit

Permalink
Backport d1099033ac63b9dd0dd6e3a7341db929e9e0e56e
Browse files Browse the repository at this point in the history
  • Loading branch information
duke committed Feb 26, 2024
1 parent 8ddff70 commit 6ac72c2
Show file tree
Hide file tree
Showing 2 changed files with 80 additions and 14 deletions.
47 changes: 40 additions & 7 deletions src/java.base/unix/classes/sun/nio/ch/SinkChannelImpl.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2000, 2022, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2000, 2024, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
Expand Down Expand Up @@ -66,6 +66,13 @@ class SinkChannelImpl
// ID of native thread doing write, for signalling
private long thread;

// True if the channel's socket has been forced into non-blocking mode
// by a virtual thread. It cannot be reset. When the channel is in
// blocking mode and the channel's socket is in non-blocking mode then
// operations that don't complete immediately will poll the socket and
// preserve the semantics of blocking operations.
private volatile boolean forcedNonBlocking;

// -- End of fields protected by stateLock


Expand All @@ -79,11 +86,34 @@ public int getFDVal() {

SinkChannelImpl(SelectorProvider sp, FileDescriptor fd) throws IOException {
super(sp);
IOUtil.configureBlocking(fd, false);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
}

/**
* Checks that the channel is open.
*
* @throws ClosedChannelException if channel is closed (or closing)
*/
private void ensureOpen() throws ClosedChannelException {
if (!isOpen())
throw new ClosedChannelException();
}

/**
* Ensures that the socket is configured non-blocking when on a virtual thread.
*/
private void configureSocketNonBlockingIfVirtualThread() throws IOException {
assert writeLock.isHeldByCurrentThread();
if (!forcedNonBlocking && Thread.currentThread().isVirtual()) {
synchronized (stateLock) {
ensureOpen();
IOUtil.configureBlocking(fd, false);
forcedNonBlocking = true;
}
}
}

/**
* Closes the write end of the pipe if there are no write operation in
* progress and the channel is not registered with a Selector.
Expand Down Expand Up @@ -183,9 +213,11 @@ protected void implConfigureBlocking(boolean block) throws IOException {
writeLock.lock();
try {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
IOUtil.configureBlocking(fd, block);
ensureOpen();
// do nothing if virtual thread has forced the socket to be non-blocking
if (!forcedNonBlocking) {
IOUtil.configureBlocking(fd, block);
}
}
} finally {
writeLock.unlock();
Expand Down Expand Up @@ -241,8 +273,7 @@ private void beginWrite(boolean blocking) throws ClosedChannelException {
begin();
}
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
ensureOpen();
if (blocking)
thread = NativeThread.current();
}
Expand Down Expand Up @@ -279,6 +310,7 @@ public int write(ByteBuffer src) throws IOException {
int n = 0;
try {
beginWrite(blocking);
configureSocketNonBlockingIfVirtualThread();
n = IOUtil.write(fd, src, -1, nd);
if (blocking) {
while (IOStatus.okayToRetry(n) && isOpen()) {
Expand Down Expand Up @@ -306,6 +338,7 @@ public long write(ByteBuffer[] srcs, int offset, int length) throws IOException
long n = 0;
try {
beginWrite(blocking);
configureSocketNonBlockingIfVirtualThread();
n = IOUtil.write(fd, srcs, offset, length, nd);
if (blocking) {
while (IOStatus.okayToRetry(n) && isOpen()) {
Expand Down
47 changes: 40 additions & 7 deletions src/java.base/unix/classes/sun/nio/ch/SourceChannelImpl.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2000, 2022, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2000, 2024, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
Expand Down Expand Up @@ -66,6 +66,13 @@ class SourceChannelImpl
// ID of native thread doing read, for signalling
private long thread;

// True if the channel's socket has been forced into non-blocking mode
// by a virtual thread. It cannot be reset. When the channel is in
// blocking mode and the channel's socket is in non-blocking mode then
// operations that don't complete immediately will poll the socket and
// preserve the semantics of blocking operations.
private volatile boolean forcedNonBlocking;

// -- End of fields protected by stateLock


Expand All @@ -79,11 +86,34 @@ public int getFDVal() {

SourceChannelImpl(SelectorProvider sp, FileDescriptor fd) throws IOException {
super(sp);
IOUtil.configureBlocking(fd, false);
this.fd = fd;
this.fdVal = IOUtil.fdVal(fd);
}

/**
* Checks that the channel is open.
*
* @throws ClosedChannelException if channel is closed (or closing)
*/
private void ensureOpen() throws ClosedChannelException {
if (!isOpen())
throw new ClosedChannelException();
}

/**
* Ensures that the socket is configured non-blocking when on a virtual thread.
*/
private void configureSocketNonBlockingIfVirtualThread() throws IOException {
assert readLock.isHeldByCurrentThread();
if (!forcedNonBlocking && Thread.currentThread().isVirtual()) {
synchronized (stateLock) {
ensureOpen();
IOUtil.configureBlocking(fd, false);
forcedNonBlocking = true;
}
}
}

/**
* Closes the read end of the pipe if there are no read operation in
* progress and the channel is not registered with a Selector.
Expand Down Expand Up @@ -183,9 +213,11 @@ protected void implConfigureBlocking(boolean block) throws IOException {
readLock.lock();
try {
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
IOUtil.configureBlocking(fd, block);
ensureOpen();
// do nothing if virtual thread has forced the socket to be non-blocking
if (!forcedNonBlocking) {
IOUtil.configureBlocking(fd, block);
}
}
} finally {
readLock.unlock();
Expand Down Expand Up @@ -241,8 +273,7 @@ private void beginRead(boolean blocking) throws ClosedChannelException {
begin();
}
synchronized (stateLock) {
if (!isOpen())
throw new ClosedChannelException();
ensureOpen();
if (blocking)
thread = NativeThread.current();
}
Expand Down Expand Up @@ -279,6 +310,7 @@ public int read(ByteBuffer dst) throws IOException {
int n = 0;
try {
beginRead(blocking);
configureSocketNonBlockingIfVirtualThread();
n = IOUtil.read(fd, dst, -1, nd);
if (blocking) {
while (IOStatus.okayToRetry(n) && isOpen()) {
Expand Down Expand Up @@ -306,6 +338,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
long n = 0;
try {
beginRead(blocking);
configureSocketNonBlockingIfVirtualThread();
n = IOUtil.read(fd, dsts, offset, length, nd);
if (blocking) {
while (IOStatus.okayToRetry(n) && isOpen()) {
Expand Down

0 comments on commit 6ac72c2

Please sign in to comment.