diff --git a/src/java.base/share/classes/java/lang/foreign/MemorySegment.java b/src/java.base/share/classes/java/lang/foreign/MemorySegment.java index 3d85e4c32cc..196f44d1abe 100644 --- a/src/java.base/share/classes/java/lang/foreign/MemorySegment.java +++ b/src/java.base/share/classes/java/lang/foreign/MemorySegment.java @@ -1120,9 +1120,7 @@ public sealed interface MemorySegment permits AbstractMemorySegmentImpl { *

* 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. *

* Finally, the resulting buffer's byte order is * {@link java.nio.ByteOrder#BIG_ENDIAN}; this can be changed using diff --git a/src/java.base/share/classes/java/nio/channels/AsynchronousByteChannel.java b/src/java.base/share/classes/java/nio/channels/AsynchronousByteChannel.java index 56003d53ec7..c65e836c99a 100644 --- a/src/java.base/share/classes/java/nio/channels/AsynchronousByteChannel.java +++ b/src/java.base/share/classes/java/nio/channels/AsynchronousByteChannel.java @@ -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 diff --git a/src/java.base/share/classes/java/nio/channels/AsynchronousFileChannel.java b/src/java.base/share/classes/java/nio/channels/AsynchronousFileChannel.java index 15af714accf..265ad80a443 100644 --- a/src/java.base/share/classes/java/nio/channels/AsynchronousFileChannel.java +++ b/src/java.base/share/classes/java/nio/channels/AsynchronousFileChannel.java @@ -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 */ diff --git a/src/java.base/share/classes/java/nio/channels/AsynchronousSocketChannel.java b/src/java.base/share/classes/java/nio/channels/AsynchronousSocketChannel.java index 736327a1825..5c65048e091 100644 --- a/src/java.base/share/classes/java/nio/channels/AsynchronousSocketChannel.java +++ b/src/java.base/share/classes/java/nio/channels/AsynchronousSocketChannel.java @@ -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 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 diff --git a/src/java.base/share/classes/sun/nio/ch/AsynchronousFileChannelImpl.java b/src/java.base/share/classes/sun/nio/ch/AsynchronousFileChannelImpl.java index 2ca1f8a87b8..28d067ebc74 100644 --- a/src/java.base/share/classes/sun/nio/ch/AsynchronousFileChannelImpl.java +++ b/src/java.base/share/classes/sun/nio/ch/AsynchronousFileChannelImpl.java @@ -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 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 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); } } diff --git a/src/java.base/share/classes/sun/nio/ch/AsynchronousSocketChannelImpl.java b/src/java.base/share/classes/sun/nio/ch/AsynchronousSocketChannelImpl.java index c53aca8b1c7..68983304735 100644 --- a/src/java.base/share/classes/sun/nio/ch/AsynchronousSocketChannelImpl.java +++ b/src/java.base/share/classes/sun/nio/ch/AsynchronousSocketChannelImpl.java @@ -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 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 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); } diff --git a/src/java.base/share/classes/sun/nio/ch/IOUtil.java b/src/java.base/share/classes/sun/nio/ch/IOUtil.java index ad1aedc2668..45f8cb2e588 100644 --- a/src/java.base/share/classes/sun/nio/ch/IOUtil.java +++ b/src/java.base/share/classes/sun/nio/ch/IOUtil.java @@ -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); } diff --git a/test/jdk/java/foreign/channels/TestAsyncSocketChannels.java b/test/jdk/java/foreign/channels/TestAsyncSocketChannels.java index 35d938ec1d3..37fef8bb4b6 100644 --- a/test/jdk/java/foreign/channels/TestAsyncSocketChannels.java +++ b/test/jdk/java/foreign/channels/TestAsyncSocketChannels.java @@ -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 IOE = IOException.class; static final Class EE = ExecutionException.class; + static final Class IAE = IllegalArgumentException.class; static final Class 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> 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)); } } } diff --git a/test/jdk/java/nio/channels/etc/MemorySegments.java b/test/jdk/java/nio/channels/etc/MemorySegments.java index 77da11759c7..2d82d855fa2 100644 --- a/test/jdk/java/nio/channels/etc/MemorySegments.java +++ b/test/jdk/java/nio/channels/etc/MemorySegments.java @@ -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 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 readTask = ch1.read(dst); + + // shared arena cannot be closed while read in progress + if (shared) { + assertThrows(IllegalStateException.class, arena::close); + } + + // async write + Future 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 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(); + 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 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(); + + 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 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 writeTask = ch.write(src, 0L); + + // finish write + int nwritten = writeTask.get(); + assertTrue(nwritten > 0); + assertTrue(nwritten <= SIZE); + assertEquals(nwritten, src.position()); + + // async read + Future 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 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 implements CompletionHandler { + 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 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 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; } } - }