worldthread.h

Go to the documentation of this file.

/*
  This file is part of MADNESS.

  Copyright (C) 2007,2010 Oak Ridge National Laboratory

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; either version 2 of the License, or
  (at your option) any later version.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

  For more information please contact:

  Robert J. Harrison
  Oak Ridge National Laboratory
  One Bethel Valley Road
  P.O. Box 2008, MS-6367

  email: harrisonrj@ornl.gov
  tel:   865-241-3937
  fax:   865-572-0680

  $Id $
*/
#ifndef MADNESS_WORLD_WORLDTHREAD_H__INCLUDED
#define MADNESS_WORLD_WORLDTHREAD_H__INCLUDED


#include <madness/TAU.h>
#include <madness/world/dqueue.h>
#include <vector>
#include <cstddef>
#include <cstdio>
#include <pthread.h>
#include <madness/world/typestuff.h>
#include <typeinfo>
#include <new>

#ifdef MADNESS_TASK_PROFILING
#include <execinfo.h> // for backtrace_symbols
#ifndef USE_LIBIBERTY
#include <cxxabi.h> // for abi::__cxa_demangle
#else
extern "C" {
  extern char * cplus_demangle (const char *mangled, int options);
#define DMGL_NO_OPTS     0              /* For readability... */
}
#endif
#include <sstream> // for std::istringstream
#include <cstring> // for strchr & strrchr
#endif // MADNESS_TASK_PROFILING

#ifdef HAVE_INTEL_TBB
#include "tbb/tbb.h"
#endif


#ifndef _SC_NPROCESSORS_CONF
// Old macs don't have necessary support thru sysconf to determine the
// no. of processors so must use sysctl
#include <sys/types.h>
#include <sys/sysctl.h>
#endif

namespace madness {

    // Forward decl.
    class Barrier;
    class ThreadPool;
    class WorldTaskQueue;
    class AtomicInt;
    void error(const char *msg);


    class ThreadBase {
        friend class ThreadPool;
        static bool bind[3];
        static int cpulo[3];
        static int cpuhi[3];
        static pthread_key_t thread_key; 

        static void* main(void* self);

        int pool_num; 
        pthread_t id;

        static void init_thread_key() {
           const int rc = pthread_key_create(&thread_key, NULL);
           if(rc != 0)
               MADNESS_EXCEPTION("pthread_key_create failed", rc);
        }

        static void delete_thread_key() {
           pthread_key_delete(thread_key);
        }

        void set_pool_thread_index(int i) { pool_num = i; }

#if defined(HAVE_IBMBGQ) and defined(HPM)
        static const int hpm_thread_id_all = -10;
        static const int hpm_thread_id_main = -2;
        static bool main_instrumented;
        static bool all_instrumented;
        static int hpm_thread_id;
#endif

    public:

        ThreadBase() : pool_num(-1) { }

        virtual ~ThreadBase() { }

        virtual void run() = 0;

        void start();

        static void exit() { pthread_exit(0); }

        const pthread_t& get_id() const { return id; }

        int get_pool_thread_index() const { return pool_num; }

        int cancel() const { return pthread_cancel(get_id()); }


        static int num_hw_processors();

        static void set_affinity_pattern(const bool bind[3], const int cpu[3]);

        static void set_affinity(int logical_id, int ind=-1);

        static ThreadBase* this_thread() {
            return static_cast<ThreadBase*>(pthread_getspecific(thread_key));
        }

#if defined(HAVE_IBMBGQ) and defined(HPM)
        static void set_hpm_thread_env(int hpm_thread_id);
#endif
    }; // class ThreadBase

    class Thread : public ThreadBase {
        void* (*f)(void *);
        void* args;

        void run() { f(args); }

    public:
        Thread() : f(0), args(0) { }

        Thread(void* (*f)(void *), void* args=0)
                : f(f), args(args) {
            ThreadBase::start();
        }

        void start(void* (*f)(void *), void* args=0) {
            this->f = f;
            this->args = args;
            ThreadBase::start();
        }

        virtual ~Thread() {}
    };



    class TaskAttributes {
        unsigned long flags;
    public:
        static const unsigned long NTHREAD      = 0xff;          // Mask for nthread byte
        static const unsigned long GENERATOR    = 1ul<<8;        // Mask for generator bit
        static const unsigned long STEALABLE    = GENERATOR<<1;  // Mask for stealable bit
        static const unsigned long HIGHPRIORITY = GENERATOR<<2;  // Mask for priority bit

