1 /*
  2  * Copyright (c) 2020, 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  * @modules jdk.incubator.foreign java.base/jdk.internal.vm.annotation java.base/jdk.internal.misc
 27  * @key randomness
 28  * @run testng/othervm TestHandshake
 29  * @run testng/othervm -Xint TestHandshake
 30  * @run testng/othervm -XX:TieredStopAtLevel=1 TestHandshake
 31  * @run testng/othervm -XX:-TieredCompilation TestHandshake
 32  */
 33 
 34 import jdk.incubator.foreign.MemorySegment;
 35 
 36 import java.lang.invoke.MethodHandles;
 37 import java.lang.invoke.VarHandle;
 38 import java.nio.ByteBuffer;
 39 import java.nio.ByteOrder;
 40 import java.util.concurrent.ExecutorService;
 41 import java.util.concurrent.Executors;
 42 import java.util.concurrent.ThreadLocalRandom;
 43 import java.util.concurrent.TimeUnit;
 44 import java.util.concurrent.atomic.AtomicBoolean;
 45 import java.util.concurrent.atomic.AtomicLong;
 46 
 47 import jdk.incubator.foreign.ResourceScope;
 48 import org.testng.annotations.DataProvider;
 49 import org.testng.annotations.Test;
 50 
 51 import static jdk.incubator.foreign.ValueLayout.JAVA_BYTE;
 52 import static org.testng.Assert.*;
 53 
 54 public class TestHandshake {
 55 
 56     static final int ITERATIONS = 5;
 57     static final int SEGMENT_SIZE = 1_000_000;
 58     static final int MAX_DELAY_MILLIS = 500;
 59     static final int MAX_EXECUTOR_WAIT_SECONDS = 20;
 60     static final int MAX_THREAD_SPIN_WAIT_MILLIS = 200;
 61 
 62     static final int NUM_ACCESSORS = Math.min(10, Runtime.getRuntime().availableProcessors());
 63 
 64     static final AtomicLong start = new AtomicLong();
 65     static final AtomicBoolean started = new AtomicBoolean();
 66 
 67     @Test(dataProvider = "accessors")
 68     public void testHandshake(String testName, AccessorFactory accessorFactory) throws InterruptedException {
 69         for (int it = 0 ; it < ITERATIONS ; it++) {
 70             ResourceScope scope = ResourceScope.newSharedScope();
 71             MemorySegment segment = MemorySegment.allocateNative(SEGMENT_SIZE, 1, scope);
 72             System.out.println("ITERATION " + it);
 73             ExecutorService accessExecutor = Executors.newCachedThreadPool();
 74             start.set(System.currentTimeMillis());
 75             started.set(false);
 76             for (int i = 0; i < NUM_ACCESSORS ; i++) {
 77                 accessExecutor.execute(accessorFactory.make(i, segment));
 78             }
 79             int delay = ThreadLocalRandom.current().nextInt(MAX_DELAY_MILLIS);
 80             System.out.println("Starting handshaker with delay set to " + delay + " millis");
 81             Thread.sleep(delay);
 82             accessExecutor.execute(new Handshaker(scope));
 83             accessExecutor.shutdown();
 84             assertTrue(accessExecutor.awaitTermination(MAX_EXECUTOR_WAIT_SECONDS, TimeUnit.SECONDS));
 85             assertTrue(!segment.scope().isAlive());
 86         }
 87     }
 88 
 89     static abstract class AbstractSegmentAccessor implements Runnable {
 90         final MemorySegment segment;
 91         final int id;
 92 
 93         AbstractSegmentAccessor(int id, MemorySegment segment) {
 94             this.id = id;
 95             this.segment = segment;
 96         }
 97 
 98         @Override
 99         public final void run() {
100             start("\"Accessor #\" + id");
101             outer: while (segment.scope().isAlive()) {
102                 try {
103                     doAccess();
104                 } catch (IllegalStateException ex) {
105                     long delay = System.currentTimeMillis() - start.get();
106                     System.out.println("Accessor #" + id + " suspending - elapsed (ms): " + delay);
107                     backoff();
108                     delay = System.currentTimeMillis() - start.get();
109                     System.out.println("Accessor #" + id + " resuming - elapsed (ms): " + delay);
110                     continue outer;
111                 }
112             }
113             long delay = System.currentTimeMillis() - start.get();
114             System.out.println("Accessor #" + id + " terminated - elapsed (ms): " + delay);
115         }
116 
117         abstract void doAccess();
118 
119         private void backoff() {
120             try {
121                 Thread.sleep(ThreadLocalRandom.current().nextInt(MAX_THREAD_SPIN_WAIT_MILLIS));
122             } catch (InterruptedException ex) {
123                 throw new AssertionError(ex);
124             }
125         }
126     }
127 
128     static void start(String name) {
129         if (started.compareAndSet(false, true)) {
130             long delay = System.currentTimeMillis() - start.get();
131             System.out.println("Started first thread: " + name + " ; elapsed (ms): " + delay);
132         }
133     }
134 
135     static abstract class AbstractBufferAccessor extends AbstractSegmentAccessor {
136         final ByteBuffer bb;
137 
138         AbstractBufferAccessor(int id, MemorySegment segment) {
139             super(id, segment);
140             this.bb = segment.asByteBuffer();
141         }
142     }
143 
144     static class SegmentAccessor extends AbstractSegmentAccessor {
145 
146         SegmentAccessor(int id, MemorySegment segment) {
147             super(id, segment);
148         }
149 
150         @Override
151         void doAccess() {
152             int sum = 0;
153             for (int i = 0; i < segment.byteSize(); i++) {
154                 sum += segment.get(JAVA_BYTE, i);
155             }
156         }
157     }
158 
159     static class SegmentCopyAccessor extends AbstractSegmentAccessor {
160 
161         MemorySegment first, second;
162 
163 
164         SegmentCopyAccessor(int id, MemorySegment segment) {
165             super(id, segment);
166             long split = segment.byteSize() / 2;
167             first = segment.asSlice(0, split);
168             second = segment.asSlice(split);
169         }
170 
171         @Override
172         public void doAccess() {
173             first.copyFrom(second);
174         }
175     }
176 
177     static class SegmentFillAccessor extends AbstractSegmentAccessor {
178 
179         SegmentFillAccessor(int id, MemorySegment segment) {
180             super(id, segment);
181         }
182 
183         @Override
184         public void doAccess() {
185             segment.fill((byte) ThreadLocalRandom.current().nextInt(10));
186         }
187     }
188 
189     static class SegmentMismatchAccessor extends AbstractSegmentAccessor {
190 
191         final MemorySegment copy;
192 
193         SegmentMismatchAccessor(int id, MemorySegment segment) {
194             super(id, segment);
195             this.copy = MemorySegment.allocateNative(SEGMENT_SIZE, 1, segment.scope());
196             copy.copyFrom(segment);
197             copy.set(JAVA_BYTE, ThreadLocalRandom.current().nextInt(SEGMENT_SIZE), (byte)42);
198         }
199 
200         @Override
201         public void doAccess() {
202             segment.mismatch(copy);
203         }
204     }
205 
206     static class BufferAccessor extends AbstractBufferAccessor {
207 
208         BufferAccessor(int id, MemorySegment segment) {
209             super(id, segment);
210         }
211 
212         @Override
213         public void doAccess() {
214             int sum = 0;
215             for (int i = 0; i < bb.capacity(); i++) {
216                 sum += bb.get(i);
217             }
218         }
219     }
220 
221     static class BufferHandleAccessor extends AbstractBufferAccessor {
222 
223         static VarHandle handle = MethodHandles.byteBufferViewVarHandle(short[].class, ByteOrder.nativeOrder());
224 
225         public BufferHandleAccessor(int id, MemorySegment segment) {
226             super(id, segment);
227         }
228 
229         @Override
230         public void doAccess() {
231             int sum = 0;
232             for (int i = 0; i < bb.capacity() / 2; i++) {
233                 sum += (short) handle.get(bb, i);
234             }
235         }
236     };
237 
238     static class Handshaker implements Runnable {
239 
240         final ResourceScope scope;
241 
242         Handshaker(ResourceScope scope) {
243             this.scope = scope;
244         }
245 
246         @Override
247         public void run() {
248             start("Handshaker");
249             while (true) {
250                 try {
251                     scope.close();
252                     break;
253                 } catch (IllegalStateException ex) {
254                     Thread.onSpinWait();
255                 }
256             }
257             long delay = System.currentTimeMillis() - start.get();
258             System.out.println("Segment closed - elapsed (ms): " + delay);
259         }
260     }
261 
262     interface AccessorFactory {
263         AbstractSegmentAccessor make(int id, MemorySegment segment);
264     }
265 
266     @DataProvider
267     static Object[][] accessors() {
268         return new Object[][] {
269                 { "SegmentAccessor", (AccessorFactory)SegmentAccessor::new },
270                 { "SegmentCopyAccessor", (AccessorFactory)SegmentCopyAccessor::new },
271                 { "SegmentMismatchAccessor", (AccessorFactory)SegmentMismatchAccessor::new },
272                 { "SegmentFillAccessor", (AccessorFactory)SegmentFillAccessor::new },
273                 { "BufferAccessor", (AccessorFactory)BufferAccessor::new },
274                 { "BufferHandleAccessor", (AccessorFactory)BufferHandleAccessor::new }
275         };
276     }
277 }