bundles/org.simantics.utils.thread/src/org/simantics/utils/threads/ua/Work.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.simantics.utils.threads.ua;
  14
  15 import java.util.concurrent.Callable;
  16 import java.util.concurrent.CancellationException;
  17 import java.util.concurrent.ExecutionException;
  18 import java.util.concurrent.RunnableFuture;
  19 import java.util.concurrent.TimeUnit;
  20 import java.util.concurrent.TimeoutException;
  21
  22 /**
  23  * Add stateful monitoring features to runnable and callable.
  24  * <p>
  25  * Executing two work instances in parallel causes RuntimeException.
  26  *
  27  * @author Toni Kalajainen (toni.kalajainen@vtt.fi)
  28  */
  29 public class Work<T> extends AbstractState<WorkState, RuntimeException> implements WorkMonitor, RunnableFuture<T> {
  30
  31         public static Work<Object> createMonitor(Runnable r)
  32         {
  33                 return new Work<Object>(r);
  34         }
  35
  36         public static Work<Object> createMonitor(Runnable r, StateListener<WorkState> listener)
  37         {
  38                 Work<Object> result = new Work<Object>(r);
  39                 result.addStateListener(listener);
  40                 return result;
  41         }
  42
  43         public static <T> Work<T> createMonitor(Callable<T> r)
  44         {
  45                 return new Work<T>(r);
  46         }
  47
  48         public static <T> Work<T> createMonitor(Callable<T> r, StateListener<WorkState> listener)
  49         {
  50                 Work<T> result = new Work<T>(r);
  51                 result.addStateListener(listener);
  52                 return result;
  53         }
  54
  55
  56         Callable<T> c;
  57         Runnable r;
  58         Thread thread;
  59         transient Exception error;
  60         transient T result;
  61         boolean canceled;
  62
  63         public Work(Runnable runnable) {
  64                 super(WorkState.Ready);
  65                 if (runnable == null)
  66                         throw new IllegalArgumentException("null arg");
  67
  68                 this.r = runnable;
  69         }
  70
  71         public Work(Runnable runnable, T result) {
  72                 super(WorkState.Ready);
  73                 if (runnable == null)
  74                         throw new IllegalArgumentException("null arg");
  75
  76                 this.r = runnable;
  77                 this.result = result;
  78         }
  79
  80         public Work(Callable<T> c) {
  81                 super(WorkState.Ready);
  82                 if (c == null)
  83                         throw new IllegalArgumentException("null arg");
  84
  85                 this.c = c;
  86         }
  87
  88         @Override
  89         public void run() {
  90                 WorkState s = getState();
  91                 if (s!=WorkState.Ready)
  92                         throw new RuntimeException("Work must be restarted before it can be reused");
  93                 thread = Thread.currentThread();
  94                 if (setState(WorkState.Working, null, WorkState.READY_STATE)==WorkState.Working) return;
  95                 try {
  96                         if (r!=null)
  97                                 r.run();
  98                         else
  99                                 result = c.call();
 100                 } catch (Exception e) {
 101                         error = e;
 102                         setState(WorkState.Error);
 103                 }
 104                 WorkState newState = WorkState.Complete;
 105                 synchronized(this) {
 106                         s = getState();
 107                         if (s==WorkState.Interrupting) {
 108                                 Thread.interrupted();
 109                                 newState = WorkState.Interrupted;
 110                         }
 111                 }
 112                 setState(newState);
 113         }
 114
 115         @Override
 116         public boolean cancel(boolean mayInterruptIfRunning) {
 117                 synchronized(this) {
 118                         WorkState s = getState();
 119                         if (s != WorkState.Ready && s != WorkState.Working) return false;
 120                 }
 121                 // Attempt cancel
 122                 if (attemptSetState(WorkState.READY_STATE, WorkState.Canceled)==WorkState.Ready) {
 123                         canceled |= true;
 124                         return true;
 125                 }
 126                 // Attempt interrupt
 127                 if (mayInterruptIfRunning &&
 128                         attemptSetState(WorkState.WORKING_STATE, WorkState.Interrupting)==WorkState.Working)
 129                 {
 130                         canceled |= true;
 131                         thread.interrupt();
 132                         return true;
 133                 }
 134                 return false;
 135         }
 136
 137         @Override
 138         protected boolean isStateTransitionAllowed(WorkState oldState, WorkState newState) {
 139                 return true;
 140         }
 141
 142         // Raise visibility
 143         @Override
 144         public void setError(RuntimeException error)
 145         {
 146                 super.setError(error);
 147         }
 148
 149         /**
 150          * Reset work for reuse.
 151          */
 152         public synchronized void restart()
 153         {
 154                 WorkState s = getState();
 155                 if (s==WorkState.Ready) return;
 156                 if (!s.isFinalState())
 157                         throw new RuntimeException("Work cannot be restarted until the previous run has completed.");
 158                 setState(WorkState.Ready);
 159         }
 160
 161         @Override
 162         public Runnable getRunnable() {
 163                 return r;
 164         }
 165
 166         @Override
 167         public T get() throws InterruptedException, ExecutionException {
 168                 WorkState s = waitForState(WorkState.FINAL_STATES);
 169                 if (s==WorkState.Canceled)
 170                         throw new CancellationException();
 171                 if (s==WorkState.Interrupted)
 172                         throw new InterruptedException();
 173                 if (s==WorkState.Error)
 174                         throw new ExecutionException(error);
 175                 return result;
 176         }
 177
 178         @Override
 179         public T get(long timeout, TimeUnit unit) throws InterruptedException,
 180                         ExecutionException, TimeoutException {
 181                 WorkState s = waitForState(WorkState.FINAL_STATES, timeout, unit);
 182                 if (s==WorkState.Canceled)
 183                         throw new CancellationException();
 184                 if (s==WorkState.Interrupted)
 185                         throw new InterruptedException();
 186                 if (s==WorkState.Error)
 187                         throw new ExecutionException(error);
 188                 return result;
 189         }
 190
 191         @Override
 192         public boolean isCancelled() {
 193                 return canceled;
 194         }
 195
 196         @Override
 197         public boolean isDone() {
 198                 return WorkState.FINAL_STATES.contains(getState());
 199         }
 200
 201 }