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

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