Futures

Introduction

A KOS "future" is a way of running tasks that don’t block the main thread and can notify other components when they are done or encounter errors. The FutureWork class is the main implementation of this concept, and it works similarly to the FutureTask class in Java.

Overview

This section gives a brief overview of KOS futures and callbacks.

Background

In KOS, there is a need to run asynchronous tasks. Requirements include:

Coordinate physical operations (as opposed to the propagation of data)
Support user feedback, such as progress and estimated completion time
Ability to group tasks together to run sequentially, in parallel, or in a nested graph
Make callbacks when task events (START, ABORT, DONE, etc) occur
Support operations that can be cancelled by the user
Support operations that can be aborted by internal software or hardware problems
Ability to handle failed, cancelled, or aborted tasks, which typically involves a good deal of hardware and state cleanup

For these and other reasons, KOS developed its own asynchronous task handling.

For a more detailed description, please read the FutureWork Javadocs.

Software Components

The components related to futures are:

FutureWork: Class that executes asynchronous tasks and issues callbacks.
FutureEvent: Enumeration of all possible FutureWork callback events.
FutureState: Enumeration of all possible FutureWork end states.
FutureRunnable: Functional interface for associating runnable code with a future.
FailedFuture: Future that immediately fails.
SequencedFuture: A grouping class that allows a list of futures to run in sequence.
ParallelFuture: A grouping class that allows a collection of futures to run in parallel.

API Components

This section gives an overview of all future-related components.

FutureWork (class)

The FutureWork class is the workhorse of this API. Here is its code overview:

API: FutureWork class

public class FutureWork implements Runnable, Tracker {

    // Constructors:
    public FutureWork(String name, long estimatedTimeMs);
    public FutureWork(String name, FutureRunnable runnable, long estimatedTimeMs);
    public FutureWork(String name, FutureRunnable runnable);

    // Misc runnable methods:
    public FutureWork setRunnable(FutureRunnable runnable);
    public final void run();
    public boolean waitUntilFinished(long timeoutMillis);

    // Mark work successful:
    public void success();

    // Mark work cancelled:
    public void cancel(String reason);
    public void cancel(String reason, Object data, Class<?> view);
    public void cancel(String reason, ReasonData reasonData);

    // Mark work aborted:
    public void abort(String reason);
    public void abort(String reason, Object data, Class<?> view);
    public void abort(String reason, ReasonData reasonData);

    // Mark work failed:
    public void fail(String reason);
    public void fail(String reason, Object data, Class<?> view);
    public void fail(String reason, ReasonData reasonData);

    // Other ways to set the state:
    public void setState(FutureState state, String reason, Object data, Class<?> view);
    public void setState(FutureState state, String reason, ReasonData reasonData);

    // General getters and setters:
    public int getId();
    public String getName();
    public void setParent(FutureWork future);
    public FutureState getEndState();
    public String getReason();
    public ReasonData getReasonData();
    public int getProgress();
    public void setProgress(int progress);
    public Object getData();
    public FutureWork setData(Object data);
    public String getTracker();
    public JsonViewWrapper getClientData();

    // Statuses:
    public boolean isRunStarted();
    public boolean isRunComplete();
    public boolean isSuccess();
    public boolean isFail();
    public boolean isCancel();
    public boolean isAbort();
    public boolean isDone();
    public boolean isTerminate();
    public boolean isInterrupted();

    // Interrupts:
    public boolean isInterruptable();
    public void setInterruptable(boolean interruptable) throws InterruptedException;
    public void interruptIfPending() throws InterruptedException;

    // Append / prepend / remove:
    public FutureWork append (String name, FutureEvent event, FutureRunnable callback);
    public FutureWork prepend(String name, FutureEvent event, FutureRunnable callback);
    public FutureWork remove (String name);

    // Timeout methods:
    public FutureWork setTimeout(long delay);
    public FutureWork setTimeout(long delay, String reason);
    public FutureWork setTimeout(long delay, FutureRunnable callback);
    public void cancelTimeout();
    public int getAbortAbandonedTimeoutMs();
    public void setAbortAbandonedTimeoutMs(int ms);
    public void disableAbortAbandoned();

    // Time methods:
    public long getStartTimeMono();
    public long getEstimatedTimeMs();
    public FutureWork setEstimatedTimeMs(long estimatedTimeMs);
    public FutureWork setRemainingTimeMs(long remainingTimeMs);
    public long getRemainingTimeMs();
    public long getEstimatedEndTimeMono();

    // Misc:
    public String toString();
}

FutureState (enum)

A future operation ends in one of four FutureState values:

Table 1. `FutureState` enumeration
State	Description
SUCCESS	the operation successfully ran to completion
FAIL	the operation ran to completion but failed
CANCEL	the operation was cancelled by the user
ABORT	the operation was aborted by the software

