Battles with CountedCompleter

import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.ExecutionList;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountedCompleter;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import javax.annotation.CheckForNull;
import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.concurrent.GuardedBy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author shevek
 */
public class ForkJoinScope implements AutoCloseable, Executor {

    @SuppressWarnings("UnusedVariable")
    private static final Logger LOG = LoggerFactory.getLogger(ForkJoinScope.class);
    private static final boolean DEBUG = false;
    private static final int EXCEPTIONS_MAX = 100;
    /** When this threshold is exceeded, submitted tasks will be run directly in the calling thread. */
    private static final int CALLER_RUNS_PENDING_COUNT_THRESHOLD = Runtime.getRuntime().availableProcessors() * 3;

    private class ManagedAction<V> extends CountedCompleter<V> implements RunnableFuture<V> {

        private final Callable<V> callable;
        @CheckForNull
        private V result;

        public ManagedAction(@Nonnull CountedCompleter<?> parent, @Nonnull Callable<V> callable) {
            super(Preconditions.checkNotNull(parent, "Parent was null."));
            this.callable = Preconditions.checkNotNull(callable, "Callable was null.");
        }

        @Override
        @CheckForNull
        public V getRawResult() {
            return result;
        }

        @Override
        protected void setRawResult(@CheckForNull V t) {
            this.result = t;
        }

        /** nothrow */
        @Override
        public void compute() {
            V v = null;
            try {
                v = callable.call();
                if (DEBUG)
                    LOG.debug(name + ": Completed normally: " + this);
            } catch (Throwable e) {
                if (DEBUG)
                    LOG.error(name + ": Completed exceptionally: " + this + ": " + e, e);
                addException(this, e);
            } finally {
                // Since this task has a completer.
                // But we might need to join() on it via invoke() if we hit CALLER_RUNS_PENDING_COUNT_THRESHOLD.
                // Therefore, we MUST call explicit complete() otherwise we can never join() on it.
                complete(v);
            }
        }

        @Override
        public void onCompletion(CountedCompleter<?> caller) {
            // if caller == this then this is a completion of this task, else a subtask.
            // if (DEBUG) LOG.error(name + ": ManagedAction.onCompletion(): " + this);
            super.onCompletion(caller);
        }

        @Override
        public boolean onExceptionalCompletion(Throwable e, CountedCompleter<?> caller) {
            // Propagate the exception to cause exceptional completion of the parent task.
            // if (DEBUG) LOG.error(name + ": ManagedAction.onExceptionalCompletion(): " + this + " from " + caller + ": " + e, e);
            // TODO: What is the effect of this on the completer (joinTask?)
            // If this aborts the joinTask, then that's NOT what we want to do.
            addException(caller, e);
            return false;
        }

        /** For AbstractExecutorService.execute(), which contains an instanceof ForkJoinTask. */
        @Override
        public final void run() {
            // We enter this if we are using a direct executor.
            // if (DEBUG) LOG.debug(name + ": " + this + ": Entering run()");
            invoke();
            // if (DEBUG) LOG.debug(name + ": " + this + ": Exiting run()");
        }

        @Override
        public String toString() {
            return MoreObjects.toStringHelper(this)
                    .add("callable", callable)
                    .add("scope", name)
                    .add("done", isDone())
                    .add("pending", getPendingCount())
                    .add("completer", getCompleter())
                    .toString();
        }
    }

    // Not static for addException()
    private class JoinTask extends CountedCompleter<Void> {

        public JoinTask() {
            super(null);
        }

        /**
         * This should never be called directly by ManagedAction,
         * since ManagedAction handles all its exceptions internally.
         * However, it is called by completeExceptionally() when a subtask
         * throws an Error.
         */
        @Override
        public boolean onExceptionalCompletion(Throwable e, CountedCompleter<?> caller) {
            if (DEBUG)
                LOG.error(name + ": " + this + ": Recording exception from " + caller + ": " + e, e);
            addException(caller, e);
            return false;
        }

        @Override
        public void compute() {
            throw new IllegalStateException("Never called.");
        }

