0
\$\begingroup\$

As stated in the javadoc below, this is intended for monitoring and debugging. I've found it especially useful for figuring out which tasks got stuck and prevented clean terminations of my Executors.
I've run some simple benchmarks for TaskTrackingThreadPoolExecutor (see below) and performance penalty is up to 1.15x for no-op tasks (()->{}) and up to 1.015x for 1ms tasks (() -> Thread.sleep(1)). Ideas for further optimization would be especially welcomed :)
Other than that I'm looking for reviews if I didn't mess something up unknowingly.
The classes are a part of a small utility lib that I've created.

// Copyright (c) Piotr Morgwai Kotarbinski, Licensed under the Apache License, Version 2.0
package pl.morgwai.base.utils.concurrent;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiConsumer;

import static java.util.stream.Collectors.toUnmodifiableList;



/**
 * {@link ExecutorService} that allows to obtain a {@link #getRunningTasks() List of currently
 * running tasks}.
 * Useful for monitoring and debugging which tasks got stuck and prevented clean
 * {@link ExecutorService#awaitTermination(long, TimeUnit) termination}.
 * @see TaskTrackingThreadPoolExecutor
 * @see ScheduledTaskTrackingThreadPoolExecutor
 */
public interface TaskTrackingExecutor extends ExecutorService {



    /**
     * Returns a {@code List} of tasks currently being run by the worker {@code Threads}.
     * Unless stated otherwise by an implementing class, this may be a subject to all kind of races
     * and thus may sometimes not even be fully consistent with any point in the past. This method
     * is intended for spotting long-running or stuck tasks or for general overview of types
     * of tasks being executed.
     */
    List<Runnable> getRunningTasks();



    // removed few unrelated default methods providing integration with other utils



