warsvrwin32ntengine.cpp

ftpserver very good sample

#include "StdAfx.h"
#include "WarSvrWin32NtEngine.h"   // class implemented


#ifndef WAR_SOCKET_IO_WIN32_NT_H
#   include "WarSocketIoWin32Nt.h"
#endif

#ifndef WAR_SVR_WIN32_NT_ENGINE_TASK_H
#   include "WarSvrWin32NtEngineTask.h"
#endif

#ifndef WAR_SHUTDOWN_ENGINE_H
#   include "WarShutdownEngine.h"
#endif

#ifndef WAR_AUTO_LOCK_H
#   include "WarAutoLock.h"
#endif

#define AUTO_LOCK WarAutoLock my_lock(mLock);

/////////////////////////////// PUBLIC ///////////////////////////////////////

WarSvrWin32NtEngine *WarSvrWin32NtEngine::mspThis;

//============================= LIFECYCLE ====================================

WarSvrWin32NtEngine::WarSvrWin32NtEngine()
: mhIocp(NULL)
{
    if (mspThis != NULL)
        WarThrow(WarError(WAR_ERR_ALREADY_INITIALIZED), NULL);

    mspThis = this;
}// WarSvrWin32NtEngine


WarSvrWin32NtEngine::~WarSvrWin32NtEngine()
{
    if (mspThis == this)
        mspThis = NULL;

    if (mhIocp != NULL)
        ::CloseHandle(mhIocp);

}// ~WarSvrWin32NtEngine


//============================= OPERATORS ====================================

//============================= OPERATIONS ===================================

void WarSvrWin32NtEngine::OnShutdown(war_cptr_t ptr)
{
    ((WarSvrWin32NtEngine *)ptr)->TerminateAllThreads(true);
}

void WarSvrWin32NtEngine::Create(war_uint32_t MinThreads, 
        war_uint32_t MaxThreads) throw(WarException)
{
    if ((mhIocp = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE,
        NULL, NULL, 0)) == NULL)
        WarThrowSyserr("CreateIoCompletionPort()");

    WarThreadPool::Create(MinThreads, MaxThreads);

    for(war_uint32_t i = 0; i < MinThreads; i++)
        PushTask(new WarSvrWin32NtEngineTask(*this));

    WarShutdownEngine::GetEngine().AddEvent(OnShutdown, this);
}

void WarSvrWin32NtEngine::QueueRequest(WarSvrWin32EventThread::task_ptr_t& taskPtr)
{
    AUTO_LOCK

    for(thread_list_t::iterator P = mIoThreads.begin()
        ; P != mIoThreads.end()
        ; P++)
    {
        if ((*P)->HasCapacity())
        {
            (*P)->QueueRequest(taskPtr);
            return;
        }
    }

    // Add new thread
    iothread_ptr_t new_thread = new WarSvrWin32EventThread;
    new_thread->Create();
    mIoThreads.push_back(new_thread);
    new_thread->QueueRequest(taskPtr);
}

void WarSvrWin32NtEngine::RemoveRequest(const war_socket_t sck)
{
    AUTO_LOCK

    // We don't know which thread that have the socket, so 
    // we have to notyfy all.
    for(thread_list_t::iterator P = mIoThreads.begin()
        ; P != mIoThreads.end()
        ; P++)
    {
        if ((*P)->HasCapacity())
        {
            (*P)->RemoveRequest(sck);
            return;
        }
    }
}

void WarSvrWin32NtEngine::RemoveIdleEventThread(WarSvrWin32EventThread *pThread)
{
    AUTO_LOCK

    if (pThread->GetNumQueued(false) == 0)
    {
        for(thread_list_t::iterator P = mIoThreads.begin()
            ; P != mIoThreads.end()
            ; P++)
        {
            if (&(**P) == pThread)
            {
                mIoThreads.erase(P);
                pThread->Shutdown();
                break;
            }
        }
    }
}

//============================= ACCESS     ===================================
//============================= INQUIRY    ===================================
/////////////////////////////// PROTECTED  ///////////////////////////////////

/////////////////////////////// PRIVATE    ///////////////////////////////////