PollThread.cpp

Go to the documentation of this file.

#include <dcl/Config.h>
 
#if __DCL_WINDOWS
    #include <winsock2.h>
    #include <windows.h>
#else
    #include <errno.h>
    #include <sys/poll.h>
#endif
 
#include <dcl/ArrayT.h>
#include <dcl/PollAble.h>
#include <dcl/PollThread.h>
 
#if __DCL_DEBUG
#undef __THIS_FILE__
static const char_t __THIS_FILE__[] = __T("dcl/PollThread.cpp");
#endif
 
__DCL_BEGIN_NAMESPACE
 
IMPLEMENT_CLASSINFO(PollThread, Thread)
 
PollThread::PollThread(const char_t *_name)
    : Thread(_name)
{
    __terminate = false;
}
 
bool PollThread::add(PollAble* _pPollAble, short _events)
{
    if (__terminate)
        return false;
 
    Thread::SingleLockMutex lock(__updateLock);
 
    _pPollAble->__events = _events;
    __updateList.add(Update(_pPollAble, true));
#if !__DCL_WINDOWS
    // 스레드가 시작되고, 호출 스레드와 다르면
    if (Thread::started() && Thread::getCurrentThread() != this)
        __pollInterrupt.set();
#endif
    return true;
}
 
bool PollThread::remove(PollAble* _pPollAble)
{
    if (__terminate)
        return false;
 
    Thread::SingleLockMutex lock(__updateLock);
 
    __updateList.add(Update(_pPollAble, false));
#if !__DCL_WINDOWS
    // 스레드가 시작되고, 호출 스레드와 다르면
    if (Thread::started() && Thread::getCurrentThread() != this)
        __pollInterrupt.set();
#endif
    return true;
}
 
void PollThread::terminate()
{
    __DCL_ASSERT(Thread::started());
    __terminate = true;
#if !__DCL_WINDOWS
    __pollInterrupt.set();
#endif
}
 
struct POLLFD : public pollfd
{
    POLLFD()
    {
        fd = 0;
        events = revents = 0;
    }
 
    POLLFD(File::HandleType fd, short events)
    {
#if __DCL_WINDOWS
        this->fd = (SOCKET) fd;
#else
        this->fd = fd;
#endif
        this->events = events;
        this->revents = 0;
    }
};
 
inline bool operator == (const POLLFD& _x, const POLLFD& _y)
{
    return _x.fd == _y.fd;
}
 
int PollThread::run()
{
    Array<PollAble*> pollAbles;
    Array<POLLFD> fds;
 
#if __DCL_WINDOWS
    // Windows에서 __pollInterrupt를 사용하지 않는다.
    pollAbles.add(NULL);
    fds.add(POLLFD((File::HandleType) INVALID_SOCKET, POLLIN));
#else
    pollAbles.add(NULL);
    fds.add(POLLFD(__pollInterrupt.handle(), POLLIN));
#endif
 
    while (!__terminate) {
        {
            Thread::SingleLockMutex lock(__updateLock);
            if (!__updateList.isEmpty()) {
#if !__DCL_WINDOWS
                __pollInterrupt.reset();
#endif
                for (UpdateList::Iterator itUpdate = __updateList.begin(); 
                        itUpdate != __updateList.end(); itUpdate++) {
                    Update& update = *itUpdate;
    //              __DCL_TRACE2(__T("add [%ls][%d]\n"), update.pPollAble->path(), update.bAdd);
                    if (update.bAdd) {
                        __DCL_TRACE1(__T("added [%ls]\n"), update.pPollAble->toString().data());
                        pollAbles.add(update.pPollAble);
                        fds.add(POLLFD(update.pPollAble->handle(), update.pPollAble->__events));
                    }
                    else {
                        Array<PollAble*>::Iterator itPollAble =
                            pollAbles.find(update.pPollAble);
                        if (itPollAble != pollAbles.end())   {
                            __DCL_TRACE1(__T("remove [%ls]\n"), update.pPollAble->toString().data());
                            onRemoved(*itPollAble);
                            pollAbles.erase(itPollAble);
                            fds.erase(pollAbles.index(itPollAble));
                        }
                    }
                }   // end of for
                __updateList.clear();
            }
        }
 
        // Wait poll
#if __DCL_WINDOWS
        int r = WSAPoll(fds.data() + 1, (ULONG)(fds.size() - 1), WSA_INFINITE);
#else
        int r = poll(fds.data(), fds.size(), INFINITE);
#endif
        if (r > 0) {
#if !__DCL_WINDOWS
            if (fds[0].revents)
                __pollInterrupt.reset();
            else
#endif
            {
                for (size_t i = 1; i < fds.size(); i++) {
                    if (fds[i].revents)
                    {
                        PollAble* pPollAble = pollAbles[i];
                        try {
                            if (!(pPollAble->onEvent(fds[i].revents, this)))
                                remove(pPollAble);
                        }
                        catch (IOException* cause) {
                            remove(pPollAble);
                            /*
                             *  복구 불가능한 치명적 포트오류가 발생했다.
                             *  USB to COM 변환기를 사용한 경우 포트가 제거된 경우일 수 있다.
                             */
                            __DCL_TRACE1(__T("onEvent Fail! %ls\n"), cause->toString().data());
                            cause->destroy();
                        }
                    }   // end of fds[i].revents
                } // end of for
            }
        }
        else if (r < 0) {
#if __DCL_DEBUG
    #if __DCL_WINDOWS
            // SOCKET_ERROR
            __DCL_TRACE1(__T("WSAPoll error [%d]\n"), WSAGetLastError());
    #else
            // fd error
            __DCL_TRACE1(__T("poll error [%d]\n"), errno);
    #endif
            Thread::sleep(1000);
#endif  
        }
        else {
            // r == 0, time out
            __DCL_ASSERT(false);
        }
    }   // end of while

    for (Array<PollAble*>::Iterator itPollAble = pollAbles.begin();
            itPollAble != pollAbles.end(); itPollAble++) {
        // NULL 인 것은 __pollInterrupt를 위한 것이다.
        if (*itPollAble != NULL) {
            __DCL_TRACE1(__T("terminate remove [%ls]\n"), (*itPollAble)->toString().data());
            onRemoved(*itPollAble);
        }
    }
 
    // __updateList에 추가된 것이 있을 수 있다.
    for (UpdateList::Iterator itUpdate = __updateList.begin(); 
            itUpdate != __updateList.end(); itUpdate++) {
        Update& update = *itUpdate;
        if (update.bAdd) {
            __DCL_TRACE1(__T("terminate remove [%ls]\n"), update.pPollAble->toString().data());
            onRemoved(update.pPollAble);
        }
    }
 
    return 0;
}
 
void PollThread::onRemoved(PollAble* _pPollAble)
{
    __DCL_TRACE1(__T("PollThread::onRemoved, destroy [%ls]\n"), _pPollAble->toString().data());
    _pPollAble->destroy();
}
 
__DCL_END_NAMESPACE