Threads.java - bookjar-utils - Project Hosting on Google Code

‹r159

r161

Source path: svn/ BookJar-utils/ src/ util/ Threads.java

package util;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.logging.Logger;
import util.io.IoUtils;

/**
 * Warning:
 * Some IO tasks don't respond to interrupts.
 * A common hang like this is using url.openStream -
 * one solution is to set the default timeout properties,
 * but that can lead to failures in slow networks.
 * Instead, in the ChainRunnable save the HttpURLConnection
 * (with a cast from url.openConnection) in a volatile field,
 * and in the doCancel(Throwable) method :
 * if(conn != null){
 * conn.setConnectTimeout(1);
 * conn.setReadTimeout(1);
 * conn.disconnect();
 * }
 *
 *
 * This class methods compose ChainRunnable, and
 * have a few Executor factories that have special
 * thread creation and survival characteristics.
 *
 * @author Owner
 */
public final class Threads {

    /**
     * This future is a null object. Its methods  do
     * nothing except log and return null (!) 
     * So if you have a bug, remember this is just
     * a initialization (volatile) and testing aid.
     * @param <T>
     * @return
     */
    public static <T> Future<T> newLoggingNullFuture() {
        return new FutureTask<T>(new Runnable() {

            @Override
            public void run() {
                Logger.getLogger(Threads.class.getName()).warning("Warning, using null object that probably shouldn't be used");
            }
        }, null);
    }

    /**
     * This runnable is a null object. Its method  does
     * nothing except log.
     * So if you have a bug, remember this is just
     * a initialization (volatile) and testing aid.
     * @param <T>
     * @return
     */
    public static Runnable newLoggingNullRunnable() {
        return new Runnable() {

            @Override
            public void run() {
                Logger.getLogger(Threads.class.getName()).warning("Warning, using null object that probably shouldn't be used");
            }
        };
    }

    /**
     * This future is a null object. Its call method  does
     * nothing, but it returns the given value.
     */
    public static <T> Future<T> newObjectFuture(T value) {
        return new FutureTask<T>(new Runnable() {

            @Override
            public void run() {
            }
        }, value);
    }

    private static ChainExecutor createExecutor(boolean shutdownOnExit, int minimumNThreads, int maximumNThreads, long secondsTimeOut, BlockingQueue<Runnable> queue, String name) {
        final ChainExecutor executor = new ChainExecutor(
                minimumNThreads,
                maximumNThreads,
                secondsTimeOut,
                TimeUnit.SECONDS,
                queue,
                IoUtils.createThreadFactory(true, "ChainExecutor-" + name));
        if (shutdownOnExit) {
            IoUtils.addShutdownHook(new Runnable() {

                @Override
                public void run() {
                    executor.shutdownNow();
                }
            });
        }
        return executor;
    }

    private Threads() {
    }

    /**
     * A pool with exactly N threads that rejects any task after
     * if there are no threads available.
     * @throws RejectedExecutionException if a task is submited
     * when all threads are busy.
     */
    public static ExecutorService newFixedRejectingExecutor(String name, int nThreads, boolean shutdownOnExit) {
        ChainExecutor executor = createExecutor(shutdownOnExit, nThreads, nThreads, 0L, new LinkedBlockingQueue<Runnable>(nThreads), name);
        return executor;
    }

    /**
     * A pool with exactly N threads that discards any task after
     * if there are no threads available.
     */
    public static ExecutorService newFixedDiscardingExecutor(String name, int nThreads, boolean shutdownOnExit) {
        ChainExecutor executor = createExecutor(shutdownOnExit, nThreads, nThreads, 0L, new LinkedBlockingQueue<Runnable>(nThreads), name);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        return executor;
    }

