4
\$\begingroup\$

To execute tasks sequentially with a timeout I use ExecutorService.submit() and Future.get(). As I don't want to block calling clients, I implement producer–consumer pattern with a queue into a Thread while(true) loop:

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ExecutorWithQueue {
    private int timeout = 1;
    private TimeUnit timeoutUnit = TimeUnit.SECONDS;
    private LinkedBlockingQueue<Task> queue = new LinkedBlockingQueue<Task>();
    ExecutorService executor = Executors.newSingleThreadExecutor();
    private TaskProcessor taskProcessor = new TaskProcessor();

    public ExecutorWithQueue(int timeout, TimeUnit timeoutUnit)
    {
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;

        taskProcessor.start();
    }

    public void push(Task task)
    {
        queue.add(task);
    }

    class TaskProcessor extends Thread
    {
        public void run() {
            while (true)
            {
                try
                {
                    final Task queueTask = queue.take();
                    Future<Boolean> future = executor.submit(queueTask);

                    try {
                        future.get(timeout, timeoutUnit);
                    } catch (TimeoutException e) {
                        System.out.println("TimeoutException");
                    }
                    catch (ExecutionException e)
                    {
                        System.out.println("ExecutionException");
                    }
                }
                catch (InterruptedException e)
                {
                    break;
                }
            }
            executor.shutdownNow();
        }
    }
}

I have two unit tests. The first to check if task is executed, the second to check that a task that takes two long doesn't block the execution flow:

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.junit.Assert.assertEquals;
import org.junit.Test;

public class TestExecutorWithQueue {

    ExecutorWithQueue executor = new ExecutorWithQueue(9, MILLISECONDS);

    @Test public void fastTaskIsExecuted() throws Exception {
        assertExecuted(0, true);
    }

    @Test public void slowTaskIsNotExecuted() throws Exception {
        assertExecuted(50, false);
    }

    public void assertExecuted(int waiting, boolean expected) throws InterruptedException
    {
        WaitingTask task = new WaitingTask(waiting);

        executor.push(task);
        Thread.sleep(100);

        assertEquals(expected, task.executed);
    }


    class WaitingTask implements Task {
        boolean executed;
        int seconds_to_wait;

        public WaitingTask(int milliseconds_to_wait) {
            this.executed = false;
            this.milliseconds_to_wait = milliseconds_to_wait;
        }

        @Override
        public Boolean call() throws Exception {
            Thread.sleep(this.milliseconds_to_wait);
            this.executed=true;
            return executed;
        }
    }
}

What do you think of this code?

\$\endgroup\$
4
\$\begingroup\$

TaskProcessor

  • I see little reason why this should extend Thread, rather than simply implement Runnable
  • as a thread it is responsive to interruption, yet there is no way a client can interrupt this thread, through API nor directly. No, I'm mistaken, a client could push a task that interrupts its own thread, yet that is not an obvious way to shut it down, nor a clean one (as the ExecutorWithQueue will not know its executor has stopped working).
  • I would replace while (true) with while (!Thread.currentThread().isInterrupted())
  • I'm not happy with the way the ExecutionException is eaten. Clients have no way to know about the Exception that occurred in the Task.
  • As an extension of the previous point : you deny your clients the benefits of a Future to monitor their submitted tasks.
  • you accept only Tasks, yet tere is no reason you couldn't handle all Callables.

ExecutorWithQueue

  • taskProcessor does not need to be a field
  • The values supplied to timeout and timeoutUnit in their declaration are always overwritten in the constructor. If you omit these initialization, you can make them final.
  • the queue and executor fields can also be final.

Test

  • Fiddling with sleep() is error prone. The test can fail if the sleep is too short, and all the time that the test sleeps in excess of what is needed delays the swift completion of the unit test suite. (in projects with many thread safety tests this can really make the total suite of tests too slow). Timing combinations are often more reliably simulated using CountDownLatches, and these will never take longer than needed.
  • I find it good practice to add a timeout value to the @Test annotation for these kinds of tests. If at some point deadlock or so occurs, the test will fail gracefully, rather than hold up the test server.

Semantics

I'm completely puzzled by what this class is intended to do. All submitted tasks will simply be executed in sequence. But that could simply be achieved by submitting them to an ExecutorService with just one thread, which already uses a queue to store tasks it has yet to run. When the timeout passes and the task isn't finished, you submit the next task, which will simply be queued by the underlying ExecutorService until the timed out task has finished. I suspect you somehow think the timeout will cancel the task, but it won't. If that is your intention, you'll need to add cancelling logic to handle a timeout, if it isn't there doesn't seem to be much point in this class.

The second test seems to assert that a timed out task does not execute. Yet it only asserts this after having slept a certain amount of time. If you increase the sleep time enough, the assertion will fail. (this is another argument pro CountDownLatch as that will communicate the test's intention better).

Hope you found this useful.

\$\endgroup\$
  • \$\begingroup\$ I'm trying to implement a "fire and forget" behavior in my push() method and future.get() blocks the submission of other task until completion. \$\endgroup\$ – Philippe Blayo Sep 5 '13 at 18:14
  • \$\begingroup\$ Nothing forces you to call future.get() prior to submitting all tasks, or to call it at all. \$\endgroup\$ – bowmore Sep 5 '13 at 18:31
  • \$\begingroup\$ Some of my tasks never end. I need to set a timeout to prevent them from blocking the next tasks. Future.get() has a timeout, executor.submit() hasn't. \$\endgroup\$ – Philippe Blayo Sep 6 '13 at 8:12
  • 1
    \$\begingroup\$ Then your current code does not achieve that, since simply waiting until a certain timeout on the Future does not cancel the corresponding task in case of a timeout. You can call Future.cancel() and those tasks should be responsive to interruption. \$\endgroup\$ – bowmore Sep 6 '13 at 9:08

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.