SimWorker.cpp

#include "Headers.h"
#include "SimWorker.h"
#include "ExceptionWorker.h"
#include "ExceptionThread.h"
#include "ExceptionSignal.h"
#include "ExceptionMutex.h"

CSimWorker::CSimWorker(
    unsigned int    id,
    unsigned int    queueSize
    )
{
    this->id = id;
    this->queueSize = queueSize;
    this->queueCount = 0;
    this->queuePtr = 0;

    // Create queue
    this->queue = new CSimWorkItem*[this->queueSize];

    // Initial state
    this->state = WORKER_STATE_STOPPED;

#ifdef WIN32
    // Initialize the thread mutex
    InitializeCriticalSection(&this->threadMutex);

    // Initialize the state signal
    if(NULL == (this->stateHandle = CreateEvent(
        NULL,   // default security attributes
        false,  // manual reset
        false,  // initial state (true is signaled)
        NULL    // event name
        ))) throw CExceptionSignal("create state signal failed", __FILE__, __LINE__);

    // Initialize the work signal
    if(NULL == (this->workHandle = CreateEvent(
        NULL,   // default security attributes
        false,  // manual reset
        false,  // initial state (true is signaled)
        NULL    // event name
        ))) throw CExceptionSignal("create work signal failed", __FILE__, __LINE__);

#elif POSIX
    // Initialize the thread mutex
    if(pthread_mutex_init(&this->threadMutex, NULL)) throw CExceptionMutex("create thread mutex failed", __FILE__, __LINE__);
    
    // Initialize the state mutex
    if(pthread_mutex_init(&this->stateMutex, NULL)) throw CExceptionMutex("create state mutex failed", __FILE__, __LINE__);
    // Initialize the state codition
    if(pthread_cond_init(&this->stateCond, NULL)) throw CExceptionSignal("create state signal failed", __FILE__, __LINE__);

    // Initialize the work mutex
    if(pthread_mutex_init(&this->workMutex, NULL)) throw CExceptionMutex("create work mutex failed", __FILE__, __LINE__);
    // Initialize the work codition
    if(pthread_cond_init(&this->workCond, NULL)) throw CExceptionSignal("create work signal failed", __FILE__, __LINE__);
#endif
}

CSimWorker::~CSimWorker()
{
#ifdef WIN32
    // Close the thread mutex
    DeleteCriticalSection(&this->threadMutex);

    // Close the state signal
    if(!CloseHandle(this->stateHandle)) throw CExceptionSignal("close state signal failed", __FILE__, __LINE__);

    // Close the work signal
    if(!CloseHandle(this->workHandle)) throw CExceptionSignal("close work signal failed", __FILE__, __LINE__);

#elif POSIX
    // Close the thread mutex
    if(pthread_mutex_destroy(&this->threadMutex)) throw CExceptionMutex("close thread mutex failed", __FILE__, __LINE__);

    // Close the state condition
    if(pthread_cond_destroy(&this->stateCond)) throw CExceptionSignal("close state signal failed", __FILE__, __LINE__);
    // Close the state mutex
    if(pthread_mutex_destroy(&this->stateMutex)) throw CExceptionMutex("close state mutex failed", __FILE__, __LINE__);

    // Close the work condition
    if(pthread_cond_destroy(&this->workCond)) throw CExceptionSignal("close work signal failed", __FILE__, __LINE__);
    // Close the work mutex
    if(pthread_mutex_destroy(&this->workMutex)) throw CExceptionMutex("close work mutex failed", __FILE__, __LINE__);
#endif

    // Delete queue
    delete[] this->queue;
}