Here’s the code:

API: FutureState enum

public enum FutureState {

    // The four possible future states:
    SUCCESS(FutureEvent.SUCCESS),
    FAIL   (FutureEvent.FAIL),
    CANCEL (FutureEvent.CANCEL),
    ABORT  (FutureEvent.ABORT);

    private final FutureEvent futureEvent;

    FutureState(FutureEvent futureEvent) {
        this.futureEvent = futureEvent;
    }

    // Convenience methods:
    public boolean isSuccess()   { return (this == SUCCESS); }
    public boolean isFail()      { return (this == FAIL);    }
    public boolean isCancel()    { return (this == CANCEL);  }
    public boolean isAbort()     { return (this == ABORT);   }
    public boolean isDone()      { return (isSuccess() || isFail());  }
    public boolean isTerminate() { return (isCancel()  || isAbort()); }

    public FutureEvent getFutureEvent();
    public String toString();
}

FutureEvent (enum)

Every future operation goes through a number of FutureEvent events:

Table 2. `FutureEvent` enumeration
Event	Description
START	when the future work is started
SUCCESS	when the future work’s `success()` method is called
FAIL	when one of the future work’s `fail()` methods is called
CANCEL	when one of the future work’s `cancel()` methods is called
ABORT	when one of the future work’s `abort()` methods is called
DONE	called after the SUCCESS or FAIL events
TERMINATE	called after the CANCEL or ABORT events
COMPLETE	called after the DONE or TERMINATE events
FINISHED	called after the COMPLETE event
INTERRUPT	called on CANCEL or ABORT if the process was still running

Here’s the code:

API: FutureEvent enum

public enum FutureEvent {
    START,
    SUCCESS,
    FAIL,
    CANCEL,
    ABORT,
    DONE,
    TERMINATE,
    COMPLETE,
    FINISHED,
    INTERRUPT
}

This is how future events flow:

Figure 1. How events flow in Futures

You can easily attach callback method(s) to any number of these events using either of the FutureWork append() or prepend() methods. A callback can be removed using the remove() method.

Use the COMPLETE event instead of FINISHED

Be aware that FINISHED is generally reserved for special logic such as chaining work together. This event assumes that all user callbacks have been completed, and all work related to the future is finished, so that other futures may safely consider this future and all related callbacks fully concluded. Using this event to do other work can result in race conditions that are very hard to debug. Instead, use the COMPLETE event.

FutureRunnable (funct interface)

The FutureRunnable is a functional interface for associating runnable code with a future.

It has a single method:

API: FutureRunnable interface

@FunctionalInterface
public interface FutureRunnable {
    void run(FutureWork future) throws Exception;
}

FailedFuture (class)

API: FailedFuture class

public class FailedFuture extends FutureWork {

    // Constructors:
    public FailedFuture(String name, String reason);
    public FailedFuture(String name, String reason, ReasonData data);
}

SequencedFuture (class)

API: SequencedFuture class

public class SequencedFuture extends FutureWork {

    // Constructor:
    public SequencedFuture(String name);

    // What to do if any child task fails:
    public void setFailState(FutureState state);

    // Add child tasks:
    public void add(FutureWork future);
    public void add(int index, FutureWork future);

    // Index of currently running task:
    public int getIndex();
    public void setIndex(int index);

    // Causes any and all remaining tasks to be skipped:
    public void skipToEnd();

    // Indicates if any child task failed:
    public boolean childFailed();
    public String getChildReason();

    // Estimated time (the sum of all child tasks):
    public long getEstimatedTimeMs()

    // These methods throw an UnsupportedOperationException:
    public FutureWork setRunnable(FutureRunnable runnable);
    public void success();
    public void fail(String reason);
}

ParallelFuture (class)

API: ParallelFuture class

public class ParallelFuture extends FutureWork {

    // Constructor:
    public ParallelFuture(String name);

    // What to do if any child task fails:
    public void setFailState(FutureState state);

    // What to do if any child task aborts:
    public void setAbortState(FutureState state);

    // Add child tasks:
    public void add(FutureWork future);

    // Estimated time (equal to the longest child task):
    public long getEstimatedTimeMs();

    // These methods throw an UnsupportedOperationException:
    public FutureWork setRunnable(FutureRunnable runnable);
    public void success();
    public void fail(String reason);
}

Example Code

This section provides sample code that demonstrates typical usages of this API.

Source code for this and most other Java reference pages is available on GitHub.

In the kos-java-examples-core sample code, make the following modification:

In the MyKosApp class, initModole() method, instantiate the FutureModule:

    public void initModule() {
        module = new FutureModule();
    }

Basic Futures

In this FutureExamples1 sample code, we are going to run four futures:

One runs to successful completion
One is failed
One is cancelled
One is aborted