        explicit TaskAttributes(unsigned long flags = 0) : flags(flags) {}

        TaskAttributes(const TaskAttributes& attr) : flags(attr.flags) {}

        virtual ~TaskAttributes() {}

        bool is_generator() const { return flags&GENERATOR; }

        bool is_stealable() const { return flags&STEALABLE; }

        bool is_high_priority() const { return flags&HIGHPRIORITY; }

        void set_generator(bool generator_hint) {
            if (generator_hint)
                flags |= GENERATOR;
            else
                flags &= ~GENERATOR;
        }

        void set_stealable(bool stealable) {
            if (stealable) flags |= STEALABLE;
            else flags &= ~STEALABLE;
        }

        void set_highpriority(bool hipri) {
            if (hipri)
                flags |= HIGHPRIORITY;
            else
                flags &= ~HIGHPRIORITY;
        }


        void set_nthread(int nthread) {
            MADNESS_ASSERT(nthread>=0 && nthread<256);
            flags = (flags & (~NTHREAD)) | (nthread & NTHREAD);
        }

        int get_nthread() const {
                int n = flags & NTHREAD;
                if (n == 0)
                    n = 1;
                return n;
        }

        template <typename Archive>
        void serialize(Archive& ar) {
            ar & flags;
        }

        static TaskAttributes generator() {
            return TaskAttributes(GENERATOR);
        }

        static TaskAttributes hipri() {
            return TaskAttributes(HIGHPRIORITY);
        }

        static TaskAttributes multi_threaded(int nthread) {
            TaskAttributes t;
            t.set_nthread(nthread);
            return t;
        }
    };

    class TaskThreadEnv {
        const int _nthread; 
        const int _id;      
        Barrier* _barrier;  

    public:
        TaskThreadEnv(int nthread, int id, Barrier* barrier)
            : _nthread(nthread), _id(id), _barrier(barrier)
        {}

#if HAVE_INTEL_TBB
        // I can not get the TaskThreadEnv to work with Barrier
        // Need to figure out why
        TaskThreadEnv(int nthread, int id)
            : _nthread(nthread), _id(id), _barrier(NULL)
        {};
#endif

        int nthread() const {return _nthread;}

        int id() const {return _id;}

        bool barrier() const {
            if (_nthread == 1)
                return true;
            else {
                MADNESS_ASSERT(_barrier);
                return _barrier->enter(_id);
            }
        }
    };


#ifdef MADNESS_TASK_PROFILING

    namespace profiling {


        class TaskEvent {
        private:
            double times_[3]; 
            std::pair<void*, unsigned short> id_; 
            unsigned short threads_; 


            static void print_demangled(std::ostream& os, const char* symbol) {
                // Get the demagled symbol name
                if(symbol) {
                    int status = 0;
#ifndef USE_LIBIBERTY
                    const char* name = abi::__cxa_demangle(symbol, 0, 0, &status);
#else
                    char* name = cplus_demangle(symbol, DMGL_NO_OPTS);
#endif
                    // Append the demangled symbol name to the output stream
                    if(status == 0) {
                        os << name << "\t";
                        free((void*)name);
                    } else {
                        os << symbol << "\t";
                    }
                } else {
                    os << "UNKNOWN\t";
                }
            }


            std::string get_name() const {

                // Get the backtrace symbol for the function address,
                // which contains the function name.
                void* const * func_ptr = const_cast<void* const *>(& id_.first);
                char** bt_sym = backtrace_symbols(func_ptr, 1);

                // Extract the mangled function name from the backtrace
                // symbol.
                std::string mangled_name;

#ifdef ON_A_MAC
                // Format of bt_sym is:
                // <frame #> <file name> <address> <mangled name> + <function offset>
                std::istringstream iss(bt_sym[0]);
                long frame;
                std::string file, address;
                iss >> frame >> file >> address >> mangled_name;
#else // Assume Linux
                // Format of bt_sym is:
                // <file>(<mangled name>+<function offset>) [<address>]
                const char* first = strchr(bt_sym[0],'(');
                if(first) {
                    ++first;
                    const char* last = strrchr(first,'+');
                    if(last)
                        mangled_name.assign(first, (last - first) - 1);
                }
#endif // ON_A_MAC

                // Free the backtrace buffer
                free(bt_sym);

                return mangled_name;
            }

        public:



