BasicTaskWrapper


1   /*
2     * JBoss, Home of Professional Open Source
3     * Copyright 2005, JBoss Inc., and individual contributors as indicated
4     * by the @authors tag. See the copyright.txt in the distribution for a
5     * full listing of individual contributors.
6     *
7     * This is free software; you can redistribute it and/or modify it
8     * under the terms of the GNU Lesser General Public License as
9     * published by the Free Software Foundation; either version 2.1 of
10    * the License, or (at your option) any later version.
11    *
12    * This software is distributed in the hope that it will be useful,
13    * but WITHOUT ANY WARRANTY; without even the implied warranty of
14    * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15    * Lesser General Public License for more details.
16    *
17    * You should have received a copy of the GNU Lesser General Public
18    * License along with this software; if not, write to the Free
19    * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
20    * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
21    */
22  package org.jboss.util.threadpool;
23  
24  import org.jboss.logging.Logger;
25  
26  /**
27   * A wrapper for the task.
28   *
29   * @author <a HREF="mailto:adrian@jboss.org">Adrian Brock</a>
30   * @version $Revision: 1958 $
31   */
32  public class BasicTaskWrapper implements TaskWrapper
33  {
34     /** The log */
35     private static final Logger log = Logger.getLogger(BasicTaskWrapper.class);
36  
37     /** The task has not been accepted */
38     public static final int TASK_NOT_ACCEPTED = 0;
39  
40     /** The task has been accepted */
41     public static final int TASK_ACCEPTED = 1;
42  
43     /** The task has been started */
44     public static final int TASK_STARTED = 2;
45  
46     /** The task has completed */
47     public static final int TASK_COMPLETED = 3;
48  
49     /** The task was rejected */
50     public static final int TASK_REJECTED = -1;
51  
52     /** The task has been stopped */
53     public static final int TASK_STOPPED = -2;
54  
55     /** The state of the task */
56     private int state = TASK_NOT_ACCEPTED;
57  
58     /** The state lock */
59     private Object   stateLock = new Object  ();
60  
61     /** The task */
62     private Task task;
63  
64     /** The task as a string */
65     private String   taskString;
66  
67     /** The start time */
68     private long startTime;
69  
70     /** The start timeout */
71     private long startTimeout;
72  
73     /** The completion timeout */
74     private long completionTimeout;
75  
76     /** The priority */
77     private int priority;
78  
79     /** The wait type */
80     private int waitType;
81  
82     /** The thread */
83     private Thread   runThread;
84  
85     /**
86      * Create a task wrapper without a task
87      */
88     protected BasicTaskWrapper()
89     {
90     }
91  
92     /**
93      * Create a new task wrapper
94      *
95      * @param task the task
96      * @throws IllegalArgumentException for a null task
97      */
98     public BasicTaskWrapper(Task task)
99     {
100       setTask(task);
101    }
102 
103    public int getTaskWaitType()
104    {
105       return waitType;
106    }
107 
108    public int getTaskPriority()
109    {
110       return priority;
111    }
112 
113    public long getTaskStartTimeout()
114    {
115       return startTimeout;
116    }
117 
118    public long getTaskCompletionTimeout()
119    {
120       return completionTimeout;
121    }
122 
123    public void acceptTask()
124    {
125       synchronized (stateLock)
126       {
127          // Not in a valid state
128          if (state != TASK_NOT_ACCEPTED)
129             return;
130       }
131 
132       // Accept the task
133       if (taskAccepted())
134          state = TASK_ACCEPTED;
135       else
136          state = TASK_REJECTED;
137 
138       // Notify the waiting task
139       synchronized (stateLock)
140       {
141          stateLock.notifyAll();
142       }
143    }
144 
145    public void rejectTask(RuntimeException   e)
146    {
147       synchronized (stateLock)
148       {
149          state = TASK_REJECTED;
150          stateLock.notifyAll();
151       }
152       taskRejected(e);
153    }
154 
155    public boolean isComplete()
156    {
157       return state == TASK_COMPLETED;
158    }
159 
160    public void stopTask()
161    {
162       boolean started;
163       synchronized (stateLock)
164       {
165          started = (state == TASK_STARTED);
166          state = TASK_STOPPED;
167       }
168       if (started)
169       {
170          // Interrupt the run thread if its not null
171          if( runThread != null )
172          {
173             runThread.interrupt();
174          }
175          taskStop();
176       }
177       else if( runThread != null && runThread.isInterrupted() )
178       {
179          /* If the thread has not been returned after being interrupted, then
180          use the deprecated stop method to try to force the thread abort.
181          */
182          runThread.stop();
183       }
184    }
185 
186    public void waitForTask()
187    {
188       switch (waitType)
189       {
190          case Task.WAIT_FOR_START:
191          {
192             boolean interrupted = false;
193             synchronized (stateLock)
194             {
195                while (state == TASK_NOT_ACCEPTED || state == TASK_ACCEPTED)
196                {
197                   try
198                   {
199                      stateLock.wait();
200                   }
201                   catch (InterruptedException   e)
202                   {
203                      interrupted = true;
204                   }
205                }
206                if (interrupted)
207                   Thread.currentThread().interrupt();
208                return;
209             }
210          }
211          default:
212          {
213             return;
214          }
215       }
216    }
217 
218    /**
219     * Called by the thread pool executor
220     */ 
221    public void run()
222    {
223       // Get the execution thread
224       this.runThread = Thread.currentThread();
225 
226       // Check for a start timeout
227       long runTime = getElapsedTime();
228       if (startTimeout > 0l && runTime >= startTimeout)
229       {
230          taskRejected(new StartTimeoutException("Start Timeout exceeded for task " + taskString));
231          return;
232       }
233 
234       // We are about to start, check for a stop
235       boolean stopped = false;
236       synchronized (stateLock)
237       {
238          if (state == TASK_STOPPED)
239          {
240             stopped = true;
241          }
242          else
243          {
244             state = TASK_STARTED;
245             taskStarted();
246             if (waitType == Task.WAIT_FOR_START)
247                stateLock.notifyAll();
248          }
249       }
250       if (stopped)
251       {
252          taskRejected(new TaskStoppedException("Task stopped for task " + taskString));
253          return;
254       }
255 
256       // Run the task
257       Throwable   throwable = null;
258       try
259       {
260          task.execute();
261       }
262       catch (Throwable   t)
263       {
264          throwable = t;
265       }
266 
267       // It is complete
268       taskCompleted(throwable);
269 
270       // We are completed
271       synchronized (stateLock)
272       {
273          state = TASK_COMPLETED;
274          if (waitType == Task.WAIT_FOR_COMPLETE)
275             stateLock.notifyAll();
276       }
277    }
278 
279    /**
280     * Set thetask for this wrapper
281     *
282     * @param task the task
283     */
284    protected void setTask(Task task)
285    {
286       if (task == null)
287          throw new IllegalArgumentException  ("Null task");
288       this.task = task;
289       this.taskString = task.toString();
290       this.startTime = System.currentTimeMillis();
291       this.waitType = task.getWaitType();
292       this.priority = task.getPriority();
293       this.startTimeout = task.getStartTimeout();
294       this.completionTimeout = task.getCompletionTimeout();
295    }
296 
297    /**
298     * Notify the task it has been accepted
299     *
300     * @return true when the notification succeeds, false otherwise
301     */
302    protected boolean taskAccepted()
303    {
304       try
305       {
306          task.accepted(getElapsedTime());
307          return true;
308       }
309       catch (Throwable   t)
310       {
311          log.warn("Unexpected error during 'accepted' for task: " + taskString, t);
312          return false;
313       }
314    }
315 
316    /**
317     * Notify the task it has been rejected
318     *
319     * @param e any error associated with the rejection
320     * @return true when the notification succeeds, false otherwise
321     */
322    protected boolean taskRejected(RuntimeException   e)
323    {
324       try
325       {
326          task.rejected(getElapsedTime(), e);
327          return true;
328       }
329       catch (Throwable   t)
330       {
331          log.warn("Unexpected error during 'rejected' for task: " + taskString, t);
332          if (e != null)
333             log.warn("Original reason for rejection of task: " + taskString, e);
334          return false;
335       }
336    }
337 
338    /**
339     * Notify the task it has started
340     *
341     * @return true when the notification succeeds, false otherwise
342     */
343    protected boolean taskStarted()
344    {
345       try
346       {
347          task.started(getElapsedTime());
348          return true;
349       }
350       catch (Throwable   t)
351       {
352          log.warn("Unexpected error during 'started' for task: " + taskString, t);
353          return false;
354       }
355    }
356 
357    /**
358     * Notify the task it has completed
359     *
360     * @param throwable any throwable associated with the completion
361     * @return true when the notification succeeds, false otherwise
362     */
363    protected boolean taskCompleted(Throwable   throwable)
364    {
365       try
366       {
367          task.completed(getElapsedTime(), throwable);
368          return true;
369       }
370       catch (Throwable   t)
371       {
372          log.warn("Unexpected error during 'completed' for task: " + taskString, t);
373          if (throwable != null)
374             log.warn("Original error during 'run' for task: " + taskString, throwable);
375          return false;
376       }
377    }
378 
379    /**
380     * Stop the task
381     *
382     * @return true when the notification succeeds, false otherwise
383     */
384    protected boolean taskStop()
385    {
386       try
387       {
388          task.stop();
389          return true;
390       }
391       catch (Throwable   t)
392       {
393          log.warn("Unexpected error during 'stop' for task: " + taskString, t);
394          return false;
395       }
396    }
397 
398    /**
399     * Calculate the elapsed time since the task was started
400     *
401     * @return the elapsed time in millis
402     */
403    protected long getElapsedTime()
404    {
405       return System.currentTimeMillis() - startTime;
406    }
407    
408    /**
409     * Get the state as a string
410     * 
411     * @return the state string
412     */
413    protected String   getStateString()
414    {
415       switch (state)
416       {
417          case TASK_NOT_ACCEPTED:
418             return "NOT_ACCEPTED";
419          case TASK_REJECTED:
420             return "REJECTED";
421          case TASK_ACCEPTED:
422             return "ACCEPTED";
423          case TASK_STARTED:
424             return "STARTED";
425          case TASK_STOPPED:
426             return "STOPPED";
427          case TASK_COMPLETED:
428             return "COMPLETED";
429          default:
430             return "???";
431       }
432    }
433 }
434
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags