2
\$\begingroup\$

The following TaskStack processes input elements asynchronously based on a function that it receives at construction:

TaskStack<String, Integer> stack = new TaskStack<>(String::length);

In a realistic scenario, this function would be long-running.

We can add input elements to our stack to have them processed:

CompletableFuture<Integer> futOutput1 = stack.add("How's it going?");
CompletableFuture<Integer> futOutput2 = stack.add("duplicate input");
CompletableFuture<Integer> futOutput3 = stack.add("duplicate input");

Now, the reason I created TaskStack and didn't use Guava caches or the Caffeine cache is that I needed the most recently added inputs to be processed first (i.e., last in, first out) and as far as I know, Guava and Caffeine don't offer a LIFO cache.

Also, if an input was already added I wanted to avoid performing the processing twice when an identical input was added later on. This is why futOutput2 and futOutput3 from the example above reference the same CompletableFuture.

A feature of TaskStack that's important to me is to be able to remove scheduled work:

stack.remove("How's it going?");

Calling remove entails different things depending on the status of processing the input "How's it going?":

  • If processing has not started yet, the input simply won't be processed
  • If it has already finished processing and we have a result, that result is thrown away and will have to be created again by a call to add
  • If the input is currently being processed, processing won't be interrupted but the future completes with a java.util.concurrent.CancellationException

This is the code of TaskStack:

import java.lang.ref.SoftReference;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * A stack that accepts input elements which are processed asynchronously in a last in, first out (LIFO) order.
 * <p>
 * Created by Matthias Braun on 2017-12-07.
 */
public class TaskStack<T, R> {

    // This function defines what it means to process input
    private final Function<T, R> func;

    /*
     * Holds the input elements and the futures containing the output elements.
     * We use soft references for the outputs to allow the garbage collector to free them if no one
     * else is referencing them. This avoids memory issues when outputs occupy a lot of memory.
     */
    private final Map<T, SoftReference<CompletableFuture<R>>> inputOutputMap = new ConcurrentHashMap<>();

    // Provides threads to turn inputs into outputs
    private final ExecutorService executor = createExecutor();

    /**
     * Creates a new {@link TaskStack}.
     *
     * @param func elements added to the {@link TaskStack} will be processed using this {@link Function}
     */
    public TaskStack(Function<T, R> func) {
        this.func = func;
    }

    /**
     * Adds and processes the {@code input}.
     *
     * @param input we process this input, turning it into a value of type {@code R} in the future
     * @return the result of processing the {@code input} wrapped in a {@link CompletableFuture}.
     * If the {@code input} was already added to the {@link TaskStack} by a prior call to this method, we return
     * the same {@link CompletableFuture} as before
     * @see #remove
     */
    public CompletableFuture<R> add(T input) {

        return fold(getOpt(inputOutputMap, input),
                // We don't have a precomputed result for the input -> Start processing
                () -> {
                    // Process the input on one of the threads provided by the executor
                    CompletableFuture<R> futureOutput = CompletableFuture.supplyAsync(() -> func.apply(input), executor);

                    // Wrap the future result in a soft reference and put it in the map so the result can be
                    // garbage collected once no one else is referencing it
                    inputOutputMap.put(input, new SoftReference<>(futureOutput));

                    return futureOutput;
                },
                // There's already a result for the input, but it may have been garbage collected
                existingSoftFuture -> fold(existingSoftFuture,
                        () -> {
                            // The result was already garbage collected ->
                            // Remove the input too and make this method reprocess the input
                            inputOutputMap.remove(input);
                            return add(input);
                        },
                        // We can return the result without reprocessing the input
                        future -> future
                )
        );
    }

    /**
     * Removes the {@code input} from this {@link TaskStack} meaning the {@code input} won't be processed.
     * If the {@code input} was already processed, subsequent calls to {@link #add} will cause the {@code input} to be
     * processed again.
     *
     * @param input                 we remove this from the {@link TaskStack}
     */
    public void remove(T input) {
        getOpt(inputOutputMap, input).ifPresent(softFuture -> {
            inputOutputMap.remove(input);

            ifPresent(softFuture,
                    // CompletableFuture ignore the mayInterruptIfRunning flag
                    future -> future.cancel(false));
        });
    }

    private static ExecutorService createExecutor() {

        // How many threads should process input at maximum. Have as many threads as there are processors minus one,
        // but at least one thread
        int maxPoolSize = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);

        // When the number of threads is greater than the core, this is the maximum time that excess idle threads will
        // wait for new tasks before terminating
        long keepAliveTime = 3;
        TimeUnit timeUnit = TimeUnit.SECONDS;