void CSimWorker::Start()
{
#ifdef WIN32
    // Lock the thread mutex
    EnterCriticalSection(&this->threadMutex);
    try
    {
        // Check the worker state
        if(WORKER_STATE_STOPPED != this->state) throw CExceptionWorker("worker is not stopped", __FILE__, __LINE__, this->id);

        // Change the worker state
        this->state = WORKER_STATE_STARTING;

        // Start thread
        if(!(this->threadHandle = CreateThread(
            NULL,                   // default security attributes
            0,                      // default stack size
            &CSimWorker::Execute,   // thread function address
            this,                   // parameter
            0,                      // creation flags
            &this->threadId         // thread ID
            ))) throw CExceptionThread("create thread failed", __FILE__, __LINE__, this->threadId);
    }
    catch(...)
    {
        // Unlock the thread mutex and re-throw exception
        LeaveCriticalSection(&this->threadMutex);
        throw;
    }
    // Unlock the thread mutex
    LeaveCriticalSection(&this->threadMutex);

    // Wait for thread to signal the state
    if(WAIT_FAILED == WaitForSingleObject(this->stateHandle, INFINITE)) throw CExceptionSignal("waiting for worker start signal failed", __FILE__, __LINE__);
#elif POSIX
    // Lock the thread mutex
    if(pthread_mutex_lock(&this->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
    // Lock the state mutex
    if(pthread_mutex_lock(&this->stateMutex)) throw CExceptionMutex("locking state mutex failed", __FILE__, __LINE__);
    try
    {
        // Check the worker state
        if(WORKER_STATE_STOPPED != this->state) throw CExceptionWorker("worker is not stopped", __FILE__, __LINE__, this->id);

        // Change the worker state
        this->state = WORKER_STATE_STARTING;

        if(pthread_create(
            &this->threadHandle,    // thread handle
            NULL,                   // default thread attribute
            &CSimWorker::Execute,   // thread function address
            (void*)this             // parameter
            )) throw CExceptionThread("create thread failed", __FILE__, __LINE__, this->threadHandle);
    }
    catch(...)
    {
        // Unlock the thread mutex and re-throw exception
        if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
        throw;
    }
    // Unlock the thread mutex
    if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);

    // Wait for thread to signal the state
    if(pthread_cond_wait(&this->stateCond, &this->stateMutex)) throw CExceptionSignal("waiting for worker start signal failed", __FILE__, __LINE__);
#endif
}

void CSimWorker::Stop()
{
#ifdef WIN32
    // Lock the thread mutex
    EnterCriticalSection(&this->threadMutex);
    try
    {
        // Check the worker state
        if((WORKER_STATE_STARTED_IDLE != this->state) && (WORKER_STATE_STARTED_BUSY != this->state))
            throw CExceptionWorker("worker is not started", __FILE__, __LINE__, this->id);

        // Save the old worker state
        EWorkerState oldState = this->state;

        // Change the worker state
        this->state = WORKER_STATE_STOPPING;

        // Send work change signal
        if(!SetEvent(this->workHandle)) throw CExceptionSignal("set work signal to signaled state failed", __FILE__, __LINE__);
    }
    catch(...)
    {
        // Unlock the thread mutex and re-throw exception
        LeaveCriticalSection(&this->threadMutex);
        throw;
    }
    // Unlock the thread mutex
    LeaveCriticalSection(&this->threadMutex);

    // Wait for thread to complete
    if(WAIT_FAILED == WaitForSingleObject(this->threadHandle, INFINITE)) throw CExceptionSignal("waiting for worker stop signal failed", __FILE__, __LINE__);

    // Close the thread handle
    if(!CloseHandle(this->threadHandle)) throw CExceptionThread("close thread failed", __FILE__, __LINE__, this->threadId);

#elif POSIX
    // Lock the thread mutex
    if(pthread_mutex_lock(&this->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
    try
    {
        // Check the worker state
        if((WORKER_STATE_STARTED_IDLE != this->state) && (WORKER_STATE_STARTED_BUSY != this->state))
            throw CExceptionWorker("worker is not started", __FILE__, __LINE__, this->id);

        // Change the worker state
        this->state = WORKER_STATE_STOPPING;

        // If can acquire a lock on the work mutex
        if(!pthread_mutex_trylock(&this->workMutex))
        {
            try
            {
                // Send work signal
                if(pthread_cond_signal(&this->workCond)) throw CExceptionSignal("set work signal to signaled state failed", __FILE__, __LINE__);
            }
            catch(...)
            {
                // Unlock the work mutex and re-throw the exception
                if(pthread_mutex_unlock(&this->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);
                throw;
            }
            // Unlock the work mutex
            if(pthread_mutex_unlock(&this->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);
        }
    }
    catch(...)
    {
        // Unlock the thread mutex and re-throw exception
        if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
        throw;
    }
    // Unlock the thread mutex
    if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);

    // Wait for thread to complete
    if(pthread_join(this->threadHandle, NULL)) throw CExceptionSignal("waiting for worker stop signal failed", __FILE__, __LINE__);
    // Unlock the state mutex
    if(pthread_mutex_unlock(&this->stateMutex)) throw CExceptionMutex("unlocking the state mutex failed", __FILE__, __LINE__);
#endif
}

void CSimWorker::Enqueue(CSimWorkItem* item)
{
#ifdef WIN32
    // Synchronize access to shared variables
    EnterCriticalSection(&this->threadMutex);
    try
    {
        // Check the worker state
        if((WORKER_STATE_STARTED_IDLE != this->state) && (WORKER_STATE_STARTED_BUSY != this->state)) throw CExceptionWorker("cannot enqueue item; worker to started", __FILE__, __LINE__, this->id);

        // Check the worker queue is not empty
        if(this->queueCount >= this->queueSize) throw CExceptionWorker("worker queue is full", __FILE__, __LINE__, this->id);

        // Add the work item to the queue
        this->queue[(this->queuePtr + this->queueCount) % this->queueSize] = item;
        this->queueCount++;

        // If the worker is in idle, send work change signal
        if(WORKER_STATE_STARTED_IDLE == this->state)
            if(!SetEvent(this->workHandle)) throw CExceptionSignal("set work signal to signaled state failed", __FILE__, __LINE__);
    }
    catch(...)
    {
        LeaveCriticalSection(&this->threadMutex);
        throw;
    }
    LeaveCriticalSection(&this->threadMutex);
#elif POSIX
    // Synchronize access to shared variables
    if(pthread_mutex_lock(&this->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
    try
    {
        // Check the worker state
        if((WORKER_STATE_STARTED_IDLE != this->state) && (WORKER_STATE_STARTED_BUSY != this->state)) throw CExceptionWorker("cannot enqueue item; worker to started", __FILE__, __LINE__, this->id);

        // Check the worker queue is not empty
        if(this->queueCount >= this->queueSize) throw CExceptionWorker("worker queue is full", __FILE__, __LINE__, this->id);

        // Add the work item to the queue
        this->queue[(this->queuePtr + this->queueCount) % this->queueSize] = item;
        this->queueCount++;

        // If can acquire a lock on the work mutex
        if(!pthread_mutex_trylock(&this->workMutex))
        {
            try
            {
                // Send work signal
                if(pthread_cond_signal(&this->workCond)) throw CExceptionSignal("set work signal to signaled state failed", __FILE__, __LINE__);
            }
            catch(...)
            {
                // Unlock the work mutex and re-throw the exception
                if(pthread_mutex_unlock(&this->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);
                throw;
            }
            // Unlock the work mutex
            if(pthread_mutex_unlock(&this->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);
        }
    }
    catch(...)
    {
        if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
        throw;
    }
    if(pthread_mutex_unlock(&this->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
#endif
}

#ifdef WIN32
DWORD CSimWorker::Execute(__in LPVOID parameter)
{
    // Get the object
    CSimWorker* worker = (CSimWorker*)(parameter);

    // Synchronize access to shared variables
    EnterCriticalSection(&worker->threadMutex);
    try
    {
        // Change the state
        worker->state = WORKER_STATE_STARTED_IDLE;

        // Send state change signal
        if(!SetEvent(worker->stateHandle)) throw CExceptionSignal("set state signal to signaled state failed", __FILE__, __LINE__);
    }
    catch(...)
    {
        LeaveCriticalSection(&worker->threadMutex);
        throw;
    }
    LeaveCriticalSection(&worker->threadMutex);

    // Worker execution loop
    do
    {
        // Wait for a work signal
        if(WAIT_FAILED == WaitForSingleObject(worker->workHandle, INFINITE)) throw CExceptionSignal("waiting for work signal failed", __FILE__, __LINE__);

        // Local copy of the work item
        CSimWorkItem* item = NULL;

        do
        {
            // Synchronize access to shared variables
            EnterCriticalSection(&worker->threadMutex);
            try
            {
                // If the worker is not stopping and the worker queue is not empty
                if(WORKER_STATE_STOPPING != worker->state)
                {
                    if(worker->queueCount)
                    {
                        // Pop the oldest work item from the queue
                        item = worker->queue[worker->queuePtr];

                        // Shift the end of the queue
                        worker->queuePtr = (++worker->queuePtr) % worker->queueSize;
                        worker->queueCount--;

                        // Set worker state to busy
                        worker->state = WORKER_STATE_STARTED_BUSY;
                    }
                    else
                    {
                        // Set item to null
                        item = NULL;

                        // Set worker state to idle
                        worker->state = WORKER_STATE_STARTED_IDLE;
                    }
                }
            }
            catch(...)
            {
                LeaveCriticalSection(&worker->threadMutex);
                throw;
            }
            LeaveCriticalSection(&worker->threadMutex);

            // If the work item is not null
            if(item && (WORKER_STATE_STOPPING != worker->state))
            {
                try
                {
                    // Execute the work item
                    item->Execute();

                    // Signal work item completion : success
                    item->Signal(CSimWorkItem::COMPLETED_SUCCESS);
                }
                catch(...)
                {
                    // Signal work item completion : fail
                    item->Signal(CSimWorkItem::COMPLETED_FAIL);
                }
            }
        }
        while(item && (WORKER_STATE_STOPPING != worker->state));

    }
    while(WORKER_STATE_STOPPING != worker->state);

    // Synchronize access to shared variables
    EnterCriticalSection(&worker->threadMutex);
    try
    {
        // Change worker state
        worker->state = WORKER_STATE_STOPPED;

        // Signal all work items remaining in the queue
        for(; worker->queueCount; worker->queueCount--)
        {
            // Pop item from the queue
            CSimWorkItem* item = worker->queue[worker->queuePtr];

            // Shift the end of the queue
            worker->queuePtr = (++worker->queuePtr) % worker->queueSize;

            // Signal work item completion : pending
            item->Signal(CSimWorkItem::PENDING);
        }
    }
    catch(...)
    {
        LeaveCriticalSection(&worker->threadMutex);
        throw;
    }
    LeaveCriticalSection(&worker->threadMutex);

    return 0;
}
#elif POSIX
void* CSimWorker::Execute(void* parameter)
{
    // Get the object
    CSimWorker* worker = (CSimWorker*)(parameter);

    // Lock the work mutex
    if(pthread_mutex_lock(&worker->workMutex)) throw CExceptionMutex("locking work mutex failed", __FILE__, __LINE__);

    try
    {
        // Synchronize access to shared variables
        if(pthread_mutex_lock(&worker->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
        try
        {
            // Change the state
            worker->state = WORKER_STATE_STARTED_IDLE;

            // Send state change signal

            // Lock the state mutex
            if(pthread_mutex_lock(&worker->stateMutex)) throw CExceptionMutex("locking state mutex failed", __FILE__, __LINE__);
            try
            {
                // Send signal
                if(pthread_cond_signal(&worker->stateCond)) throw CExceptionSignal("set state signal to signaled state failed", __FILE__, __LINE__);
            }
            catch(...)
            {
                // Unlock the state mutex
                if(pthread_mutex_unlock(&worker->stateMutex)) throw CExceptionMutex("unlocking state mutex failed", __FILE__, __LINE__);
                throw;
            }
            if(pthread_mutex_unlock(&worker->stateMutex)) throw CExceptionMutex("unlocking state mutex failed", __FILE__, __LINE__);
        }
        catch(...)
        {
            if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
            throw;
        }
        if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);

        // Worker execution loop
        do
        {
            // Wait for a work signal
            if(pthread_cond_wait(&worker->workCond, &worker->workMutex)) throw CExceptionSignal("waiting for work signal failed", __FILE__, __LINE__);

            // Local copy of the work item
            CSimWorkItem* item = NULL;

            do
            {
                // Synchronize access to shared variables
                if(pthread_mutex_lock(&worker->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
                try
                {
                    // If the worker is not stopping and the worker queue is not empty
                    if(WORKER_STATE_STOPPING != worker->state)
                    {
                        if(worker->queueCount)
                        {
                            // Pop the oldest work item from the queue
                            item = worker->queue[worker->queuePtr];

                            // Shift the end of the queue
                            worker->queuePtr = (++worker->queuePtr) % worker->queueSize;
                            worker->queueCount--;

                            // Set worker state to busy
                            worker->state = WORKER_STATE_STARTED_BUSY;
                        }
                        else
                        {
                            // Set item to null
                            item = NULL;

                            // Set worker state to idle
                            worker->state = WORKER_STATE_STARTED_IDLE;
                        }
                    }
                }
                catch(...)
                {
                    if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
                    throw;
                }
                if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);

                // If the work item is not null
                if(item && (WORKER_STATE_STOPPING != worker->state))
                {
                    try
                    {
                        // Execute the work item
                        item->Execute();

                        // Signal work item completion : success
                        item->Signal(CSimWorkItem::COMPLETED_SUCCESS);
                    }
                    catch(...)
                    {
                        // Signal work item completion : fail
                        item->Signal(CSimWorkItem::COMPLETED_FAIL);
                    }
                }
            }
            while(item && (WORKER_STATE_STOPPING != worker->state));

        }
        while(WORKER_STATE_STOPPING != worker->state);

        // Synchronize access to shared variables
        if(pthread_mutex_lock(&worker->threadMutex)) throw CExceptionMutex("locking thread mutex failed", __FILE__, __LINE__);
        try
        {
            // Change worker state
            worker->state = WORKER_STATE_STOPPED;

            // Signal all work items remaining in the queue
            for(; worker->queueCount; worker->queueCount--)
            {
                // Pop item from the queue
                CSimWorkItem* item = worker->queue[worker->queuePtr];

                // Shift the end of the queue
                worker->queuePtr = (++worker->queuePtr) % worker->queueSize;

                // Signal work item completion : pending
                item->Signal(CSimWorkItem::PENDING);
            }
        }
        catch(...)
        {
            if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
            throw;
        }
        if(pthread_mutex_unlock(&worker->threadMutex)) throw CExceptionMutex("unlocking thread mutex failed", __FILE__, __LINE__);
    }
    catch(...)
    {
        // Unlock the work mutex and re-throw exception
        if(pthread_mutex_unlock(&worker->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);
        throw;
    }
    // Unlock the work mutex
    if(pthread_mutex_unlock(&worker->workMutex)) throw CExceptionMutex("unlocking work mutex failed", __FILE__, __LINE__);

    return 0;
}
#endif