infernap12/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2026-07-28 20:03:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

8358958: (aio) AsynchronousByteChannel.read/write should throw IAE if buffer is thread-confined

Browse Source 
Reviewed-by: jpai, bpb
This commit is contained in:
Alan Bateman 2025-06-13 06:33:42 +00:00
parent 991097b7bf
commit 9aeacf2de5
9 changed files with 437 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/java.base/share/classes/java/lang/foreign/MemorySegment.java
View File

@ -1120,9 +1120,7 @@ public sealed interface MemorySegment permits AbstractMemorySegmentImpl {
* <p>
* If this segment is {@linkplain #isAccessibleBy(Thread) accessible} from a single
* thread, calling read/write I/O operations on the resulting buffer might result in
* unspecified exceptions being thrown. Examples of such problematic operations are
* {@link java.nio.channels.AsynchronousSocketChannel#read(ByteBuffer)} and
* {@link java.nio.channels.AsynchronousSocketChannel#write(ByteBuffer)}.
* unspecified exceptions being thrown.
* <p>
* Finally, the resulting buffer's byte order is
* {@link java.nio.ByteOrder#BIG_ENDIAN}; this can be changed using

16
src/java.base/share/classes/java/nio/channels/AsynchronousByteChannel.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2007, 2013, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2007, 2025, 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
@ -25,6 +25,8 @@
package java.nio.channels;
import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.nio.ByteBuffer;
import java.util.concurrent.Future;
@ -97,7 +99,8 @@ public interface AsynchronousByteChannel
* The completion handler
*
* @throws IllegalArgumentException
* If the buffer is read-only
* If the buffer is read-only or a view of a {@link MemorySegment}
* allocated from a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws ReadPendingException
* If the channel does not allow more than one read to be outstanding
* and a previous read has not completed
@ -128,7 +131,8 @@ public interface AsynchronousByteChannel
* @return A Future representing the result of the operation
*
* @throws IllegalArgumentException
* If the buffer is read-only
* If the buffer is read-only or a view of a {@link MemorySegment}
* allocated from a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws ReadPendingException
* If the channel does not allow more than one read to be outstanding
* and a previous read has not completed
@ -177,6 +181,9 @@ public interface AsynchronousByteChannel
* @param handler
* The completion handler object
*
* @throws IllegalArgumentException
* If the buffer is a view of a {@link MemorySegment} allocated from a
* {@linkplain Arena#ofConfined() thread-confined arena}
* @throws WritePendingException
* If the channel does not allow more than one write to be outstanding
* and a previous write has not completed
@ -205,6 +212,9 @@ public interface AsynchronousByteChannel
*
* @return A Future representing the result of the operation
*
* @throws IllegalArgumentException
* If the buffer is a view of a {@link MemorySegment} allocated from a
* {@linkplain Arena#ofConfined() thread-confined arena}
* @throws WritePendingException
* If the channel does not allow more than one write to be outstanding
* and a previous write has not completed

20
src/java.base/share/classes/java/nio/channels/AsynchronousFileChannel.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2007, 2024, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2007, 2025, 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
@ -25,6 +25,8 @@
package java.nio.channels;
import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.nio.file.*;
import java.nio.file.attribute.FileAttribute;
import java.nio.file.spi.*;
@ -693,7 +695,9 @@ public abstract class AsynchronousFileChannel
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the position is negative or the buffer is read-only
* If the position is negative, or the buffer is read-only or a view of a
* {@link MemorySegment} allocated from a {@linkplain Arena#ofConfined()
* thread-confined arena}
* @throws NonReadableChannelException
* If this channel was not opened for reading
*/
@ -728,7 +732,9 @@ public abstract class AsynchronousFileChannel
* @return A {@code Future} object representing the pending result
*
* @throws IllegalArgumentException
* If the position is negative or the buffer is read-only
* If the position is negative, or the buffer is read-only or a view of a
* {@link MemorySegment} allocated from a {@linkplain Arena#ofConfined()
* thread-confined arena}
* @throws NonReadableChannelException
* If this channel was not opened for reading
*/
@ -759,7 +765,9 @@ public abstract class AsynchronousFileChannel
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the position is negative
* If the position is negative or the buffer is a view of a {@link
* MemorySegment} allocated from a {@linkplain Arena#ofConfined()
* thread-confined arena}
* @throws NonWritableChannelException
* If this channel was not opened for writing
*/
@ -795,7 +803,9 @@ public abstract class AsynchronousFileChannel
* @return A {@code Future} object representing the pending result
*
* @throws IllegalArgumentException
* If the position is negative
* If the position is negative or the buffer is a view of a {@link
* MemorySegment} allocated from a {@linkplain Arena#ofConfined()
* thread-confined arena}
* @throws NonWritableChannelException
* If this channel was not opened for writing
*/

18
src/java.base/share/classes/java/nio/channels/AsynchronousSocketChannel.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2007, 2024, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2007, 2025, 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
@ -25,6 +25,8 @@
package java.nio.channels;
import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.nio.channels.spi.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Future;
@ -379,7 +381,8 @@ public abstract class AsynchronousSocketChannel
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the buffer is read-only
* If the buffer is read-only or a view of a {@link MemorySegment}
* allocated from a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws ReadPendingException
* If a read operation is already in progress on this channel
* @throws NotYetConnectedException
@ -488,7 +491,8 @@ public abstract class AsynchronousSocketChannel
* If the pre-conditions for the {@code offset} and {@code length}
* parameter aren't met
* @throws IllegalArgumentException
* If the buffer is read-only
* If any of the buffers is read-only or a view of a {@link MemorySegment}
* allocated from a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws ReadPendingException
* If a read operation is already in progress on this channel
* @throws NotYetConnectedException
@ -538,6 +542,9 @@ public abstract class AsynchronousSocketChannel
* @param handler
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If the buffer is a view of a {@link MemorySegment} allocated from
* a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws WritePendingException
* If a write operation is already in progress on this channel
* @throws NotYetConnectedException
@ -552,6 +559,7 @@ public abstract class AsynchronousSocketChannel
CompletionHandler<Integer,? super A> handler);
/**
* @throws IllegalArgumentException {@inheritDoc}
* @throws WritePendingException {@inheritDoc}
* @throws NotYetConnectedException
* If this channel is not yet connected
@ -568,6 +576,7 @@ public abstract class AsynchronousSocketChannel
}
/**
* @throws IllegalArgumentException {@inheritDoc}
* @throws WritePendingException {@inheritDoc}
* @throws NotYetConnectedException
* If this channel is not yet connected
@ -638,6 +647,9 @@ public abstract class AsynchronousSocketChannel
* @param handler
* The handler for consuming the result
*
* @throws IllegalArgumentException
* If any of the buffers is a view of a {@link MemorySegment}
* allocated from a {@linkplain Arena#ofConfined() thread-confined arena}
* @throws IndexOutOfBoundsException
* If the pre-conditions for the {@code offset} and {@code length}
* parameter aren't met

18
src/java.base/share/classes/sun/nio/ch/AsynchronousFileChannelImpl.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2008, 2018, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2008, 2025, 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
@ -32,6 +32,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.locks.*;
import java.io.FileDescriptor;
import java.io.IOException;
import jdk.internal.access.JavaNioAccess;
import jdk.internal.access.SharedSecrets;
/**
* Base implementation of AsynchronousFileChannel.
@ -40,6 +42,8 @@ import java.io.IOException;
abstract class AsynchronousFileChannelImpl
extends AsynchronousFileChannel
{
private static final JavaNioAccess NIO_ACCESS = SharedSecrets.getJavaNioAccess();
// close support
protected final ReadWriteLock closeLock = new ReentrantReadWriteLock();
protected volatile boolean closed;
@ -215,6 +219,10 @@ abstract class AsynchronousFileChannelImpl
@Override
public final Future<Integer> read(ByteBuffer dst, long position) {
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (NIO_ACCESS.isThreadConfined(dst))
throw new IllegalArgumentException("Buffer is thread confined");
return implRead(dst, position, null, null);
}
@ -226,6 +234,10 @@ abstract class AsynchronousFileChannelImpl
{
if (handler == null)
throw new NullPointerException("'handler' is null");
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (NIO_ACCESS.isThreadConfined(dst))
throw new IllegalArgumentException("Buffer is thread confined");
implRead(dst, position, attachment, handler);
}
@ -237,6 +249,8 @@ abstract class AsynchronousFileChannelImpl
@Override
public final Future<Integer> write(ByteBuffer src, long position) {
if (NIO_ACCESS.isThreadConfined(src))
throw new IllegalArgumentException("Buffer is thread confined");
return implWrite(src, position, null, null);
}
@ -248,6 +262,8 @@ abstract class AsynchronousFileChannelImpl
{
if (handler == null)
throw new NullPointerException("'handler' is null");
if (NIO_ACCESS.isThreadConfined(src))
throw new IllegalArgumentException("Buffer is thread confined");
implWrite(src, position, attachment, handler);
}
}

29
src/java.base/share/classes/sun/nio/ch/AsynchronousSocketChannelImpl.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2008, 2024, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2008, 2025, 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
@ -39,6 +39,8 @@ import java.util.Objects;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;
import jdk.internal.access.JavaNioAccess;
import jdk.internal.access.SharedSecrets;
import sun.net.NetHooks;
import sun.net.ext.ExtendedSocketOptions;
@ -50,6 +52,8 @@ abstract class AsynchronousSocketChannelImpl
extends AsynchronousSocketChannel
implements Cancellable, Groupable
{
private static final JavaNioAccess NIO_ACCESS = SharedSecrets.getJavaNioAccess();
protected final FileDescriptor fd;
// protects state, localAddress, and remoteAddress
@ -281,6 +285,8 @@ abstract class AsynchronousSocketChannelImpl
public final Future<Integer> read(ByteBuffer dst) {
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (NIO_ACCESS.isThreadConfined(dst))
throw new IllegalArgumentException("Buffer is thread confined");
return read(false, dst, null, 0L, TimeUnit.MILLISECONDS, null, null);
}
@ -295,6 +301,8 @@ abstract class AsynchronousSocketChannelImpl
throw new NullPointerException("'handler' is null");
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (NIO_ACCESS.isThreadConfined(dst))
throw new IllegalArgumentException("Buffer is thread confined");
read(false, dst, null, timeout, unit, attachment, handler);
}
@ -310,12 +318,14 @@ abstract class AsynchronousSocketChannelImpl
if (handler == null)
throw new NullPointerException("'handler' is null");
Objects.checkFromIndexSize(offset, length, dsts.length);
ByteBuffer[] bufs = Util.subsequence(dsts, offset, length);
for (int i=0; i<bufs.length; i++) {
if (bufs[i].isReadOnly())
dsts = Util.subsequence(dsts, offset, length);
for (ByteBuffer dst : dsts) {
if (dst.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (NIO_ACCESS.isThreadConfined(dst))
throw new IllegalArgumentException("Buffer is thread confined");
}
read(true, null, bufs, timeout, unit, attachment, handler);
read(true, null, dsts, timeout, unit, attachment, handler);
}
/**
@ -384,6 +394,8 @@ abstract class AsynchronousSocketChannelImpl
@Override
public final Future<Integer> write(ByteBuffer src) {
if (NIO_ACCESS.isThreadConfined(src))
throw new IllegalArgumentException("Buffer is thread confined");
return write(false, src, null, 0L, TimeUnit.MILLISECONDS, null, null);
}
@ -396,6 +408,8 @@ abstract class AsynchronousSocketChannelImpl
{
if (handler == null)
throw new NullPointerException("'handler' is null");
if (NIO_ACCESS.isThreadConfined(src))
throw new IllegalArgumentException("Buffer is thread confined");
write(false, src, null, timeout, unit, attachment, handler);
}
@ -412,6 +426,11 @@ abstract class AsynchronousSocketChannelImpl
throw new NullPointerException("'handler' is null");
Objects.checkFromIndexSize(offset, length, srcs.length);
srcs = Util.subsequence(srcs, offset, length);
for (ByteBuffer src : srcs) {
if (NIO_ACCESS.isThreadConfined(src)) {
throw new IllegalArgumentException("Buffer is thread confined");
}
}
write(true, null, srcs, timeout, unit, attachment, handler);
}

2
src/java.base/share/classes/sun/nio/ch/IOUtil.java
View File

@ -479,7 +479,7 @@ public final class IOUtil {
static void acquireScope(ByteBuffer bb, boolean async) {
if (async && NIO_ACCESS.isThreadConfined(bb)) {
throw new IllegalStateException("Confined session not supported");
throw new IllegalArgumentException("Buffer is thread confined");
}
NIO_ACCESS.acquireSession(bb);
}

12
test/jdk/java/foreign/channels/TestAsyncSocketChannels.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2021, 2023, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2021, 2025, 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
@ -63,6 +63,7 @@ public class TestAsyncSocketChannels extends AbstractChannelsTest {
static final Class<IOException> IOE = IOException.class;
static final Class<ExecutionException> EE = ExecutionException.class;
static final Class<IllegalArgumentException> IAE = IllegalArgumentException.class;
static final Class<IllegalStateException> ISE = IllegalStateException.class;
/** Tests that confined sessions are not supported. */
@ -79,8 +80,8 @@ public class TestAsyncSocketChannels extends AbstractChannelsTest {
var bb = segment.asByteBuffer();
var bba = new ByteBuffer[] { bb };
List<ThrowingConsumer<TestHandler,?>> ioOps = List.of(
handler -> handler.propagateHandlerFromFuture(channel.write(bb)),
handler -> handler.propagateHandlerFromFuture(channel.read(bb)),
handler -> channel.write(bb),
handler -> channel.read(bb),
handler -> channel.write(bb, null, handler),
handler -> channel.read( bb, null, handler),
handler -> channel.write(bb , 0L, SECONDS, null, handler),
@ -91,10 +92,7 @@ public class TestAsyncSocketChannels extends AbstractChannelsTest {
for (var ioOp : ioOps) {
out.println("testAsyncWithConfined - op");
var handler = new TestHandler();
ioOp.accept(handler);
handler.await()
.assertFailedWith(ISE)
.assertExceptionMessage("Confined session not supported");
expectThrows(IAE, () -> ioOp.accept(handler));
}
}
}

351
test/jdk/java/nio/channels/etc/MemorySegments.java
View File

@ -1,5 +1,5 @@
/*
* Copyright (c) 2024, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2024, 2025, 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
@ -23,9 +23,9 @@
/*
* @test
* @bug 8333849
* @summary Test ByteChannel implementations of read and write with ByteBuffers that are
* backed by MemorySegments allocated from an Arena
* @bug 8333849 8358958
* @summary Test ByteChannel and AsycnhronousByteChannel implementations with ByteBuffers
* that are views of a MemorySegment
* @run junit MemorySegments
*/
@ -35,6 +35,10 @@ import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.channels.DatagramChannel;
import java.nio.channels.FileChannel;
import java.nio.channels.ServerSocketChannel;
@ -42,10 +46,18 @@ import java.nio.channels.SocketChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Stream;
import static java.nio.file.StandardOpenOption.*;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import static org.junit.jupiter.api.Assertions.*;
@ -325,6 +337,177 @@ class MemorySegments {
}
}
/**
* AsynchronousSocketChannel read/write(ByteBuffer).
*/
@ParameterizedTest
@MethodSource("arenaSuppliers")
void testAsyncSocketChannelReadWrite(Supplier<Arena> arenaSupplier) throws Exception {
boolean shared = isShared(arenaSupplier);
boolean confined = isConfined(arenaSupplier);
Arena arena = arenaSupplier.get();
try (var listener = AsynchronousServerSocketChannel.open();
AsynchronousSocketChannel ch1 = AsynchronousSocketChannel.open()) {
listener.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
ch1.connect(listener.getLocalAddress()).get();
try (AsynchronousSocketChannel ch2 = listener.accept().get()) {
ByteBuffer dst = arena.allocate(SIZE + 100).asByteBuffer();
ByteBuffer src = arena.allocate(SIZE).asByteBuffer();
fillRandom(src);
if (confined) {
// read and write should fail with IAE
assertThrows(IllegalArgumentException.class, () -> ch1.read(dst));
assertThrows(IllegalArgumentException.class, () -> ch2.write(src));
} else {
// async read
Future<Integer> readTask = ch1.read(dst);
// shared arena cannot be closed while read in progress
if (shared) {
assertThrows(IllegalStateException.class, arena::close);
}
// async write
Future<Integer> writeTask = ch2.write(src);
// finish write
int nwritten = writeTask.get();
assertTrue(nwritten > 0);
assertTrue(nwritten <= SIZE);
assertEquals(nwritten, src.position());
// finish read
int nread = readTask.get();
assertTrue(nread > 0);
assertTrue(nread <= nwritten);
assertEquals(nread, dst.position());
// check contents
dst.flip();
assertEquals(src.slice(0, nread), dst);
}
}
} finally {
tryClose(arena);
}
}
/**
* AsynchronousSocketChannel write(ByteBuffer[]).
*/
@ParameterizedTest
@MethodSource("arenaSuppliers")
void testAsyncSocketChannelGatheringWrite(Supplier<Arena> arenaSupplier) throws Throwable {
boolean confined = isConfined(arenaSupplier);
Arena arena = arenaSupplier.get();
try (var listener = AsynchronousServerSocketChannel.open();
AsynchronousSocketChannel ch1 = AsynchronousSocketChannel.open()) {
listener.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
ch1.connect(listener.getLocalAddress()).get();
try (AsynchronousSocketChannel ch2 = listener.accept().get()) {
ByteBuffer src = arena.allocate(SIZE * 2).asByteBuffer();
fillRandom(src);
ByteBuffer src1 = src.slice(0, SIZE);
ByteBuffer src2 = src.slice(SIZE, SIZE);
var srcs = new ByteBuffer[] { src1, src2 };
var writeHandler = new Handler<Long>();
if (confined) {
assertThrows(IllegalArgumentException.class,
() -> ch1.write(srcs, 0, 2, 0, TimeUnit.SECONDS, null, writeHandler));
} else {
// async gathering write
ch1.write(srcs, 0, 2, 0, TimeUnit.SECONDS, null, writeHandler);
// finish gathering write
int nwritten = (int) (long) writeHandler.join();
assertTrue(nwritten > 0);
assertEquals(Math.min(nwritten, SIZE), src1.position());
assertEquals(nwritten, src1.position() + src2.position());
// async read + finish read
ByteBuffer dst = arena.allocate(SIZE * 2 + 50).asByteBuffer();
int nread = ch2.read(dst).get();
assertTrue(nread > 0);
assertTrue(nread <= nwritten);
assertEquals(nread, dst.position());
// check contents
dst.flip();
assertEquals(src.slice(0, nread), dst);
}
}
} finally {
tryClose(arena);
}
}
/**
* AsynchronousSocketChannel read(ByteBuffer[]).
*/
@ParameterizedTest
@MethodSource("arenaSuppliers")
void testAsyncSocketChannelScatteringRead(Supplier<Arena> arenaSupplier) throws Throwable {
boolean shared = isShared(arenaSupplier);
boolean confined = isConfined(arenaSupplier);
Arena arena = arenaSupplier.get();
try (var listener = AsynchronousServerSocketChannel.open();
AsynchronousSocketChannel ch1 = AsynchronousSocketChannel.open()) {
listener.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
ch1.connect(listener.getLocalAddress()).get();
try (AsynchronousSocketChannel ch2 = listener.accept().get()) {
ByteBuffer dst = arena.allocate(SIZE + 50).asByteBuffer();
ByteBuffer dst1 = dst.slice(0, 50);
ByteBuffer dst2 = dst.slice(50, dst.capacity() - 50);
var dsts = new ByteBuffer[]{dst1, dst2};
var readHandler = new Handler<Long>();
if (confined) {
assertThrows(IllegalArgumentException.class,
() -> ch1.read(dsts, 0, 2, 0, TimeUnit.SECONDS, null, readHandler));
} else {
// async scattering read
ch1.read(dsts, 0, 2, 0, TimeUnit.SECONDS, null, readHandler);
// shared arena cannot be closed while read in progress
if (shared) {
assertThrows(IllegalStateException.class, arena::close);
}
// async write + finish write
ByteBuffer src = arena.allocate(SIZE).asByteBuffer();
fillRandom(src);
int nwritten = ch2.write(src).get();
assertTrue(nwritten > 0);
assertTrue(nwritten <= SIZE);
assertEquals(nwritten, src.position());
// finish scattering read
int nread = (int) (long) readHandler.join();
assertTrue(nread > 0);
assertTrue(nread <= nwritten);
assertTrue(dst1.position() > 0);
assertEquals(nread, dst1.position() + dst2.position());
// check contents
src.flip();
assertEquals(src, dst.slice(0, nread));
}
}
} finally {
tryClose(arena);
}
}
/**
* FileChannel read/write(ByteBuffer).
*/
@ -467,6 +650,129 @@ class MemorySegments {
}
}
/**
* AsynchronousFileChannel read/write(ByteBuffer).
*/
@DisabledOnOs(OS.WINDOWS)
@ParameterizedTest
@MethodSource("arenaSuppliers")
void testAsyncFileChannelReadWrite(Supplier<Arena> arenaSupplier) throws Throwable {
boolean confined = isConfined(arenaSupplier);
Arena arena = arenaSupplier.get();
Path file = Files.createTempFile(Path.of(""), "foo", ".dat");
try (AsynchronousFileChannel ch = AsynchronousFileChannel.open(file, READ, WRITE)) {
ByteBuffer dst = arena.allocate(SIZE + 100).asByteBuffer();
ByteBuffer src = arena.allocate(SIZE).asByteBuffer();
fillRandom(src);
if (confined) {
// read and write should fail with IAE
assertThrows(IllegalArgumentException.class, () -> ch.read(dst, 0L));
assertThrows(IllegalArgumentException.class, () -> ch.write(src, 0L));
} else {
// async write
Future<Integer> writeTask = ch.write(src, 0L);
// finish write
int nwritten = writeTask.get();
assertTrue(nwritten > 0);
assertTrue(nwritten <= SIZE);
assertEquals(nwritten, src.position());
// async read
Future<Integer> readTask = ch.read(dst, 0L);
// finish read
int nread = readTask.get();
assertTrue(nread > 0);
assertTrue(nread <= nwritten);
assertEquals(nread, dst.position());
// check contents
dst.flip();
assertEquals(src.slice(0, nread), dst);
}
} finally {
tryClose(arena);
}
}
/**
* Test closing a shared arena while AsynchronousFileChannel.write in progress.
*/
@Disabled
@RepeatedTest(20)
void testAsyncFileChannelWriteRacingArenaClose() throws Exception {
Path file = Files.createTempFile(Path.of(""), "foo", ".dat");
// use SYNC option to cause write operation to be slow
try (AsynchronousFileChannel ch = AsynchronousFileChannel.open(file, READ, WRITE, SYNC)) {
Arena arena = Arena.ofShared();
boolean closed = false;
try {
ByteBuffer src = arena.allocate(SIZE).asByteBuffer();
fillRandom(src);
// need copy of source buffer so that writing can be tested after arena is closed
ByteBuffer srcCopy = copyOf(src);
// async write
Future<Integer> writeTask = ch.write(src, 0L);
// attempt to close arena, races with write operation
try {
arena.close();
closed = true;
} catch (IllegalStateException e) {
// in use
}
// finish write
int nwritten = writeTask.get();
assertTrue(nwritten > 0);
assertTrue(nwritten <= SIZE);
// read and check contents
ByteBuffer dst = ByteBuffer.allocate(SIZE + 100);
int nread = ch.read(dst, 0L).get();
dst.flip();
assertEquals(srcCopy.slice(0, nread), dst);
} finally {
if (!closed) {
arena.close();
}
}
}
}
/**
* CompletionHandler with a join method to wait for operation to complete.
*/
private static class Handler<V> implements CompletionHandler<V, Void> {
volatile V result;
volatile Throwable ex;
final CountDownLatch latch = new CountDownLatch(1);
@Override
public void completed(V result, Void att) {
this.result = result;
latch.countDown();
}
@Override
public void failed(Throwable ex, Void att) {
this.ex = ex;
latch.countDown();
}
V join() throws ExecutionException, InterruptedException {
latch.await();
Throwable ex = this.ex;
if (ex != null) {
throw new ExecutionException(ex);
}
return result;
}
}
/**
* Fill the buffer with random bytes.
*/
@ -479,6 +785,42 @@ class MemorySegments {
bb.position(pos);
}
/**
* Return a copy of a buffer.
*/
private ByteBuffer copyOf(ByteBuffer buf) {
ByteBuffer copy = ByteBuffer.allocate(buf.capacity());
buf.put(copy);
buf.flip();
copy.flip();
assertEquals(buf, copy);
return copy;
}
/**
* Returns true if the supplier produces shared arenas.
*/
private boolean isShared(Supplier<Arena> arenaSupplier) {
if (!isConfined(arenaSupplier)) {
try {
arenaSupplier.get().close();
return true;
} catch (UnsupportedOperationException e) { }
}
// confined or non-closeable
return false;
}
/**
* Returns true if the supplier produces thread-confined arenas.
*/
private boolean isConfined(Supplier<Arena> arenaSupplier) {
Arena arena = arenaSupplier.get();
boolean confined = !arena.allocate(0).isAccessibleBy(new Thread());
tryClose(arena);
return confined;
}
/**
* Attempt to close the given Arena.
*/
@ -490,5 +832,4 @@ class MemorySegments {
return false;
}
}
}