    /**
     * A pool with exactly nThreads that don't timeout
     * @param nThreads > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newFIFOFixedExecutor(String name, int nThreads, boolean shutdownOnExit) {
        return createExecutor(shutdownOnExit, nThreads, nThreads, 0L, new LinkedBlockingQueue<Runnable>(), name);
    }

    /**
     * A pool with exactly nThreads that don't timeout
     * @param nThreads > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newLIFOFixedExecutor(String name, int nThreads, boolean shutdownOnExit) {
        return createExecutor(shutdownOnExit, nThreads, nThreads, 0L, new LifoQueue<Runnable>(), name);
    }

    /**
     * A pool that has no maximum number of threads
     * and will kill the threads after a timeout after the last task
     * @param secondsTimeOut > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newFIFOCachedExecutor(String name, long secondsTimeout, boolean shutdownOnExit) {
        return createExecutor(shutdownOnExit, 0, Integer.MAX_VALUE, secondsTimeout, new SynchronousQueue<Runnable>(), name);
    }

    /**
     * A pool that has a maximum number of threads
     * and will kill the threads after a timeout after the last task
     * @param maximumNThreads > 0
     * @param secondsTimeOut > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newFIFOScalingExecutor(String name, int maximumNThreads, long secondsTimeOut, boolean shutdownOnExit) {
        ScalingQueue<Runnable> queue = new ScalingQueue<Runnable>();
        ChainExecutor executor = createExecutor(shutdownOnExit, 0, maximumNThreads, secondsTimeOut, queue, name);
        executor.setRejectedExecutionHandler(new ForceQueuePolicy());
        queue.setThreadPoolExecutor(executor);
        return executor;
    }

    /**
     * A pool that has a maximum number of threads
     * and will kill the threads after a timeout after the last task
     * @param maximumNThreads > 0
     * @param secondsTimeOut > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newLIFOScalingExecutor(String name, int maximumNThreads, long secondsTimeOut, boolean shutdownOnExit) {
        LifoScalingQueue<Runnable> queue = new LifoScalingQueue<Runnable>();
        ChainExecutor executor = createExecutor(shutdownOnExit, 0, maximumNThreads, secondsTimeOut, queue, name);
        executor.setRejectedExecutionHandler(new ForceQueuePolicy());
        queue.setThreadPoolExecutor(executor);
        return executor;
    }


    /**
     * A pool that has a minimum number of threads, a maximum number of threads
     * and will kill maximumNThreads - minimumNThreads after a timeout after the last task
     * @param minimumNThreads >= 0
     * @param maximumNThreads > 0
     * @param secondsTimeOut > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newFIFOScalingExecutor(String name, int minimumNThreads, int maximumNThreads, long secondsTimeOut, boolean shutdownOnExit) {
        ScalingQueue<Runnable> queue = new ScalingQueue<Runnable>();
        ChainExecutor executor = createExecutor(shutdownOnExit, minimumNThreads, maximumNThreads, secondsTimeOut, queue, name);
        executor.setRejectedExecutionHandler(new ForceQueuePolicy());
        queue.setThreadPoolExecutor(executor);
        return executor;
    }

    /**
     * A pool that has a minimum number of threads, a maximum number of threads
     * and will kill maximumNThreads - minimumNThreads after a timeout after the last task
     * @param minimumNThreads >= 0
     * @param maximumNThreads > 0
     * @param secondsTimeOut > 0
     * @param shutdownOnExit run a Shutdown Hook to shutdown on exit.
     * @return
     */
    public static ExecutorService newLIFOScalingExecutor(String name, int minimumNThreads, int maximumNThreads, long secondsTimeOut, boolean shutdownOnExit) {
        LifoScalingQueue<Runnable> queue = new LifoScalingQueue<Runnable>();
        ChainExecutor executor = createExecutor(shutdownOnExit, minimumNThreads, maximumNThreads, secondsTimeOut, queue, name);
        executor.setRejectedExecutionHandler(new ForceQueuePolicy());
        queue.setThreadPoolExecutor(executor);
        return executor;
    }