    /**
     * Decorator for an {@link ExecutorService} that makes it a {@link TaskTrackingExecutor}.
     * @see TaskTrackingThreadPoolExecutor
     * @see ScheduledTaskTrackingThreadPoolExecutor
     */
    class TaskTrackingExecutorDecorator extends AbstractExecutorService
            implements TaskTrackingExecutor {



        final ExecutorService backingExecutor;

        final Set<TaskHolder> runningTasks;
        static class TaskHolder { volatile Runnable task; }



        /**
         * Decorates {@code executorToDecorate}.
         * {@code executorToDecorate} must be idle at the time of decoration.
         * Afterwards {@code executorToDecorate} must be used only via the constructed decorator.
         * @param threadPoolSize used as a concurrency-level hint.
         */
        public TaskTrackingExecutorDecorator(ExecutorService executorToDecorate, int threadPoolSize)
        {
            this(executorToDecorate, false, threadPoolSize);
        }



        /** See {@link #TaskTrackingExecutorDecorator(ExecutorService, int)}. */
        public TaskTrackingExecutorDecorator(ExecutorService executorToDecorate) {
            this(executorToDecorate, -1);
        }



        /**
         * Decorates {@code executorToDecorate}.
         * Calls {@link #decorateRejectedExecutionHandler(ThreadPoolExecutor)
         * decorateRejectedExecutionHandler(executorToDecorate)} and
         * {@link #decorateThreadFactory(ThreadFactory)
         * executorToDecorate.setThreadFactory(
         * decorateThreadFactory(executorToDecorate.getThreadFactory()))} in the process.
         * Afterwards {@code executorToDecorate} must be used only via the constructed decorator.
         * @throws IllegalStateException if {@code executorToDecorate} is not idle.
         */
        public TaskTrackingExecutorDecorator(ThreadPoolExecutor executorToDecorate) {
            this(executorToDecorate, false, executorToDecorate.getCorePoolSize());
            if (executorToDecorate.getActiveCount() > 0) {
                throw new IllegalStateException(
                        "executor must be idle to decorate it with TaskTrackingExecutorDecorator");
            }
            decorateRejectedExecutionHandler(executorToDecorate);
            executorToDecorate.setThreadFactory(
                    decorateThreadFactory(executorToDecorate.getThreadFactory()));
            final int corePoolSize = executorToDecorate.getCorePoolSize();
            executorToDecorate.setCorePoolSize(0);
            executorToDecorate.setCorePoolSize(corePoolSize);
        }



        /**
         * Decorates {@code executor}'s {@link RejectedExecutionHandler} to unwrap tasks from
         * {@link TrackableTask} before passing them to the original handler.
         */
        public static void decorateRejectedExecutionHandler(ThreadPoolExecutor executor) {
            final var originalHandler = executor.getRejectedExecutionHandler();
            executor.setRejectedExecutionHandler(
                (wrappedTask, rejectingExecutor) -> originalHandler.rejectedExecution(
                    ((TrackableTask) wrappedTask).wrappedTask,
                    rejectingExecutor
                )
            );
        }



        /**
         * Decorates {@code factoryToDecorate} to create worker {@link Thread}s, that remove their
         * monitoring hooks from this {@code TaskTrackingExecutorDecorator} after they exit.
         */
        public ThreadFactory decorateThreadFactory(ThreadFactory factoryToDecorate) {
            return (task) -> factoryToDecorate.newThread(() -> {
                try {
                    task.run();
                } finally {
                    final var taskHolder = threadLocalTaskHolder.get();
                    if (taskHolder != null) runningTasks.remove(taskHolder);
                }
            });
        }



        /** Hooking capabilities allow to avoid wrapping tasks with {@link TrackableTask}. */
        public interface HookableExecutor extends ExecutorService {
            void addBeforeExecuteHook(BiConsumer<Thread, Runnable> hook);
            void addAfterExecuteHook(BiConsumer<Runnable, Throwable> hook);
        }



        /** See {@link #TaskTrackingExecutorDecorator(ExecutorService, int)}. */
        public TaskTrackingExecutorDecorator(
            HookableExecutor executorToDecorate,
            int threadPoolSize
        ) {
            this(executorToDecorate, true, threadPoolSize);
            executorToDecorate.addBeforeExecuteHook(
                    (worker, task) -> storeTaskIntoHolderBeforeExecute(task));
            executorToDecorate.addAfterExecuteHook((task, error) -> clearTaskHolderAfterExecute());
        }



        public TaskTrackingExecutorDecorator(HookableExecutor executorToDecorate) {
            this(executorToDecorate, -1);
        }



        final boolean backingExecutorHookable;



        TaskTrackingExecutorDecorator(
            ExecutorService executorToDecorate,
            boolean backingExecutorHookable,
            int threadPoolSize
        ) {
            runningTasks = (threadPoolSize > 0)
                    ? ConcurrentHashMap.newKeySet(threadPoolSize) : ConcurrentHashMap.newKeySet();
            this.backingExecutor = executorToDecorate;
            this.backingExecutorHookable = backingExecutorHookable;
        }



        @Override
        public List<Runnable> shutdownNow() {
            return backingExecutorHookable
                    ? backingExecutor.shutdownNow()
                    : backingExecutor.shutdownNow().stream()
                        .map(TrackableTask.class::cast)
                        .map(TrackableTask::getWrappedTask)
                        .collect(toUnmodifiableList());
        }



        @Override
        public List<Runnable> getRunningTasks() {
            return runningTasks.stream()
                .map((holder) -> holder.task)
                .filter(Objects::nonNull)
                .collect(toUnmodifiableList());
        }



        ThreadLocal<TaskHolder> threadLocalTaskHolder = new ThreadLocal<>();



        void storeTaskIntoHolderBeforeExecute(Runnable task) {
            var taskHolder = threadLocalTaskHolder.get();
            if (taskHolder == null) {
                taskHolder = new TaskHolder();
                threadLocalTaskHolder.set(taskHolder);
                runningTasks.add(taskHolder);
            }
            taskHolder.task = task;
        }



        void clearTaskHolderAfterExecute() {
            threadLocalTaskHolder.get().task = null;
        }



        /**
         * Wraps {@code task} with a {@link TrackableTask} if needed and passes it to its backing
         * executor.
         * If the backing executor is a {@link HookableExecutor}, no wrapping is needed and
         * {@code task} is directly passed to the backing executor.
         */
        @Override
        public void execute(Runnable task) {
            backingExecutor.execute(backingExecutorHookable ? task : new TrackableTask(task));
        }



        /** A decorator that tracks execution of its wrapped task. */
        public class TrackableTask implements Runnable {

            public Runnable getWrappedTask() { return wrappedTask; }
            final Runnable wrappedTask;

            protected TrackableTask(Runnable taskToWrap) {
                wrappedTask = taskToWrap;
            }

            @Override public void run() {
                storeTaskIntoHolderBeforeExecute(wrappedTask);
                try {
                    wrappedTask.run();
                } finally {
                    clearTaskHolderAfterExecute();
                }
            }

            @Override public String toString() {
                return  wrappedTask.toString();
            }
        }



        /**
         * If {@code task} is an instance of {@link TrackableTask} returns
         * {@link TrackableTask#getWrappedTask() task.getWrappedTask()}, otherwise just
         * {@code task}.
         */
        public static Runnable unwrapTask(Runnable task) {
            return task instanceof TrackableTask ? ((TrackableTask) task).wrappedTask : task;
        }



        // only dumb delegations to backingExecutor below:

        @Override
        public void shutdown() {
            backingExecutor.shutdown();
        }

        @Override
        public boolean isShutdown() {
            return backingExecutor.isShutdown();
        }

        @Override
        public boolean isTerminated() {
            return backingExecutor.isTerminated();
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
            return backingExecutor.awaitTermination(timeout, unit);
        }
    }
}

