Modified the serial executor to support timing out of hung executor tasks.

git-svn-id: https://samskivert.googlecode.com/svn/trunk@1749 6335cc39-0255-0410-8fd6-9bcaacd3b74c

projects/samskivert/src/java/com/samskivert/util/SerialExecutor.java

@@ -8,9 +8,11 @@ import java.util.ArrayList;
 import com.samskivert.Log;
 
 /**
- * A simple serial executor, similar to what can be done in java 1.5, only
- * using 1.4 and samskivert-style tasks. This is like RunQueue only slightly
- * more sophisticated. Bloat!
+ * Executes tasks serially, but each one on a separate thread. If a task times
+ * out, the executor will attempt to interrupt the thread and abort the task,
+ * but will abandon the thread in any case after the abort attempt so that
+ * subsequent tasks can be processed. The threads created are daemon threads so
+ * that they will not block the eventual termination of the virtual machine.
  */
 public class SerialExecutor
 {
@@ -28,9 +30,17 @@ public class SerialExecutor
          */
         public boolean merge (ExecutorTask other);
 
+        /**
+         * Returns the number of milliseconds after which this task should be
+         * considered a lost cause. If the task times out, {@link #timedOut}
+         * will be called instead of {@link #resultReceived}.
+         */
+        public long getTimeout ();
+
         /**
          * The portion of the task that will be executed on the executor's
-         * thread.
+         * thread. This method should handle {@link InterruptedException} as
+         * meaning that the task should be aborted.
          */
         public void executeTask ();
 
@@ -40,6 +50,12 @@ public class SerialExecutor
          * task.
          */
         public void resultReceived ();
+
+        /**
+         * This method is called instead of {@link #resultReceived} if the task
+         * does not complete within the requisite time.
+         */
+        public void timedOut ();
     }
 
     /**
@@ -88,9 +104,19 @@ public class SerialExecutor
     {
         _executingNow = !_queue.isEmpty();
         if (_executingNow) {
+            // start up a thread to execute the task in question
             ExecutorTask task = (ExecutorTask) _queue.remove(0);
-            ExecutorThread thread = new ExecutorThread(task);
+            final ExecutorThread thread = new ExecutorThread(task);
             thread.start();
+
+            // start up a timer that will abort this thread after the specified
+            // timeout
+            new Interval() {
+                public void expired () {
+                    // this will NOOP if the task has already completed
+                    thread.abort();
+                }
+            }.schedule(task.getTimeout());
         }
     }
 
@@ -105,22 +131,54 @@ public class SerialExecutor
             _task = task;
         }
 
-        // documentation inherited
+        public synchronized void abort ()
+        {
+            if (_task != null) {
+                final ExecutorTask task = _task;
+                // clear out the task reference which will let the running
+                // thread know to stop if/when executeTask() returns
+                _task = null;
+
+                // let the task know that it timed out
+                _receiver.postRunnable(new Runnable() {
+                    public void run () {
+                        try {
+                            task.timedOut();
+                        } catch (Throwable t) {
+                            Log.logStackTrace(t);
+                        }
+                        checkNext();
+                    }
+                });
+
+                // finally interrupt the thread in hopes of waking it up from
+                // it's hangitude
+                interrupt();
+            }
+        }
+
         public void run ()
         {
+            final ExecutorTask task = _task;
             try {
                 _task.executeTask();
+                synchronized (this) {
+                    if (_task == null) {
+                        // we were aborted, abandon ship
+                        System.err.println("Aborted!");
+                        return;
+                    }
+                    _task = null;
+                }
             } catch (Throwable t) {
-                Log.warning("ExecutorTask choked: " + t);
                 Log.logStackTrace(t);
             }
 
             _receiver.postRunnable(new Runnable() {
                 public void run () {
                     try {
-                        _task.resultReceived();
+                        task.resultReceived();
                     } catch (Throwable t) {
-                        Log.warning("ExecutorTask choked: " + t);
                         Log.logStackTrace(t);
                     }
                     checkNext();
@@ -128,7 +186,6 @@ public class SerialExecutor
             });
         }
 
-        /** The task to execute. */
         protected ExecutorTask _task;
     }
 

