From f527e6d68a94ba3ca0fba997238f90657102fcf4 Mon Sep 17 00:00:00 2001
From: brinn <brinn>
Date: Tue, 17 Jun 2008 19:19:18 +0000
Subject: [PATCH] change: - perform a major refactoring on TerminableCallable
by putting the thread guard stuff to a class of its own (ThreadGuard) -
ensure the TerminableCallable doesn't get started if it is cancelled or
terminated before it starts running - remove the hasCleanedUp() and
waitForCleanedUp() methods - remove the waitForStarted() method
SVN: 6652
---
.../common/concurrent/TerminableCallable.java | 251 ++++-------------
.../common/concurrent/TerminableFuture.java | 41 +--
.../cisd/common/concurrent/ThreadGuard.java | 257 ++++++++++++++++++
.../concurrent/TerminableCallableTest.java | 14 +-
4 files changed, 316 insertions(+), 247 deletions(-)
create mode 100644 common/source/java/ch/systemsx/cisd/common/concurrent/ThreadGuard.java
diff --git a/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableCallable.java b/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableCallable.java
index 233dcc36361..cb14208fd83 100644
--- a/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableCallable.java
+++ b/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableCallable.java
@@ -17,10 +17,6 @@
package ch.systemsx.cisd.common.concurrent;
import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import ch.systemsx.cisd.common.exceptions.CheckedExceptionTunnel;
import ch.systemsx.cisd.common.exceptions.HardStopException;
@@ -36,8 +32,8 @@ import ch.systemsx.cisd.common.utilities.ITerminable;
* this class, but create a new one instead!
* <p>
* All code in the delegate {@link #call()} that cannot be interrupted but safely be stopped (see
- * {@link Thread#interrupt()} and <code>Thread.stop()</code> for details) should be executed in
- * the <var>stoppableExecutor</var>.
+ * {@link Thread#interrupt()} and <code>Thread.stop()</code>) should be executed in the
+ * <var>stoppableExecutor</var>.
* <p>
* <strong>Note: Code executed in the <var>stoppableExecutor</var> must <i>not</i> change
* variables or data structures used by several threads or else the problems described in <a
@@ -112,25 +108,15 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
*/
public static final long DEFAULT_WAIT_INTERRUPT_MILLIS = 100L;
+ /**
+ * The guard of the thread that runs the task.
+ */
+ private final ThreadGuard threadGuard = new ThreadGuard();
+
private final ICallable<V> delegate;
private final ICleaner cleanerOrNull;
- /** The thread to interrupt or stop. */
- private volatile Thread thread;
-
- /** This latch signals that the callable has actually started. */
- private final CountDownLatch started = new CountDownLatch(1);
-
- /** This latch signals that the callable has finished. */
- private final CountDownLatch finished = new CountDownLatch(1);
-
- /** This latch signals that the callable has finished clean up after an interrupt or stop. */
- private final CountDownLatch cleanedUp = new CountDownLatch(1);
-
- /** The lock that guards stopping the thread. */
- private final Lock stopLock = new ReentrantLock();
-
/** The time (in milli-seconds) to wait for {@link Thread#interrupt()} to work. */
private final long waitInterruptMillis;
@@ -143,13 +129,13 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
/** The callable completed normally. */
COMPLETED,
- /** The thread of the callable got interrupted. */
+ /** The thread that runs the callable got interrupted. */
INTERRUPTED,
- /** The thread of the callable got stopped. */
+ /** The thread that runs the callable got stopped. */
STOPPED,
- /** The callable was terminated with an exception. */
+ /** The callable got terminated with an exception. */
EXCEPTION,
}
@@ -280,12 +266,6 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
* href="http://java.sun.com/j2se/1.5.0/docs/guide/misc/threadPrimitiveDeprecation.html">"Why is
* <code>Thread.stop()</code> deprecated?"</a> apply to your code! Watch out for static
* thread-safe variables like e.g. the ones of type {@link ThreadLocal}!</strong>
- * <p>
- * Convenience wrapper for
- * {@link #create(ch.systemsx.cisd.common.concurrent.TerminableCallable.ICallable,
- * ch.systemsx.cisd.common.concurrent.TerminableCallable.ICleaner, long, long)} with
- * <var>waitInterruptMillis</var> set to 0 and <var>timeoutTerminateMillis</var> set to
- * {@link #WAIT_FOREVER_MILLIS}.
*/
public static <V> TerminableCallable<V> createStoppable(final Callable<V> delegate)
{
@@ -320,53 +300,6 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
this.timeoutTerminateMillis = timeoutTerminateMillis;
}
- /** Return <code>true</code>, if the <var>latch</var> becomes 0 in due time. */
- private static boolean wait(CountDownLatch latch, long timeoutMillis) throws StopException
- {
- try
- {
- return latch.await(timeoutMillis, TimeUnit.MILLISECONDS);
- } catch (InterruptedException ex)
- {
- throw new StopException(ex);
- }
- }
-
- private boolean stop(Thread t, long timeoutMillis) throws StopException
- {
- final boolean gotIt;
- try
- {
- gotIt = stopLock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS);
- } catch (InterruptedException ex)
- {
- throw new StopException(ex);
- }
- if (gotIt == false)
- {
- return false;
- }
- try
- {
- stopNow(t);
- return true;
- } finally
- {
- stopLock.unlock();
- }
- }
-
- @SuppressWarnings("deprecation")
- private void stopNow(Thread t)
- {
- t.stop(new HardStopException());
- }
-
- private void cleanUp()
- {
- cleanUp(null);
- }
-
private void cleanUp(Throwable throwableOrNull)
{
if (cleanerOrNull != null)
@@ -390,23 +323,6 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
}
}
- private void markStarted()
- {
- thread = Thread.currentThread();
- started.countDown();
- }
-
- private void markFinished()
- {
- thread = null;
- finished.countDown();
- }
-
- private void markCleanedUp()
- {
- cleanedUp.countDown();
- }
-
private InterruptedException getOrCreateInterruptedException(StopException stopEx)
{
final InterruptedException causeOrNull = (InterruptedException) stopEx.getCause();
@@ -415,57 +331,54 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
public V call() throws InterruptedException
{
- stopLock.lock();
+ Throwable throwableOrNull = null;
try
{
final V result;
- markStarted();
+ threadGuard.startGuardOrInterrupt();
try
{
result = delegate.call(new IStoppableExecutor<V>()
{
public V execute(Callable<V> callable) throws Exception
{
- stopLock.unlock();
+ threadGuard.allowStopping();
try
{
return callable.call();
} finally
{
- stopLock.lock();
+ threadGuard.preventStopping();
}
}
public void execute(Runnable runnable) throws Exception
{
- stopLock.unlock();
+ threadGuard.allowStopping();
try
{
runnable.run();
} finally
{
- stopLock.lock();
+ threadGuard.preventStopping();
}
}
});
} catch (Throwable th)
{
- markFinished();
- cleanUp(th);
- if (th instanceof StopException)
- {
- throw getOrCreateInterruptedException((StopException) th);
- } else
- {
- throw CheckedExceptionTunnel.wrapIfNecessary(th);
- }
+ throwableOrNull = th;
+ threadGuard.markFinishingOrInterrupt();
+ throw CheckedExceptionTunnel.wrapIfNecessary(th);
}
- markFinished();
- cleanUp();
+ threadGuard.markFinishingOrInterrupt();
return result;
+ } catch (StopException ex)
+ {
+ throw getOrCreateInterruptedException(ex);
} finally
{
- markCleanedUp();
+ cleanUp(throwableOrNull);
+ threadGuard.markFinished();
}
}
@@ -490,28 +403,20 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
}
/**
- * Returns <code>true</code>, if the callable has already started running.
+ * Returns <code>true</code>, if the callable is currently running and <code>false</code>
+ * otherwise.
*/
- public boolean hasStarted()
+ public boolean isRunning()
{
- return waitForStarted(NO_WAIT_MILLIS);
+ return threadGuard.isRunning();
}
/**
- * Waits for the callable to start running. The method waits at most <var>timeoutMillis</var>
- * milli-seconds.
- *
- * @return <code>true</code>, if the callable has started running when the method returns.
+ * Returns <code>true</code>, if the callable has already started running.
*/
- public boolean waitForStarted(long timeoutMillis) throws StopException
+ public boolean hasStarted()
{
- try
- {
- return started.await(timeoutMillis, TimeUnit.MILLISECONDS);
- } catch (InterruptedException ex)
- {
- throw new StopException(ex);
- }
+ return threadGuard.hasStarted();
}
/**
@@ -532,7 +437,7 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
{
try
{
- return finished.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ return threadGuard.waitForFinished(timeoutMillis);
} catch (InterruptedException ex)
{
throw new StopException(ex);
@@ -540,39 +445,14 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
}
/**
- * Returns <code>true</code>, if the callable has already called the
- * {@link ICleaner#cleanUp(ch.systemsx.cisd.common.concurrent.TerminableCallable.FinishCause)}
- * method, if any.
- */
- public boolean hasCleanedUp()
- {
- return waitForCleanedUp(NO_WAIT_MILLIS);
- }
-
- /**
- * Waits for the callable to finish cleaning up. The method waits at most <var>timeoutMillis</var>
- * milli-seconds.
+ * Cancels the callable if it is not yet running.
*
- * @return <code>true</code>, if the callable has finished cleaning up when the method returns.
+ * @return <code>true</code>, if the callable is cancelled and <code>false</code>
+ * otherwise.
*/
- public boolean waitForCleanedUp(long timeoutMillis) throws StopException
+ public boolean cancel()
{
- try
- {
- return cleanedUp.await(timeoutMillis, TimeUnit.MILLISECONDS);
- } catch (InterruptedException ex)
- {
- throw new StopException(ex);
- }
- }
-
- /**
- * Returns <code>true</code>, if the callable is currently running and <code>false</code>
- * otherwise.
- */
- public boolean isRunning()
- {
- return (thread != null);
+ return threadGuard.cancel();
}
/**
@@ -593,58 +473,25 @@ public final class TerminableCallable<V> implements Callable<V>, ITerminable
/**
* Tries to terminate this {@link TerminableCallable}. Note that this is a synchronous call
- * that returns only when either the callable has been terminated (and cleaned up) or when a
- * timeout has occurred. Note also that even when providing <var>timeoutMillis</var> as 0, this
- * method may wait up to <var>timeoutInterruptMillis</var> milli-seconds for the
+ * that returns only when either the callable has been terminated or finished or when a timeout
+ * has occurred. Note also that even when providing <var>timeoutMillis</var> as 0, this method
+ * may wait up to <var>waitInterruptMillis</var> milli-seconds for the
* {@link Thread#interrupt()} call to terminate the callable.
- * <p>
- * The following steps are performed:
- * <ol>
- * <li> Wait for the callable to start up, if that has not happened. If it doesn't start up in
- * due time, return with <code>false</code>. </li>
- * <li> If the callable has already terminated, wait for the clean up to finish and return
- * <code>true</code>, if the clean up finishes in due time and <code>false</code>
- * otherwise. </li>
- * <li> Call <code>Thread.terminate()</code> on the thread that is running the callable. </li>
- * <li> Wait for the callable to terminate in response to the call above (wait for
- * <var>timeoutInterruptMillis</var> milli-seconds, as provided to the
- * {@link #create(ch.systemsx.cisd.common.concurrent.TerminableCallable.ICallable,
- * ch.systemsx.cisd.common.concurrent.TerminableCallable.ICleaner, long, long)} method. </li>
- * <li> If the callable didn't terminate, try to call <code>Thread.stop()</code> on the thread
- * that is running the callable. This can only succeed, if the callable is currently running
- * code in the <var>stoppableExecutor</var>.</li>
- * <li> If either interrupt or stop took effect in due time, wait for clean up to finish. </li>
- * <li> If the clean up finishes in due time, return <code>true</code>, otherwise
- * <code>false</code>. </li>
- * </ol>
*
- * @param timeoutMillis If <code>== 0</code>, the method will wait for the callable to stop,
- * if <code>> 0</code>, the method will wait at most this time.
- * @return <code>true</code>, if the callable is confirmed to be terminated and cleaned up
- * successfully in due time, or <code>false</code>, if a timeout has occurred.
+ * @param timeoutMillis The method will wait at most this time (in milli-seconds).
+ * @return <code>true</code>, if the callable is confirmed to be terminated or finished, or
+ * <code>false</code>, if a timeout has occurred.
* @throws StopException If the current thread is interrupted.
*/
- public synchronized boolean terminate(long timeoutMillis) throws StopException
+ public boolean terminate(long timeoutMillis) throws StopException
{
- final long start = System.currentTimeMillis();
- if (wait(started, timeoutMillis) == false)
- {
- return false;
- }
- final Thread t = thread;
- if (t == null)
+ try
{
- return wait(cleanedUp, timeoutMillis - (System.currentTimeMillis() - start));
- }
- t.interrupt();
- if (wait(finished, waitInterruptMillis) == false)
+ return threadGuard.terminateAndWait(waitInterruptMillis, timeoutMillis);
+ } catch (InterruptedException ex)
{
- if (stop(t, timeoutMillis - (System.currentTimeMillis() - start)) == false)
- {
- return false;
- }
+ throw new StopException(ex);
}
- return wait(cleanedUp, timeoutMillis - (System.currentTimeMillis() - start));
}
}
diff --git a/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableFuture.java b/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableFuture.java
index 3e633555c4f..dc782131a83 100644
--- a/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableFuture.java
+++ b/common/source/java/ch/systemsx/cisd/common/concurrent/TerminableFuture.java
@@ -24,16 +24,12 @@ import java.util.concurrent.TimeoutException;
import ch.systemsx.cisd.common.exceptions.StopException;
/**
- * Implementation of a {@link ITerminableFuture} that delegates to appropriate classes. Note that
- * there is some additional logic in the {@link #terminate()} and {@link #terminate(long)} methods
- * to make them equivalent to {@link Future#cancel(boolean)} if the task did not yet start running.
+ * Implementation of a {@link ITerminableFuture} that delegates to appropriate classes.
*
* @author Bernd Rinn
*/
final class TerminableFuture<V> implements ITerminableFuture<V>
{
- private static final long TINY_PERIOD_MILLIS = 20L;
-
private final Future<V> delegateFuture;
private final TerminableCallable<V> delegateTerminableCallable;
@@ -76,11 +72,6 @@ final class TerminableFuture<V> implements ITerminableFuture<V>
return delegateTerminableCallable.isRunning();
}
- public boolean waitForStarted(long timeoutMillis) throws StopException
- {
- return delegateTerminableCallable.waitForStarted(timeoutMillis);
- }
-
public boolean hasStarted()
{
return delegateTerminableCallable.hasStarted();
@@ -96,42 +87,16 @@ final class TerminableFuture<V> implements ITerminableFuture<V>
return delegateTerminableCallable.hasFinished();
}
- public boolean waitForCleanedUp(long timeoutMillis) throws StopException
- {
- return delegateTerminableCallable.waitForCleanedUp(timeoutMillis);
- }
-
- public boolean hasCleanedUp()
- {
- return delegateTerminableCallable.hasCleanedUp();
- }
-
public boolean terminate()
{
cancel(false);
- // Wait for a very short period of time to ensure that the callable didn't just start
- // running at the very moment when we canceled.
- if (waitForStarted(TINY_PERIOD_MILLIS))
- {
- return delegateTerminableCallable.terminate();
- } else
- {
- return true;
- }
+ return delegateTerminableCallable.terminate();
}
public final boolean terminate(long timeoutMillis) throws StopException
{
cancel(false);
- // Wait for a very short period of time to ensure that the callable didn't just start
- // running at the very moment when we canceled.
- if (waitForStarted(TINY_PERIOD_MILLIS))
- {
- return delegateTerminableCallable.terminate(timeoutMillis);
- } else
- {
- return true;
- }
+ return delegateTerminableCallable.terminate(timeoutMillis);
}
}
diff --git a/common/source/java/ch/systemsx/cisd/common/concurrent/ThreadGuard.java b/common/source/java/ch/systemsx/cisd/common/concurrent/ThreadGuard.java
new file mode 100644
index 00000000000..943411a6ca5
--- /dev/null
+++ b/common/source/java/ch/systemsx/cisd/common/concurrent/ThreadGuard.java
@@ -0,0 +1,257 @@
+/*
+ * Copyright 2008 ETH Zuerich, CISD
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ch.systemsx.cisd.common.concurrent;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import ch.systemsx.cisd.common.exceptions.HardStopException;
+
+/**
+ * A class that provides the framework for guarding a {@link Thread} such that it can be stopped
+ * safely. It will guard the current thread of whoever runs {@link #startGuardOrInterrupt()}.
+ * <p>
+ * A {@link ThreadGuard} instance can only guard a thread once, it can not be reset.
+ *
+ * @author Bernd Rinn
+ */
+final class ThreadGuard
+{
+ private enum State
+ {
+ INITIAL, CANCELLED, TERMINATING, RUNNING, FINISHING
+ }
+
+ /** The lock that guards stopping the thread. */
+ private final Lock stopLock = new ReentrantLock();
+
+ /** This latch signals that the task is finished. */
+ private final CountDownLatch finishedLatch = new CountDownLatch(1);
+
+ private Thread thread;
+
+ private ThreadGuard.State state = State.INITIAL;
+
+ @SuppressWarnings("deprecation")
+ private static void stopNow(Thread t)
+ {
+ t.stop(new HardStopException());
+ }
+
+ // Do not synchronize this or things will stop working!
+ private boolean stop(Thread t, long timeoutMillis) throws InterruptedException
+ {
+ final boolean gotIt;
+ gotIt = stopLock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS);
+ if (waitForFinished(TerminableCallable.NO_WAIT_MILLIS))
+ {
+ // interrupt() took effect, we don't need to stop()
+ return true;
+ }
+ if (gotIt == false)
+ {
+ return false;
+ }
+ try
+ {
+ stopNow(t);
+ return true;
+ } finally
+ {
+ stopLock.unlock();
+ }
+ }
+
+ private synchronized Thread getThreadForTermination()
+ {
+ if (state == State.RUNNING)
+ {
+ state = State.TERMINATING;
+ return thread;
+ } else
+ {
+ return null;
+ }
+ }
+
+ // No need to synchronize these.
+
+ /**
+ * Call this method to ensure the thread can't be stopped.
+ */
+ void preventStopping()
+ {
+ stopLock.lock();
+ }
+
+ /**
+ * Call this method to ensure the thread can be stopped.
+ */
+ void allowStopping()
+ {
+ stopLock.unlock();
+ }
+
+ /**
+ * Mark the guard as being finished. Stopping must not be allowed when calling this method.
+ */
+ void markFinished()
+ {
+ finishedLatch.countDown();
+ stopLock.unlock(); // Just in case we happen to hang in stop()
+ }
+
+ /**
+ * Wait for the guard to be marked finished.
+ *
+ * @return <code>true</code>, if the guard was marked finished in due time and
+ * <code>false</code>, if the wait timed out.
+ * @throws InterruptedException If this thread got interrupted.
+ */
+ boolean waitForFinished(long timeoutMillis) throws InterruptedException
+ {
+ return finishedLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+
+ // Whatever manipulates state or thread needs to be synchronized.
+
+ /**
+ * Mark the guard as being started. The current thread when running this method will be guarded
+ * from now on. Implies {@link #preventStopping()}.
+ *
+ * @throws InterruptedException If the guard is not in initial state (e.g. because it got
+ * cancelled).
+ */
+ synchronized void startGuardOrInterrupt() throws InterruptedException
+ {
+ if (state != State.INITIAL)
+ {
+ Thread.interrupted(); // Clear interrupt flag
+ throw new InterruptedException();
+ }
+ stopLock.lock();
+ state = State.RUNNING;
+ thread = Thread.currentThread();
+ }
+
+ /**
+ * Mark the guard as being in state finishing.
+ *
+ * @throws InterruptedException If this thread is not in state running.
+ */
+ synchronized void markFinishingOrInterrupt() throws InterruptedException
+ {
+ if (state != State.RUNNING)
+ {
+ Thread.interrupted(); // Clear interrupt flag
+ throw new InterruptedException();
+ }
+ state = State.FINISHING;
+ thread = null;
+ }
+
+ /**
+ * Returns <code>true</code>, if the guard has been started.
+ */
+ synchronized boolean hasStarted()
+ {
+ return (state != State.INITIAL && state != State.CANCELLED);
+ }
+
+ /**
+ * Returns <code>true</code>, if the guard is in state running.
+ */
+ synchronized boolean isRunning()
+ {
+ return (state == State.RUNNING);
+ }
+
+ /**
+ * Tries to cancel the guard, i.e. prevent it from running if it doesn't run yet.
+ *
+ * @return <code>true</code>, if the guard has been cancelled successfully.
+ */
+ synchronized boolean cancel()
+ {
+ if (state == State.INITIAL)
+ {
+ state = State.CANCELLED;
+ return true;
+ } else
+ {
+ return false;
+ }
+ }
+
+ /**
+ * Tries to terminate task running in the thread. Note that this is a synchronous call that
+ * returns only when either the guard is marked finished or when a timeout has occurred. Note
+ * also that even when providing <var>timeoutMillis</var> as 0, this method may wait up to
+ * <var>waitInterruptMillis</var> milli-seconds for the {@link Thread#interrupt()} call to
+ * work.
+ * <p>
+ * The following steps are performed:
+ * <ol>
+ * <li> If the guard got cancelled, return with cod<code>true</code>. </li>
+ * <li> If the guard is already in state finishing, wait for the guard to be set to finished and
+ * return <code>true</code>, if that happens in due time and <code>false</code> otherwise.
+ * </li>
+ * <li> Call <code>Thread.terminate()</code> on the thread. </li>
+ * <li> Wait for the guard to be marked finished in response to the call above (wait for
+ * <var>timeoutInterruptMillis</var> milli-seconds. </li>
+ * <li> If the guard didn't get marked as finished, try to call <code>Thread.stop()</code> on
+ * the thread. This can only succeed, if stopping becomes allowed in due time.</li>
+ * <li> If either interrupt or stop took effect in due time, wait for guard to be marked as
+ * finished. </li>
+ * <li> If the guard is marked finished in due time, return <code>true</code>, otherwise
+ * <code>false</code>. </li>
+ * </ol>
+ *
+ * @param waitInterruptMillis The time to wait for <code>interrupt()</code> to terminate the
+ * task.
+ * @param timeoutMillis The method will wait for at most this time for the task to stop.
+ * @return <code>true</code>, if the guard has been marked finished, or <code>false</code>,
+ * if a timeout has occurred.
+ * @throws InterruptedException If the current thread is interrupted.
+ */
+ // Do not synchronize this or things will stop working!
+ boolean terminateAndWait(long waitInterruptMillis, long timeoutMillis)
+ throws InterruptedException
+ {
+ if (cancel())
+ {
+ return true;
+ }
+ final long start = System.currentTimeMillis();
+ final Thread t = getThreadForTermination();
+ if (t != null)
+ {
+ t.interrupt();
+ if (waitForFinished(waitInterruptMillis) == false)
+ {
+ if (stop(t, timeoutMillis - (System.currentTimeMillis() - start)) == false)
+ {
+ return false;
+ }
+ }
+ }
+ return waitForFinished(timeoutMillis - (System.currentTimeMillis() - start));
+ }
+
+}
\ No newline at end of file
diff --git a/common/sourceTest/java/ch/systemsx/cisd/common/concurrent/TerminableCallableTest.java b/common/sourceTest/java/ch/systemsx/cisd/common/concurrent/TerminableCallableTest.java
index 0f5b7832078..6395ba9b004 100644
--- a/common/sourceTest/java/ch/systemsx/cisd/common/concurrent/TerminableCallableTest.java
+++ b/common/sourceTest/java/ch/systemsx/cisd/common/concurrent/TerminableCallableTest.java
@@ -141,7 +141,7 @@ public class TerminableCallableTest
new TestRunnable(launchLatch, milestoneLatch, Strategy.COMPLETE_IMMEDIATELY,
finishLatch);
final TerminableCallable<Object> callableUnderTest = TerminableCallable.create(sensor);
- new Thread(callableUnderTest.asRunnable()).start();
+ new Thread(callableUnderTest.asRunnable(), "complete").start();
finishLatch.await(500L, TimeUnit.MILLISECONDS);
assertTrue(milestoneLatch.await(0, TimeUnit.MILLISECONDS));
assertTrue(describe(sensor.cause), FinishCause.COMPLETED.equals(sensor.cause));
@@ -156,9 +156,9 @@ public class TerminableCallableTest
final TestRunnable sensor =
new TestRunnable(launchLatch, milestoneLatch, Strategy.SLEEP_FOREVER);
final TerminableCallable<Object> callableUnderTest = TerminableCallable.create(sensor);
- new Thread(callableUnderTest.asRunnable()).start();
+ new Thread(callableUnderTest.asRunnable(), "interrupt").start();
launchLatch.await();
- callableUnderTest.terminate();
+ assertTrue(callableUnderTest.terminate(200L));
assertTrue(milestoneLatch.await(0, TimeUnit.MILLISECONDS));
assertTrue(describe(sensor.cause), FinishCause.INTERRUPTED.equals(sensor.cause));
assertEquals(1, sensor.cleanUpCount);
@@ -173,7 +173,7 @@ public class TerminableCallableTest
final TestRunnable sensor =
new TestRunnable(launchLatch, milestoneLatch, Strategy.KEEP_SPINNING);
final TerminableCallable<Object> callableUnderTest = TerminableCallable.create(sensor);
- final Thread t = new Thread(callableUnderTest.asRunnable());
+ final Thread t = new Thread(callableUnderTest.asRunnable(), "terminate failed");
t.start();
launchLatch.await();
assertFalse(callableUnderTest.terminate(200L));
@@ -191,7 +191,7 @@ public class TerminableCallableTest
final TestRunnable sensor =
new TestRunnable(launchLatch, milestoneLatch, Strategy.KEEP_SPINNING_STOPPABLE);
final TerminableCallable<Object> callableUnderTest = TerminableCallable.create(sensor);
- final Thread t = new Thread(callableUnderTest.asRunnable());
+ final Thread t = new Thread(callableUnderTest.asRunnable(), "stop");
t.start();
launchLatch.await();
assertTrue(callableUnderTest.terminate(200L));
@@ -208,11 +208,11 @@ public class TerminableCallableTest
final TestRunnable sensor =
new TestRunnable(launchLatch, milestoneLatch, Strategy.THROW_EXCEPTION);
final TerminableCallable<Object> callableUnderTest = TerminableCallable.create(sensor);
- final Thread t = new Thread(callableUnderTest.asRunnable());
+ final Thread t = new Thread(callableUnderTest.asRunnable(), "throw exception");
t.start();
launchLatch.await();
milestoneLatch.await();
- assertTrue(callableUnderTest.terminate());
+ assertTrue(callableUnderTest.terminate(200L));
assertTrue(milestoneLatch.await(0, TimeUnit.MILLISECONDS));
assertTrue(describe(sensor.cause), FinishCause.EXCEPTION.equals(sensor.cause));
assertEquals(1, sensor.cleanUpCount);
--
GitLab