        // It's the stack that makes the executor assign threads to the submitted tasks in a last in, first out order
        return new ThreadPoolExecutor(0, maxPoolSize, keepAliveTime, timeUnit, new BlockingStack<>());
    }

    /**
     * Applies the referent of {@code ref} to {@code ifPresent} if {@code ref} is not null. Otherwise, does nothing.
     *
     * @param ref       the {@link SoftReference} whose referent we apply to {@code ifPresent} if it's not null
     * @param ifPresent the {@link Consumer} to which we pass the non-null referent of {@code ref}
     * @param <T>       the type of {@code ref}'s referent
     */
    private static <T> void ifPresent(SoftReference<T> ref, Consumer<T> ifPresent) {
        T referent = ref.get();
        if (referent != null) {
            ifPresent.accept(referent);
        }
    }

    /**
     * Applies the referent of {@code ref} to {@code ifPresent} if {@code ref} is not null. Otherwise,
     * calls {@code ifAbsent}.
     *
     * @param ref       the {@link SoftReference} whose referent we apply to {@code ifPresent} if it's not null
     * @param ifAbsent  if {@code ref}'s referent is null, we call this {@link Supplier} to produce a value of type
     *                  {@code Res}
     * @param ifPresent the {@link Function} to which we pass the non-null referent of {@code ref} to produce a value
     *                  of type {@code Res}
     * @param <T>       the type of {@code ref}'s referent
     * @param <Res>     the type of the value produced by both {@code ifAbsent} and {@code ifPresent}
     * @return a value of type {@code Res}
     */
    private static <T, Res> Res fold(SoftReference<T> ref, Supplier<Res> ifAbsent, Function<T, Res> ifPresent) {

        T referent = ref.get();
        final Res result;
        if (referent == null) {
            result = ifAbsent.get();
        } else {
            result = ifPresent.apply(referent);
        }
        return result;
    }

    /**
     * Gets the value associated with {@code key} from the {@code map} or an {@link Optional#empty()} if the {@code map}
     * doesn't contain the {@code key}.
     *
     * @param map we get the value from this {@link Map}
     * @param key we get the value associated with this {@code key}
     * @param <K> the {@code key}'s type
     * @param <V> the value's type
     * @return the value associated with {@code key} or an {@link Optional#empty()} if the {@code map}
     * doesn't contain the {@code key}
     */
    private static <K, V> Optional<V> getOpt(final Map<K, V> map, final K key) {
        return Optional.ofNullable(map.get(key));
    }

    /**
     * Applies the value of {@code optional} to {@code ifPresent} if it's present. Otherwise, calls {@code ifAbsent}.
     *
     * @param optional  the {@link Optional} whose value we apply to {@code ifPresent} if it's present
     * @param ifAbsent  if {@code optional}'s value is absent, we call this {@link Supplier} to produce a value of type
     *                  {@code Res}
     * @param ifPresent the {@link Function} to which we pass the value of {@code optional} to produce a value
     *                  of type {@code Res}
     * @param <T>       the type of {@code optional}'s value
     * @param <Res>     the type of the value produced by both {@code ifAbsent} and {@code ifPresent}
     * @return a value of type {@code Res}
     */
    private static <T, Res> Res fold(Optional<T> optional, Supplier<Res> ifAbsent, Function<T, Res> ifPresent) {
        final Res result;
        if (optional.isPresent()) {
            result = ifPresent.apply(optional.get());
        } else {
            result = ifAbsent.get();
        }
        return result;
    }

    /**
     * A stack that will block when it's full and clients try to add new elements to it.
     * Being a stack, it adds new elements in a last in first out manner: We put the most recently added elements at the
     * first position in the stack.
     * <p>
     * If its capacity is unspecified, it defaults to {@link Integer#MAX_VALUE}.
     *
     * @param <E> the elements inside the {@link BlockingStack}
     */
    public static class BlockingStack<E> extends LinkedBlockingDeque<E> {

        @Override
        public boolean offer(E e) {
            return offerFirst(e);
        }

        @Override
        public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
            return offerFirst(e, timeout, unit);
        }

        @Override
        public boolean add(E e) {
            return offerFirst(e);
        }

        @Override
        public void put(E e) throws InterruptedException {
            putFirst(e);
        }
    }
}

I welcome feedback on every aspect of the code, especially regarding concurrency bugs I might have missed.

\$\endgroup\$
2
\$\begingroup\$

You have what, based on your description, I would consider to be a concurrency bug. It's a bit of an edge case, but then most concurrency bugs are.

You've said:

Also, if an input was already added I wanted to avoid performing the processing twice when an identical input was added later on. This is why futOutput2 and futOutput3 from the example above reference the same CompletableFuture.

So, you want to avoid performing processing twice, but do you have to avoid performing the processing twice in all cases? You're using a ConcurrentHashMap, but it's not clear if you'll have multiple threads adding to the stack. If you're using multiple threads and can't do the processing twice, then you may have an issue.