        @Override
        public String toString() {
            return "JoinTask(" + name + ", done=" + isDone() + ", pending=" + getPendingCount() + ")";
        }
    }

    @Nonnull
    private final String name;
    private final Executor executor;
    @Nonnull
    private final CountedCompleter<Void> joinTask;
    @GuardedBy("exceptions")
    private final List<Throwable> exceptions = new ArrayList<>(EXCEPTIONS_MAX);

    public ForkJoinScope(@Nonnull String name, @Nonnull Executor executor) {
        this.name = Preconditions.checkNotNull(name, "Name was null.");
        this.executor = Preconditions.checkNotNull(executor, "Executor was null.");
        // this.parentTask = parentTask;

        // We never provide the parent task as the completer for the wrapper;
        // doing so will cause completions to propagate immediately up to the root
        // of the scope stack, without giving us a chance to finish joining intermediate scopes.
        // Shevek says: "the only issue with not having the completer as the parent is that
        // when a thread steals work, it can't steal the "right" work. But meh, let's move on."
        this.joinTask = new JoinTask();
    }

    private void addException(@Nonnull CountedCompleter<?> task, @Nonnull Throwable e) {
        if (DEBUG)
            LOG.debug(name + ": " + task + " -> " + e, e);
        if (e instanceof Error)
            joinTask.completeExceptionally(e);
        synchronized (exceptions) {
            if (exceptions.size() < EXCEPTIONS_MAX)
                exceptions.add(e);
        }
    }

    @Nonnull
    private <V, T extends ManagedAction<V>> T submit(@Nonnull T task) {
        Preconditions.checkState(!joinTask.isDone(), "Parent is already done.");
        joinTask.addToPendingCount(1);
        // if (!(Thread.currentThread() instanceof ForkJoinWorkerThread))
        if (joinTask.getPendingCount() > CALLER_RUNS_PENDING_COUNT_THRESHOLD) {
            if (DEBUG)
                LOG.debug(name + ": Direct invoke " + task + " join-pending " + joinTask.getPendingCount());
            task.invoke();
            if (DEBUG)
                LOG.debug(name + ": Direct return " + task + " join-pending " + joinTask.getPendingCount());
        } else {
            executor.execute(task);   // ForkJoinPool avoids a rewrap, given a ForkJoinTask (ManagedAction).
        }
        return task;
    }

    /** Like submit, but does not return the future, avoiding a warning. */
    @SuppressWarnings("FutureReturnValueIgnored")
    public void execute(@Nonnull Callable<?> r) {
        submit(new ManagedAction<>(joinTask, r));
    }

    @Override
    public void execute(@Nonnull Runnable r) {
        execute(Executors.callable(r, null));
    }

    private void join() throws InterruptedException, ExecutionException {
        // if (DEBUG) LOG.debug(name + ": Joining (calling invoke()) on wrapper " + joinTask);
        // Decrement a ref - this is the last ref if all children already completed.
        joinTask.tryComplete();
        // Wait for the refcount to be zero, which is either the immediately preceding tryComplete() or the last outstanding child.
        joinTask.join();
        // if (DEBUG) LOG.debug(name + ": DONE Join (invoke()) done on wrapper " + joinTask);
        synchronized (exceptions) {
            if (exceptions.isEmpty())
                return;
            Throwable e = exceptions.get(0);
            for (int i = 1; i < exceptions.size(); i++)
                e.addSuppressed(exceptions.get(i));
            exceptions.clear(); // GC only.

            Throwables.propagateIfPossible(e, InterruptedException.class, ExecutionException.class);
            throw new ExecutionException(e);
        }
    }

    @Override
    public void close() throws InterruptedException, ExecutionException, RuntimeException {
        // LOG.debug("Closing ExecutorManager", new Exception());
        join();
        // LOG.debug("Closed ExecutorManager", new Exception());
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("name", name)
                .add("executor", executor)
                .add("callerRunsPendingCountThreshold", callerRunsPendingCountThreshold)
                .add("joinTask", joinTask)
                .toString();
    }
}