Java Executor partial shutdown
Question
Lets have one classic Executor
in application. Many parts of application use this executor for some computations, each computation can be cancelled, for this I can call shutdown()
or shutdownNow()
on Executor.
But I want to shutdown only part of tasks in Executor. Sadly I can't have access to Future
objects, they are private part of computation implementation (actually computation is backed by actor framework jetlang)
I want something like Executor wrapper, which I could pass to computation and which should be backed by real Executor. Something like this:
// main application executor
Executor applicationExecutor = Executors.newCachedThreadPool();
// starting computation
Executor computationExecutor = new ExecutorWrapper(applicationExecutor);
Computation computation = new Computation(computationExecutor);
computation.start();
// cancelling computation
computation.cancel();
// shutting down only computation tasks
computationExecutor.shutdown();
// applicationExecutor remains running and happy
Or any other idea?
Solution
For those, who wants good ends: there is final solution, partially based of Ivan Sopov's answer. Luckily jetlang uses for running its tasks only Executor
interface (not ExecutorService
), so I make wrapper class which supports stopping tasks created only by this wrapper.
static class StoppableExecutor implements Executor {
final ExecutorService executor;
final List<Future<?>> futures = Lists.newArrayList();
boolean stopped;
public StoppableExecutor(ExecutorService executor) {
this.executor = executor;
}
void stop() {
this.stopped = true;
synchronized (futures) {
for (Iterator<Future<?>> iterator = futures.iterator(); iterator.hasNext();) {
Future<?> future = iterator.next();
if (!future.isDone() && !future.isCancelled()) {
System.out.println(future.cancel(true));
}
}
futures.clear();
}
}
@Override
public void execute(Runnable command) {
if (!stopped) {
synchronized (futures) {
Future<?> newFuture = executor.submit(command);
for (Iterator<Future<?>> iterator = futures.iterator(); iterator.hasNext();) {
Future<?> future = iterator.next();
if (future.isDone() || future.isCancelled())
iterator.remove();
}
futures.add(newFuture);
}
}
}
}
Using this is pretty straightforward:
ExecutorService service = Executors.newFixedThreadPool(5);
StoppableExecutor executor = new StoppableExecutor(service);
// doing some actor stuff with executor instance
PoolFiberFactory factory = new PoolFiberFactory(executor);
// stopping tasks only created on executor instance
// executor service is happily running other tasks
executor.stop();
That's all. Works nice.
OTHER TIPS
How about having your Computation
be a Runnable
(and run using the provided Executor
) until a boolean flag is set? Something along the lines of :
public class Computation
{
boolean volatile stopped;
public void run(){
while(!stopped){
//do magic
}
public void cancel)(){stopped=true;}
}
What you are doing is essentially stopping the thread. However, it does not get garbage-collected, but is instead re-used because it is managed by the Executor. Look up "what is the proper way to stop a thread?".
EDIT: please note the code above is quite primitive in the sense it assumes the body of the while loop takes a short amount of time. If it does not, the check will be executed infrequently and you will notice a delay between canceling a task and it actually stopping.
Something like this? You may do partial shutdown:
for (Future<?> future : %ExecutorServiceWrapperInstance%.getFutures()) {
if (%CONDITION%) {
future.cancel(true);
}
}
Here is the code:
package com.sopovs.moradanen;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class ExecutorServiceWrapper implements ExecutorService {
private final ExecutorService realService;
private List<Future<?>> futures = new ArrayList<Future<?>>();
public ExecutorServiceWrapper(ExecutorService realService) {
this.realService = realService;
}
@Override
public void execute(Runnable command) {
realService.execute(command);
}
@Override
public void shutdown() {
realService.shutdown();
}
@Override
public List<Runnable> shutdownNow() {
return realService.shutdownNow();
}
@Override
public boolean isShutdown() {
return realService.isShutdown();
}
@Override
public boolean isTerminated() {
return realService.isTerminated();
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return realService.awaitTermination(timeout, unit);
}
@Override
public <T> Future<T> submit(Callable<T> task) {
Future<T> future = realService.submit(task);
synchronized (this) {
futures.add(future);
}
return future;
}
public synchronized List<Future<?>> getFutures() {
return Collections.unmodifiableList(futures);
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
Future<T> future = realService.submit(task, result);
synchronized (this) {
futures.add(future);
}
return future;
}
@Override
public Future<?> submit(Runnable task) {
Future<?> future = realService.submit(task);
synchronized (this) {
futures.add(future);
}
return future;
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
List<Future<T>> future = realService.invokeAll(tasks);
synchronized (this) {
futures.addAll(future);
}
return future;
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
List<Future<T>> future = realService.invokeAll(tasks, timeout, unit);
synchronized (this) {
futures.addAll(future);
}
return future;
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
//don't know what to do here. Maybe this method is not needed by the framework
//than just throw new NotImplementedException();
return realService.invokeAny(tasks);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
//don't know what to do here. Maybe this method is not needed by the framework
//than just throw new NotImplementedException();
return realService.invokeAny(tasks, timeout, unit);
}
}