8264200: java/nio/channels/DatagramChannel/SRTest.java fails intermittently

Reviewed-by: dfuchs, rriggs, msheppar

Author: c-cleary

test/jdk/java/nio/channels/DatagramChannel/SRTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2001, 2021, 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,185 +23,216 @@
 
 /* @test
  * @summary Test DatagramChannel's send and receive methods
- * @author Mike McCloskey
+ * @run testng/othervm/timeout=20 SRTest
  */
 
 import java.io.*;
 import java.net.*;
 import java.nio.*;
 import java.nio.channels.*;
-import java.nio.charset.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Stream;
+import static java.nio.charset.StandardCharsets.US_ASCII;
 
+import org.testng.annotations.*;
 
 public class SRTest {
 
+    ExecutorService executorService;
     static PrintStream log = System.err;
 
-    public static void main(String[] args) throws Exception {
-        test();
+    static final String DATA_STRING = "hello";
+
+    @BeforeClass
+    public void beforeClass() {
+        executorService = Executors.newCachedThreadPool();
     }
 
-    static void test() throws Exception {
-        ClassicReader classicReader;
-        NioReader nioReader;
+    @AfterClass
+    public void afterClass() {
+        executorService.shutdown();
+    }
 
-        classicReader = new ClassicReader();
-        invoke(classicReader, new ClassicWriter(classicReader.port()));
+    @Test
+    public void classicReaderClassicWriter() throws Exception {
+        try (ClassicReader cr = new ClassicReader();
+             ClassicWriter cw = new ClassicWriter(cr.port())) {
+            invoke(executorService, cr, cw);
+        }
         log.println("Classic RW: OK");
+    }
 
-        classicReader = new ClassicReader();
-        invoke(classicReader, new NioWriter(classicReader.port()));
+    @Test
+    public void classicReaderNioWriter() throws Exception {
+        try (ClassicReader cr = new ClassicReader();
+             NioWriter nw = new NioWriter(cr.port())) {
+            invoke(executorService, cr, nw);
+        }
         log.println("Classic R, Nio W: OK");
+    }
 
-        nioReader = new NioReader();
-        invoke(nioReader, new ClassicWriter(nioReader.port()));
+    @Test
+    public void nioReaderClassicWriter() throws Exception {
+        try (NioReader nr = new NioReader();
+             ClassicWriter cw = new ClassicWriter(nr.port())) {
+            invoke(executorService, nr, cw);
+        }
         log.println("Classic W, Nio R: OK");
+    }
 
-        nioReader = new NioReader();
-        invoke(nioReader, new NioWriter(nioReader.port()));
+    @Test
+    public void nioReaderNioWriter() throws Exception {
+        try (NioReader nr = new NioReader();
+             NioWriter nw = new NioWriter(nr.port())) {
+            invoke(executorService, nr, nw);
+        }
         log.println("Nio RW: OK");
     }
 
-    static void invoke(Sprintable reader, Sprintable writer) throws Exception {
-        Thread readerThread = new Thread(reader);
-        readerThread.start();
-        Thread.sleep(50);
-
-        Thread writerThread = new Thread(writer);
-        writerThread.start();
-
-        writerThread.join();
-        readerThread.join();
-
-        reader.throwException();
-        writer.throwException();
+    private static void invoke(ExecutorService e, Runnable reader, Runnable writer) {
+        CompletableFuture<Void> f1 = CompletableFuture.runAsync(writer, e);
+        CompletableFuture<Void> f2 = CompletableFuture.runAsync(reader, e);
+        wait(f1, f2);
     }
 
-    public interface Sprintable extends Runnable {
-        public void throwException() throws Exception;
+    // Exit with CompletionException if any passed futures complete exceptionally
+    private static void wait(CompletableFuture<?>... futures) throws CompletionException {
+        CompletableFuture<?> future = CompletableFuture.allOf(futures);
+        Stream.of(futures)
+                .forEach(f -> f.exceptionally(ex -> {
+                    future.completeExceptionally(ex);
+                    return null;
+                }));
+        future.join();
     }
 
-    public static class ClassicWriter implements Sprintable {
-        final int port;
-        Exception e = null;
-
-        ClassicWriter(int port) {
-            this.port = port;
-        }
+    public static class ClassicWriter implements Runnable, AutoCloseable {
+        final DatagramSocket ds;
+        final int dstPort;
 
-        public void throwException() throws Exception {
-            if (e != null)
-                throw e;
+        ClassicWriter(int dstPort) throws SocketException {
+            this.dstPort = dstPort;
+            this.ds = new DatagramSocket();
         }
 
         public void run() {
             try {
-                DatagramSocket ds = new DatagramSocket();
-                String dataString = "hello";
-                byte[] data = dataString.getBytes();
-                InetAddress address = InetAddress.getLocalHost();
+                byte[] data = DATA_STRING.getBytes(US_ASCII);
+                InetAddress address = InetAddress.getLoopbackAddress();
                 DatagramPacket dp = new DatagramPacket(data, data.length,
-                                                       address, port);
+                                                       address, dstPort);
                 ds.send(dp);
-                Thread.sleep(50);
-                ds.send(dp);
-            } catch (Exception ex) {
-                e = ex;
+            } catch (Exception e) {
+                log.println("ClassicWriter [" + ds.getLocalAddress() + "]");
+                throw new RuntimeException("ClassicWriter threw exception: " + e);
+            } finally {
+                log.println("ClassicWriter finished");
             }
         }
-    }
-
-    public static class NioWriter implements Sprintable {
-        final int port;
-        Exception e = null;
 
-        NioWriter(int port) {
-            this.port = port;
+        @Override
+        public void close() throws IOException {
+            ds.close();
         }
+    }
+
+    public static class NioWriter implements Runnable, AutoCloseable {
+        final DatagramChannel dc;
+        final int dstPort;
 
-        public void throwException() throws Exception {
-            if (e != null)
-                throw e;
+        NioWriter(int dstPort) throws IOException {
+            this.dc = DatagramChannel.open();
+            this.dstPort = dstPort;
         }
 
         public void run() {
             try {
-                DatagramChannel dc = DatagramChannel.open();
                 ByteBuffer bb = ByteBuffer.allocateDirect(256);
-                bb.put("hello".getBytes());
+                bb.put(DATA_STRING.getBytes(US_ASCII));
                 bb.flip();
-                InetAddress address = InetAddress.getLocalHost();
-                InetSocketAddress isa = new InetSocketAddress(address, port);
-                dc.send(bb, isa);
-                Thread.sleep(50);
+                InetAddress address = InetAddress.getLoopbackAddress();
+                InetSocketAddress isa = new InetSocketAddress(address, dstPort);
                 dc.send(bb, isa);
             } catch (Exception ex) {
-                e = ex;
+                log.println("NioWriter [" + dc.socket().getLocalAddress() + "]");
+                throw new RuntimeException("NioWriter threw exception: " + ex);
+            } finally {
+                log.println("NioWriter finished");
             }
         }
+
+        @Override
+        public void close() throws IOException {
+            dc.close();
+        }
     }
 
-    public static class ClassicReader implements Sprintable {
+    public static class ClassicReader implements Runnable, AutoCloseable {
         final DatagramSocket ds;
-        Exception e = null;
 
         ClassicReader() throws IOException {
-            this.ds = new DatagramSocket();
+            InetSocketAddress address = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0);
+            this.ds = new DatagramSocket(address);
         }
 
         int port() {
             return ds.getLocalPort();
         }
 
-        public void throwException() throws Exception {
-            if (e != null)
-                throw e;
-        }
-
         public void run() {
             try {
                 byte[] buf = new byte[256];
                 DatagramPacket dp = new DatagramPacket(buf, buf.length);
                 ds.receive(dp);
-                String received = new String(dp.getData());
-                log.println(received);
-                ds.close();
+                String received = new String(dp.getData(), dp.getOffset(), dp.getLength(), US_ASCII);
+                log.println("ClassicReader received: " + received);
             } catch (Exception ex) {
-                e = ex;
+                log.println("ClassicReader [" + ds.getLocalAddress() +"]");
+                throw new RuntimeException("ClassicReader threw exception: " + ex);
+            } finally {
+                log.println("ClassicReader finished");
             }
         }
+
+        @Override
+        public void close() throws IOException {
+            ds.close();
+        }
     }
 
-    public static class NioReader implements Sprintable {
+    public static class NioReader implements Runnable, AutoCloseable {
         final DatagramChannel dc;
-        Exception e = null;
 
         NioReader() throws IOException {
-            this.dc = DatagramChannel.open().bind(new InetSocketAddress(0));
+            InetSocketAddress address = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0);
+            this.dc = DatagramChannel.open().bind(address);
         }
 
         int port() {
             return dc.socket().getLocalPort();
         }
 
-        public void throwException() throws Exception {
-            if (e != null)
-                throw e;
-        }
-
         public void run() {
             try {
                 ByteBuffer bb = ByteBuffer.allocateDirect(100);
-                SocketAddress sa = dc.receive(bb);
+                dc.receive(bb);
                 bb.flip();
-                CharBuffer cb = Charset.forName("US-ASCII").
-                    newDecoder().decode(bb);
-                log.println("From: "+sa+ " said " +cb);
-                dc.close();
+                CharBuffer cb = US_ASCII.newDecoder().decode(bb);
+                log.println("NioReader received: " + cb);
             } catch (Exception ex) {
-                e = ex;
+                log.println("NioReader [" + dc.socket().getLocalAddress() +"]");
+                throw new RuntimeException("NioReader threw exception: " + ex);
+            } finally {
+                log.println("NioReader finished");
             }
         }
-    }
 
+        @Override
+        public void close() throws IOException {
+            dc.close();
+        }
+    }
 }