Package de.aristaflow.adept2.util.threading.executor

Class AbstractObjectOneWorkExecutor<O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>

  • Type Parameters:
    O - The type of object for which work is registered.
    R - The runtime data stored for each object.
    All Implemented Interfaces:
    Executor, ExecutorService
    Direct Known Subclasses:
    ObjectOneWorkExecutor, ScheduledObjectOneWorkExecutor
    public abstract class AbstractObjectOneWorkExecutor<O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>
extends AbstractObjectSpecificExecutor<O,​O,​R>
    A one work executor allows for an arbitrary amount of objects having work, but only one piece of work is allowed for each object at a time. If work is added for an object that has pending work, the new work will be silently ignored and cancelled immediately. Instead of the completion of previous work this can also last until the next flush.
    It is also possible to add new work that overrides (and cancel) previous pending work. This will not override running work.

    As object-specific executor, the indexed objects are the same as the registered objects. The runtime data is an own class AbstractObjectOneWorkExecutor.RuntimeData which is a pair of runnables that are registered for the corresponding object. The first runnable is the one pending to be executed after the current work but before a flush. This work will be cancelled if flushing is with abortion. It will be null if there is no more work or if the work should block until the next flush. The second runnable is work which is added while a flush is pending. This is executed right after all pending flushs have been executed.
    To prevent starvation, each piece of work will be executed by an arbitrary thread; work for the same object will only be executed by the same thread when shutting down.

    • Field Detail

      • blockUntilFlush

        protected final boolean blockUntilFlush
        Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.

      • overridingWork

        protected final Collection<Thread> overridingWork
        The threads that are currently submitting or executing a piece of work that should override existing (pending) work. Overriding needs to be considered within addRunnable(Object, Object, Runnable), using this collection avoids the need for an additional method parameter (which would require an ugly interface change).

    • Constructor Detail

      • AbstractObjectOneWorkExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        Adept2ThreadFactory threadFactory)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectOneWorkExecutor(boolean, int, Adept2ThreadFactory) instead.
        Creates a new thread pool executor with an unlimited amount of threads and objects, accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        threadFactory - The factory to use when the executor creates a new thread.

      • AbstractObjectOneWorkExecutor

        protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int maxPoolSize,
                                        Adept2ThreadFactory threadFactory)
        Creates a new thread pool executor with an unlimited amount objects and the designated maximum amount of threads, accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        maxPoolSize - The maximum number of threads allowed in the pool. Set this to the number of objects allowed to be executed concurrently or less.
        threadFactory - The factory to use when the executor creates a new thread.

      • AbstractObjectOneWorkExecutor

        protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        Adept2ThreadFactory threadFactory)
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads (and unlimited objects), accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool all the time, even if they are idle.
        maxPoolSize - The maximum number of threads allowed in the pool. Set this to the number of objects allowed to be executed concurrently or less.
        threadFactory - The factory to use when the executor creates a new thread.

      • AbstractObjectOneWorkExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        boolean prestart,
                                        Adept2ThreadFactory threadFactory)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectOneWorkExecutor(boolean, int, int, Adept2ThreadFactory) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads (and unlimited objects), accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool, even if they are idle.
        maxPoolSize - The maximum number of threads to allow in the pool.
        prestart - Whether all core threads should be prestarted.
        threadFactory - The factory to use when the executor creates a new thread.

      • AbstractObjectOneWorkExecutor

        protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        Adept2ThreadFactory threadFactory,
                                        int queueLimitPoolStatistics,
                                        Level minLogLevelPoolStatistics)
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads (and unlimited objects), accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool all the time, even if they are idle.
        maxPoolSize - The maximum number of threads allowed in the pool. Set this to the number of objects allowed to be executed concurrently or less.
        threadFactory - The factory to use when the executor creates a new thread.
        queueLimitPoolStatistics - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

      • AbstractObjectOneWorkExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        boolean prestart,
                                        Adept2ThreadFactory threadFactory,
                                        int queueLimitPoolStatistics,
                                        Level minLogLevelPoolStatistics)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectOneWorkExecutor(boolean, int, int, Adept2ThreadFactory, int, Level) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads (and unlimited objects), accepting exactly one piece of work per object. It uses the designated thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool, even if they are idle.
        maxPoolSize - The maximum number of threads to allow in the pool.
        prestart - Whether all core threads should be prestarted.
        threadFactory - The factory to use when the executor creates a new thread.
        queueLimitPoolStatistics - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

      • AbstractObjectOneWorkExecutor

        protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        Comparator<? super Runnable> comparator,
                                        Adept2ThreadFactory threadFactory,
                                        Level minLogLevelPoolStatistics)
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads, accepting exactly one piece of work per object. The work will be prioritised using the designated comparator, work exceeding the maximum pool size will be enqueued. The executor uses the designated thread factory for creating threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool all the time, even if they are idle.
        maxPoolSize - The maximum number of threads allowed in the pool. Set this to the number of objects allowed to be executed concurrently or less.
        comparator - The comparator used to determine the priority of work.
        threadFactory - The factory to use when the executor creates a new thread. Also the logger from this thread factory will be used by this executor.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

      • AbstractObjectOneWorkExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        boolean prestart,
                                        Comparator<? super Runnable> comparator,
                                        Adept2ThreadFactory threadFactory,
                                        Level minLogLevelPoolStatistics)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectOneWorkExecutor(boolean, int, int, Comparator, Adept2ThreadFactory, Level) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads, accepting exactly one piece of work per object. The work will be prioritised using the designated comparator, work exceeding the maximum pool size will be enqueued. The executor uses the designated thread factory for creating threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool, even if they are idle.
        maxPoolSize - The maximum number of threads to allow in the pool.
        prestart - Whether all core threads should be prestarted.
        comparator - The comparator used to determine the priority of work.
        threadFactory - The factory to use when the executor creates a new thread. Also the logger from this thread factory will be used by this executor.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

      • AbstractObjectOneWorkExecutor

        protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        Comparator<? super Runnable> comparator,
                                        Adept2ThreadFactory threadFactory,
                                        int queueLimitPoolStatistics,
                                        Level minLogLevelPoolStatistics)
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads, accepting exactly one piece of work per object. The work will be prioritised using the designated comparator, work exceeding the maximum pool size will be enqueued. The executor uses the designated thread factory for creating threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool all the time, even if they are idle.
        maxPoolSize - The maximum number of threads allowed in the pool. Set this to the number of objects allowed to be executed concurrently or less.
        comparator - The comparator used to determine the priority of work.
        threadFactory - The factory to use when the executor creates a new thread. Also the logger from this thread factory will be used by this executor.
        queueLimitPoolStatistics - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

      • AbstractObjectOneWorkExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectOneWorkExecutor​(boolean blockUntilFlush,
                                        int corePoolSize,
                                        int maxPoolSize,
                                        boolean prestart,
                                        Comparator<? super Runnable> comparator,
                                        Adept2ThreadFactory threadFactory,
                                        int queueLimitPoolStatistics,
                                        Level minLogLevelPoolStatistics)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectOneWorkExecutor(boolean, int, int, Comparator, Adept2ThreadFactory, int, Level) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads, accepting exactly one piece of work per object. The work will be prioritised using the designated comparator, work exceeding the maximum pool size will be enqueued. The executor uses the designated thread factory for creating threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        blockUntilFlush - Whether work should be blocked for each object until the next flush. Otherwise work is allowed as soon as the work of the object has started.
        corePoolSize - The number of threads to keep in the pool, even if they are idle.
        maxPoolSize - The maximum number of threads to allow in the pool.
        prestart - Whether all core threads should be prestarted.
        comparator - The comparator used to determine the priority of work.
        threadFactory - The factory to use when the executor creates a new thread. Also the logger from this thread factory will be used by this executor.
        queueLimitPoolStatistics - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.
        minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

    • Method Detail

      • submitOverride

        public Future<?> submitOverride​(O object,
                                Runnable task)
        Submits the designated task for the designated object and overrides existing pending work.
        Parameters:
        object - The object for which to execute the designated task.
        task - The task which to execute.
        Returns:
        A Future representing pending completion of the task
        Throws:
        RejectedExecutionException - if the executor has been shut down, a RejectedExecutionException will be thrown.
        See Also:
        AbstractObjectSpecificExecutor.submit(Object, Runnable)

      • submitOverride

        public <T> Future<T> submitOverride​(O object,
                                    Callable<T> task)
        Submits the designated task for the designated object and overrides existing pending work.
        Parameters:
        object - The object for which to execute the designated task.
        task - The task which to execute.
        Returns:
        A Future representing pending completion of the task
        Throws:
        RejectedExecutionException - if the executor has been shut down, a RejectedExecutionException will be thrown.
        See Also:
        AbstractObjectSpecificExecutor.submit(Object, Callable)

      • addRunnable

        protected ObjectRunnable<O,​O> addRunnable​(O registeredObject,
                                                O indexObject,
                                                Runnable task)
        Description copied from class: AbstractObjectSpecificExecutor
        Adds the designated runnable for the designated object and index object to this executor.
        Implementations may store the work internally for later execution or assign it to an already existing active element (usually a thread). If they need a new active element for the designated runnable, they have to return a new runnable which will then be added to this executor using the normal CachedThreadPoolExecutor.execute(Runnable). If they return null, they have to assign the runnable to an active element eventually and arbitrarily.

        The caller has the lock on objects.

        Specified by:
        addRunnable in class AbstractObjectSpecificExecutor<O,​O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>
        Parameters:
        registeredObject - The object for which this runnable has been registered.
        indexObject - The object with which this runnable is managed and executed within the corresponding object-specific executor service.
        task - The task which to execute.
        Returns:
        A runnable which to add to the executor so that this gets a new active element (thread). This will be handled via CachedThreadPoolExecutor.execute(Runnable), which means the runnable will be executed independently from other work and a new thread may be started for it.

      • clearAbortedPendingWork

        protected void clearAbortedPendingWork​(boolean shutdownRequest)
        Removes all pending work that should be aborted for an aborting flush, that is, it removes the from the data structures, e. g. objects. This method only affects data structures, it does not affect any executed work. The caller will take care of this shortly.

        The caller has the lock on objects. Only removes the entries which do not have work pending for after a flush in case this is no shutdown request. Otherwise all objects and all work will be removed.

        Specified by:
        clearAbortedPendingWork in class AbstractObjectSpecificExecutor<O,​O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>
        Parameters:
        shutdownRequest - Whether this flush is caused by a AbstractObjectSpecificExecutor.shutdownNow(). In this case all work has to be aborted, even the work pending after a flush.

      • cancelAndFlush

        protected <T> Pair<Future<T>,​Map<O,​R>> cancelAndFlush​(boolean shutdownRequest,
                                                                  Map<O,​R> pending,
                                                                  Callable<T> task,
                                                                  Collection<ObjectRunnable<O,​O>> objectRunnables)
        Description copied from class: AbstractObjectSpecificExecutor
        Cancels the designated pending work (if appropriate), flushes (i. e. waits for or aborts the designated runnables) and executes the designated task.

        The caller has the lock on objects.

        Specified by:
        cancelAndFlush in class AbstractObjectSpecificExecutor<O,​O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>
        Parameters:
        shutdownRequest - Whether this flush is caused by a AbstractObjectSpecificExecutor.shutdownNow(). In this case all work has to be aborted, even the work pending after a flush.
        pending - The pending work that will not be executed due to abort or null if running work should not be aborted.
        task - The task which to execute without any object work running.
        objectRunnables - The currently running work. This will either be aborted or awaited for flushing.
        Returns:
        A pair containing the future representing the designated task and as second element either null in case no abortion is requested otherwise the managed objects and the runtime data of the aborted work. Note that the runtime data may contain work that is not cancelled by the flush. This depends on the runtime data.

      • getOutstandingRunnables

        protected List<Runnable> getOutstandingRunnables​(Map<O,​R> outstandingWork)
        Description copied from class: AbstractObjectSpecificExecutor
        Gets a list of runnables which have not been executed based on the designated outstanding work.
        This method will be called after an immediate shutdown which aborts pending work. Depending on the implementation, the designated runtime data may contain runnables which have not been executed and should therefore be returned to conform to the interface of an executor service.
        The caller does not have the lock on objects.
        Specified by:
        getOutstandingRunnables in class AbstractObjectSpecificExecutor<O,​O,​R extends AbstractObjectOneWorkExecutor.RuntimeData>
        Parameters:
        outstandingWork - The pending work that has been aborted due to the immediate shutdown.
        Returns:
        A list of runnables stemming from the designated outstanding work.

      • createObjectOneWorkRunnable

        protected ObjectOneWorkRunnable<O,​R> createObjectOneWorkRunnable​(O object)
        Creates a new object runnable. This method allows subclasses to create more specific runnables.
        Parameters:
        object - The object for which to create a runnable.
        Returns:
        A new runnable for the designated object.

      • createPendingBeforeFlush

        protected abstract R createPendingBeforeFlush​(Runnable pendingBeforeFlush)
        Creates runtime data having the designated runnable as pending before flush.
        Parameters:
        pendingBeforeFlush - The runnable which to have as pending before flush.
        Returns:
        Newly created runtime data having the designated runnable as pending before flush.

      • createPendingAfterFlush

        protected abstract R createPendingAfterFlush​(Runnable pendingAfterFlush)
        Creates runtime data having the designated runnable as pending after flush.
        Parameters:
        pendingAfterFlush - The runnable which to have as pending after flush.
        Returns:
        Newly created runtime data having the designated runnable as pending after flush.

      • addNewPendingBeforeFlush

        protected abstract R addNewPendingBeforeFlush​(Runnable pendingBeforeFlush,
                                              R rtData)
        Creates runtime data having the same data as the designated runtime data except for the pending before flush which will be the designated runnable.
        Parameters:
        pendingBeforeFlush - The runnable which to have as pending before flush.
        rtData - The runtime data from which to take over most of the data.
        Returns:
        Runtime data having the same data as the designated runtime data except for the pending before flush which will be the designated runnable.

      • addNewPendingAfterFlush

        protected abstract R addNewPendingAfterFlush​(R rtData,
                                             Runnable pendingAfterFlush)
        Creates runtime data having the same data as the designated runtime data except for the pending after flush which will be the designated runnable.
        Parameters:
        pendingAfterFlush - The runnable which to have as pending after flush.
        rtData - The runtime data from which to take over most of the data.
        Returns:
        Runtime data having the same data as the designated runtime data except for the pending after flush which will be the designated runnable.

      • removePendingBeforeFlush

        protected abstract R removePendingBeforeFlush​(R rtData)
        Creates runtime data having the same data as the designated runtime data except for the pending before flush which will be null.
        Parameters:
        rtData - The runtime data from which to take over most of the data.
        Returns:
        Runtime data having the same data as the designated runtime data except for the pending before flush which will be null.

      • switchPendingAfterFlush

        protected abstract R switchPendingAfterFlush​(R rtData)
        Creates runtime data having pending before flush taken from pending after flush from the designated runtime and pending after flush being null.
        This has to be called when restarting after a flush.
        Parameters:
        rtData - The runtime data from which to take over the data.
        Returns:
        Runtime data having pending before flush taken from pending after flush from the designated runtime and pending after flush being null.