1 /*
  2  * Copyright (c) 2021, Oracle and/or its affiliates. All rights reserved.
  3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  4  *
  5  * This code is free software; you can redistribute it and/or modify it
  6  * under the terms of the GNU General Public License version 2 only, as
  7  * published by the Free Software Foundation.
  8  *
  9  * This code is distributed in the hope that it will be useful, but WITHOUT
 10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 12  * version 2 for more details (a copy is included in the LICENSE file that
 13  * accompanied this code).
 14  *
 15  * You should have received a copy of the GNU General Public License version
 16  * 2 along with this work; if not, write to the Free Software Foundation,
 17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 18  *
 19  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 20  * or visit www.oracle.com if you need additional information or have any
 21  * questions.
 22  */
 23 
 24 /*
 25  * @test
 26  * @library /test/lib
 27  * @modules java.base/sun.nio.ch
 28  *          jdk.incubator.foreign/jdk.internal.foreign
 29  * @key randomness
 30  * @run testng/othervm TestAsyncSocketChannels
 31  * @run testng/othervm -Dsun.nio.ch.disableSynchronousRead=true TestAsyncSocketChannels
 32  * @run testng/othervm -Dsun.nio.ch.disableSynchronousRead=false TestAsyncSocketChannels
 33  */
 34 
 35 import java.io.IOException;
 36 import java.net.InetAddress;
 37 import java.net.InetSocketAddress;
 38 import java.net.StandardSocketOptions;
 39 import java.nio.ByteBuffer;
 40 import java.nio.channels.AsynchronousServerSocketChannel;
 41 import java.nio.channels.AsynchronousSocketChannel;
 42 import java.nio.channels.CompletionHandler;
 43 import java.util.List;
 44 import java.util.concurrent.CountDownLatch;
 45 import java.util.concurrent.ExecutionException;
 46 import java.util.concurrent.Future;
 47 import java.util.concurrent.atomic.AtomicBoolean;
 48 import java.util.concurrent.atomic.AtomicInteger;
 49 import java.util.concurrent.atomic.AtomicLong;
 50 import java.util.function.Supplier;
 51 
 52 import jdk.incubator.foreign.MemorySegment;
 53 import jdk.incubator.foreign.ResourceScope;
 54 import org.testng.annotations.*;
 55 import static java.lang.System.out;
 56 import static java.util.concurrent.TimeUnit.SECONDS;
 57 import static jdk.incubator.foreign.ValueLayout.JAVA_BYTE;
 58 import static org.testng.Assert.*;
 59 
 60 /**
 61  * Tests consisting of buffer views with asynchronous NIO network channels.
 62  */
 63 public class TestAsyncSocketChannels extends AbstractChannelsTest {
 64 
 65     static final Class<IOException> IOE = IOException.class;
 66     static final Class<ExecutionException> EE = ExecutionException.class;
 67     static final Class<IllegalStateException> ISE = IllegalStateException.class;
 68 
 69     /** Tests that confined scopes are not supported. */
 70     @Test(dataProvider = "confinedScopes")
 71     public void testWithConfined(Supplier<ResourceScope> scopeSupplier)
 72         throws Throwable
 73     {
 74         try (var channel = AsynchronousSocketChannel.open();
 75              var server = AsynchronousServerSocketChannel.open();
 76              var connectedChannel = connectChannels(server, channel);
 77              var scope = scopeSupplier.get()) {
 78             var segment = MemorySegment.allocateNative(10, 1, scope);
 79             var bb = segment.asByteBuffer();
 80             var bba = new ByteBuffer[] { bb };
 81             List<ThrowingConsumer<TestHandler,?>> ioOps = List.of(
 82                     handler -> handler.propagateHandlerFromFuture(channel.write(bb)),
 83                     handler -> handler.propagateHandlerFromFuture(channel.read(bb)),
 84                     handler -> channel.write(bb, null, handler),
 85                     handler -> channel.read( bb, null, handler),
 86                     handler -> channel.write(bb , 0L, SECONDS, null, handler),
 87                     handler -> channel.read( bb,  0L, SECONDS, null, handler),
 88                     handler -> channel.write(bba, 0, bba.length, 0L, SECONDS, null, handler),
 89                     handler -> channel.read( bba, 0, bba.length, 0L, SECONDS, null, handler)
 90             );
 91             for (var ioOp : ioOps) {
 92                 out.println("testAsyncWithConfined - op");
 93                 var handler = new TestHandler();
 94                 ioOp.accept(handler);
 95                 handler.await()
 96                         .assertFailedWith(ISE)
 97                         .assertExceptionMessage("Confined scope not supported");
 98             }
 99         }
100     }
101 
102     /** Tests that I/O with a closed scope throws a suitable exception. */
103     @Test(dataProvider = "sharedScopesAndTimeouts")
104     public void testIOWithClosedSharedScope(Supplier<ResourceScope> scopeSupplier, int timeout)
105         throws Exception
106     {
107         try (var channel = AsynchronousSocketChannel.open();
108              var server = AsynchronousServerSocketChannel.open();
109              var connectedChannel = connectChannels(server, channel)) {
110             ResourceScope scope = scopeSupplier.get();
111             ByteBuffer bb = segmentBufferOfSize(scope, 64);
112             ByteBuffer[] buffers = segmentBuffersOfSize(8, scope, 32);
113             scope.close();
114             {
115                 assertCauses(expectThrows(EE, () -> connectedChannel.read(bb).get()), IOE, ISE);
116             }
117             {
118                 var handler = new TestHandler<Integer>();
119                 connectedChannel.read(bb, null, handler);
120                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
121             }
122             {
123                 var handler = new TestHandler<Integer>();
124                 connectedChannel.read(bb, timeout, SECONDS, null, handler);
125                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
126             }
127             {
128                 var handler = new TestHandler<Long>();
129                 connectedChannel.read(buffers, 0, buffers.length, timeout, SECONDS, null, handler);
130                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
131             }
132             {
133                 assertCauses(expectThrows(EE, () -> connectedChannel.write(bb).get()), IOE, ISE);
134             }
135             {
136                 var handler = new TestHandler<Integer>();
137                 connectedChannel.write(bb, null, handler);
138                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
139             }
140             {
141                 var handler = new TestHandler<Integer>();
142                 connectedChannel.write(bb, timeout, SECONDS, null, handler);
143                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
144             }
145             {
146                 var handler = new TestHandler<Long>();
147                 connectedChannel.write(buffers, 0, buffers.length, timeout, SECONDS, null, handler);
148                 handler.await().assertFailedWith(ISE).assertExceptionMessage("Already closed");
149             }
150         }
151     }
152 
153     /** Tests basic I/O operations work with views over implicit and shared scopes. */
154     @Test(dataProvider = "sharedAndImplicitScopes")
155     public void testBasicIOWithSupportedScope(Supplier<ResourceScope> scopeSupplier)
156         throws Exception
157     {
158         ResourceScope scope;
159         try (var asc1 = AsynchronousSocketChannel.open();
160              var assc = AsynchronousServerSocketChannel.open();
161              var asc2 = connectChannels(assc, asc1);
162              var scp = closeableScopeOrNull(scope = scopeSupplier.get())) {
163             MemorySegment segment1 = MemorySegment.allocateNative(10, 1, scope);
164             MemorySegment segment2 = MemorySegment.allocateNative(10, 1, scope);
165             for (int i = 0; i < 10; i++) {
166                 segment1.set(JAVA_BYTE, i, (byte) i);
167             }
168             {   // Future variants
169                 ByteBuffer bb1 = segment1.asByteBuffer();
170                 ByteBuffer bb2 = segment2.asByteBuffer();
171                 assertEquals((int)asc1.write(bb1).get(), 10);
172                 assertEquals((int)asc2.read(bb2).get(), 10);
173                 assertEquals(bb2.flip(), ByteBuffer.wrap(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}));
174             }
175             {   // CompletionHandler variants
176                 ByteBuffer bb1 = segment1.asByteBuffer();
177                 ByteBuffer bb2 = segment2.asByteBuffer();
178                 var writeHandler = new TestHandler();
179                 asc1.write(new ByteBuffer[]{bb1}, 0, 1, 30L, SECONDS, null, writeHandler);
180                 writeHandler.await().assertCompleteWith(10L);
181                 var readHandler = new TestHandler();
182                 asc2.read(new ByteBuffer[]{bb2}, 0, 1, 30L, SECONDS, null, readHandler);
183                 readHandler.await().assertCompleteWith(10L);
184                 assertEquals(bb2.flip(), ByteBuffer.wrap(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}));
185             }
186             {   // Gathering/Scattering variants
187                 var writeBuffers = mixedBuffersOfSize(16, scope, 32);
188                 var readBuffers = mixedBuffersOfSize(16, scope, 32);
189                 long expectedCount = remaining(writeBuffers);
190                 var writeHandler = new TestHandler();
191                 asc1.write(writeBuffers, 0, 16, 30L, SECONDS, null, writeHandler);
192                 writeHandler.await().assertCompleteWith(expectedCount);
193                 var readHandler = new TestHandler();
194                 asc2.read(readBuffers, 0, 16, 30L, SECONDS, null, readHandler);
195                 readHandler.await().assertCompleteWith(expectedCount);
196                 assertEquals(flip(readBuffers), clear(writeBuffers));
197             }
198         }
199     }
200 
201     /** Tests that a scope is not closeable when there is an outstanding read operation. */
202     @Test(dataProvider = "sharedScopesAndTimeouts")
203     public void testCloseWithOutstandingRead(Supplier<ResourceScope> scopeSupplier, int timeout)
204         throws Throwable
205     {
206         try (var asc1 = AsynchronousSocketChannel.open();
207              var assc = AsynchronousServerSocketChannel.open();
208              var asc2 = connectChannels(assc, asc1);
209              var scope = scopeSupplier.get()) {
210             var segment = MemorySegment.allocateNative(10, 1, scope);
211             var bb = segment.asByteBuffer();
212             var bba = new ByteBuffer[] { bb };
213             List<ThrowingConsumer<TestHandler,?>> readOps = List.of(
214                     handler -> handler.propagateHandlerFromFuture(asc1.read(bb)),
215                     handler -> asc1.read(bb, null, handler),
216                     handler -> asc1.read(bb,  timeout, SECONDS, null, handler),
217                     handler -> asc1.read(bba, 0, bba.length, timeout, SECONDS, null, handler)
218             );
219             for (var ioOp : readOps) {
220                 out.println("testCloseWithOutstandingRead - op");
221                 var handler = new TestHandler<Long>();
222                 ioOp.accept(handler);
223                 assertFalse(handler.isDone());
224                 assertTrue(scope.isAlive());
225                 assertMessage(expectThrows(ISE, () -> scope.close()), "Scope is kept alive by");
226 
227                 // write to allow the blocking read complete, which will
228                 // in turn unlock the scope and allow it to be closed.
229                 asc2.write(ByteBuffer.wrap(new byte[] { 0x01 })).get();
230                 handler.await().assertCompleteWith(1L);
231                 assertTrue(scope.isAlive());
232             }
233         }
234     }
235 
236     /** Tests that a scope is not closeable when there is an outstanding write operation. */
237     // Note: limited scenarios are checked, given the 5 sec sleep!
238     @Test(dataProvider = "sharedScopesAndTimeouts")
239     public void testCloseWithOutstandingWrite(Supplier<ResourceScope> scopeSupplier, int timeout)
240          throws Throwable
241     {
242         try (var asc1 = AsynchronousSocketChannel.open();
243              var assc = AsynchronousServerSocketChannel.open();
244              var asc2 = connectChannels(assc, asc1);
245              var scope = scopeSupplier.get()) {
246 
247             // number of bytes written
248             final AtomicLong bytesWritten = new AtomicLong(0);
249             // set to true to signal that no more buffers should be written
250             final AtomicBoolean continueWriting = new AtomicBoolean(true);
251             final AtomicInteger outstandingWriteOps = new AtomicInteger(0);
252 
253             // write until socket buffer is full so as to create the conditions
254             // for when a write does not complete immediately
255             var bba = segmentBuffersOfSize(32, scope, 128);
256             TestHandler<Long> handler;
257             outstandingWriteOps.getAndIncrement();
258             asc1.write(bba, 0, bba.length, timeout, SECONDS, null,
259                     (handler = new TestHandler<>() {
260                         public void completed(Long result, Void att) {
261                             super.completed(result, att);
262                             bytesWritten.addAndGet(result);
263                             if (continueWriting.get()) {
264                                 var bba = segmentBuffersOfSize(32, scope, 128);
265                                 outstandingWriteOps.getAndIncrement();
266                                 asc1.write(bba, 0, bba.length, timeout, SECONDS, null, this);
267                             }
268                             outstandingWriteOps.getAndDecrement();
269                         }
270                     }));
271             // give time for socket buffer to fill up.
272             awaitNoFurtherWrites(bytesWritten);
273 
274             assertMessage(expectThrows(ISE, () -> scope.close()), "Scope is kept alive by");
275             assertTrue(scope.isAlive());
276 
277             // signal handler to stop further writing
278             continueWriting.set(false);
279 
280             // read to allow the outstanding write complete, which will
281             // in turn unlock the scope and allow it to be closed.
282             readNBytes(asc2, bytesWritten.get());
283             assertTrue(scope.isAlive());
284             awaitOutstandingWrites(outstandingWriteOps);
285             handler.await();
286         }
287     }
288 
289     /** Waits for outstandingWriteOps to complete (become 0). */
290     static void awaitOutstandingWrites(AtomicInteger outstandingWriteOps) {
291         boolean initial = true;
292         while (outstandingWriteOps.get() > 0 )  {
293             if (initial) {
294                 out.print("awaiting outstanding writes");
295                 initial = false;
296             }
297             out.print(".");
298             Thread.onSpinWait();
299         }
300         out.println("outstanding writes: " + outstandingWriteOps.get());
301     }
302 
303     /** Waits, at most 20secs, for bytesWritten to stabilize. */
304     static void awaitNoFurtherWrites(AtomicLong bytesWritten) throws Exception {
305         int i;
306         long prevN = 0;
307         for (i=0; i<10; i++) {
308             long n = bytesWritten.get();
309             Thread.sleep(2 * 1000);
310             if (bytesWritten.get() == n && prevN == n) {
311                 break;
312             }
313             prevN = n;
314         }
315         out.println("awaitNoFurtherWrites: i=" + i +" , bytesWritten=" + bytesWritten.get());
316     }
317 
318     /** Completion handler that exposes conveniences to assert results. */
319     static class TestHandler<V extends Number> implements CompletionHandler<V, Void> {
320         volatile V result;
321         volatile Throwable throwable;
322         final CountDownLatch latch = new CountDownLatch(1);
323 
324         /** Starts a thread that complete the handled with the Future result. */
325         TestHandler propagateHandlerFromFuture(Future<Integer> future) {
326             Runnable runnable = () -> {
327                 try {
328                     this.completed((V)future.get(), null);
329                 } catch (Throwable t) {
330                     // assert and unwrap exception added by Future
331                     assertTrue(ExecutionException.class.isInstance(t));
332                     t = t.getCause();
333                     assertTrue(IOException.class.isInstance(t));
334                     t = t.getCause();
335                     this.failed(t, null);
336                 }
337             };
338             Thread t = new Thread(runnable);
339             t.start();
340             return this;
341         }
342 
343         @Override
344         public void completed(V result, Void att) {
345             assert result.longValue() >= 0;
346             this.result = result;
347             latch.countDown();
348         }
349         @Override
350         public void failed(Throwable exc, Void att){
351             this.throwable = tolerateIOEOnWindows(exc);
352             latch.countDown();
353         }
354 
355         TestHandler await() throws InterruptedException{
356             latch.await();
357             return this;
358         }
359 
360         TestHandler assertCompleteWith(V value) {
361             assertEquals(result.longValue(), value.longValue());
362             assertEquals(throwable, null);
363             return this;
364         }
365 
366         TestHandler assertFailedWith(Class<? extends Exception> expectedException) {
367             assertTrue(expectedException.isInstance(throwable),
368                        "Expected type:%s, got:%s".formatted(expectedException, throwable) );
369             assertEquals(result, null, "Unexpected result: " + result);
370             return this;
371         }
372 
373         TestHandler assertExceptionMessage(String expectedMessage) {
374             assertEquals(throwable.getMessage(), expectedMessage);
375             return this;
376         }
377 
378         boolean isDone() {
379             return latch.getCount() == 0;
380         }
381     }
382 
383     static AsynchronousSocketChannel connectChannels(AsynchronousServerSocketChannel assc,
384                                                      AsynchronousSocketChannel asc)
385         throws Exception
386     {
387         setBufferSized(assc, asc);
388         assc.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
389         asc.connect(assc.getLocalAddress()).get();
390         return assc.accept().get();
391     }
392 
393     /** Sets the send/receive buffer sizes in an attempt/hint to limit the
394      * accepted/connected socket buffer sizes. Actual buffer sizes in use will
395      * likely be larger due to TCP auto-tuning, but the hint typically reduces
396      * the overall scaled sizes. This is primarily to stabilize outstanding
397      * write operations.
398      */
399     static void setBufferSized(AsynchronousServerSocketChannel assc,
400                                AsynchronousSocketChannel asc)
401         throws Exception
402     {
403         assc.setOption(StandardSocketOptions.SO_RCVBUF, 32 * 1024);
404         asc.setOption(StandardSocketOptions.SO_SNDBUF, 32 * 1024);
405         asc.setOption(StandardSocketOptions.SO_RCVBUF, 32 * 1024);
406     }
407 
408     /** Tolerate the additional level of IOException wrapping of unchecked exceptions
409      * On Windows, when completing the completion handler with a failure. */
410     static Throwable tolerateIOEOnWindows(Throwable t) {
411         if (System.getProperty("os.name").startsWith("Windows")) {
412             if (t instanceof IOException)
413                 return t.getCause();
414         }
415         return t;
416     }
417 
418     static void readNBytes(AsynchronousSocketChannel channel, long len)
419         throws Exception
420     {
421         var buf = ByteBuffer.allocateDirect(4096);
422         long total = 0L;
423         do {
424             int n = channel.read(buf).get();
425             assertTrue(n > 0, "got:" + n);
426             buf.clear();
427             total += n;
428         } while (total < len);
429     }
430 }