/* * Copyright (c) 2013, 2015, 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 * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ import java.util.concurrent.Phaser; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.LockSupport; import jdk.testlibrary.LockFreeLogManager; import jdk.testlibrary.Utils; /** * ThreadStateController allows a thread to request this thread to transition * to a specific thread state. The {@linkplain #transitionTo request} is * a blocking call that the calling thread will wait until this thread is about * going to the new state. Only one request of state transition at a time * is supported (the Phaser expects only parties of 2 to arrive and advance * to next phase). */ public class ThreadStateController extends Thread { // used to achieve waiting states private final Object lock; public ThreadStateController(String name, Object lock) { super(name); this.lock = lock; } public void checkThreadState(Thread.State expected) { // maximum number of retries when checking for thread state. final int MAX_RETRY = 500; // wait for the thread to transition to the expected state. // There is a small window between the thread checking the state // and the thread actual entering that state. Thread.State state; int retryCount=0; while ((state = getState()) != expected && retryCount < MAX_RETRY) { pause(10); retryCount++; } if (state == null) { throw new RuntimeException(getName() + " expected to have " + expected + " but got null."); } if (state != expected) { throw new RuntimeException(String.format("%s expected in %s state but got %s " + "(iterations %d interrupted %d)%n", getName(), expected, state, iterations.get(), interrupted.get())); } } public static void pause(long ms) { try { Thread.sleep(Utils.adjustTimeout(ms)); } catch (InterruptedException e) { throw new RuntimeException(e); } } // Phaser to sync between the main thread putting // this thread into various states private final Phaser phaser = new Phaser(2); private volatile int newState = S_RUNNABLE; private volatile int state = 0; private boolean done = false; private static final int S_RUNNABLE = 1; private static final int S_BLOCKED = 2; private static final int S_WAITING = 3; private static final int S_TIMED_WAITING = 4; private static final int S_PARKED = 5; private static final int S_TIMED_PARKED = 6; private static final int S_SLEEPING = 7; private static final int S_TERMINATE = 8; // for debugging private final AtomicInteger iterations = new AtomicInteger(); private final AtomicInteger interrupted = new AtomicInteger(); private final LockFreeLogManager logManager = new LockFreeLogManager(); @Override public void run() { // this thread has started while (!done) { // state transition int nextState = state; if (newState != state) { nextState = newState; iterations.set(0); interrupted.set(0); } iterations.incrementAndGet(); switch (nextState) { case S_RUNNABLE: { stateChange(nextState); double sum = 0; for (int i = 0; i < 1000; i++) { double r = Math.random(); double x = Math.pow(3, r); sum += x - r; } break; } case S_BLOCKED: { log("%d: %s is going to block (iterations %d)%n", getId(), getName(), iterations.get()); stateChange(nextState); // going to block on lock synchronized (lock) { log("%d: %s acquired the lock (iterations %d)%n", getId(), getName(), iterations.get()); try { // this thread has escaped the BLOCKED state // release the lock and a short wait before continue lock.wait(Utils.adjustTimeout(10)); } catch (InterruptedException e) { // ignore interrupted.incrementAndGet(); } } break; } case S_WAITING: { synchronized (lock) { log("%d: %s is going to waiting (iterations %d interrupted %d)%n", getId(), getName(), iterations.get(), interrupted.get()); try { stateChange(nextState); lock.wait(); log("%d: %s wakes up from waiting (iterations %d interrupted %d)%n", getId(), getName(), iterations.get(), interrupted.get()); } catch (InterruptedException e) { // ignore interrupted.incrementAndGet(); } } break; } case S_TIMED_WAITING: { synchronized (lock) { log("%d: %s is going to timed waiting (iterations %d interrupted %d)%n", getId(), getName(), iterations.get(), interrupted.get()); try { stateChange(nextState); lock.wait(Integer.MAX_VALUE); log("%d: %s wakes up from timed waiting (iterations %d interrupted %d)%n", getId(), getName(), iterations.get(), interrupted.get()); } catch (InterruptedException e) { // ignore interrupted.incrementAndGet(); } } break; } case S_PARKED: { log("%d: %s is going to park (iterations %d)%n", getId(), getName(), iterations.get()); stateChange(nextState); LockSupport.park(); break; } case S_TIMED_PARKED: { log("%d: %s is going to timed park (iterations %d)%n", getId(), getName(), iterations.get()); long deadline = System.currentTimeMillis() + Utils.adjustTimeout(10000*1000); stateChange(nextState); LockSupport.parkUntil(deadline); break; } case S_SLEEPING: { log("%d: %s is going to sleep (iterations %d interrupted %d)%n", getId(), getName(), iterations.get(), interrupted.get()); try { stateChange(nextState); Thread.sleep(Utils.adjustTimeout(1000000)); } catch (InterruptedException e) { // finish sleeping interrupted.incrementAndGet(); } break; } case S_TERMINATE: { done = true; stateChange(nextState); break; } default: break; } } } /** * Change the state if it matches newState. */ private void stateChange(int nextState) { // no state change if (state == nextState) return; // transition to the new state if (newState == nextState) { state = nextState; phaser.arrive(); log("%d: state change: %s %s%n", getId(), toStateName(nextState), phaserToString(phaser)); return; } // should never reach here throw new RuntimeException("current " + state + " next " + nextState + " new state " + newState); } /** * Blocks until this thread transitions to the given state */ public void transitionTo(Thread.State tstate) throws InterruptedException { switch (tstate) { case RUNNABLE: nextState(S_RUNNABLE); break; case BLOCKED: nextState(S_BLOCKED); break; case WAITING: nextState(S_WAITING); break; case TIMED_WAITING: nextState(S_TIMED_WAITING); break; case TERMINATED: nextState(S_TERMINATE); break; default: break; } } /** * Blocks until this thread transitions to sleeping */ public void transitionToSleep() throws InterruptedException { nextState(S_SLEEPING); } /** * Blocks until this thread transitions to park or timed park */ public void transitionToPark(boolean timed) throws InterruptedException { nextState(timed ? S_TIMED_PARKED : S_PARKED); } private void nextState(int s) throws InterruptedException { final long id = Thread.currentThread().getId(); log("%d: wait until the thread transitions to %s %s%n", id, toStateName(s), phaserToString(phaser)); this.newState = s; int phase = phaser.arrive(); log("%d: awaiting party arrive %s %s%n", id, toStateName(s), phaserToString(phaser)); for (;;) { // when this thread has changed its state before it waits or parks // on a lock, a potential race might happen if it misses the notify // or unpark. Hence await for the phaser to advance with timeout // to cope with this race condition. switch (state) { case S_WAITING: case S_TIMED_WAITING: synchronized (lock) { lock.notify(); } break; case S_PARKED: case S_TIMED_PARKED: LockSupport.unpark(this); break; case S_SLEEPING: this.interrupt(); break; case S_BLOCKED: default: break; } try { phaser.awaitAdvanceInterruptibly(phase, 100, TimeUnit.MILLISECONDS); log("%d: arrived at %s %s%n", id, toStateName(s), phaserToString(phaser)); return; } catch (TimeoutException ex) { // this thread hasn't arrived at this phase log("%d: Timeout: %s%n", id, phaser); } } } private String phaserToString(Phaser p) { return "[phase = " + p.getPhase() + " parties = " + p.getRegisteredParties() + " arrived = " + p.getArrivedParties() + "]"; } private String toStateName(int state) { switch (state) { case S_RUNNABLE: return "runnable"; case S_WAITING: return "waiting"; case S_TIMED_WAITING: return "timed waiting"; case S_PARKED: return "parked"; case S_TIMED_PARKED: return "timed parked"; case S_SLEEPING: return "sleeping"; case S_BLOCKED: return "blocked"; case S_TERMINATE: return "terminated"; default: return "unknown " + state; } } private void log(String msg, Object ... params) { logManager.log(msg, params); } /** * Waits for the controller to complete the test run and returns the * generated log * @return The controller log * @throws InterruptedException */ public String getLog() throws InterruptedException { this.join(); return logManager.toString(); } }