From fc4d94a70ca58c849175d8b639e6708bca819db0 Mon Sep 17 00:00:00 2001 From: mdb Date: Tue, 3 Jan 2006 21:50:50 +0000 Subject: [PATCH] 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 --- .../com/samskivert/util/SerialExecutor.java | 77 ++++++++-- .../samskivert/util/SerialExecutorTest.java | 141 ++++++++++++++++++ 2 files changed, 208 insertions(+), 10 deletions(-) create mode 100644 projects/samskivert/tests/src/java/com/samskivert/util/SerialExecutorTest.java diff --git a/projects/samskivert/src/java/com/samskivert/util/SerialExecutor.java b/projects/samskivert/src/java/com/samskivert/util/SerialExecutor.java index 8875fd48..e1769550 100644 --- a/projects/samskivert/src/java/com/samskivert/util/SerialExecutor.java +++ b/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; } diff --git a/projects/samskivert/tests/src/java/com/samskivert/util/SerialExecutorTest.java b/projects/samskivert/tests/src/java/com/samskivert/util/SerialExecutorTest.java new file mode 100644 index 00000000..015f09a3 --- /dev/null +++ b/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; +}