You're not explicitly doing anything to protect your HashMap, you're relying on its internal protection to make sure it doesn't get into an inconsistent state. This is fine, but it doesn't have any context of what you're doing. As a consequence, it's possible to add the same item to the TaskStack from different threads in such a way that the task will be run twice. This can be demonstrated with the following code:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class TSReviewApp {
    private class ThreadedEvaluator extends Thread {
        private final String valueToAdd;
        private final TaskStack<String, Integer> stack;
        private CompletableFuture<Integer> result;

        public ThreadedEvaluator(TaskStack<String, Integer> stack, String valueToAdd) {
            this.stack = stack;
            this.valueToAdd = valueToAdd;
        }

        public void run() {
            result = stack.add(valueToAdd);
        }

        public Integer result() throws ExecutionException, InterruptedException {
            return result.get();
        }

        public CompletableFuture<Integer> future() {
            return result;
        }
    }

    private static int GetLengthQuick(String str) {
        return str.length();
    }
    private static int GetLengthSlow(String str) {
        try {
            System.out.println("Thinking about: " + str);
            Thread.sleep(2000);
            System.out.println("Evaluated: " + str);
        } catch (InterruptedException ex) {
            System.out.println("Sleep interrupted");
        }
        return str.length();
    }

    private void Go() {
        TaskStack<String, Integer> stack = new TaskStack<>(TSReviewApp::GetLengthSlow);

        ThreadedEvaluator futOutput1 = new ThreadedEvaluator(stack,"How is it going?");
        ThreadedEvaluator duplicateFuture1 = new ThreadedEvaluator(stack,"duplicate input");
        ThreadedEvaluator duplicateFuture2 = new ThreadedEvaluator(stack,"duplicate input");
        ThreadedEvaluator duplicateFuture3 = new ThreadedEvaluator(stack,"duplicate input");

        futOutput1.start();
        duplicateFuture1.start();
        duplicateFuture2.start();

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            System.out.println("MainMethod sleep exception");
            e.printStackTrace();
        }

        try {
            System.out.println(futOutput1.result());
            // If the same future was used, this would evaluate to *true*, it doesn't.
            // because two futures were created, one that is in the map still and one
            // that isn't.
            System.out.println("future1 == future 2? " + (duplicateFuture1.future() == duplicateFuture2.future()));
            System.out.println(duplicateFuture1.result());
            System.out.println(duplicateFuture2.result());
        } catch (Exception ex) {
            System.out.println(ex);
        }

        duplicateFuture3.start();

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            System.out.println("MainMethod sleep 2 exception");
            e.printStackTrace();
        }

        try {
            System.out.println(duplicateFuture3.result());
            // Because one of the previous calls will still be in the map,
            // One of the next two checks will return true
            System.out.println("future3 == future 1? " + (duplicateFuture1.future() == duplicateFuture3.future()));
            System.out.println("future3 == future 2? " + (duplicateFuture2.future() == duplicateFuture3.future()));
        } catch (Exception ex) {
            System.out.println(ex);
        }

    }

    public static void main(String[] args) {
        TSReviewApp app = new TSReviewApp();
        app.Go();
    }

}

Which outputs something like:

Thinking about: How is it going?
Evaluated: How is it going?
Thinking about: duplicate input
16
future1 == future 2? false
Evaluated: duplicate input
Thinking about: duplicate input
Evaluated: duplicate input
15
15
15
future3 == future 1? true
future3 == future 2? false

The application processes two strings "How is it going?" and "duplicate input". The thing to notice in the output is that when the initial three items are added and executed (from different threads), it results in three sets of processing ("duplicate input" is processed twice). If at a later point, "duplicate input" is added again, the existing item in the map is used and so the processing doesn't have to be done again. How important is this?

You can detect that you're in this situation from within your TaskStack, by checking the value of the put call:

public CompletableFuture<R> add(T input) {

    return fold(getOpt(inputOutputMap, input),
            // We don't have a precomputed result for the input -> Start processing
            () -> {
                // Process the input on one of the threads provided by the executor
                CompletableFuture<R> futureOutput = CompletableFuture.supplyAsync(() -> func.apply(input), executor);

                // Wrap the future result in a soft reference and put it in the map so the result can be
                // garbage collected once no one else is referencing it
//---->
                SoftReference<CompletableFuture<R>> previousValue = inputOutputMap.put(input, new SoftReference<>(futureOutput));
                if(null != previousValue) {
                    System.out.println("Added '" + input + "' to map when it was already present");
                }
//---->

                return futureOutput;
            },
            // There's already a result for the input, but it may have been garbage collected
            existingSoftFuture -> fold(existingSoftFuture,
                    () -> {
                        // The result was already garbage collected ->
                        // Remove the input too and make this method reprocess the input
                        inputOutputMap.remove(input);
                        return add(input);
                    },
                    // We can return the result without reprocessing the input
                    future -> future
            )
    );
}

Again, this may be acceptable for your usage scenario (it doesn't matter if processing is occasionally done multiple times, you only add from one thread, the likelihood of a duplicate input is low), or you may need to think about some kind of locking around read/writes.

\$\endgroup\$

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.