SocketPollThread.cpp

Go to the documentation of this file.

#include <dcl/Config.h>
 
#if __DCL_WINDOWS
#include <winsock2.h>
 
#include <dcl/ArrayT.h>
#include <dcl/Socket.h>
#include <dcl/SocketPollThread.h>
 
#if __DCL_DEBUG
    #undef __THIS_FILE__
    static const char_t __THIS_FILE__[] = __T("dcl/SocketPollThread.cpp");
#endif
 
__DCL_BEGIN_NAMESPACE
 
IMPLEMENT_CLASSINFO(SocketPollThread, PollThread)
 
SocketPollThread::SocketPollThread(const char_t* _name)
    : PollThread(_name)
{
    __pollInterrupt = WSACreateEvent();
}
 
SocketPollThread::~SocketPollThread()
{
    WSACloseEvent(__pollInterrupt);
}
 
bool SocketPollThread::add(PollAble* _pPollAble, short _events)
        __DCL_THROWS1(IOException*)
{
#if __DCL_NO_RTTI
    __DCL_ASSERT(_pPollAble->isKindOf(CLASSINFO(Socket)));
#else
    __DCL_ASSERT(dynamic_cast<Socket*>(_pPollAble) != NULL);
#endif
 
    if (__terminate)
        return false;
 
    Thread::SingleLockMutex lock(__updateLock);
 
    if (((Socket*)_pPollAble)->__waitEvent == WSA_INVALID_EVENT)
    {
        ((Socket*)_pPollAble)->__waitEvent = WSACreateEvent();
        if (((Socket*)_pPollAble)->__waitEvent == WSA_INVALID_EVENT)
            throw new IOException(_pPollAble->toString(), WSAGetLastError());
    }
 
    if (WSAEventSelect((SOCKET) (_pPollAble->handle()),
        ((Socket*)_pPollAble)->__waitEvent, (long) _events))
        throw new IOException(_pPollAble->toString(), WSAGetLastError());
 
    __updateList.add(Update(_pPollAble, true));
 
    // 스레드가 시작되고, 호출 스레드와 다르면
    if (Thread::started() && Thread::getCurrentThread() != this)
        WSASetEvent(__pollInterrupt);
 
    return true;
}
 
bool SocketPollThread::remove(PollAble* _pPollAble)
{
    if (__terminate)
        return false;
 
    Thread::SingleLockMutex lock(__updateLock);
 
    __updateList.add(Update(_pPollAble, false));
 
    // 스레드가 시작되고, 호출 스레드와 다르면
    if (Thread::started() && Thread::getCurrentThread() != this)
        WSASetEvent(__pollInterrupt);
 
    return true;
}
 
void SocketPollThread::terminate()
{
    __DCL_ASSERT(Thread::started());
    __terminate = true;
    WSASetEvent(__pollInterrupt);
}
 
int SocketPollThread::run()
{
    Array<Socket*> pollAbles;
    Array<WSAEVENT> fds;
 
    pollAbles.add(NULL);
    fds.add(__pollInterrupt);
 
    while (!__terminate) {
        {
            Thread::SingleLockMutex lock(__updateLock);
            if (!__updateList.isEmpty()) {
                WSAResetEvent(__pollInterrupt);
                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((Socket*) update.pPollAble);
                        fds.add(((Socket*) update.pPollAble)->__waitEvent);
                    }
                    else {
                        Array<Socket*>::Iterator itPollAble =
                            pollAbles.find((Socket*) 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
    //  __DCL_TRACE2(__T("%d, %p\n"), fds.size(), fds.data());
        DWORD r = WSAWaitForMultipleEvents(
            (DWORD) fds.size(),     // __in     DWORD nCount
            fds.data(),     // __in     const WSAEVENT* lpHandle
            FALSE,                     // __in     BOOL bWaitAll
            WSA_INFINITE,           // __in     DWORD dwTimeout
            FALSE                      // __in     BOOL fAlertable
            );
 
        switch (r) {
            case WSA_WAIT_FAILED:
#if __DCL_DEBUG
                __DCL_TRACE1(__T("WaitFail! %u\n"), WSAGetLastError());
                Thread::sleep(1000);
#endif
                break;
            case WSA_WAIT_TIMEOUT:
                // unreachable code. INFINITE
                break;
            case WSA_WAIT_EVENT_0:
                WSASetEvent(__pollInterrupt);
                break;
            default: {
                size_t index = r - WSA_WAIT_EVENT_0;
                __DCL_ASSERT(1 <= index && index < fds.size());
                Socket* pSocket = pollAbles[index];
 
                WSANETWORKEVENTS events;
                __DCL_VERIFY(
                    ::WSAEnumNetworkEvents(
                        (SOCKET)(pSocket->handle()),  pSocket->__waitEvent, &events)
                    == 0);
 
                try {
                    if (! pSocket->onEvent((short) events.lNetworkEvents, this))
                        remove(pSocket);
                }
                catch (IOException* cause) {
                    remove(pSocket);
                    /*
                     *  복구 불가능한 치명적 포트오류가 발생했다.
                     *  USB to COM 변환기를 사용한 경우 포트가 제거된 경우일 수 있다.
                     */
                    __DCL_TRACE1(__T("onEvent or waitCommEvent Fail! %ls\n"),
                        cause->toString().data());
                    cause->destroy();
                }
            }
        }   // end of switch
    }   // end of while(!__terminate)

    for (Array<Socket*>::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;
}
 
__DCL_END_NAMESPACE
 
#endif      // __DCL_WINDOWS