ThreadFlock defines the {@link #open(String) open} method to open a new flock, * the {@link #start(Thread) start} method to start a thread in the flock, and the * {@link #close() close} method to close the flock. The {@code close} waits for all * threads in the flock to finish. The {@link #awaitAll() awaitAll} method may be used * to wait for all threads to finish without closing the flock. The {@link #wakeup()} * method will cause {@code awaitAll} method to complete early, which can be used to * support cancellation in higher-level APIs. ThreadFlock also defines the {@link * #shutdown() shutdown} method to prevent new threads from starting while allowing * existing threads in the flock to continue. * *
Thread flocks are intended to be used in a structured manner. The * thread that opens a new flock is the {@link #owner() owner}. The owner closes the * flock when done, failure to do so may result in a resource leak. The {@code open} * and {@code close} should be matched to avoid closing an enclosing flock * while a nested flock is open. A ThreadFlock can be used with the * try-with-resources construct if required but more likely, the close method of a * higher-level API that implements {@link AutoCloseable} will close the flock. * *
Thread flocks are conceptually nodes in a tree. A thread {@code T} started in * flock "A" may itself open a new flock "B", implicitly forming a tree where flock * "A" is the parent of flock "B". When nested, say where thread {@code T} opens * flock "B" and then invokes a method that opens flock "C", then the enclosing * flock "B" is conceptually the parent of the nested flock "C". ThreadFlock does * not define APIs that exposes the tree structure. It does define the {@link * #containsThread(Thread) containsThread} method to test if a flock contains a * thread, a test that is equivalent to testing membership of flocks in the tree. * The {@code start} and {@code shutdown} methods are confined to the flock * owner or threads contained in the flock. The confinement check is equivalent to * invoking the {@code containsThread} method to test if the caller is contained * in the flock. * *
Unless otherwise specified, passing a {@code null} argument to a method
* in this class will cause a {@link NullPointerException} to be thrown.
*/
public class ThreadFlock implements AutoCloseable {
private static final JavaLangAccess JLA = SharedSecrets.getJavaLangAccess();
private static final VarHandle THREAD_COUNT;
private static final VarHandle PERMIT;
static {
MethodHandles.Lookup l = MethodHandles.lookup();
THREAD_COUNT = MhUtil.findVarHandle(l, "threadCount", int.class);
PERMIT = MhUtil.findVarHandle(l, "permit", boolean.class);
}
private final Set This method captures the current thread's {@linkplain ScopedValue scoped value}
* bindings for inheritance by threads created in the flock.
*
* For the purposes of containment, monitoring, and debugging, the parent
* of the new flock is determined as follows:
* The thread is started with the scoped value bindings that were captured
* when opening the flock. The bindings must match the current thread's bindings.
*
* This method may only be invoked by the flock owner or threads {@linkplain
* #containsThread(Thread) contained} in the flock.
*
* @param thread the unstarted thread
* @return the thread, started
* @throws IllegalStateException if this flock is shutdown or closed
* @throws IllegalThreadStateException if the given thread was already started
* @throws WrongThreadException if the current thread is not the owner or a thread
* contained in the flock
* @throws StructureViolationException if the current scoped value bindings are
* not the same as when the flock was created
*/
public Thread start(Thread thread) {
ensureOwnerOrContainsThread();
JLA.start(thread, container);
return thread;
}
/**
* Shutdown this flock so that no new threads can be started, existing threads
* in the flock will continue to run. This method is a no-op if the flock is
* already shutdown or closed.
*/
public void shutdown() {
if (!shutdown) {
shutdown = true;
}
}
/**
* Wait for all threads in the flock to finish executing their tasks. This method
* waits until all threads finish, the {@link #wakeup() wakeup} method is invoked,
* or the current thread is interrupted.
*
* This method may only be invoked by the flock owner. The method trivially
* returns true when the flock is closed.
*
* This method clears the effect of any previous invocations of the
* {@code wakeup} method.
*
* @return true if there are no threads in the flock, false if wakeup was invoked
* and there are unfinished threads
* @throws InterruptedException if interrupted while waiting
* @throws WrongThreadException if invoked by a thread that is not the owner
*/
public boolean awaitAll() throws InterruptedException {
ensureOwner();
if (getAndSetPermit(false))
return (threadCount == 0);
while (threadCount > 0 && !permit) {
LockSupport.park();
if (Thread.interrupted())
throw new InterruptedException();
}
clearPermit();
return (threadCount == 0);
}
/**
* Wait, up to the given waiting timeout, for all threads in the flock to finish
* executing their tasks. This method waits until all threads finish, the {@link
* #wakeup() wakeup} method is invoked, the current thread is interrupted, or
* the timeout expires.
*
* This method may only be invoked by the flock owner. The method trivially
* returns true when the flock is closed.
*
* This method clears the effect of any previous invocations of the {@code wakeup}
* method.
*
* @param timeout the maximum duration to wait
* @return true if there are no threads in the flock, false if wakeup was invoked
* and there are unfinished threads
* @throws InterruptedException if interrupted while waiting
* @throws TimeoutException if the wait timed out
* @throws WrongThreadException if invoked by a thread that is not the owner
*/
public boolean awaitAll(Duration timeout)
throws InterruptedException, TimeoutException {
Objects.requireNonNull(timeout);
ensureOwner();
if (getAndSetPermit(false))
return (threadCount == 0);
long startNanos = System.nanoTime();
long nanos = NANOSECONDS.convert(timeout);
long remainingNanos = nanos;
while (threadCount > 0 && remainingNanos > 0 && !permit) {
LockSupport.parkNanos(remainingNanos);
if (Thread.interrupted())
throw new InterruptedException();
remainingNanos = nanos - (System.nanoTime() - startNanos);
}
boolean done = (threadCount == 0);
if (!done && remainingNanos <= 0 && !permit) {
throw new TimeoutException();
} else {
clearPermit();
return done;
}
}
/**
* Causes the call to {@link #awaitAll()} or {@link #awaitAll(Duration)} by the
* {@linkplain #owner() owner} to return immediately.
*
* If the owner is blocked in {@code awaitAll} then it will return immediately.
* If the owner is not blocked in {@code awaitAll} then its next call to wait
* will return immediately. The method does nothing when the flock is closed.
*/
public void wakeup() {
if (!getAndSetPermit(true) && Thread.currentThread() != owner()) {
LockSupport.unpark(owner());
}
}
/**
* Closes this flock. This method first shuts down the flock to prevent
* new threads from starting. It then waits for the threads in the flock
* to finish executing their tasks. In other words, this method blocks until
* all threads in the flock finish.
*
* This method may only be invoked by the flock owner.
*
* If interrupted then this method continues to wait until all threads
* finish, before completing with the interrupt status set.
*
* A ThreadFlock is intended to be used in a structured manner. If
* this method is called to close a flock before nested flocks are closed then it
* closes the nested flocks (in the reverse order that they were created in),
* closes this flock, and then throws {@code StructureViolationException}.
* Similarly, if this method is called to close a thread flock while executing with
* scoped value bindings, and the thread flock was created before the scoped values
* were bound, then {@code StructureViolationException} is thrown after closing the
* thread flock.
*
* @throws WrongThreadException if invoked by a thread that is not the owner
* @throws StructureViolationException if a structure violation was detected
*/
public void close() {
ensureOwner();
if (closed)
return;
// shutdown, if not already shutdown
if (!shutdown)
shutdown = true;
// wait for threads to finish
boolean interrupted = false;
try {
while (threadCount > 0) {
LockSupport.park();
if (Thread.interrupted()) {
interrupted = true;
}
}
} finally {
try {
container.close(); // may throw
} finally {
closed = true;
if (interrupted) Thread.currentThread().interrupt();
}
}
}
/**
* {@return true if the flock has been {@linkplain #shutdown() shut down}}
*/
public boolean isShutdown() {
return shutdown;
}
/**
* {@return true if the flock has been {@linkplain #close() closed}}
*/
public boolean isClosed() {
return closed;
}
/**
* {@return a stream of the threads in this flock}
* The elements of the stream are threads that were started in this flock
* but have not terminated. The stream will reflect the set of threads in the
* flock at some point at or since the creation of the stream. It may or may
* not reflect changes to the set of threads subsequent to creation of the
* stream.
*/
public Stream
*
*
* @param name the name of the flock, can be null
* @return a new thread flock
*/
public static ThreadFlock open(String name) {
var flock = new ThreadFlock(name);
flock.container.push();
return flock;
}
/**
* {@return the name of this flock or {@code null} if unnamed}
*/
public String name() {
return name;
}
/**
* {@return the owner of this flock}
*/
public Thread owner() {
return container.owner();
}
/**
* Starts the given unstarted thread in this flock.
*
*