            void start(const std::pair<void*, unsigned short>& id,
                    const unsigned short threads, const double submit_time)
            {
                id_ = id;
                threads_ = threads;
                times_[0] = submit_time;
                times_[1] = wall_time();
            }

            void stop() { times_[2] = wall_time(); }


            friend std::ostream& operator<<(std::ostream& os, const TaskEvent& te) {
                // Add address to output stream
                os << std::hex << std::showbase << te.id_.first <<
                        std::dec << std::noshowbase << "\t";

                // Print the name
                switch(te.id_.second) {
                    case 1:
                        {
                            const std::string mangled_name = te.get_name();

                            // Print the demangled name
                            if(! mangled_name.empty())
                                print_demangled(os, mangled_name.c_str());
                            else
                                os << "UNKNOWN\t";
                        }
                        break;
                    case 2:
                        print_demangled(os, static_cast<const char*>(te.id_.first));
                        break;
                    default:
                        os << "UNKNOWN\t";
                }

                // Print:
                // # of threads, submit time, start time, stop time
                os << te.threads_;
                const std::streamsize precision = os.precision();
                os.precision(6);
                os << std::fixed << "\t" << te.times_[0]
                        << "\t" << te.times_[1] << "\t" << te.times_[2];
                os.precision(precision);
                return os;
            }

        }; // class TaskEvent


        class TaskEventListBase {
        private:
            TaskEventListBase* next_;

            TaskEventListBase(const TaskEventListBase&);
            TaskEventListBase& operator=(const TaskEventListBase&);

        public:

            TaskEventListBase() : next_(NULL) { }

            virtual ~TaskEventListBase() { }

            TaskEventListBase* next() const { return next_; }


            void insert(TaskEventListBase* list) {
                if(next_)
                    list->next_ = next_;
                next_ = list;
            }


            friend inline std::ostream& operator<<(std::ostream& os, const TaskEventListBase& tel) {
                return tel.print_events(os);
            }

        private:

            virtual std::ostream& print_events(std::ostream&) const = 0;

        }; // class TaskEventList


        class TaskEventList : public TaskEventListBase {
        private:
            unsigned int n_; 
            TaskEvent* events_; 

            // Not allowed
            TaskEventList(const TaskEventList&);
            TaskEventList& operator=(const TaskEventList&);

        public:

            TaskEventList(const unsigned int nmax) :
                TaskEventListBase(), n_(0ul), events_(new TaskEvent[nmax])
            { }

            virtual ~TaskEventList() { delete [] events_; }


            TaskEvent* event() { return events_ + (n_++); }

        private:

            virtual std::ostream& print_events(std::ostream& os) const {
                const int thread_id = ThreadBase::this_thread()->get_pool_thread_index();
                for(std::size_t i = 0; i < n_; ++i)
                    os << thread_id << "\t" << events_[i] << std::endl;
                return os;
            }

        }; // class TaskEventList


        class TaskProfiler {
        private:
            TaskEventListBase* head_; 
            TaskEventListBase* tail_; 

            static Mutex output_mutex_; 

            // Not allowed
            TaskProfiler(const TaskProfiler&);
            TaskProfiler& operator=(const TaskProfiler&);

        public:
            static const char* output_file_name_; 
        public:
            TaskProfiler() : head_(NULL), tail_(NULL) { }

            ~TaskProfiler() {
                // Cleanup linked list
                TaskEventListBase* next = NULL;
                while(head_ != NULL) {
                    next = head_->next();
                    delete head_;
                    head_ = next;
                }
            }


            TaskEventList* new_list(const std::size_t nmax) {
                // Create a new event list
                TaskEventList* list = new TaskEventList(nmax);

                // Append the list to the tail of the linked list
                if(head_ != NULL) {
                    tail_->insert(list);
                    tail_ = list;
                } else {
                    head_ = list;
                    tail_ = list;
                }
                return list;
            }


            void write_to_file();
        }; // class TaskProfiler

    } // namespace profiling

#endif // MADNESS_TASK_PROFILING



