Gang.h

Go to the documentation of this file.

/*  Copyright (C) 2003-2021 Free Electron Organization
    Any use of this software requires a license.  If a valid license
    was not distributed with this file, visit freeelectron.org. */

/** @file */

#ifndef __thread_Gang_h__
#define __thread_Gang_h__

namespace fe
{
namespace ext
{

/**************************************************************************//**
    @brief Group of Worker threads

    @ingroup thread
*//***************************************************************************/
template<typename WORKER,typename T>
class Gang: public JobQueue<T>
{
    public:
                Gang(void):
                    m_pGroup(NULL),
                    m_ppThread(NULL),
                    m_ppWorker(NULL)

                {   Counted::setName("Gang"); }

virtual         ~Gang(void)
                {
                    if(JobQueue<T>::m_workers)
                    {
                        finish();
                    }
                }

        BWORD   start(sp<Counted> a_spObject=sp<Counted>(NULL),
                        U32 a_threads=0,String a_stage="")
                {
                    FEASSERT(!JobQueue<T>::m_workers);

                    //* if threads arg is zero, create one thread per core
                    if(!a_threads)
                    {
                        a_threads=Thread::hardwareConcurrency();
                    }

                    JobQueue<T>::m_workers=a_threads;
                    sp< JobQueue<T> > spJobQueue(this);

                    FEASSERT(!m_pGroup);
                    m_pGroup=new Thread::Group();
                    m_ppWorker=new WORKER*[a_threads];
                    m_ppThread=new Thread*[a_threads];

                    for(U32 m=0;m<a_threads;m++)
                    {
                        m_ppWorker[m]=new WORKER(spJobQueue,m,a_stage);
                        m_ppWorker[m]->setObject(a_spObject);
                        m_ppThread[m]=NULL;
                    }

                    for(U32 m=0;m<a_threads;m++)
                    {
                        U32 tries=0;
                        const U32 maxTries=3;
                        while(TRUE)
                        {
                            try
                            {
                                //* boost can throw boost::thread_resource_error
                                m_ppThread[m]=m_pGroup->createThread(
                                        m_ppWorker[m]);
                                if(!m_ppThread[m])
                                {
                                    feX(e_cannotCreate,"Gang::start",
                                            "failed to create thread");
                                }

                            }
                            catch (std::exception& e)
                            {
#if !FE_RTTI_HOMEBREW
                                Exception* pException=fe_cast<Exception>(&e);
                                if(pException &&
                                        pException->getResult()==e_cannotCreate)
                                {
#if FE_CODEGEN<=FE_DEBUG
                                    feLog("Gang::start"
                                            " failed to create thread\n");
#endif
                                    return FALSE;
                                }
#endif

                                feLog("Gang::start"
                                        " failed to create thread "
                                        " worker %d/%d try %d/%d"
                                        " (\"%s\")\n",
                                        m,a_threads,tries,maxTries,e.what());
                                if(++tries>=maxTries)
                                {
                                    feLog("Gang::start"
                                            " giving up creating thread\n");

                                    for(U32 n=0;n<m;n++)
                                    {
                                        if(m_ppThread[n])
                                        {
                                            feLog("Gang::start"
                                                    " interrupt %d/%d\n",n,m);
                                            m_ppThread[n]->interrupt();
                                        }
                                    }

                                    feLog("Gang::start cleaning up\n");
                                    discardWorkers();
                                    feLog("Gang::start done\n");

                                    return FALSE;
                                }
                                milliSleep(1000);
                                feLog("Gang::start retrying\n");
                                continue;
                            }
                            if(tries)
                            {
                                feLog("Gang::start retry succeeded\n");
                            }
                            break;
                        }
                    }
                    return TRUE;
                }

        void    finish(void)
                {
                    FEASSERT(JobQueue<T>::m_workers);
                    FEASSERT(m_pGroup);

                    m_pGroup->joinAll();

                    discardWorkers();

                    //* empty out queue
                    T job;
                    while(JobQueue<T>::take(job)){}
                }

    private:

        void    discardWorkers(void)
                {
                    for(U32 m=0;m<JobQueue<T>::m_workers;m++)
                    {
//                      if(m_ppThread[m])
//                      {
//                          m_pGroup->remove_thread(m_ppThread[m]);
//                          m_ppThread[m]=NULL;
//                      }
                        if(m_ppThread[m])
                        {
                            delete m_ppThread[m];
                            m_ppThread[m]=NULL;
                        }

                        delete m_ppWorker[m];
                        m_ppWorker[m]=NULL;
                    }

                    delete[] m_ppThread;
                    m_ppThread=NULL;

                    delete[] m_ppWorker;
                    m_ppWorker=NULL;

                    delete m_pGroup;
                    m_pGroup=NULL;

                    JobQueue<T>::m_workers=0;
                }

        Thread::Group*  m_pGroup;
        Thread**        m_ppThread;
        WORKER**        m_ppWorker;
};

} /* namespace ext */
} /* namespace fe */

#endif // __thread_Gang_h__