Optimized application to ThreadPoolExecutor:

// Copyright (c) Piotr Morgwai Kotarbinski, Licensed under the Apache License, Version 2.0
package pl.morgwai.base.utils.concurrent;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiConsumer;

import pl.morgwai.base.utils.concurrent.TaskTrackingExecutor.TaskTrackingExecutorDecorator;

import static java.util.concurrent.TimeUnit.DAYS;



/** A {@link TaskTrackingExecutor} based on a {@link ThreadPoolExecutor}. */
public class TaskTrackingThreadPoolExecutor extends ThreadPoolExecutor
        implements TaskTrackingExecutor, TaskTrackingExecutorDecorator.HookableExecutor {



    final TaskTrackingExecutorDecorator taskTrackingDecorator;



    /**
     * Calls {@link ThreadPoolExecutor#ThreadPoolExecutor(int, int, long, TimeUnit, BlockingQueue,
     * ThreadFactory, RejectedExecutionHandler) super} and sets up
     * {@link TaskTrackingExecutor.TaskTrackingExecutorDecorator}.
     */
    public TaskTrackingThreadPoolExecutor(
        int corePoolSize,
        int maximumPoolSize,
        long keepAliveTime,
        TimeUnit unit,
        BlockingQueue<Runnable> workQueue,
        ThreadFactory threadFactory,
        RejectedExecutionHandler handler
    ) {
        super(
            corePoolSize,
            maximumPoolSize,
            keepAliveTime,
            unit,
            workQueue,
            threadFactory,
            handler
        );
        taskTrackingDecorator = createAndSetupTaskTrackingDecorator();
    }

    TaskTrackingExecutorDecorator createAndSetupTaskTrackingDecorator() {
        final var decorator = new TaskTrackingExecutorDecorator(this, getCorePoolSize());
        setThreadFactory(decorator.decorateThreadFactory(getThreadFactory()));
        return decorator;
    }



    @Override
    public List<Runnable> getRunningTasks() {
        return taskTrackingDecorator.getRunningTasks();
    }



    final Deque<BiConsumer<Thread, Runnable>> beforeExecuteHooks = new LinkedList<>();

    /** Adds {@code hook} to be executed in {@link #beforeExecute(Thread, Runnable)}. */
    public void addBeforeExecuteHook(BiConsumer<Thread, Runnable> hook) {
        beforeExecuteHooks.addFirst(hook);
    }

    /**
     * Executes all {@link #addBeforeExecuteHook(BiConsumer) added hooks} in the reverse order they
     * were added.
     */
    @Override
    protected final void beforeExecute(Thread worker, Runnable task) {
        beforeExecuteHooks.forEach((hook) -> hook.accept(worker, task));
    }



    final List<BiConsumer<Runnable, Throwable>> afterExecuteHooks = new LinkedList<>();

    /** Adds hook to be executed in {@link #afterExecute(Runnable, Throwable)}. */
    public void addAfterExecuteHook(BiConsumer<Runnable, Throwable> hook) {
        afterExecuteHooks.add(hook);
    }

    /**
     * Executes all {@link #addAfterExecuteHook(BiConsumer)}  added hooks} in the order they were
     * added.
     */
    @Override
    protected final void afterExecute(Runnable task, Throwable error) {
        afterExecuteHooks.forEach((hook) -> hook.accept(task, error));
    }



    // removed some boilerplate constructors
}
\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Having the constructors of TaskTrackingExecutorDecorator sprinkled around the class with other methods in between makes it hard to follow. Often the order of elements is inner classes first, then fields, then constructors, then other methods. Partly because of this, it took me longer than it should have to figure out that you have a requirement for executorToDecorate to be inactive for it to be decorated, but the code does not go through that check in all possible constructor invocation paths. That looks like a possible bug.