    class PoolTaskInterface :
#ifdef HAVE_INTEL_TBB
            public tbb::task,
#endif // HAVE_INTEL_TBB
            public TaskAttributes
    {
        friend class ThreadPool;

    private:
        Barrier* barrier;     
        AtomicInt count;  

#ifdef MADNESS_TASK_PROFILING
        profiling::TaskEvent* task_event_;
        double submit_time_;
        std::pair<void*, unsigned short> id_;

        void set_event(profiling::TaskEvent* task_event) {
            task_event_ = task_event;
        }

        void submit() {
            submit_time_ = wall_time();
            this->get_id(id_);
        }
#endif // MADNESS_TASK_PROFILING


        template <typename T>
        union FunctionPointerGrabber {
            T in;
            void* out;
        };

    protected:


        template <typename fnT>
        static typename enable_if_c<detail::function_traits<fnT>::value ||
                detail::memfunc_traits<fnT>::value>::type
        make_id(std::pair<void*,unsigned short>& id, fnT fn) {
            FunctionPointerGrabber<fnT> poop;
            poop.in = fn;
            id.first = poop.out;
            id.second = 1ul;
        }

        template <typename fnobjT>
        static typename disable_if_c<detail::function_traits<fnobjT>::value ||
                detail::memfunc_traits<fnobjT>::value>::type
        make_id(std::pair<void*,unsigned short>& id, const fnobjT&) {
            id.first = reinterpret_cast<void*>(const_cast<char*>(typeid(fnobjT).name()));
            id.second = 2ul;
        }

    private:

        virtual void get_id(std::pair<void*,unsigned short>& id) const {
            id.first = NULL;
            id.second = 0ul;
        }

        bool run_multi_threaded() {
#ifdef HAVE_INTEL_TBB
            MADNESS_EXCEPTION("run_multi_threaded should not be called when using Intel TBB", 1);
#else
            // As a thread enters this routine it increments the shared counter
            // to generate a unique id without needing any thread-local storage.
            // A downside is this does not preserve any relationships between thread
            // numbering and the architecture ... more work ahead.
            int nthread = get_nthread();
            if (nthread == 1) {
#ifdef MADNESS_TASK_PROFILING
                task_event_->start(id_, nthread, submit_time_);
#endif // MADNESS_TASK_PROFILING
                TAU_START("PoolTaskInterface::run_multithreaded run(TaskThreadEnv)");
                run(TaskThreadEnv(1,0,0));
                TAU_STOP("PoolTaskInterface::run_multithreaded run(TaskThreadEnv)");
#ifdef MADNESS_TASK_PROFILING
                task_event_->stop();
#endif // MADNESS_TASK_PROFILING
                return true;
            }
            else {
                int id = count++;
                volatile bool barrier_flag;
                barrier->register_thread(id, &barrier_flag);

#ifdef MADNESS_TASK_PROFILING
                if(id == 0)
                    task_event_->start(id_, nthread, submit_time_);
#endif // MADNESS_TASK_PROFILING

                TAU_START("PoolTaskInterface::run_multithreaded run(TaskThreadEnv)");
                run(TaskThreadEnv(nthread, id, barrier));
                TAU_STOP("PoolTaskInterface::run_multithreaded run(TaskThreadEnv)");

#ifdef MADNESS_TASK_PROFILING
                const bool cleanup = barrier->enter(id);
                if(cleanup) task_event_->stop();
                return cleanup;
#else
                return barrier->enter(id);
#endif // MADNESS_TASK_PROFILING
            }
#endif // HAVE_INTEL_TBB
        }

    public:
        PoolTaskInterface()
            : TaskAttributes()
            , barrier(0)
        {
            count = 0;
        }

        explicit PoolTaskInterface(const TaskAttributes& attr)
            : TaskAttributes(attr)
            , barrier(attr.get_nthread()>1 ? new Barrier(attr.get_nthread()) : 0)
        {
            count = 0;
        }


        void set_nthread(int nthread) {
            if (nthread != get_nthread()) {
                TaskAttributes::set_nthread(nthread);
                delete barrier;
                if (nthread > 1)
                    barrier = new Barrier(nthread);
                else
                    barrier = 0;
            }
        }

#if HAVE_INTEL_TBB
        tbb::task* execute() {
            int nthread = get_nthread();
            int id = count++;
//            volatile bool barrier_flag;
//            barrier->register_thread(id, &barrier_flag);

            run( TaskThreadEnv(nthread, id) );
            return NULL;
        }

        static inline void * operator new(std::size_t size) throw(std::bad_alloc);

#endif // HAVE_INTEL_TBB

        static inline void operator delete(void* p, std::size_t size) throw() {
            if(p != NULL) {
#ifdef HAVE_INTEL_TBB
                tbb::task::destroy(*reinterpret_cast<tbb::task*>(p));
#else
                ::operator delete(p);
#endif // HAVE_INTEL_TBB
            }
        }


        virtual void run(const TaskThreadEnv& info) = 0;

        virtual ~PoolTaskInterface() {
            delete barrier;
        }
    };

