include/amino/tp_exec.h

Go to the documentation of this file.

/*
 * (c) Copyright 2008, IBM Corporation.
 * Licensed 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.
 *
 *  Change History:
 *
 *  yy-mm-dd  Developer  Defect     Description
 *  --------  ---------  ------     -----------
 *  08-07-28  ganzhi     N/A        Initial version
 */
#ifndef AMINO_UTIL_THREAD_EXECUTOR
#define AMINO_UTIL_THREAD_EXECUTOR

#include <pthread.h>
#include <string>
#include <stdio.h>
#include <memory>

#include <amino/thread.h>
#include <amino/exec_serv.h>
#include <amino/condition.h>
#include <amino/mutex.h>
#include <amino/lock.h>
#include <amino/ws_scheduler.h>
#include <amino/util.h>
#include <amino/future.h>
#include <amino/ftask.h>

#ifdef DEBUG_EXEC
#include <sstream>
#endif

namespace amino {
    using namespace std;

    class ThreadPoolExecutor;
    typedef ws_scheduler<Runnable> Scheduler;
    typedef Scheduler * pScheduler;

    class ThreadPoolExecutor:public ExecutorService{
        class WorkerThread:public Thread{
            private:
                ThreadPoolExecutor * pExec;
                int threadId;
            public:
                WorkerThread(ThreadPoolExecutor * tp, int index){
                    pExec = tp;
                    threadId = index;
                }

                virtual void* run() {
                    while(true){
                        Runnable * task = pExec->scheduler->getTask(threadId);
                        //NULL means end of task queue
                        if(task==NULL){
#ifdef DEBUG_EXEC
                            ostringstream output;
                            output<<"Thread Exit: "<<pExec->active_count->load()<<endl;
                            cout<<output.str();
#endif
                            unique_lock<mutex> llock(pExec->f_mutex_finish);
                            int count = pExec->active_count->fetch_sub(1, memory_order_relaxed);
                            assert(count>0);
                            SC_FENCE;
                            if(count==1){
#ifdef DEBUG
                                cout<<"I'm notifying finish of running()\n";
#endif
                                pExec->notify_finish();
                            }
                            return NULL;
                        }

#ifdef DEBUG_EXEC
                        cout<<"A real task\n";
#endif
                        task->run();
                    }
                }
        };

        friend class WorkerThread;
        typedef WorkerThread * pThread;

        private:
        pScheduler scheduler;

        /* Number of working threads */
        int fThreadNum;

        /* Array of threads */
        WorkerThread ** threads;

        /* Count of working threads */
        atomic<int> * active_count;

        void init(int threadNum){
            fShutdown = false;

            fThreadNum = threadNum;

            scheduler = new Scheduler(fThreadNum);
            active_count = new atomic<int>();
            active_count->store(threadNum, memory_order_relaxed);

            threads = new pThread[threadNum];
            for(int i=0;i<threadNum;i++){
                threads[i] = new WorkerThread(this, i);
                threads[i]->start();
            }
        }

        public:
        ThreadPoolExecutor(){
            init(getProcessNum());
        }

        ThreadPoolExecutor(int threadNum){
            init(threadNum);
        }

        virtual ~ThreadPoolExecutor(){
            if(!finished()){
                if(!fShutdown)
                    shutdown();
                waitTermination();
                // There is no problem if you used waitTermination() and get this error.
                // Some exceptions might skipped your waitTermination()
                // throw logic_error("Some threads are still running");
            }

            for(int i=0;i<fThreadNum;i++){
                delete threads[i];
            }

            delete [] threads;
            delete active_count;

            delete scheduler;
        }

        void execute(Runnable* task){
            if(fShutdown)
                throw logic_error("Already shutdown\n!");
            //TODO what will happen if shutdown becomes true just at this moments?
            scheduler->addTask(task);
        }

        virtual void shutdown(){
            fShutdown = true;
            scheduler->shutdown();
        }

        virtual void halt(){
            fShutdown = true;
        }

        virtual void waitTermination(){
            for(int i=0;i<fThreadNum;i++){
                threads[i]->join();
            }
        }

        virtual bool waitTermination(int timeOut){
            bool res=ExecutorService::waitTermination(timeOut);
            if(res)
                for(int i=0;i<fThreadNum;i++){
                    threads[i]->join();
                }
#ifdef DEBUG
            if(!res)
                assert(!finished());

            if(res)
                assert(finished());

            stringstream ss;
            ss<<"Timed Wait Output: "<< res<<endl;
            cout<<ss.str();
#endif
            return res;
        }

        protected:
        virtual bool finished(){
#ifdef DEBUG
            ostringstream output;
            output<<"Finish_Active_Count: "<<active_count->load(memory_order_relaxed)<<" in "<<this<<endl;
            cout<<output.str()<<flush;
            assert(active_count->load(memory_order_relaxed)>=0);
#endif
            unique_lock<mutex> llock(f_mutex_finish);
            return active_count->load(memory_order_relaxed)==0;
        }
    };
}
#endif

---