1 /*
  2  * Copyright (c) 2019, 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  * @bug 8236246
 27  * @modules java.base/sun.nio.ch
 28  * @run testng InterruptibleOrNot
 29  * @summary Test SelectorProviderImpl.openDatagramChannel(boolean) to create
 30  *     DatagramChannel objects that optionally support interrupt
 31  */
 32 
 33 import java.io.Closeable;
 34 import java.io.IOException;
 35 import java.net.InetAddress;
 36 import java.net.InetSocketAddress;
 37 import java.net.SocketAddress;
 38 import java.nio.ByteBuffer;
 39 import java.nio.channels.AsynchronousCloseException;
 40 import java.nio.channels.ClosedByInterruptException;
 41 import java.nio.channels.DatagramChannel;
 42 import java.time.Duration;
 43 import java.util.concurrent.Executors;
 44 import java.util.concurrent.Future;
 45 import java.util.concurrent.ScheduledExecutorService;
 46 import java.util.concurrent.TimeUnit;
 47 import sun.nio.ch.DefaultSelectorProvider;
 48 
 49 import org.testng.annotations.Test;
 50 import static org.testng.Assert.*;
 51 
 52 @Test
 53 public class InterruptibleOrNot {
 54 
 55     public void testInterruptBeforeInterruptibleReceive() throws Exception {
 56         testInterruptBeforeReceive(true);
 57     }
 58 
 59     public void testInterruptDuringInterruptibleReceive() throws Exception {
 60         testInterruptDuringReceive(true);
 61     }
 62 
 63     public void testInterruptBeforeUninterruptibleReceive() throws Exception {
 64         testInterruptBeforeReceive(false);
 65     }
 66 
 67     public void testInterruptDuringUninterruptibleReceive() throws Exception {
 68         testInterruptDuringReceive(false);
 69     }
 70 
 71     public void testInterruptBeforeInterruptibleSend() throws Exception {
 72         testInterruptBeforeSend(true);
 73     }
 74 
 75     public void testInterruptBeforeUninterruptibleSend() throws Exception {
 76         testInterruptBeforeSend(false);
 77     }
 78 
 79     /**
 80      * Test invoking DatagramChannel receive with interrupt status set
 81      */
 82     static void testInterruptBeforeReceive(boolean interruptible)
 83         throws Exception
 84     {
 85         try (DatagramChannel dc = openDatagramChannel(interruptible)) {
 86             dc.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
 87             Future<?> timeout = scheduleClose(dc, Duration.ofSeconds(2));
 88             try {
 89                 ByteBuffer buf = ByteBuffer.allocate(100);
 90                 Thread.currentThread().interrupt();
 91                 assertThrows(expectedException(interruptible), () -> dc.receive(buf));
 92             } finally {
 93                 timeout.cancel(false);
 94             }
 95         } finally {
 96             Thread.interrupted();  // clear interrupt
 97         }
 98     }
 99 
100     /**
101      * Test Thread.interrupt when target thread is blocked in DatagramChannel receive
102      */
103     static void testInterruptDuringReceive(boolean interruptible)
104         throws Exception
105     {
106         try (DatagramChannel dc = openDatagramChannel(interruptible)) {
107             dc.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
108             Future<?> timerTask = scheduleClose(dc, Duration.ofSeconds(5));
109             Future<?> interruptTask = scheduleInterrupt(Thread.currentThread(), Duration.ofSeconds(1));
110             try {
111                 ByteBuffer buf = ByteBuffer.allocate(100);
112                 assertThrows(expectedException(interruptible), () -> dc.receive(buf));
113             } finally {
114                 timerTask.cancel(false);
115                 interruptTask.cancel(false);
116             }
117         } finally {
118             Thread.interrupted();  // clear interrupt
119         }
120     }
121 
122     /**
123      * Test invoking DatagramChannel send with interrupt status set
124      */
125     static void testInterruptBeforeSend(boolean interruptible)
126         throws Exception
127     {
128         try (DatagramChannel dc = openDatagramChannel(interruptible)) {
129             dc.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
130             Future<?> timeout = scheduleClose(dc, Duration.ofSeconds(2));
131             try {
132                 ByteBuffer buf = ByteBuffer.allocate(100);
133                 SocketAddress target = dc.getLocalAddress();
134                 Thread.currentThread().interrupt();
135                 if (interruptible) {
136                     assertThrows(ClosedByInterruptException.class, () -> dc.send(buf, target));
137                 } else {
138                     int n = dc.send(buf, target);
139                     assertTrue(n == 100);
140                 }
141             } finally {
142                 timeout.cancel(false);
143             }
144         } finally {
145             Thread.interrupted();  // clear interrupt
146         }
147     }
148 
149     /**
150      * Creates a DatagramChannel that is interruptible or not.
151      */
152     static DatagramChannel openDatagramChannel(boolean interruptible) throws IOException {
153         if (interruptible) {
154             return DatagramChannel.open();
155         } else {
156             return DefaultSelectorProvider.get().openUninterruptibleDatagramChannel();
157         }
158     }
159 
160     /**
161      * Expect ClosedByInterruptException if interruptible.
162      */
163     static Class<? extends Exception> expectedException(boolean expectInterrupt) {
164         if (expectInterrupt) {
165             return ClosedByInterruptException.class;
166         } else {
167             return AsynchronousCloseException.class;
168         }
169     }
170 
171     /**
172      * Schedule the given object to be closed.
173      */
174     static Future<?> scheduleClose(Closeable c, Duration timeout) {
175         long nanos = TimeUnit.NANOSECONDS.convert(timeout);
176         return STPE.schedule(() -> {
177             c.close();
178             return null;
179         }, nanos, TimeUnit.NANOSECONDS);
180     }
181 
182     /**
183      * Schedule the given thread to be interrupted.
184      */
185     static Future<?> scheduleInterrupt(Thread t, Duration timeout) {
186         long nanos = TimeUnit.NANOSECONDS.convert(timeout);
187         return STPE.schedule(t::interrupt, nanos, TimeUnit.NANOSECONDS);
188     }
189 
190     static final ScheduledExecutorService STPE = Executors.newScheduledThreadPool(0);
191 }