    class PoolTaskNull : public PoolTaskInterface {
    public:
        void run(const TaskThreadEnv& /*info*/) {}
        virtual ~PoolTaskNull() {}
    private:
        virtual void get_id(std::pair<void*,unsigned short>& id) const {
            PoolTaskInterface::make_id(id, &PoolTaskNull::run);
        }
    };


    class ThreadPoolThread : public Thread {
    private:
        // Thread local data for thread pool
#ifdef MADNESS_TASK_PROFILING
        profiling::TaskProfiler profiler_;
#endif // MADNESS_TASK_PROFILING

    public:
        ThreadPoolThread() : Thread() { }

        virtual ~ThreadPoolThread() { }

#ifdef MADNESS_TASK_PROFILING
        profiling::TaskProfiler& profiler() { return profiler_; }
#endif // MADNESS_TASK_PROFILING
    };


    class ThreadPool {
    private:
        friend class WorldTaskQueue;

        // Thread pool data
        ThreadPoolThread *threads; 
        ThreadPoolThread main_thread; 
        DQueue<PoolTaskInterface*> queue; 
        int nthreads; 
        volatile bool finish; 
        AtomicInt nfinished; 

        // Static data
        static ThreadPool* instance_ptr; 
#ifdef __bgq__
#warning WE NEED TO TUNE THE nmax PARAMETER
#endif
        static const int nmax=128; // WAS 100 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! DEBUG
        static double await_timeout; 

#if defined(HAVE_IBMBGQ) and defined(HPM)
        static unsigned int main_hpmctx; // HPM context for main thread
#endif
        ThreadPool(int nthread=-1);

        ThreadPool(const ThreadPool&);           // Verboten
        void operator=(const ThreadPool&);       // Verboten

        int default_nthread();

        bool run_task(bool wait, ThreadPoolThread* this_thread) {
#if HAVE_INTEL_TBB
            MADNESS_EXCEPTION("run_task should not be called when using Intel TBB", 1);
#else

            if (!wait && queue.empty()) return false;
            std::pair<PoolTaskInterface*,bool> t = queue.pop_front(wait);
#ifdef MADNESS_TASK_PROFILING
            profiling::TaskEventList* event_list =
                    this_thread->profiler().new_list(1);
#endif // MADNESS_TASK_PROFILING
            // Task pointer might be zero due to stealing
            if (t.second && t.first) {
#ifdef MADNESS_TASK_PROFILING
                t.first->set_event(event_list->event());
#endif // MADNESS_TASK_PROFILING
                if (t.first->run_multi_threaded())         // What we are here to do
                    delete t.first;
            }
            return t.second;
#endif
        }

        bool run_tasks(bool wait, ThreadPoolThread* this_thread) {
#if HAVE_INTEL_TBB
//            if (!wait && tbb_task_list->empty()) return false;
//            tbb::task* t = &tbb_task_list->pop_front();
//            if (t) {
//                tbb_parent_task->increment_ref_count();
//                tbb_parent_task->enqueue(*t);
//            }

//            wait = (tbb_parent_task->ref_count() >= 1) ? false : true;
//            return wait;

            MADNESS_EXCEPTION("run_tasks should not be called when using Intel TBB", 1);
#else

            PoolTaskInterface* taskbuf[nmax];
            int ntask = queue.pop_front(nmax, taskbuf, wait);
#ifdef MADNESS_TASK_PROFILING
            profiling::TaskEventList* event_list =
                    this_thread->profiler().new_list(ntask);
#endif // MADNESS_TASK_PROFILING
            for (int i=0; i<ntask; ++i) {
                if (taskbuf[i]) { // Task pointer might be zero due to stealing
#ifdef MADNESS_TASK_PROFILING
                    taskbuf[i]->set_event(event_list->event());
#endif // MADNESS_TASK_PROFILING
                    if (taskbuf[i]->run_multi_threaded()) {
                        delete taskbuf[i];
                    }
                }
            }
            return (ntask>0);
#endif
        }

        void thread_main(ThreadPoolThread* const thread);

        static void* pool_thread_main(void *v);


        static ThreadPool* instance() {
#ifndef MADNESS_ASSERTIONS_DISABLE
            if(! instance_ptr) {
                std::cerr << "!!! ERROR: The thread pool has not been initialized.\n"
                          << "!!! ERROR: Call madness::initialize before submitting tasks to the task queue.\n";
                MADNESS_EXCEPTION("ThreadPool::instance_ptr is NULL", 0);
            }
#endif
            return instance_ptr;
        }