    /**
     * As you can see, we are going to reject the addition of a new task if
     * there are no threads to handle it. This will cause the thread pool executor
     * to try and allocate a new thread (up to the maximum threads).
     * If there are no threads, the task will be rejected.
     * In our case, if the task is rejected, we would like to put it back to the queue.
     * This is a simple thing to do with ThreadPoolExecutor since we can implement
     * our own RejectedExecutionHandler
     */
    private static class ScalingQueue<E> extends LinkedBlockingQueue<E> {

        /**
         * The executor this Queue belongs to
         */
        private ThreadPoolExecutor executor;

        /**
         * Creates a <tt>TaskQueue</tt> with a capacity of
         * {@link Integer#MAX_VALUE}.
         */
        public ScalingQueue() {
            super();
        }

        /**
         * Creates a <tt>TaskQueue</tt> with the given (fixed) capacity.
         *
         * @param capacity the capacity of this queue.
         */
        public ScalingQueue(int capacity) {
            super(capacity);
        }

        /**
         * Sets the executor this queue belongs to.
         */
        public void setThreadPoolExecutor(ThreadPoolExecutor executor) {
            this.executor = executor;
        }

        /**
         * Inserts the specified element at the tail of this queue if there is at
         * least one available thread to run the current task. If all pool threads
         * are actively busy, it rejects the offer.
         *
         * @param o the element to add.
         * @return <tt>true</tt> if it was possible to add the element to this
         *         queue, else <tt>false</tt>
         * @see ThreadPoolExecutor#execute(Runnable)
         */
        @Override
        public boolean offer(E o) {
            int allWorkingThreads = executor.getActiveCount() + super.size();
            return allWorkingThreads < executor.getPoolSize() && super.offer(o);
        }
    }

    /**
     * A queue that acts like a lifo
     */
    private static class LifoQueue<E> extends LinkedBlockingDeque<E> {

        public LifoQueue(int capacity) {
            super(capacity);
        }

        public LifoQueue() {
            super();
        }

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

        @Override
        public E take() throws InterruptedException {
            return takeFirst();
        }

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

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

        @Override
        public E peek() {
            return peekFirst();
        }

        @Override
        public E poll() {
            return pollFirst();
        }

        @Override
        public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            return pollFirst(timeout, unit);
        }

        @Override
        public E pop() {
            return removeFirst();
        }