With each of these situations, we examine the 10 future events and their associated callbacks.

FutureExamples1 class

package com.example.futures;

import com.tccc.kos.commons.util.KosUtil;
import com.tccc.kos.commons.util.concurrent.future.FutureEvent;
import com.tccc.kos.commons.util.concurrent.future.FutureWork;
import com.tccc.kos.commons.util.misc.Str;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class FutureExamples1 {

    public static final int DURATION = 2000;
    public static final int EXTRA = 200;
    public static final int TIMEOUT = DURATION + EXTRA;

    public void run() {
        log.info("> FutureExamples1.run()");
        runWithAllFourEndStates();
    }

    /**
     * Runs four tasks, demonstrating the different end states and event calls.
     */
    private void runWithAllFourEndStates() {

        FutureWork future1 = createFuture("fw-1"); (1)
        future1.run();
        future1.waitUntilFinished(TIMEOUT);

        FutureWork future2 = createFuture("fw-2"); (2)
        future2.setTimeout(1000, f -> f.fail("Something went wrong"));
        future2.run();
        future2.waitUntilFinished(TIMEOUT);

        FutureWork future3 = createFuture("fw-3"); (3)
        future3.run();
        KosUtil.sleep(1000);
        future3.cancel("The user changed his/her mind");
        future3.waitUntilFinished(TIMEOUT);

        FutureWork future4 = createFuture("fw-4"); (4)
        future4.run();
        KosUtil.sleep(1000);
        future4.abort("An internal operation unexpectedly failed");
        future4.waitUntilFinished(TIMEOUT);
    }

    protected static FutureWork createFuture(String name) { (5)
        FutureWork future = new FutureWork(name, f -> {
            KosUtil.scheduleCallback(f::success, DURATION);
        }, DURATION);
        appendAllEvents(future);
        return future;
    }

    protected static void appendAllEvents(FutureWork future) {
        //@formatter:off
        future.append("started",     FutureEvent.START,     f -> logIt(f, "STARTED"    ));
        future.append("successful",  FutureEvent.SUCCESS,   f -> logIt(f, "SUCCESSFUL" ));
        future.append("failed",      FutureEvent.FAIL,      f -> logIt(f, "FAILED"     ));
        future.append("cancelled",   FutureEvent.CANCEL,    f -> logIt(f, "CANCELLED"  ));
        future.append("aborted",     FutureEvent.ABORT,     f -> logIt(f, "ABORTED"    ));
        future.append("done",        FutureEvent.DONE,      f -> logIt(f, "DONE"       ));
        future.append("terminated",  FutureEvent.TERMINATE, f -> logIt(f, "TERMINATED" ));
        future.append("complete",    FutureEvent.COMPLETE,  f -> logIt(f, "COMPLETED"  ));
        future.append("finished",    FutureEvent.FINISHED,  f -> logIt(f, "FINISHED"   ));
        future.append("interrupted", FutureEvent.INTERRUPT, f -> logIt(f, "INTERRUPTED"));
        //@formatter:on
    }

    protected static void logIt(FutureWork future, String eventName) {
        log.info(String.format("> %s: %-10s: %s",
                future.getName(), eventName, Str.notNull(future.getReason())));
    }
}

1	Create a future and let it run to successful completion
2	Create a future but fail it halfway through its operation
3	Create a future but cancel it halfway through its operation
4	Create a future but abort it halfway through its operation
5	Creates a simple future that logs every event

In the kos-java-examples-core sample code, make the following modification:

In the FutureModule class, run() method, comment out all everything except the FutureExamples1 code.

Run this application in debug mode inside your IDE, then let’s look at the log output:

FutureExamples1 output: futures that are successful, failed, cancelled, and aborted

> ====== BEGIN =========================================
> MyKosApp.load()
> FutureModule.init()
> MyKosApp.start()
> FutureModule.run()
> FutureExamples1.run()
> fw-1: STARTED   : (1)
> fw-1: SUCCESSFUL:
> fw-1: DONE      :
> fw-1: COMPLETED :
> fw-1: FINISHED  :
> fw-2: STARTED   : (2)
> fw-2: FAILED    : Something went wrong
> fw-2: DONE      : Something went wrong
> fw-2: COMPLETED : Something went wrong
> fw-2: FINISHED  : Something went wrong
> fw-3: STARTED   : (3)
> fw-3: CANCELLED : The user changed his/her mind
> fw-3: TERMINATED: The user changed his/her mind
> fw-3: COMPLETED : The user changed his/her mind
> fw-3: FINISHED  : The user changed his/her mind
> fw-4: STARTED   : (4)
> fw-4: ABORTED   : An internal operation unexpectedly failed
> fw-4: TERMINATED: An internal operation unexpectedly failed
> fw-4: COMPLETED : An internal operation unexpectedly failed
> fw-4: FINISHED  : An internal operation unexpectedly failed
> ====== END ===========================================

