CAmSocketHandler.cpp

Go to the documentation of this file.

#include "CAmSocketHandler.h"
#include <cassert>
#include <sys/fcntl.h>
#include <sys/errno.h>
#include <sys/poll.h>
#include <time.h>
#include <algorithm>
#include <features.h>
#include <csignal>
#include <unistd.h>
#include "CAmDltWrapper.h"

namespace am
{

CAmSocketHandler::CAmSocketHandler() :
        receiverCallbackT(this, &CAmSocketHandler::receiverCallback),//
        checkerCallbackT(this, &CAmSocketHandler::checkerCallback),//
        mPipe(), //
        mDispatchDone(1),//
        mListPoll(), //
        mListTimer(), //
        mListActiveTimer(), //
        mLastInsertedHandle(0), //
        mLastInsertedPollHandle(0), //
        mRecreatePollfds(true), //
        mStartTime() //
{
    if (pipe(mPipe) == -1)
    {
        logError("Sockethandler could not create pipe!");
    }

    //add the pipe to the poll - nothing needs to be proccessed here we just need the pipe to trigger the ppoll
    short event = 0;
    sh_pollHandle_t handle;
    event |= POLLIN;
    addFDPoll(mPipe[0], event, NULL, &receiverCallbackT, &checkerCallbackT, NULL, NULL, handle);
}

CAmSocketHandler::~CAmSocketHandler()
{
    close(mPipe[0]);
    close(mPipe[1]);
}

//todo: maybe have some: give me more time returned?
void CAmSocketHandler::start_listenting()
{
    mDispatchDone = 0;
    int16_t pollStatus;

    //prepare the signalmask
    sigset_t sigmask;
    sigemptyset(&sigmask);
    sigaddset(&sigmask, SIGINT);
    sigaddset(&sigmask, SIGQUIT);
    sigaddset(&sigmask, SIGTERM);
    sigaddset(&sigmask, SIGHUP);
    sigaddset(&sigmask, SIGQUIT);

    clock_gettime(CLOCK_MONOTONIC, &mStartTime);
    while (!mDispatchDone)
    {
        //first we go through the registered filedescriptors and check if someone needs preparation:
        std::for_each(mListPoll.begin(), mListPoll.end(), CAmShCallPrep());

        if (mRecreatePollfds)
        {
            mfdPollingArray.clear();
            //there was a change in the setup, so we need to recreate the fdarray from the list
            std::for_each(mListPoll.begin(), mListPoll.end(), CAmShCopyPollfd(mfdPollingArray));
            mRecreatePollfds = false;
        }

        timerCorrection();

        //block until something is on a filedescriptor

        timespec buffertime;
        if ((pollStatus = ppoll(&mfdPollingArray[0], mfdPollingArray.size(), insertTime(buffertime), &sigmask)) < 0)
        {
            if (errno == EINTR)
            {
                //a signal was received, that means it's time to go...
                pollStatus = 0;
            }
            else
            {
                logError("SocketHandler::start_listenting ppoll returned with error", errno);
                throw std::runtime_error(std::string("SocketHandler::start_listenting ppoll returned with error."));
            }
        }

        if (pollStatus != 0) //only check filedescriptors if there was a change
        {
            //todo: here could be a timer that makes sure naughty plugins return!

            //freeze mListPoll by copying it - otherwise we get problems when we want to manipulate it during the next lines
            std::list<sh_poll_s> listPoll;
            mListPoll_t::iterator listmPollIt;

            //remove all filedescriptors who did not fire
            std::vector<pollfd>::iterator it = mfdPollingArray.begin();
            do
            {
                it = std::find_if(it, mfdPollingArray.end(), eventFired);
                if (it != mfdPollingArray.end())
                {
                    listmPollIt = mListPoll.begin();
                    std::advance(listmPollIt, std::distance(mfdPollingArray.begin(), it));
                    listPoll.push_back(*listmPollIt);
                    listPoll.back().pollfdValue = *it;
                    it++;
                }
            } while (it != mfdPollingArray.end());

            //stage 1, call firedCB
            std::for_each(listPoll.begin(), listPoll.end(), CAmShCallFire());

            //stage 2, lets ask around if some dispatching is necessary, the ones who need stay on the list
            listPoll.remove_if(noDispatching);

            //stage 3, the ones left need to dispatch, we do this as long as there is something to dispatch..
            do
            {
                listPoll.remove_if(dispatchingFinished);
            } while (!listPoll.empty());

        }
        else //Timerevent
        {
            //this was a timer event, we need to take care about the timers
            timerUp();
        }
    }
}

void CAmSocketHandler::stop_listening()
{
    mDispatchDone = 1;

    //this is for all running timers only - we need to handle the additional offset here
    if (!mListActiveTimer.empty())
    {
        timespec currentTime, correctionTime;
        clock_gettime(CLOCK_MONOTONIC, &currentTime);
        correctionTime = timespecSub(currentTime, mStartTime);
        std::for_each(mListActiveTimer.begin(), mListActiveTimer.end(), CAmShSubstractTime(correctionTime));
    }

}

am_Error_e CAmSocketHandler::addFDPoll(const int fd, const short event, IAmShPollPrepare *prepare, IAmShPollFired *fired, IAmShPollCheck *check, IAmShPollDispatch *dispatch, void *userData, sh_pollHandle_t & handle)
{
    if (!fdIsValid(fd))
        return (E_NON_EXISTENT);
        
     //create a new handle for the poll
    sh_pollHandle_t lastHandle(mLastInsertedPollHandle);
    do
    {
        ++mLastInsertedPollHandle;
        if (mLastInsertedPollHandle == MAX_POLLHANDLE)
        {
                mLastInsertedPollHandle = 1;
        }
        if (mLastInsertedPollHandle==lastHandle)
        {
            logError(__func__,"Could not create new polls, too many open!");
            return (am_Error_e::E_NOT_POSSIBLE);
        }
            
    } while (mSetPollKeys.find(mLastInsertedPollHandle) != mSetPollKeys.end());
    
    mSetPollKeys.insert(mLastInsertedPollHandle);

    sh_poll_s pollData;
    pollData.pollfdValue.fd = fd;
    pollData.handle = mLastInsertedPollHandle;
    pollData.pollfdValue.events = event;
    pollData.pollfdValue.revents = 0;
    pollData.userData = userData;
    pollData.prepareCB = prepare;
    pollData.firedCB = fired;
    pollData.checkCB = check;
    pollData.dispatchCB = dispatch;

    //add new data to the list
    mListPoll.push_back(pollData);

    mRecreatePollfds = true;

    handle = pollData.handle;
    return (E_OK);
}

am_Error_e CAmSocketHandler::removeFDPoll(const sh_pollHandle_t handle)
{
    mListPoll_t::iterator iterator = mListPoll.begin();

    for (; iterator != mListPoll.end(); ++iterator)
    {
        if (iterator->handle == handle)
        {
            iterator = mListPoll.erase(iterator);
            mSetPollKeys.erase(handle);
            mRecreatePollfds = true;
            return (E_OK);
        }
    }
    return (E_UNKNOWN);
}

am_Error_e CAmSocketHandler::addTimer(const timespec timeouts, IAmShTimerCallBack* callback, sh_timerHandle_t& handle, void * userData)
{
    assert(!((timeouts.tv_sec==0) && (timeouts.tv_nsec==0)));
    assert(callback!=NULL);

    sh_timer_s timerItem;

    //create a new handle for the timer
    sh_timerHandle_t lastTimerHandle(mLastInsertedHandle);
    do
    {
        ++mLastInsertedHandle;
        if (mLastInsertedHandle == MAX_TIMERHANDLE)
        {
                mLastInsertedHandle = 1;
        }
        if (lastTimerHandle==mLastInsertedHandle)
        {
            logError(__func__,"Could not create new timers, too many open!");
            return (am_Error_e::E_NOT_POSSIBLE);    
        }
            
    } while (mSetTimerKeys.find(mLastInsertedHandle) != mSetTimerKeys.end());
    
    mSetTimerKeys.insert(mLastInsertedHandle);
    handle=mLastInsertedHandle;
    timerItem.handle = handle;
    timerItem.countdown = timeouts;
    timerItem.callback = callback;
    timerItem.userData = userData;

    mListTimer.push_back(timerItem);

    //we add here the time difference between startTime and currenttime, because this time will be substracted later on in timecorrection
    timespec currentTime;
    clock_gettime(CLOCK_MONOTONIC, &currentTime);
    if (!mDispatchDone) //the mainloop is started
        timerItem.countdown = timespecAdd(timeouts, timespecSub(currentTime, mStartTime));

    mListActiveTimer.push_back(timerItem);
    mListActiveTimer.sort(compareCountdown);
    return (E_OK);
}

am_Error_e CAmSocketHandler::removeTimer(const sh_timerHandle_t handle)
{
    assert(handle!=0);

    //stop the current timer
    stopTimer(handle);

    std::list<sh_timer_s>::iterator it(mListTimer.begin());
    for (; it != mListTimer.end(); ++it)
    {
        if (it->handle == handle)
        {
            it = mListTimer.erase(it);
            mSetTimerKeys.erase(handle);
            return (E_OK);
        }
    }
    return (E_UNKNOWN);
}

am_Error_e CAmSocketHandler::updateTimer(const sh_timerHandle_t handle, const timespec timeouts)
{
    //update the mList ....
    sh_timer_s timerItem;
    std::list<sh_timer_s>::iterator it(mListTimer.begin()), activeIt(mListActiveTimer.begin());
    bool found(false);
    for (; it != mListTimer.end(); ++it)
    {
        if (it->handle == handle)
        {
            it->countdown = timeouts;
            timerItem = *it;
            found = true;
            break;
        }
    }
    if (!found)
        return (E_NON_EXISTENT);

    found = false;

    //we add here the time difference between startTime and currenttime, because this time will be substracted later on in timecorrection
    timespec currentTime, timeoutsCorrected;
    currentTime.tv_nsec=timeoutsCorrected.tv_nsec=0;
    currentTime.tv_sec=timeoutsCorrected.tv_sec=0;
    clock_gettime(CLOCK_MONOTONIC, &currentTime);
    if (!mDispatchDone) //the mainloop is started
        timeoutsCorrected = timespecAdd(timeouts, timespecSub(currentTime, mStartTime));

    for (; activeIt != mListActiveTimer.end(); ++activeIt)
    {
        if (activeIt->handle == handle)
        {
            activeIt->countdown = timeoutsCorrected;
            found = true;
            break;
        }
    }

    if (!found)
        timerItem.countdown = timeoutsCorrected;
    mListActiveTimer.push_back(timerItem);

    mListActiveTimer.sort(compareCountdown);
    return (E_OK);
}

am_Error_e CAmSocketHandler::restartTimer(const sh_timerHandle_t handle)
{
    sh_timer_s timerItem; 
    //find the original value
    std::list<sh_timer_s>::iterator it(mListTimer.begin()), activeIt(mListActiveTimer.begin());
    bool found(false);
    for (; it != mListTimer.end(); ++it)
    {
        if (it->handle == handle)
        {
            timerItem = *it;
            found = true;
            break;
        }
    }
    if (!found)
        return (E_NON_EXISTENT);

    found = false;

    //we add here the time difference between startTime and currenttime, because this time will be substracted later on in timecorrection
    timespec currentTime, timeoutsCorrected;
    clock_gettime(CLOCK_MONOTONIC, &currentTime);
    if (!mDispatchDone) //the mainloop is started
    {
        timeoutsCorrected = timespecAdd(timerItem.countdown, timespecSub(currentTime, mStartTime));
        timerItem.countdown = timeoutsCorrected;
    }

    for (; activeIt != mListActiveTimer.end(); ++activeIt)
    {
        if (activeIt->handle == handle)
        {
            activeIt->countdown = timerItem.countdown;
            found = true;
            break;
        }
    }

    if (!found)
        mListActiveTimer.push_back(timerItem);

    mListActiveTimer.sort(compareCountdown);

    return (E_OK);
}

am_Error_e CAmSocketHandler::stopTimer(const sh_timerHandle_t handle)
{
    //go through the list and remove the timer with the handle
    std::list<sh_timer_s>::iterator it(mListActiveTimer.begin());
    for (; it != mListActiveTimer.end(); ++it)
    {
        if (it->handle == handle)
        {
            it = mListActiveTimer.erase(it);
            return (E_OK);
        }
    }
    return (E_NON_EXISTENT);
}

am_Error_e CAmSocketHandler::updateEventFlags(const sh_pollHandle_t handle, const short events)
{
    mListPoll_t::iterator iterator = mListPoll.begin();

    for (; iterator != mListPoll.end(); ++iterator)
    {
        if (iterator->handle == handle)
        {
            iterator->pollfdValue.events = events;
            mRecreatePollfds = true;
            return (E_OK);
        }
    }
    return (E_UNKNOWN);
}

bool CAmSocketHandler::fdIsValid(const int fd) const
{
    return (fcntl(fd, F_GETFL) != -1 || errno != EBADF);
}

void CAmSocketHandler::timerUp()
{
    //find out the timedifference to starttime
    timespec currentTime, diffTime;
    clock_gettime(CLOCK_MONOTONIC, &currentTime);
    diffTime = timespecSub(currentTime, mStartTime);

    //now we need to substract the diffTime from all timers and see if they are up
    std::list<sh_timer_s>::reverse_iterator overflowIter = std::find_if(mListActiveTimer.rbegin(), mListActiveTimer.rend(), CAmShCountdownUp(diffTime));

    //copy all fired timers into a list
    std::vector<sh_timer_s> tempList(overflowIter, mListActiveTimer.rend());

    //erase all fired timers
    std::list<sh_timer_s>::iterator it(overflowIter.base());
    mListActiveTimer.erase(mListActiveTimer.begin(), it);

    //call the callbacks for the timers
    std::for_each(tempList.begin(), tempList.end(), CAmShCallTimer());
}

void CAmSocketHandler::timerCorrection()
{
    //get the current time and calculate the correction value
    timespec currentTime, correctionTime;
    clock_gettime(CLOCK_MONOTONIC, &currentTime);
    correctionTime = timespecSub(currentTime, mStartTime);
    mStartTime = currentTime;

    if (!mListActiveTimer.empty())
    {

        //subtract the correction value from all items in the list
        std::for_each(mListActiveTimer.begin(), mListActiveTimer.end(), CAmShSubstractTime(correctionTime));

        //find the last occurrence of zero -> timer overflowed
        std::list<sh_timer_s>::reverse_iterator overflowIter = std::find_if(mListActiveTimer.rbegin(), mListActiveTimer.rend(), CAmShCountdownZero());

        //only if a timer overflowed
        if (overflowIter != mListActiveTimer.rend())
        {
            //copy all timers that need to be called to a new list
            std::vector<sh_timer_s> tempList(overflowIter, mListActiveTimer.rend());

            //erase all fired timers
            std::list<sh_timer_s>::iterator it(overflowIter.base());
            mListActiveTimer.erase(mListActiveTimer.begin(), it);

            //call the callbacks for the timers
            std::for_each(tempList.begin(), tempList.end(), CAmShCallTimer());
        }
    }
}

void CAmSocketHandler::exit_mainloop()
{
    //end the while loop
    stop_listening();

    //fire the ending filedescriptor
    int p(1);
    ssize_t result = write(mPipe[1], &p, sizeof(p));
}

inline timespec* CAmSocketHandler::insertTime(timespec& buffertime)
{
    if (!mListActiveTimer.empty())
    {
        buffertime = mListActiveTimer.front().countdown;
        return (&buffertime);
    }
    else
    {
        return (NULL);
    }
}

void CAmSocketHandler::CAmShCallFire::operator()(sh_poll_s& row)
{   
    try
    {
        row.firedCB->Call(row.pollfdValue, row.handle, row.userData);
    }
    catch (std::exception& e)
    {
        logError("Sockethandler: Exception in FireCallback,caught",e.what());
    }
}

void CAmSocketHandler::CAmShCallPrep::operator()(sh_poll_s& row)
{
    if (row.prepareCB)
    {
        try
        {
            row.prepareCB->Call(row.handle, row.userData);
        }
        catch (std::exception& e)
        {
            logError("Sockethandler: Exception in Preparecallback,caught",e.what());
        }
    }
}

void CAmSocketHandler::CAmShCallTimer::operator()(sh_timer_s& row)
{
    try
    {
        row.callback->Call(row.handle, row.userData);
    }
    catch (std::exception& e)
    {
        logError("Sockethandler: Exception in Timercallback,caught",e.what());
    }
}

void CAmSocketHandler::CAmShCopyPollfd::operator()(const sh_poll_s& row)
{
    pollfd temp = row.pollfdValue;
    temp.revents = 0;
    mArray.push_back(temp);
}

bool CAmSocketHandler::CAmShCountdownUp::operator()(const sh_timer_s& row)
{
    timespec sub = timespecSub(row.countdown, mDiffTime);
    if (sub.tv_nsec == 0 && sub.tv_sec == 0)
        return (true);
    return (false);
}

bool CAmSocketHandler::CAmShCountdownZero::operator()(const sh_timer_s& row)
{
    if (row.countdown.tv_nsec == 0 && row.countdown.tv_sec == 0)
        return (true);
    return (false);
}


}