Rhythmic task controller

Question

The purpose of a rhythmic task controller is to accept a task and execute all the queued tasks in a cycle or "rhythm" defined by the client code. A task returns true or false depending on whether or not it is "finished". A finished task should be removed from the controller's control and won't be executed again while an unfinished task should be executed on the next cycle.

I don't think my solution is as well optimized as it could be, and this is a solution that needs to be as fast as possible as it serves as the core task controller in my game framework!

import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public final class RhythmicTaskController implements TaskController, Runnable {

    private final ExecutorService service;
    private final ConcurrentLinkedQueue<Task> queue = new ConcurrentLinkedQueue<>();

    public RhythmicTaskController(int threads, int delay, TimeUnit unit) {
        service = Executors.newFixedThreadPool(threads);

        newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this, delay, delay, unit);
    }

    @Override
    public void offer(Task task) {
        queue.offer(task);
    }

    @Override
    public void run() {
        int initial = queue.size();
        ConcurrentMap<Future<Boolean>, Task> futures = new ConcurrentHashMap<>(initial);
        queue.forEach(task -> futures.put(service.submit(task::finish), task));
        futures.forEach((r, task) -> {
            try {
                if (r.get())
                    queue.remove(task);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

}

Answer 1

and this is a solution that needs to be as fast as possible as it serves as the core task controller in my game framework!

Not necessarily. It all depends on how much a single task takes on the average. Unless your tasks are very fast, fooling around with their queue doesn't matter.

---

@Override
public void offer(Task task) {
    queue.offer(task);
}

Don't use offer without checking the return value. It's fine as long as you use ConcurrentLinkedQueue, but when you switch implementations, it stops working without any warning. Use something like

com.google.common.base.Verify(queue.offer(task));

---

Unless removals are common, I guess that your original implementation is faster than SpiderPig's solution removing and re-adding tasks on each iterations. Usually, iterating a collection is faster than modifying it (twice). But it's just a guess... measuring would tell you, but measuring Java performance is hard.

Answer 2

You could remove all elements from the queue and then offer them back if necessary.

public void run() {
    ArrayList<Task> tasks = new ArrayList<>();
    {
      Task task;
      while((task = queue.poll()) != null) {
        tasks.add(task);
      }
    }
    for(Task task: tasks) {
      service.submit(() -> {
        if(!task.finish()) offer(task);
      });
    }
}