ScheduledThreadPoolExecutor scheduleWithFixedDelay and "urgent" execution

Question

I have the following problem that the standard library doesn't solve well, and I'm wondering if anybody has seen another library out there than can do it so I don't need to hack together a custom solution. I have a task that is currently scheduled on a thread pool using scheduleWithFixedDelay(), and I need to modify the code to handle requests for "urgent" execution of the task related to asynchronous events. Thus, if the task is scheduled to occur with a delay of 5 minutes between executions, and an event occurs 2 minutes after the last completed execution, I would like to execute the task immediately and then have it wait for 5 minutes after the completion of the urgent execution before it runs again. Right now the best solution that I can come up with is to have the event handler call cancel() on the ScheduledFuture object returned by scheduleWithFixedDelay() and execute the task immediately, and then set a flag in the task to tell it to reschedule itself with the same delay parameters. Is this functionality available already and I'm just missing something in the documentation?

Answer 1

If you are using ScheduledThreadPoolExecutor there is a method decorateTask (well in fact there are two, for Runnable and Callable tasks) that you can override to store a reference to the task somewhere.

When you need urgent execution, you just call run() on that reference which makes it run and rescheduled with same delay.

A quick hack-up attempt:

public class UrgentScheduledThreadPoolExecutor extends
        ScheduledThreadPoolExecutor {
    RunnableScheduledFuture scheduledTask;

    public UrgentScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize);
    }

    @Override
    protected  RunnableScheduledFuture decorateTask(Runnable runnable,
            RunnableScheduledFuture task) {
        scheduledTask = task;
        return super.decorateTask(runnable, task);
    }

    public void runUrgently() {
        this.scheduledTask.run();
    }
}

which can be used like this:

public class UrgentExecutionTest {

    public static void main(String[] args) throws Exception {
        UrgentScheduledThreadPoolExecutor pool = new UrgentScheduledThreadPoolExecutor(5);

        pool.scheduleWithFixedDelay(new Runnable() {
            SimpleDateFormat format = new SimpleDateFormat("ss"); 

            @Override
            public void run() {
                System.out.println(format.format(new Date()));
            }
        }, 0, 2L, TimeUnit.SECONDS);
        Thread.sleep(7000);
        pool.runUrgently();
        pool.awaitTermination(600, TimeUnit.SECONDS);
    }
}

and produces the following output: 06 08 10 11 13 15

Answer 2

as requested (soz, in a hurry) my EventBasedExecutor

Warning: This currently only works for tasks that are scheduled in a periodic run. You can change the code to handle all the tasks, I so far haven't because I only have the periodically run task. I also run this in a signle-threaded threadpool (I only need one scheduled runner thread that is that runs in one dedicated thread all the time every X seconds)

Here we go:

public class EventBasedExecutor extends ScheduledThreadPoolExecutor implements EventBasedExecutorService {

    private List<RunnableScheduledFuture<?>> workers = new ArrayList<>();

    private int index;

    public EventBasedExecutor(int corePoolSize) {
        super(corePoolSize,  new ThreadFactoryBuilder().setDaemon(true).setNameFormat("message-sender-%d").build());
    }

    @Override
    protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
        if(!workers.contains(runnable)) {
            workers.add(task);
        }
        return super.decorateTask(runnable, task);
    }

    @Override
    public void executeEarly() {
        if(index >= workers.size()) {
            index = 0;
        }

        if(workers.size() == 0) {
            return;
        }

        RunnableScheduledFuture<?> runnableScheduledFuture = workers.get(index);
        index ++;
        execute(runnableScheduledFuture);
        System.out.println("Executing");
    }


    public static void main(String[] args) throws InterruptedException {
        EventBasedExecutor executor = new EventBasedExecutor(10);

        long currentTimeMillis = System.currentTimeMillis();

        // this will never run
        executor.scheduleAtFixedRate(() -> {
            System.out.println("hello");
        }, 5000, 5000, TimeUnit.HOURS);

        executor.executeEarly();

        System.out.println("Run after: " + (System.currentTimeMillis() - currentTimeMillis));
    }

}

This will execute the task in the dedicated worker thread.

It will print:

Executing
hello
Run after: 39

Have fun hacking :)

artur