1	Future #1 ran through these events: STARTED → SUCCESSFUL → DONE → COMPLETED → FINISHED
2	Future #2 ran through these events: STARTED → FAILED → DONE → COMPLETED → FINISHED
3	Future #3 ran through these events: STARTED → CANCELLED → TERMINATED → COMPLETED → FINISHED
4	Future #4 ran through these events: STARTED → ABORTED → TERMINATED → COMPLETED → FINISHED

Note that SUCCESS and FAIL both proceed to DONE, while CANCEL and ABORT go to TERMINATE.

Sequential and Parallel

In this FutureExamples2 example, we are going to:

Run three futures in sequence using the SequencedFuture grouping class, and
Then run those same three futures in parallel using the ParallelFuture grouping class.

FutureExamples2 class

package com.example.futures;

import com.tccc.kos.commons.util.concurrent.future.ParallelFuture;
import com.tccc.kos.commons.util.concurrent.future.SequencedFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.StopWatch;

@Slf4j
public class FutureExamples2 extends FutureExamples1 {

    public void run() {
        log.info("> FutureExamples2.run()");
        runInSequence();
        runInParallel();
    }

    /**
     * Runs three tasks in sequence, which finishes in six seconds.
     */
    private void runInSequence() { (1)
        log.info("> runInSequence(): start");
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        SequencedFuture sequencedFuture = new SequencedFuture("sf-1");
        sequencedFuture.add(createFuture("fw-1"));
        sequencedFuture.add(createFuture("fw-2"));
        sequencedFuture.add(createFuture("fw-3"));
        sequencedFuture.run();
        boolean workFinished = sequencedFuture.waitUntilFinished(3 * DURATION + EXTRA);
        stopWatch.stop();
        log.info("> runInSequence(): end: workFinished = {}, time = {}", workFinished, stopWatch);
    }

    /**
     * Runs same three tasks in parallel, which finishes in two seconds.
     */
    private void runInParallel() { (2)
        log.info("> runInParallel(): start");
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        ParallelFuture parallelFuture = new ParallelFuture("pf-1");
        parallelFuture.add(createFuture("fw-1"));
        parallelFuture.add(createFuture("fw-2"));
        parallelFuture.add(createFuture("fw-3"));
        parallelFuture.run();
        boolean workFinished = parallelFuture.waitUntilFinished(TIMEOUT);
        stopWatch.stop();
        log.info("> runInParallel(): end: workFinished = {}, time = {}", workFinished, stopWatch);
    }
}

1	Run three tasks in sequence, one after the other
2	Run those same three tasks in parallel

Now let’s look at the second half of the logs, which represents the output of the FutureExample2 operation:

FutureExamples2 output

> ====== BEGIN =========================================
> MyKosApp.load()
> FutureModule.init()
> MyKosApp.start()
> FutureModule.run()
> FutureExamples2.run()
> runInSequence(): start
> fw-1: STARTED   : (1)
> fw-1: SUCCESSFUL:
> fw-1: DONE      :
> fw-1: COMPLETED :
> fw-1: FINISHED  :
> fw-2: STARTED   :
> fw-2: SUCCESSFUL:
> fw-2: DONE      :
> fw-2: COMPLETED :
> fw-2: FINISHED  :
> fw-3: STARTED   :
> fw-3: SUCCESSFUL:
> fw-3: DONE      :
> fw-3: COMPLETED :
> fw-3: FINISHED  :
> runInSequence(): end: workFinished = true, time = 00:00:06.007 (2)
> runInParallel(): start
> fw-2: STARTED   : (3)
> fw-1: STARTED   :
> fw-3: STARTED   :
> fw-2: SUCCESSFUL:
> fw-1: SUCCESSFUL:
> fw-3: SUCCESSFUL:
> fw-2: DONE      :
> fw-2: COMPLETED :
> fw-2: FINISHED  :
> fw-1: DONE      :
> fw-1: COMPLETED :
> fw-1: FINISHED  :
> fw-3: DONE      :
> fw-3: COMPLETED :
> fw-3: FINISHED  :
> runInParallel(): end: workFinished = true, time = 00:00:02.005 (4)
> ====== END ===========================================

1	In the sequence example, the first future runs to completion, followed by the second future, followed by the third.
2	The sequenced futures took about 6 seconds to execute, as expected.
3	In the parallel example, all three of the futures are started at the same time, in random order.
4	The parallel futures took about 2 seconds to execute, as expected.

Summary

In this article, we looked a KOS "futures", how they assist in writing asynchronous code, and saw some examples of these. Be sure to run this code yourself, making modifications to examine what happens in different situations.