    public:

#if HAVE_INTEL_TBB
        // all tasks run as children of tbb_parent_task
        // be sure to allocate tasks with tbb_parent_task->allocate_child()
        static tbb::empty_task* tbb_parent_task;
        static tbb::task_scheduler_init* tbb_scheduler;
#endif

        static void begin(int nthread=-1);

        static void end();

        static void add(PoolTaskInterface* task) {
#ifdef MADNESS_TASK_PROFILING
            task->submit();
#endif // MADNESS_TASK_PROFILING
#if HAVE_INTEL_TBB
            ThreadPool::tbb_parent_task->increment_ref_count();
            if (task->is_high_priority()) {
                ThreadPool::tbb_parent_task->spawn(*task);
            }
            else {
                ThreadPool::tbb_parent_task->enqueue(*task);
            }
#else
            if (!task) MADNESS_EXCEPTION("ThreadPool: inserting a NULL task pointer", 1);
            int task_threads = task->get_nthread();
            // Currently multithreaded tasks must be shoved on the end of the q
            // to avoid a race condition as multithreaded task is starting up
            if (task->is_high_priority() && (task_threads == 1)) {
                instance()->queue.push_front(task);
            }
            else {
                instance()->queue.push_back(task, task_threads);
            }
#endif // HAVE_INTEL_TBB
        }

        template <typename opT>
        void scan(opT& op) {
            queue.scan(op);
        }


        static void add(const std::vector<PoolTaskInterface*>& tasks) {
#if HAVE_INTEL_TBB
            MADNESS_EXCEPTION("Do not add tasks to the madness task queue when using Intel TBB.", 1);
#else
            typedef std::vector<PoolTaskInterface*>::const_iterator iteratorT;
            for (iteratorT it=tasks.begin(); it!=tasks.end(); ++it) {
                add(*it);
            }
#endif
        }


        static bool run_task() {
#ifdef MADNESS_TASK_PROFILING
            return instance()->run_tasks(false, static_cast<ThreadPoolThread*>(ThreadBase::this_thread()));
#else
            return instance()->run_tasks(false, NULL);
#endif // MADNESS_TASK_PROFILING
        }

        static std::size_t size() {
            return instance()->nthreads;
        }

        static std::size_t queue_size() {
            return instance()->queue.size();
        }

        static const DQStats& get_stats();


        template <typename Probe>
        static void await(const Probe& probe, bool dowork = true) {
            double start = cpu_time();
            const double timeout = await_timeout;
            int counter = 0;

            MutexWaiter waiter;
            while (!probe()) {

#if HAVE_INTEL_TBB
                // TODO: Enable busy waiting with TBB.
                const bool working = false;
#else
                const bool working = (dowork ? ThreadPool::run_task() : false);
#endif // HAVE_INTEL_TBB
                const double current_time = cpu_time();

                if (working) {
                    // Reset timeout logic
                    waiter.reset();
                    start = current_time;
                    counter = 0;
                } else {
                    // Check for timeout
                    if(((current_time - start) > timeout) && (timeout > 1.0)) {
                        std::cout << "!!MADNESS: Hung queue?\n";
                        std::cout.flush();

                        if(counter++ > 3)
                            throw madness::MadnessException("ThreadPool::await() timeout",
                                    0, 1, __LINE__, __FUNCTION__, __FILE__);
                    }

                    waiter.wait();
                }
            }
        }

        ~ThreadPool() {
#if HAVE_INTEL_TBB
            tbb_parent_task->decrement_ref_count();
            tbb::task::destroy(*tbb_parent_task);
            tbb_scheduler->terminate();
            delete(tbb_scheduler);
#endif
        }
    };

#ifdef HAVE_INTEL_TBB
    inline void * PoolTaskInterface::operator new(std::size_t size) throw(std::bad_alloc)
    {
        if(! ThreadPool::tbb_parent_task) {
            std::cerr << "!!! Error: Cannot allocate task object because the thread pool has not been initialized.\n";
            throw std::bad_alloc();
        }
        return ::operator new(size, ThreadPool::tbb_parent_task->allocate_child());
    }
#endif // HAVE_INTEL_TBB

}

#endif // MADNESS_WORLD_WORLDTHREAD_H__INCLUDED