        @Override
        public void push(E item) {
            addFirst(item);
        }
    }

    private static class LifoScalingQueue<E> extends LifoQueue<E> {

        /**
         * The executor this Queue belongs to
         */
        private ThreadPoolExecutor executor;

        /**
         * Creates a <tt>TaskQueue</tt> with a capacity of
         * {@link Integer#MAX_VALUE}.
         */
        public LifoScalingQueue() {
            super();
        }

        /**
         * Creates a <tt>TaskQueue</tt> with the given (fixed) capacity.
         *
         * @param capacity the capacity of this queue.
         */
        public LifoScalingQueue(int capacity) {
            super(capacity);
        }

        /**
         * Sets the executor this queue belongs to.
         */
        public void setThreadPoolExecutor(ThreadPoolExecutor executor) {
            this.executor = executor;
        }

        /**
         * Inserts the specified element at the tail of this queue if there is at
         * least one available thread to run the current task. If all pool threads
         * are actively busy, it rejects the offer.
         *
         * @param o the element to add.
         * @return <tt>true</tt> if it was possible to add the element to this
         *         queue, else <tt>false</tt>
         * @see ThreadPoolExecutor#execute(Runnable)
         */
        @Override
        public boolean offer(E o) {
            int allWorkingThreads = executor.getActiveCount() + super.size();
            return allWorkingThreads < executor.getPoolSize() && super.offer(o);
        }
    }

    private static class ForceQueuePolicy implements RejectedExecutionHandler {


        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                //should never happen since we never wait
                throw new RejectedExecutionException(e);
            }
        }
    }

    /**
     * Link conditional chains. If the first fail,
     * the second tries to execute, etc.
     * If one succeds the value returned is that
     * calculated there. No values are passed between
     * chains (they start with their value, default null).
     * It will ignore exceptions thrown by the composed
     * results. If all are cancelled/throw/return null,
     * it returns null.
     */
    public static <T> T compose(ChainCallable<T>... composed) {
        for (ChainCallable<T> chain : composed) {
            try {
                T output = chain.call();
                if (output != null) {
                    return output;
                }
            } catch (Exception t) {
                //ignore it, try the next.
            }
        }
        return null;
    }

    /**
     *  Creates a chain with a null first arg value
     * @param <T> the links arguments
     * @param firstArg the first argument
     * @param root the first link
     * @param links the rest of the links
     * @return a complete chain of execution
     */
    public static <T> ChainCallable<T> appendChains(ChainCallable<T> root, ChainCallable<T>... links) {
        return appendChains(null, root, links);
    }

    /**
     * Creates a chain with a T first arg value
     * @param <T> the links arguments
     * @param firstArg the first argument
     * @param root the first link
     * @param links the rest of the links
     * @return a complete chain of execution
     */
    public static <T> ChainCallable<T> appendChains(T firstArg, ChainCallable<T> root, ChainCallable<T>... links) {
        root.arg = firstArg;
        ChainCallable<T> last = root;
        if (links != null) {
            for (int i = 0; i < links.length; i++) {
                while (last.inner != null) {
                    last = last.inner;
                }
                last.inner = links[i];
                last = links[i];
            }
        }
        return root;
    }

    private static final class ChainExecutor extends ThreadPoolExecutor implements ExecutorService {
        //needs to be protected by synchronized
        //(including the iterator, and Collections doesn't do that)

        private final Object lock = new Object();
        private Collection<DelegateFutureTask> tasks = new HashSet<DelegateFutureTask>();

        /**
         * {@link  java.util.concurrent.ThreadPoolExecutor#ThreadPoolExecutor(int corePoolSize,
         * int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory) superclass contructor passthruu}
         */
        public ChainExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                ThreadFactory threadFactory) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }

        private final static class DelegateFutureTask<T> extends FutureTask<T> {

            private final ChainCallable innerComputation;

            public DelegateFutureTask(ChainCallable<T> callable) {
                super(callable);
                innerComputation = callable;
            }

            public ChainCallable getInnerComputation() {
                return innerComputation;
            }
        }

        /**
         * Override so that we can recognize the Chain.
         * @param <T>
         * @param callable
         * @return
         */
        @Override
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
            if (callable instanceof ChainCallable) {
                return new DelegateFutureTask<T>((ChainCallable<T>) callable);
            }
            return super.newTaskFor(callable);
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
            if (r instanceof DelegateFutureTask) {
                synchronized (lock) {
                    tasks.remove((DelegateFutureTask) r);
                }
            }
        }

        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            if (r instanceof DelegateFutureTask) {
                synchronized (lock) {
                    tasks.add((DelegateFutureTask) r);
                }
            }

            super.beforeExecute(t, r);
        }

        @Override
        public List<Runnable> shutdownNow() {
            List<Runnable> unExecuted = super.shutdownNow();
            InterruptedException interrupt = new InterruptedException("shutdown interrupt");
            synchronized (lock) {
                for (DelegateFutureTask c : tasks) {
                    ChainCallable computation = c.getInnerComputation();
                    computation.doCancel(interrupt);
                }
                tasks.clear();
            }

            return unExecuted;
        }
    }
}

Show details Hide details

Change log

r161 by i30817 on Today (2 hours ago) Diff

Refactoring. Fixed a bug with images
processing, and deleted some unused
functions.

Go to:

Older revisions

r159 by i30817 on Today (8 hours ago) Diff

Allow multiple body tags in html
files, don't allow url without a local
file to be saved in the books list

r147 by i30817 on Jan 06, 2010 Diff

New factory methods in Threads. A
small serialization fix.

r145 by i30817 on Jan 05, 2010 Diff

Remove possible exception in wrapper
because of mis-read javadoc

All revisions of this file

File info

Size: 19536 bytes, 558 lines

View raw file

File properties

svn:mergeinfo