On the other hand, does that inactivity requirement make sense? Whoever created the underlying executor still holds a reference to it and is able to bypass the decorator by calling it directly.

Having nested classes three deep seems a bit unnecessary and looks cumbersome to use.

Should you try to isolate the before and after execution hooks from each other? Trying to make it bulletproof isn't trivial, though. An error in one will now prevent the others hooks from executing, which is worth documenting.

Worrying about performance seems like premature optimization. It's a debugging tool that places a small one time cost on each task and a feature toggle will prevent the decoration from occurring in production, right?

\$\endgroup\$
5
  • \$\begingroup\$ Thanks for looking into it! :) inactivity requirement: If the decorator is applied to an active Executor, the results of getRunningTasks will be completely inaccurate and useless, hence the requirement. Every public constructor states this requirement in its javadoc or links to the docsn of a one that states it. It is impossible to verify in constructors other than the one accepting ThreadPoolExecutor due to interface limitations. I was considering adding sth like isActive() to HookableExecutor: what do you think? (still the ExecutorService variants can do nothing) \$\endgroup\$
    – morgwai
    Jan 17 at 11:04
  • \$\begingroup\$ hook isolation: oh yeah! I'll definitely document it and will put some thought into isolation. 3-level nesting: I was actually considering making HookableExecutor package-protected, changed it as I thought someone may want to make his custom ExecutorService implementation to be task-tracking capable, in which case HookableExecutor will come handy. Otherwise it was not intended for general use. Do you think I should consider it general-purpose worthy? \$\endgroup\$
    – morgwai
    Jan 17 at 11:04
  • \$\begingroup\$ performance: I'm actually using it in production also for monitoring purposes. If someone wants to use it to decorate a ForkJoinPool dealing with large amount of tiny tasks, performance may be important and I guess 1.15x may be too much... helper methods between constructors: I follow the convention that if a helper is used by 1 single "higher-level" method (constructor in this case), the helper should be defined right after its "higher-level" method to make it easier to understand the "higher-level" method. Many thanks again! :) \$\endgroup\$
    – morgwai
    Jan 17 at 11:04
  • \$\begingroup\$ hmm, I've just realized that the last public constructor lacks any javadoc whatsoever... ;-] my bad: I'll fix it :) \$\endgroup\$
    – morgwai
    Jan 17 at 11:16
  • \$\begingroup\$ regarding helper between constructors: I will reorder constructors, so that the one using these helpers will be the last one: this way the helpers will not get in between :) \$\endgroup\$
    – morgwai
    Jan 17 at 11:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.