projects/samskivert/tests/src/java/com/samskivert/util/SerialExecutorTest.java

@@ -0,0 +1,141 @@
+//
+// $Id$
+
+package com.samskivert.util;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+
+/**
+ * Tests the {@link SerialExecutor} class.
+ */
+public class SerialExecutorTest extends TestCase
+    implements RunQueue
+{
+    public SerialExecutorTest ()
+    {
+        super(SerialExecutorTest.class.getName());
+        _main = Thread.currentThread();
+    }
+
+    public void runTest ()
+    {
+        SerialExecutor executor = new SerialExecutor(this);
+        int added = 0;
+
+        // _sleeps++, _exits++, _results++
+        executor.addTask(new Sleeper(500L, false));
+        added++;
+
+        // _interrupts++, _exits++, _timeouts++
+        executor.addTask(new Sleeper(1500L, false));
+        added++;
+
+        // _interrupts++, _timeouts++
+        executor.addTask(new Sleeper(1500L, true));
+        added++;
+
+        // _sleeps++, _exits++, _results++
+        executor.addTask(new Sleeper(500L, false));
+        added++;
+
+        // process the results posted on our run queue
+        for (int ii = 0; ii < added; ii++) {
+            Runnable r = (Runnable)_queue.get();
+            r.run();
+        }
+
+        // give the executor threads a second to run to completion
+        try {
+            Thread.sleep(1000L);
+        } catch (InterruptedException ie) {
+        }
+
+        // make sure we got the expected number of sleeps, interrupts, etc.
+        assertCount("_sleeps", _sleeps, 2);
+        assertCount("_interrupts", _interrupts, 2);
+        assertCount("_exits", _exits, 3);
+        assertCount("_results", _results, 2);
+        assertCount("_timeouts", _timeouts, 2);
+        assertCount("_doubleints", _doubleints, 0);
+    }
+
+    protected void assertCount (String field, int value, int expected)
+    {
+        assertTrue(field + " != " + expected + " (" + value + ")",
+                   value == expected);
+    }
+
+    public void postRunnable (Runnable r)
+    {
+        _queue.append(r);
+    }
+
+    public boolean isDispatchThread ()
+    {
+        return Thread.currentThread() == _main;
+    }
+
+    public static Test suite ()
+    {
+        return new SerialExecutorTest();
+    }
+
+    public static void main (String[] args)
+    {
+        SerialExecutorTest test = new SerialExecutorTest();
+        test.runTest();
+    }
+
+    protected class Sleeper implements SerialExecutor.ExecutorTask
+    {
+        public Sleeper (long sleepFor, boolean hang) {
+            _sleepFor = sleepFor;
+            _hang = hang;
+        }
+
+        public boolean merge (SerialExecutor.ExecutorTask task) {
+            return false;
+        }
+
+        public long getTimeout () {
+            return 1000L;
+        }
+
+        public void executeTask () {
+            try {
+                Thread.sleep(_sleepFor);
+                _sleeps++;
+            } catch (InterruptedException ie) {
+                _interrupts++;
+            }
+
+            if (_hang) {
+                try {
+                    Thread.sleep(100000L);
+                } catch (InterruptedException ie) {
+                    _doubleints++;
+                }
+            }
+
+            _exits++;
+        }
+
+        public void resultReceived () {
+            _results++;
+        }
+
+        public void timedOut () {
+            _timeouts++;
+        }
+
+        protected long _sleepFor;
+        protected boolean _hang;
+    }
+
+    protected Thread _main;
+    protected Queue _queue = new Queue();
+
+    protected int _sleeps, _interrupts, _doubleints, _exits;
+    protected int _results, _timeouts;
+}