JobQueue.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_JobQueue_h__
#define __thread_JobQueue_h__

#define FE_JQ_LOCK_DEBUG    FALSE

namespace fe
{
namespace ext
{

class JobQueuePre
{
    public:
                JobQueuePre(void)   { Mutex::confirm(); }
};

/**************************************************************************//**
    @brief Queue of tasks

    @ingroup thread
*//***************************************************************************/
template<typename T>
class JobQueue: public JobQueuePre, public Handled< JobQueue<T> >
{
    public:
                JobQueue(void):
                    m_workers(0),
                    m_barrierCount(0),
                    m_barrierMet(FALSE),
                    m_pException(NULL)
                {   Counted::setName("JobQueue"); }

                ~JobQueue(void)
                {
                    if(m_pException)
                    {
                        delete m_pException;
                        m_pException=NULL;
                    }
                }

        U32     workers(void) const         { return m_workers; }

        void    post(T job)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::post %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_jobMonitor);
//                  feLog("JobQueue::post %d\n",job);
                    m_posted.push(job);
                    m_jobAvailable.notifyOne();
                }
        void    post(T first_job,T last_job)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::post %s %s LOCK\n",
                            c_print(first_job),c_print(last_job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_jobMonitor);
//                  feLog("JobQueue::post %d\n",job);
                    for(T job=first_job;job<=last_job;job++)
                    {
                        m_posted.push(job);
                        m_jobAvailable.notifyOne();
                    }
                }

        BWORD   take(T& job)
                {
//                  feLog("JobQueue::take\n");
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::take %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_jobMonitor);
//                  feLog("JobQueue::take locked\n");
                    if(m_posted.empty())
                    {
                        return FALSE;
                    }
                    job=m_posted.front();
                    m_posted.pop();
                    return TRUE;
                }
        void    waitForJob(T& job)
                {
//                  feLog("JobQueue::waitForJob\n");
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::waitForJob %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_jobMonitor);
//                  feLog("JobQueue::waitForJob locked\n");
                    while(m_posted.empty())
                    {
//                      feLog("JobQueue::waitForJob empty\n");
                        m_jobAvailable.wait(lock);
                    }
                    FEASSERT(!m_posted.empty());
                    job=m_posted.front();
//                  feLog("JobQueue::waitForJob job=%d\n",job);
                    m_posted.pop();
                }

        void    deliver(T job)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::deliver %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_workerMonitor);
//                  feLog("JobQueue::deliver %d\n",job);
                    m_delivered.push(job);
                    m_workerDelivered.notifyOne();
                }

        BWORD   acceptDelivery(T& job)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::acceptDelivery %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_workerMonitor);
//                  feLog("JobQueue::acceptDelivery\n");
                    if(m_delivered.empty())
                    {
                        return FALSE;
                    }
                    job=m_delivered.front();
                    m_delivered.pop();
                    return TRUE;
                }
        void    waitForDelivery(T& job)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::waitForDelivery %s LOCK\n",c_print(job));
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_workerMonitor);
//                  feLog("JobQueue::waitForDelivery\n");
                    while(m_delivered.empty())
                    {
//                      feLog("JobQueue::waitForDelivery empty\n");
                        m_workerDelivered.wait(lock);
                    }
                    FEASSERT(!m_delivered.empty());
                    job=m_delivered.front();
                    m_delivered.pop();
                }

                /** @brief waits until all threads call sync

                    Returns true to exactly one of the threads. */
        BWORD   synchronize(BWORD a_supervised=FALSE)
                {
                    const U32 participants=m_workers+a_supervised;
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::synchronize LOCK\n");
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_barrierMonitor);
//                  feLog("JobQueue::sync %d\n",m_barrierCount);
                    while(m_barrierMet)
                    {
//                      feLog("JobQueue::sync previous %d/%d\n",
//                              m_barrierCount,participants);
                        m_barrierHit.wait(lock);
                    }
//                  feLog("JobQueue::sync %d++\n",m_barrierCount);
                    m_barrierCount++;
                    while(!m_barrierMet && m_barrierCount<participants)
                    {
//                      feLog("JobQueue::sync wait %d/%d\n",
//                              m_barrierCount,participants);
                        m_barrierHit.wait(lock);
                    }
                    BWORD result=(m_barrierCount==participants);
                    m_barrierCount--;
                    m_barrierMet=(m_barrierCount>0);
                    m_barrierHit.notifyAll();
//                  feLog("JobQueue::sync complete\n");
                    return result;
                }

        void    relayException(const Exception& a_rException)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::relayException LOCK\n");
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_workerMonitor);

                    if(!m_pException)
                    {
                        m_pException=new Exception(a_rException);
                    }
                };

        void    resolveException(void)
                {
#if FE_JQ_LOCK_DEBUG
                    feLog("JobQueue::resolveException LOCK\n");
#endif
                    Thread::interruptionPoint();

                    RecursiveMutex::Guard lock(m_workerMonitor);

                    if(m_pException)
                    {
                        Exception exception=*m_pException;

                        delete m_pException;
                        m_pException=NULL;

                        throw exception;
                    }
                };

    protected:
        U32                         m_workers;

    private:
        RecursiveMutex              m_jobMonitor;
        RecursiveMutex              m_workerMonitor;
        RecursiveMutex::Condition   m_jobAvailable;
        RecursiveMutex::Condition   m_workerDelivered;

        std::queue<T>               m_posted;
        std::queue<T>               m_delivered;

        RecursiveMutex              m_barrierMonitor;
        RecursiveMutex::Condition   m_barrierHit;
        U32                         m_barrierCount;
        U32                         m_barrierMet;

        Exception*                  m_pException;
};

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

#endif // __thread_JobQueue_h__