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

Class AbstractObjectSpecificExecutor<O1,​O2,​R>

  • Type Parameters:
    O1 - The type of objects for which runnables are registered.
    O2 - The type of objects with which runnables are managed and executed within this executor.
    R - The runtime data for managed objects. This allows for arbitrary data for each managed object.
    All Implemented Interfaces:
    Executor, ExecutorService
    Direct Known Subclasses:
    AbstractObjectOneWorkExecutor, AbstractPipeliningExecutor
    public abstract class AbstractObjectSpecificExecutor<O1,​O2,​R>
extends CachedThreadPoolExecutor
    A thread pool executor that accepts work for specific objects. How the objects are handled depends on the executor implementation. When submitting work without an object, this executor behaves just like any ThreadPoolExecutor.
    This executor provides the means to flush, that is, wait for pending work or abort pending work, execute a specific runnable and continue with new work arriving after the flush. Therefore flush synchronises the work of all objects.

    • Field Detail

      • objects

        protected final Map<O2,​Pair<ObjectRunnable<O1,​O2>,​R>> objects
        All managed objects, their runnables as well as their runtime data.

      • flushPending

        protected final AtomicInteger flushPending
        A counter for pending flushs. With this running object runnables will be terminated/suspended and concurrent flushs will be synchronised. That is, the threads causing the flush will block until all current pending flushs have been executed.
        Access has to be synchronised via objects.

      • abortCount

        protected final AtomicInteger abortCount
        A counter for all flushs that aborted pending work.
        Access has to be synchronised via objects.

      • shutdownPending

        protected final AtomicBoolean shutdownPending
        Whether a shutdown has been requested.

      • logger

        protected final Logger logger
        The logger for log messages, especially runtime exceptions occurring when executing work.

    • Constructor Detail

      • AbstractObjectSpecificExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectSpecificExecutor​(Adept2ThreadFactory threadFactory)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectSpecificExecutor(int, Adept2ThreadFactory) instead.
        Creates a new thread pool executor with an unlimited amount of threads for different objects using the designate thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        threadFactory - The factory to use when the executor creates a new thread.

      • AbstractObjectSpecificExecutor

        protected AbstractObjectSpecificExecutor​(int maxPoolSize,
                                         Adept2ThreadFactory threadFactory)
        Creates a new thread pool executor with the designated maximum amount of threads and threads for different objects. Threads are created using the designated thread factory. The threads will stay idle for 60 seconds before they terminate, no idle thread will live forever.
        Parameters:
        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. Also the logger from this thread factory will be used by this executor.

      • AbstractObjectSpecificExecutor

        protected AbstractObjectSpecificExecutor​(int corePoolSize,
                                         int maxPoolSize,
                                         Adept2ThreadFactory threadFactory)
        Creates a new thread pool executor with the designated number of threads and threads for different objects using the designate thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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. Also the logger from this thread factory will be used by this executor.

      • AbstractObjectSpecificExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectSpecificExecutor​(int corePoolSize,
                                         int maxPoolSize,
                                         boolean prestart,
                                         Adept2ThreadFactory threadFactory)
        Deprecated, for removal: This API element is subject to removal in a future version.
        Use AbstractObjectSpecificExecutor(int, int, Adept2ThreadFactory) instead.
        Creates a new thread pool executor with the designated number of threads and threads for different objects using the designate thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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. Also the logger from this thread factory will be used by this executor.

      • AbstractObjectSpecificExecutor

        protected AbstractObjectSpecificExecutor​(int corePoolSize,
                                         int maxPoolSize,
                                         Adept2ThreadFactory threadFactory,
                                         int queueLimitPoolStatistics,
                                         Level minLogLevelPoolStatistics)
        Creates a new thread pool executor with the designated number of threads and threads for different objects using the designate thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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. 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.

      • AbstractObjectSpecificExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectSpecificExecutor​(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 AbstractObjectSpecificExecutor(int, int, Adept2ThreadFactory, int, Level) instead.
        Creates a new thread pool executor with the designated number of threads and threads for different objects using the designate thread factory for creating threads. The threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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. 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.

      • AbstractObjectSpecificExecutor

        protected AbstractObjectSpecificExecutor​(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 using the designated comparator for prioritising work. More work will enqueued. The designated thread factory will be used for creating new threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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.

      • AbstractObjectSpecificExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectSpecificExecutor​(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 AbstractObjectSpecificExecutor(int, int, Comparator, Adept2ThreadFactory, Level) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads using the designated comparator for prioritising work. More work will enqueued. The designated thread factory will be used for creating new threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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.

      • AbstractObjectSpecificExecutor

        protected AbstractObjectSpecificExecutor​(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 using the designated comparator for prioritising work. More work will enqueued. The designated thread factory will be used for creating new threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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.

      • AbstractObjectSpecificExecutor

        @Deprecated(since="14.4.0",
            forRemoval=true)
protected AbstractObjectSpecificExecutor​(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 AbstractObjectSpecificExecutor(int, int, Comparator, Adept2ThreadFactory, int, Level) instead.
        Creates a new thread pool executor with the designated (maximum) amount of (core) threads using the designated comparator for prioritising work. More work will enqueued. The designated thread factory will be used for creating new threads. The non-core threads will stay idle for 60 seconds before they terminate.
        Parameters:
        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

      • execute

        public void execute​(O1 object,
                    Runnable task)
        Executes the designated task for the designated object. This just creates the index object, adds the runnable and executes it if not terminated or waiting for a flush.
        How the runnable and the objects are handled depends on the actual implementation.
        Parameters:
        object - The object for which to execute the designated task.
        task - The task which to execute.
        Throws:
        RejectedExecutionException - if the executor has been shut down, a RejectedExecutionException will be thrown.
        See Also:
        CachedThreadPoolExecutor.execute(Runnable)

      • getIndexObjectFor

        protected abstract O2 getIndexObjectFor​(O1 object)
        Gets the object used for managing and executing runnables within this executor.
        Parameters:
        object - The object for which a runnable is registered.
        Returns:
        The object used for managing and executing runnables (registered for the designated object) within this executor.

      • addRunnable

        protected abstract ObjectRunnable<O1,​O2> addRunnable​(O1 registeredObject,
                                                           O2 indexObject,
                                                           Runnable task)
        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.

        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.

      • flushAndExecute

        public <T> Pair<Future<T>,​Map<O2,​R>> flushAndExecute​(boolean abortPending,
                                                                 Callable<T> task)
        Flushes all work and executes the designated task. When aborting, all registered work will be removed and returned. Otherwise null will be returned and the object work will restart after the designated task has been executed.
        This method can be called concurrently. It is ensured that all concurrent flushs will be executed before the object work will restart. The tasks of concurrent flushs will be executed concurrently!
        Parameters:
        abortPending - Whether the present work should be cancelled. In this case the corresponding runtime data will be returned.
        task - The task which to execute without any object work running.
        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.
        Throws:
        RejectedExecutionException - if the executor has been shut down, a RejectedExecutionException will be thrown.

      • flushAndExecuteUnchecked

        protected <T> Pair<Future<T>,​Map<O2,​R>> flushAndExecuteUnchecked​(boolean abortPending,
                                                                             boolean shutdownRequest,
                                                                             Callable<T> task)
        The same as flushAndExecute(boolean, Callable) but without the check for a pending shutdown. This is required for the shutdown procedure.
        Parameters:
        abortPending - Whether the present work should be cancelled. In this case the corresponding runtime data will be returned.
        shutdownRequest - Whether this flush is caused by a shutdownNow(). In this case all work has to be aborted, even the work pending after a flush.
        task - The task which to execute without any object work running.
        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.
        Throws:
        RejectedExecutionException - if the executor has been shut down, a RejectedExecutionException will be thrown.

      • clearAbortedPendingWork

        protected abstract 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.

        Parameters:
        shutdownRequest - Whether this flush is caused by a shutdownNow(). In this case all work has to be aborted, even the work pending after a flush.

      • cancelAndFlush

        protected abstract <T> Pair<Future<T>,​Map<O2,​R>> cancelAndFlush​(boolean shutdownRequest,
                                                                            Map<O2,​R> pending,
                                                                            Callable<T> task,
                                                                            Collection<ObjectRunnable<O1,​O2>> objectRunnables)
        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.

        Parameters:
        shutdownRequest - Whether this flush is caused by a 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.

      • submit

        public Future<?> submit​(O1 object,
                        Runnable task)
        Submits the designated task for the designated object.
        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:
        AbstractExecutorService.submit(Runnable)

      • submit

        public <T> Future<T> submit​(O1 object,
                            Callable<T> task)
        Submits the designated task for the designated object.
        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:
        AbstractExecutorService.submit(Callable)

      • submitAndFlush

        public <T> Pair<Future<T>,​Map<O2,​R>> submitAndFlush​(boolean abortPending,
                                                                Callable<T> task)
        Submits the designated task and flushes all work. When aborting, all registered work will be removed and returned. Otherwise null will be returned and the work will restart after the designated task has been executed.
        This method can be called concurrently. It is ensured that all concurrent flushs will be executed before the work will restart. The tasks of concurrent flushs will be executed concurrently!
        Parameters:
        abortPending - Whether the present work should be cancelled. In this case the corresponding runnables will be returned.
        task - The task which to execute without any object work running.
        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.
        Throws:
        RejectedExecutionException - if the executor has been shut down.
        See Also:
        flushAndExecute(boolean, Callable)

      • restart

        protected boolean restart​(boolean abortedPending)
        Restarts the work after a flush. If there are several concurrent flushs the work will be started after the last flush.
        Parameters:
        abortedPending - Whether the flush leading to this restart has aborted pending work.
        Returns:
        Whether the work has been restarted. Otherwise there are concurrent flushs or the executor has been shut down.

      • restartPendingWork

        protected void restartPendingWork()
        Actually restarts by inspecting all existing objects for whether there is still a runnable for it. If so, this runnable will be executed, that is, the work will be added to the thread pool. If there is no such runnable, the object will be removed.

        The caller has the lock on objects.

      • shutdown

        public void shutdown()
        Shutting down will reject new work (normal or flushing). But the current work may terminate. If there is currently a flush pending, the actual shutdown will wait until the suppressed work has been executed.
        Specified by:
        shutdown in interface ExecutorService
        Overrides:
        shutdown in class ThreadPoolExecutor

      • executeShutdown

        protected abstract void executeShutdown​(Runnable shutdown)
        Executes the designated shutdown task which contains a flush calling ThreadPoolExecutor.shutdown() without abortion.
        This method is only called if there is pending work. So the implementation may decide on how and when to execute the shutdown runnable.

        The caller has the lock on objects.

        Parameters:
        shutdown - The task which to perform to shutdown.

      • shutdownRunnableRun

        protected void shutdownRunnableRun()
        This method is the run() method of the shutdown runnable. It shuts down this executor. Subclasses may override it for additional shutdown behaviour.

      • getOutstandingRunnables

        protected abstract List<Runnable> getOutstandingRunnables​(Map<O2,​R> outstandingWork)
        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.
        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.