src/main/decaf/util/concurrent/ThreadPoolExecutor.h

Go to the documentation of this file.

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef _DECAF_UTIL_CONCURRENT_THREADPOOLEXECUTOR_H_
#define _DECAF_UTIL_CONCURRENT_THREADPOOLEXECUTOR_H_

#include <decaf/lang/Runnable.h>
#include <decaf/lang/Throwable.h>
#include <decaf/util/concurrent/ThreadFactory.h>
#include <decaf/util/concurrent/BlockingQueue.h>
#include <decaf/util/concurrent/TimeUnit.h>
#include <decaf/util/concurrent/AbstractExecutorService.h>
#include <decaf/util/concurrent/RejectedExecutionHandler.h>
#include <decaf/util/concurrent/RejectedExecutionException.h>
#include <decaf/util/LinkedList.h>
#include <decaf/util/ArrayList.h>
#include <decaf/util/Config.h>

#include <vector>

namespace decaf{
namespace util{
namespace concurrent{

    using decaf::lang::Pointer;

    class ExecutorKernel;

    class DECAF_API ThreadPoolExecutor : public AbstractExecutorService {
    private:

        ThreadPoolExecutor( const ThreadPoolExecutor& );
        ThreadPoolExecutor& operator= ( const ThreadPoolExecutor& );

    private:

        friend class ExecutorKernel;
        ExecutorKernel* kernel;

    public:

        ThreadPoolExecutor(int corePoolSize, int maxPoolSize,
                           long long keepAliveTime, const TimeUnit& unit,
                           BlockingQueue<decaf::lang::Runnable*>* workQueue);

        ThreadPoolExecutor(int corePoolSize, int maxPoolSize,
                           long long keepAliveTime, const TimeUnit& unit,
                           BlockingQueue<decaf::lang::Runnable*>* workQueue,
                           RejectedExecutionHandler* handler);

        ThreadPoolExecutor(int corePoolSize, int maxPoolSize,
                           long long keepAliveTime, const TimeUnit& unit,
                           BlockingQueue<decaf::lang::Runnable*>* workQueue,
                           ThreadFactory* threadFactory);

        ThreadPoolExecutor(int corePoolSize, int maxPoolSize,
                           long long keepAliveTime, const TimeUnit& unit,
                           BlockingQueue<decaf::lang::Runnable*>* workQueue,
                           ThreadFactory* threadFactory,
                           RejectedExecutionHandler* handler);

        virtual ~ThreadPoolExecutor();

        virtual void execute(decaf::lang::Runnable* task);

        virtual void shutdown();

        virtual ArrayList<decaf::lang::Runnable*> shutdownNow();

        virtual bool awaitTermination(long long timeout, const decaf::util::concurrent::TimeUnit& unit);

        virtual bool isShutdown() const;

        virtual bool isTerminated() const;

        virtual int getPoolSize() const;

        virtual int getCorePoolSize() const;

        virtual void setCorePoolSize(int poolSize);

        virtual int getMaximumPoolSize() const;

        virtual void setMaximumPoolSize(int maxSize);

        virtual long long getTaskCount() const;

        virtual int getActiveCount() const;

        virtual long long getCompletedTaskCount() const;

        virtual int getLargestPoolSize() const;

        virtual BlockingQueue<decaf::lang::Runnable*>* getQueue();

        virtual bool isTerminating() const;

        virtual void allowCoreThreadTimeout(bool value);

        virtual bool allowsCoreThreadTimeout() const;

        virtual long long getKeepAliveTime(const TimeUnit& unit) const;

        virtual void setKeepAliveTime(long long timeout, const TimeUnit& unit);

        virtual void setThreadFactory(ThreadFactory* factory);

        virtual ThreadFactory* getThreadFactory() const;

        virtual RejectedExecutionHandler* getRejectedExecutionHandler() const;

        virtual void setRejectedExecutionHandler(RejectedExecutionHandler* handler);

        virtual bool prestartCoreThread();

        virtual int prestartAllCoreThreads();

        bool remove(decaf::lang::Runnable* task);

        virtual void purge();

    protected:

        virtual void beforeExecute(decaf::lang::Thread* thread, decaf::lang::Runnable* task);

        virtual void afterExecute(decaf::lang::Runnable* task, decaf::lang::Throwable* error);

        virtual void terminated();

    public:  // RejectedExecutionHandler implementations.

        class AbortPolicy : public RejectedExecutionHandler {
        public:

            AbortPolicy() : RejectedExecutionHandler() {
            }

            virtual ~AbortPolicy() {
            }

            virtual void rejectedExecution(decaf::lang::Runnable* task, ThreadPoolExecutor* executer DECAF_UNUSED) {
                delete task;
                throw RejectedExecutionException(__FILE__, __LINE__, "Unable to execute task.");
            }

        };

        class CallerRunsPolicy : public RejectedExecutionHandler {
        public:

            CallerRunsPolicy() : RejectedExecutionHandler() {
            }

            virtual ~CallerRunsPolicy() {
            }

            virtual void rejectedExecution(decaf::lang::Runnable* task, ThreadPoolExecutor* executer DECAF_UNUSED) {

                if (executer->isShutdown()) {
                    delete task;
                    return;
                }

                try{
                    task->run();
                } catch(decaf::lang::Exception& ex) {
                    delete task;
                    throw ex;
                }
            }

            class DiscardPolicy : public RejectedExecutionHandler {
            public:

                DiscardPolicy() : RejectedExecutionHandler() {
                }

                virtual ~DiscardPolicy() {
                }

                virtual void rejectedExecution(decaf::lang::Runnable* task, ThreadPoolExecutor* executer DECAF_UNUSED) {
                    delete task;
                }

            };

            class DiscardOldestPolicy : public RejectedExecutionHandler {
            public:

                DiscardOldestPolicy() : RejectedExecutionHandler() {
                }

                virtual ~DiscardOldestPolicy() {
                }

                virtual void rejectedExecution( decaf::lang::Runnable* task, ThreadPoolExecutor* executer ) {

                    if (executer->isShutdown()) {
                        delete task;
                        return;
                    }

                    try{

                        decaf::lang::Runnable* oldest = NULL;
                        executer->getQueue()->poll(oldest);
                        delete oldest;

                        executer->execute(task);
                    } catch(decaf::lang::Exception& ex) {
                        delete task;
                        throw ex;
                    }
                }

            };

        };
    };

}}}

#endif /*_DECAF_UTIL_CONCURRENT_THREADPOOLEXECUTOR_H_*/