1 /*

     1 /*

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     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

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.

     7  * published by the Free Software Foundation.

    25 import java.util.concurrent.TimeUnit;

    25 import java.util.concurrent.TimeUnit;

    26 import java.util.concurrent.TimeoutException;

    26 import java.util.concurrent.TimeoutException;

    27 import java.util.concurrent.atomic.AtomicInteger;

    27 import java.util.concurrent.atomic.AtomicInteger;

    28 import java.util.concurrent.locks.LockSupport;

    28 import java.util.concurrent.locks.LockSupport;

    29 

    29 

    30 /**

    32 /**

    31  * ThreadStateController allows a thread to request this thread to transition

    33  * ThreadStateController allows a thread to request this thread to transition

    32  * to a specific thread state.  The {@linkplain #transitionTo request} is

    34  * to a specific thread state.  The {@linkplain #transitionTo request} is

    33  * a blocking call that the calling thread will wait until this thread is about

    35  * a blocking call that the calling thread will wait until this thread is about

    34  * going to the new state.  Only one request of state transition at a time

    36  * going to the new state.  Only one request of state transition at a time

    92     private static final int S_TIMED_PARKED = 6;

    94     private static final int S_TIMED_PARKED = 6;

    93     private static final int S_SLEEPING = 7;

    95     private static final int S_SLEEPING = 7;

    94     private static final int S_TERMINATE = 8;

    96     private static final int S_TERMINATE = 8;

    95 

    97 

    96     // for debugging

    98     // for debugging

    99     public void run() {

   105     public void run() {

   100         // this thread has started

   106         // this thread has started

   101         while (!done) {

   107         while (!done) {

   102             // state transition

   108             // state transition

   103             int nextState = state;

   109             int nextState = state;

   117                        sum += x - r;

   123                        sum += x - r;

   118                     }

   124                     }

   119                     break;

   125                     break;

   120                 }

   126                 }

   121                 case S_BLOCKED: {

   127                 case S_BLOCKED: {

   124                     stateChange(nextState);

   130                     stateChange(nextState);

   125                     // going to block on lock

   131                     // going to block on lock

   126                     synchronized (lock) {

   132                     synchronized (lock) {

   129                         try {

   135                         try {

   130                             // this thread has escaped the BLOCKED state

   136                             // this thread has escaped the BLOCKED state

   131                             // release the lock and a short wait before continue

   137                             // release the lock and a short wait before continue

   132                             lock.wait(10);

   138                             lock.wait(10);

   133                         } catch (InterruptedException e) {

   139                         } catch (InterruptedException e) {

   137                     }

   143                     }

   138                     break;

   144                     break;

   139                 }

   145                 }

   140                 case S_WAITING: {

   146                 case S_WAITING: {

   141                     synchronized (lock) {

   147                     synchronized (lock) {

   144                         try {

   150                         try {

   145                             stateChange(nextState);

   151                             stateChange(nextState);

   146                             lock.wait();

   152                             lock.wait();

   149                         } catch (InterruptedException e) {

   155                         } catch (InterruptedException e) {

   150                             // ignore

   156                             // ignore

   151                             interrupted.incrementAndGet();

   157                             interrupted.incrementAndGet();

   152                         }

   158                         }

   153                     }

   159                     }

   154                     break;

   160                     break;

   155                 }

   161                 }

   156                 case S_TIMED_WAITING: {

   162                 case S_TIMED_WAITING: {

   157                     synchronized (lock) {

   163                     synchronized (lock) {

   160                         try {

   166                         try {

   161                             stateChange(nextState);

   167                             stateChange(nextState);

   162                             lock.wait(10000);

   168                             lock.wait(10000);

   165                         } catch (InterruptedException e) {

   171                         } catch (InterruptedException e) {

   166                             // ignore

   172                             // ignore

   167                             interrupted.incrementAndGet();

   173                             interrupted.incrementAndGet();

   168                         }

   174                         }

   169                     }

   175                     }

   170                     break;

   176                     break;

   171                 }

   177                 }

   172                 case S_PARKED: {

   178                 case S_PARKED: {

   175                     stateChange(nextState);

   181                     stateChange(nextState);

   176                     LockSupport.park();

   182                     LockSupport.park();

   177                     break;

   183                     break;

   178                 }

   184                 }

   179                 case S_TIMED_PARKED: {

   185                 case S_TIMED_PARKED: {

   182                     long deadline = System.currentTimeMillis() + 10000*1000;

   188                     long deadline = System.currentTimeMillis() + 10000*1000;

   183                     stateChange(nextState);

   189                     stateChange(nextState);

   184                     LockSupport.parkUntil(deadline);

   190                     LockSupport.parkUntil(deadline);

   185                     break;

   191                     break;

   186                 }

   192                 }

   187                 case S_SLEEPING: {

   193                 case S_SLEEPING: {

   190                     try {

   196                     try {

   191                         stateChange(nextState);

   197                         stateChange(nextState);

   192                         Thread.sleep(1000000);

   198                         Thread.sleep(1000000);

   193                     } catch (InterruptedException e) {

   199                     } catch (InterruptedException e) {

   194                         // finish sleeping

   200                         // finish sleeping

   217 

   223 

   218         // transition to the new state

   224         // transition to the new state

   219         if (newState == nextState) {

   225         if (newState == nextState) {

   220             state = nextState;

   226             state = nextState;

   221             phaser.arrive();

   227             phaser.arrive();

   224             return;

   230             return;

   225         }

   231         }

   226 

   232 

   227         // should never reach here

   233         // should never reach here

   228         throw new RuntimeException("current " + state + " next " + nextState +

   234         throw new RuntimeException("current " + state + " next " + nextState +

   268         nextState(timed ? S_TIMED_PARKED : S_PARKED);

   274         nextState(timed ? S_TIMED_PARKED : S_PARKED);

   269     }

   275     }

   270 

   276 

   271     private void nextState(int s) throws InterruptedException {

   277     private void nextState(int s) throws InterruptedException {

   272         final long id = Thread.currentThread().getId();

   278         final long id = Thread.currentThread().getId();

   275         this.newState = s;

   281         this.newState = s;

   276         int phase = phaser.arrive();

   282         int phase = phaser.arrive();

   279         for (;;) {

   285         for (;;) {

   280             // when this thread has changed its state before it waits or parks

   286             // when this thread has changed its state before it waits or parks

   281             // on a lock, a potential race might happen if it misses the notify

   287             // on a lock, a potential race might happen if it misses the notify

   282             // or unpark.  Hence await for the phaser to advance with timeout

   288             // or unpark.  Hence await for the phaser to advance with timeout

   283             // to cope with this race condition.

   289             // to cope with this race condition.

   299                 default:

   305                 default:

   300                     break;

   306                     break;

   301             }

   307             }

   302             try {

   308             try {

   303                 phaser.awaitAdvanceInterruptibly(phase, 100, TimeUnit.MILLISECONDS);

   309                 phaser.awaitAdvanceInterruptibly(phase, 100, TimeUnit.MILLISECONDS);

   306                 return;

   312                 return;

   307             } catch (TimeoutException ex) {

   313             } catch (TimeoutException ex) {

   308                 // this thread hasn't arrived at this phase

   314                 // this thread hasn't arrived at this phase

   310             }

   316             }

   311         }

   317         }

   312     }

   318     }

   313     private String phaserToString(Phaser p) {

   320     private String phaserToString(Phaser p) {

   314         return "[phase = " + p.getPhase() +

   321         return "[phase = " + p.getPhase() +

   315                " parties = " + p.getRegisteredParties() +

   322                " parties = " + p.getRegisteredParties() +

   316                " arrived = " + p.getArrivedParties() + "]";

   323                " arrived = " + p.getArrivedParties() + "]";

   317     }

   324     }

   318     private String toStateName(int state) {

   326     private String toStateName(int state) {

   319         switch (state) {

   327         switch (state) {

   320             case S_RUNNABLE:

   328             case S_RUNNABLE:

   321                 return "runnable";

   329                 return "runnable";

   322             case S_WAITING:

   330             case S_WAITING:

   335                 return "terminated";

   343                 return "terminated";

   336             default:

   344             default:

   337                 return "unknown " + state;

   345                 return "unknown " + state;

   338         }

   346         }

   339     }

   347     }

   340 }

   364 }