1 /*
  2  * Copyright (c) 2007, 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 package nsk.share.runner;
 24 
 25 import java.util.concurrent.locks.LockSupport;
 26 import nsk.share.gc.OOMStress;
 27 import nsk.share.log.*;
 28 import nsk.share.test.Stresser;
 29 import nsk.share.test.ExecutionController;
 30 import nsk.share.TestBug;
 31 import java.util.List;
 32 import java.util.ArrayList;
 33 import java.util.concurrent.atomic.AtomicInteger;
 34 
 35 /**
 36  *  Helper to assist in running threads.
 37  *
 38  *  This class starts a number of threads which run some tasks in cycle.
 39  *  They exit after some time or after some iterations as
 40  *  determined by RunParams.
 41  */
 42 public class ThreadsRunner implements MultiRunner, LogAware, RunParamsAware {
 43 
 44     private Log log;
 45     private RunParams runParams;
 46     private List<Runnable> runnables = new ArrayList<Runnable>();
 47     private List<ManagedThread> threads = new ArrayList<ManagedThread>();
 48     private AtomicInteger notStarted;
 49     private AtomicInteger finished;
 50     private boolean started = false;
 51     private boolean successful = true;
 52 
 53     public ThreadsRunner() {
 54         this(RunParams.getInstance());
 55     }
 56 
 57     public ThreadsRunner(RunParams runParams) {
 58         setRunParams(runParams);
 59     }
 60 
 61     public final void setLog(Log log) {
 62         this.log = log;
 63     }
 64 
 65     private static class ManagedThreadFactory {
 66 
 67         private RunParams params;
 68 
 69         static ManagedThreadFactory createFactory(RunParams params) {
 70             return new ManagedThreadFactory(params);
 71         }
 72 
 73         private ManagedThreadFactory(RunParams params) {
 74             this.params = params;
 75         }
 76 
 77         public Thread newThread(Runnable runnable, String name, int num) {
 78             Thread t;
 79             if (this.params.useVirtualThreads()) {
 80                 t = unstartedVirtualThread(runnable);
 81             } else {
 82                 t = new Thread(runnable);
 83             }
 84             t.setName(name);
 85             return t;
 86         }
 87 
 88         private Thread unstartedVirtualThread(Runnable task) {
 89             try {
 90                 Object builder = Thread.class.getMethod("ofVirtual").invoke(null);
 91                 Class<?> clazz = Class.forName("java.lang.Thread$Builder");
 92                 java.lang.reflect.Method unstarted = clazz.getMethod("unstarted", Runnable.class);
 93                 return (Thread) unstarted.invoke(builder, task);
 94             } catch (RuntimeException | Error e) {
 95                 throw e;
 96             } catch (Exception e) {
 97                 throw new RuntimeException(e);
 98             }
 99         }
100 
101     }
102 
103     private class ManagedThread implements Runnable {
104 
105         private Stresser stresser;
106         private Throwable exception;
107         private Runnable test;
108         private boolean shouldWait;
109         private Thread thread;
110 
111 
112         public ManagedThread(ManagedThreadFactory threadFactory, Runnable test, int num) {
113             this.test = test;
114             this.shouldWait = true;
115             this.thread = threadFactory.newThread(this, test.toString(), num);
116             this.stresser = new Stresser(thread.getName(), runParams.getStressOptions());
117         }
118 
119         @Override
120         public void run() {
121             notStarted.decrementAndGet();
122             while (notStarted.get() != 0) {
123                 LockSupport.parkNanos(1);
124             }
125             try {
126                 stresser.start(runParams.getIterations());
127                 while (!this.thread.isInterrupted() && stresser.iteration()) {
128                     test.run();
129                     LockSupport.parkNanos(1);
130                 }
131             } catch (OutOfMemoryError oom) {
132                 if (test instanceof OOMStress) {
133                     // Test stressing OOM, not a failure.
134                     log.info("Caught OutOfMemoryError in OOM stress test, omitting exception.");
135                 } else {
136                     failWithException(oom);
137                 }
138             } catch (Throwable t) {
139                 failWithException(t);
140             } finally {
141                 waitForOtherThreads();
142                 stresser.finish();
143             }
144         }
145 
146         private void waitForOtherThreads() {
147             if (shouldWait) {
148                 shouldWait = false;
149                 finished.decrementAndGet();
150                 while (finished.get() != 0) {
151                     try {
152                         Thread.sleep(100);
153                     } catch (InterruptedException ie) {
154                     }
155                 }
156             } else {
157                 throw new TestBug("Waiting a second time is not premitted");
158             }
159         }
160 
161         private void failWithException(Throwable t) {
162             log.debug("Exception in ");
163             log.debug(test);
164             log.debug(t);
165             exception = t;
166         }
167 
168         public void forceFinish() {
169             stresser.forceFinish();
170             if (runParams.isInterruptThreads()) {
171                 log.debug("Interrupting: " + this);
172                 this.thread.interrupt();
173             }
174         }
175 
176         public final Throwable getException() {
177             return exception;
178         }
179 
180         public final ExecutionController getExecutionController() {
181             return stresser;
182         }
183     }
184 
185     public void add(Runnable runnable) {
186         runnables.add(runnable);
187     }
188 
189     public void remove(Runnable runnable) {
190         runnables.remove(runnable);
191     }
192 
193     public void removeAll() {
194         runnables.clear();
195     }
196 
197     private Runnable get(int index) {
198         return (Runnable) runnables.get(index);
199     }
200 
201     public Thread getThread(int index) {
202         return threads.get(index).thread;
203     }
204 
205     private int getCount() {
206         return runnables.size();
207     }
208 
209     private void prepare() {
210     }
211 
212     private void create() {
213         int threadCount = runnables.size();
214         notStarted = new AtomicInteger(threadCount);
215         finished = new AtomicInteger(threadCount);
216         ManagedThreadFactory factory = ManagedThreadFactory.createFactory(runParams);
217         for (int i = 0; i < threadCount; ++i) {
218             threads.add(new ManagedThread(factory, get(i), i));
219         }
220     }
221 
222     /**
223      * Start threads that run the tasks.
224      */
225     public void start() {
226         if (started) {
227             return;
228         }
229         create();
230         prepare();
231         for (int i = 0; i < threads.size(); ++i) {
232             Thread t = threads.get(i).thread;
233             log.debug("Starting " + t);
234             t.start();
235         }
236         started = true;
237     }
238 
239     /**
240      * Stop threads that run the tasks.
241      */
242     public void forceFinish() {
243         log.info("Forcing threads to finish");
244         for (int i = 0; i < threads.size(); i++) {
245             ManagedThread thread = threads.get(i);
246             thread.forceFinish();
247         }
248     }
249 
250     /**
251      * Join threads that run the tasks.
252      */
253     public void join() throws InterruptedException {
254         for (int i = 0; i < threads.size(); ++i) {
255             Thread t = threads.get(i).thread;
256             //log.debug("Joining " + t);
257             t.join();
258         }
259     }
260 
261     private int dumpFailures() {
262         int n = 0;
263         for (int i = 0; i < threads.size(); i++) {
264             ManagedThread thread = threads.get(i);
265             Throwable exception = thread.getException();
266             if (exception != null) {
267                 if (n == 0) {
268                     log.error("Failures summary:");
269                 }
270                 ++n;
271                 log.error(exception);
272             }
273         }
274         if (n == 0) {
275             log.info("No unexpected exceptions/errors are thrown");
276         }
277         return n;
278     }
279 
280     private ManagedThread findManagedThread(Thread t) {
281         for (int i = 0; i < threads.size(); i++) {
282             ManagedThread mt = threads.get(i);
283             if (mt.thread == t) {
284                 return mt;
285             }
286         }
287         return null;
288     }
289 
290     /**
291      * Run threads as determined by RunParams.
292      *
293      * Start threads, run for some time or for some number of iterations,
294      * then join and report if there were any exceptions.
295      *
296      * This method may additionally run other threads (as determined by RunParams):
297      * - thread that does System.gc() in cycle, @see GCRunner
298      * - thread that prints memory information in cycle, @see MemDiag
299      * - thread that prints information about FinMemoryObject's in cycle, @see FinDiag
300      * - thread that prints information about AllMemoryObject's in cycle, @see AllDiag
301      *
302      * @return true if there were no exceptions, false otherwise
303      */
304     public void run() {
305         if (runParams.isRunGCThread()) {
306             add(new GCRunner());
307         }
308         if (runParams.isRunFinThread()) {
309             add(new FinRunner());
310         }
311         if (runParams.isRunMemDiagThread()) {
312             add(new MemDiag());
313         }
314         try {
315             start();
316             join();
317             successful = dumpFailures() == 0;
318         } catch (Throwable t) {
319             log.info("Unexpected exception during the run.");
320             log.info(t);
321             successful = false;
322         }
323     }
324 
325     public boolean isSuccessful() {
326         return successful;
327     }
328 
329     public ExecutionController getExecutionController() {
330         Thread ct = Thread.currentThread();
331         ManagedThread t = findManagedThread(ct);
332         if (t != null) {
333             return t.getExecutionController();
334         } else {
335             throw new TestBug("Unable to find managed thread for thread (this method should be called from one of managed threads): " + ct);
336         }
337     }
338 
339     public void runForever() {
340         start();
341     }
342 
343     public final void setRunParams(RunParams runParams) {
344         this.runParams = runParams;
345     }
346 }