How to Implement a DAG-like Scheduler in Java?

Question

I want to implement a simple DAG-like scheduler in Java (no result needed), described as the following graph:

I can simply use manual code to achieve this:

ExecutorService executor = Executors.newCachedThreadPool();
Future<?> futureA = executor.submit(new Task("A"));
Future<?> futureC = executor.submit(new Task("C"));
futureA.get();
Future<?> futureB = executor.submit(new Task("B"));
futureB.get();
futureC.get();
Future<?> futureD = executor.submit(new Task("D"));
futureD.get();

But I'm looking for a more general way to do this, so I can use the scheduler like this:

Container container = new Container();
container.addTask("A", new Task("A"));
container.addTask("B", new Task("B"), "A");
container.addTask("C", new Task("C"));
container.addTask("D", new Task("D"), "B", "C");
container.waitForCompletion();

And actually I've already implement a simple one:

https://github.com/jizhang/micro-scheduler/blob/master/src/main/java/com/shzhangji/micro_scheduler/App.java

But I need to iterate all the tasks every 100ms to see which one is ready to be submitted. Also in this implementation there's no exception checking.

I also checkout the Guava lib's ListenableFuture, but I don't know how to use it properly.

Any suggestions on how to implement a DAG, or recommending an existing opensource scheduler will be appreciated.

Answer 1

What you're looking for can be done using the google's guava library and it's listenable future interface. ListenableFutures allow you to have complex chains of asynchronous operations. You should implement a listenable future to execute task D once task B and C are completed using the allAsList method.

The documentation for Listenable Futures: https://code.google.com/p/guava-libraries/wiki/ListenableFutureExplained

A tutorial on Listenable Futures: http://www.javacodegeeks.com/2013/02/listenablefuture-in-guava.html

An example of using the allAsList, chain, and transform methods: http://codingjunkie.net/google-guava-futures/

Answer 2

Dexecutor (Disclaimer : I am the owner) is the library you are looking for, Here is an example

public class WorkFlowManager {

    private final Dexecutor<String, Boolean> dexecutor;

    public WorkFlowManager(ExecutorService executorService) {
        this.dexecutor = buildDexecutor(executorService);

        buildGraph();
    }

    private Dexecutor<String, Boolean> buildDexecutor(final ExecutorService executorService) {
        DexecutorConfig<String, Boolean> config = new DexecutorConfig<>(executorService, new WorkFlowTaskProvider());
        return new DefaultDexecutor<>(config);
    }

    private void buildGraph() {
        this.dexecutor.addDependency(TaskOne.NAME, TaskTwo.NAME);
        this.dexecutor.addDependency(TaskTwo.NAME, TaskThree.NAME);
        this.dexecutor.addDependency(TaskTwo.NAME, TaskFour.NAME);
        this.dexecutor.addDependency(TaskTwo.NAME, TaskFive.NAME);
        this.dexecutor.addDependency(TaskFive.NAME, TaskSix.NAME);
        this.dexecutor.addAsDependentOnAllLeafNodes(TaskSeven.NAME);
    }

    public void execute() {
        this.dexecutor.execute(ExecutionConfig.TERMINATING);
    }
}

Which would build the following graph, and execution would proceed accordingly.

Refer How Do I? for more detail

Why Dexecutor

• Ultra light weight
• Ultra Fast
• Immediate/Scheduled Retry Logic supported
• Non terminating behaviour supported
• Conditionally skip the task execution
• Good test coverage to keep you safe from harm
• Available in maven central
• Good amount of documentation
• Distribute Execution Supported (Ignite, Hazelcast, Infinispan)

Usefull links

• Dexecutor Blogs
• Dexecutor Website
• Dexecutor Wiki
• Dexecutor Source Code