/*
 KX High Capacity High Performance Epoll Socket Daemon Service Class
 ---------------------------------------------------------------------------
 $Id: $ $Revision: $
 $HeadURL: $
 Copyright(c) 1990 - 2018 KRKIM(DURUMUL) Allrights reserved.
 */

/**
 * @file
 * @brief 
 * @detail 
 */

/*
 * History:
 * ===========================================================================
 * 2004.08.13  KRKIM   Initialized.
 *                     Relay Server Manager (epoll edge-triggered / worker thread)
 *
 */

/*
refer site:
http://chonga.pe.kr/blog/index.php?pl=972
http://www.joinc.co.kr/modules/moniwiki/wiki.php/Site/Network_Programing/AdvancedComm/epoll24
http://www.joinc.co.kr/modules/moniwiki/wiki.php/epoll
http://durumul.tistory.com
*/

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <errno.h>
#include <signal.h>

#include <list>
#include <queue>
#include <vector>
#include <pthread.h>

#include "network.h"
#include "epollcomn.h"
#ifdef USE_INTERDBPOOL
#include "dbpool.h"
#endif

#include "kxHashMap.h"
#include "kxThread.h"
#include "iocq.h"
#include "tokenizer.h"

#include "kxSockSession.h"
#include "kxPollService.h"
#include "kxServerService.h"
#include "kxtracelog.h"

#define EPOLL_NEW

using namespace std;

//void *CKxServerService::sockworkerthread_function(void *arg)
//{
//    // do not use THREAD CANCELLATION
//    //CKxServerService *pServer = (CKxServerService *)arg;
//    CKxThreadCtx *lThread = (CKxThreadCtx *)arg;
//    //lThread->SocketJob(lThread);
//    //SocketJob(lThread);
//    //pServer->workerthreadloop(pthread_self());
//    //2018-09-28
//    CKxService *pService = (CKxService *)lThread->GetService();
//    pService->OnThreadLoop(pthread_self());
//    pthread_exit(NULL);
//}

CKxServerService::CKxServerService() : CKxService()
{
    m_nBacklog = 5;
    m_pPollService = NULL;
    m_queuemode = QUEUE_MODE_WORKQUEUE | QUEUE_MODE_USE_CLIENT;
    m_listensockfd = INVALID_SOCKET;
}

CKxServerService::~CKxServerService()
{
    Destroy();
}

void CKxServerService::SetListenBacklog(short backlog)
{
    m_nBacklog = backlog;
}

void CKxServerService::SetListenPort(short port)
{
    m_nListenPort = port;
}

void CKxServerService::SetPollService(CKxPollService *pPollService)
{
    m_pPollService = pPollService;
}

void CKxServerService::SetMaxClient(int maxclient)
{
    m_nMaxClient = maxclient;
    int system_limit_fdcount = get_limit_fd();
    kxTraceLog(0, "SetMaxClient: Request MaxClient(%d),system limit max fd(%d)\n", maxclient, system_limit_fdcount);
    if (m_nMaxClient > system_limit_fdcount){
        m_nMaxClient = system_limit_fdcount;
        kxTraceLog(0, "SetMaxClient: over limit max fd(%d)\n", system_limit_fdcount);
    }
}

bool CKxServerService::Init(int listenport, int maxclient, int maxthreadpool, int epolltimeout)
{
    SetListenPort(listenport);
    SetMaxClient(maxclient);
    SetMaxThreadPool(maxthreadpool);
    m_epolltimeout = epolltimeout;
    kxTraceLog(0, "Init:listenport=%d maxclient=%d m_epolltimeout=%d maxthreadpool=%d\n",
        listenport, maxclient, m_epolltimeout, maxthreadpool);
#if 0
    //m_pThreadPool = new CKxThreadPool(m_nMaxThreadPool);
    //if (NULL == m_pThreadPool) {
    //    return false;
    //}
    //m_pThreadPool->setthreadfuncptr(sockworkerthread_function);
    //m_pThreadPool->setthreadparam(this);
#else
    if (m_nMaxThreadPool > 0){
        InitThreadPool(maxthreadpool);
    }
#endif
    return true;
}

bool CKxServerService::StartService()
{
    InitSockServer();
    return CKxService::StartService();
}

//bool CKxServerService::StartNetwork()
//{
//    int rc;
//    //SetServiceRun(true);
//    // run thread pool
//    //if (!m_pThreadPool->Start((void*)this))
//    //{
//    //    return false;
//    //}
//
//    //if (!InitSockServer())
//    //{
//    //    return false;
//    //}
//
//    //// create server loop thread
//    //rc = pthread_create(&m_serverthread, NULL, epollthread_function, this);
//    //if (rc != 0)
//    //{
//    //    kxTraceLog(0, "error:create server epoll thread\n");
//    //    return false;
//    //}
//
//    return rc;
//}

bool CKxServerService::InitSockServer()
{
    int rc;
    int epollsize = m_nMaxClient, poolsize = m_nMaxClient;

    // ignore sigpipe signal for socket
    //signal (SIGPIPE, SIG_IGN);

    m_eventfd = init_epoll(epollsize);
    if (m_eventfd < 0)
    {
        kxTraceLog(0, "initsockserver:init_epoll error\n");
        return false;
    }

    // init pool
    m_pEpollEvents = new struct epoll_event[sizeof (*m_pEpollEvents) * poolsize];
    if (m_pEpollEvents == NULL)
    {
        kxTraceLog(0, "initsockserver:memory alloc failed.\n");
        return false;
    }

    //backlog 인데 maxclient를 설정할 필요가 있나,backlog와 max connection socket 갯수와는
    //무관하다.backlog는 연결대기 통로.
    m_listensockfd = init_acceptsock(m_nListenPort, m_nBacklog);
    if (m_listensockfd < 0)
    {
        kxTraceLog(0, "initsockserver:init_acceptsock error\n");
        return false;
    }
    /*no one shot */
#ifndef EPOLL_NEW
    rc = add_epollfd(m_eventfd, m_listensockfd,EPEVT_READ|EPEVT_HANGUP|EPEVT_ERROR|EPEVT_ET);
#else
    kxTraceLog(0, "initsockserver:server this=%x\n", this);
    rc = Epoll_Add(m_eventfd, m_listensockfd, this, EPEVT_READ | EPEVT_HANGUP | EPEVT_ERROR | EPEVT_ET);
#endif
    if (rc < 0)
    {
        kxTraceLog(0, "initsockserver:Epoll_Add error\n");
        return false;
    }
    return true;
}

bool CKxServerService::DestroySockServer()
{
    // close accept , close epoll
#ifdef EPOLL_NEW
    Epoll_Del(m_eventfd, m_listensockfd);
#endif
    close(m_listensockfd);
    close(m_eventfd);//모든 작업이 끝났다면 close()를 호출해서 닫아주어야 한다. 
    if (m_pEpollEvents)
        delete[]m_pEpollEvents;
   
    m_pEpollEvents = NULL;

    CKxService::DestroyService();
    return true;
}

//메인쓰레드 시작직후 호출
//리턴: 0 성공 0이 아니면 종료
int CKxServerService::OnServiceBegin()
{
    
    return CKxService::OnServiceBegin();
}

//메인쓰레드 종료전 호출
int CKxServerService::OnServiceEnd()
{
    
    return CKxService::OnServiceEnd();
}

//메인쓰레드 시작직후 호출
//리턴: 0 성공 0이 아니면 종료
int CKxServerService::OnThreadBegin(int nID)
{
    return CKxService::OnThreadBegin(nID);
}

//메인쓰레드 종료전 호출
int CKxServerService::OnThreadEnd(int nID)
{
    
    return CKxService::OnThreadEnd(nID);
}

int CKxServerService::OnServiceLoop(int threadid)
{
    int	i;
    int	clientfd;
    int rc,start;
    int	poolsize = m_nMaxClient;
    int	max_got_events;
    CKxQueue *pQueue = NULL;

    start = OnServiceBegin();
    while (start == 0 && m_bServiceRunning)
    {
        max_got_events = epoll_wait(m_eventfd, m_pEpollEvents, poolsize, m_epolltimeout/*-1*/);
        for (i = 0; i < max_got_events; i++)
        {
#ifndef EPOLL_NEW
            if(m_pEpollEvents[i].data.fd == m_listensockfd)
                //tracelog(0,"currentsock %x: this=%x\n",currentsock,this);
#else
            if (m_pEpollEvents[i].data.ptr == this)
#endif
            {
                int n = 0;
                //ET 일때는 모든 가능한 ACCEPT 가 여러개가 한번에 이벤트 발생하므로
                //EAGAIN을 만날때까지 loop 로 모두 처리해줘야 한다.
                while (1)
                {
                    n++;
                    kxTraceLog(0, "call do_accept count %d:\n", n);
                    struct sockaddr_in clientaddr;
                    clientfd = do_accept(m_listensockfd, &clientaddr);
                    if (clientfd < 0){
                        if (errno == EAGAIN || clientfd == -2){//next retry..no more client try to connect yet.
                            break;
                        }
                    }
#ifndef EPOLL_NEW
                    rc = add_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_WRITE|EPEVT_ET|EPEVT_ONESHOT);
                    if(rc < 0)
                    {
                        kxTraceLog(0,"ERROR:add client to epool,so close client sock now.\n");
                        close(clientfd);
                        break;
                    }
#endif

                    //CKxSockSession *currentuser = new CKxSockSession();
#ifdef USE_LOOSE_SOCKDATA_ACCEPT_ASAP
                    CKxSockSession *currentuser = AddClientSock(clientfd, &clientaddr);
#else
                    CKxSockSession *currentuser = SetClientSock(clientfd);
#endif
                    if (currentuser != NULL)
                    {
#ifdef USE_LOOSE_SOCKDATA_ACCEPT_ASAP
#else
                        AddSocketPool(currentuser,&clientaddr);
#endif
#ifdef EPOLL_NEW
                        rc = Epoll_Add(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
                        if (rc < 0)
                        {
                            kxTraceLog(0, "ERROR:add client to epool,so close client sock now.\n");
                            currentuser->OnDisconnected();
                            close(clientfd);
                            break;
                        }
#endif

                        char *remoteaddr = inet_ntoa(clientaddr.sin_addr);
                        //가장 적은 소켓을 처리하는 쓰레드를 찾는다.					
                        int thread_index = 0;
                        thread_index = m_pThreadPool->FindMinClients();//m_ClientSockList.GetCount() % m_nMaxThreadPool;
                        kxTraceLog(0, "new client sock(%d) ip=%s thread_index=%d\n",
                            clientfd, remoteaddr, thread_index);

                        currentuser->m_threadno = thread_index;

                        CKxThreadCtx * lThread = m_pThreadPool->GetThread(thread_index);
                        lThread->Lock();
                        lThread->m_clientnum++;
                        lThread->UnLock();
                        kxTraceLog(0, "Accept OK:client sock(%d) currentuser=%x ip=%s dispatch to Thread[%d] clients=%d\n",
                            currentuser->GetSockFD(), currentuser, remoteaddr, thread_index, lThread->m_clientnum);
                        currentuser->OnConnected();
                    }
                    else{
                        kxTraceLog(0, "ERROR:add client.\n");
                    }
                }
            }
            else
            {
                //소유쓰레드의 이벤트큐에 입력한다.
                //int clientfd = ev.data.fd;
                CKxSockSession *currentuser = (CKxSockSession *)m_pEpollEvents[i].data.ptr;//m_ClientSockList.Find(clientfd);
                CKxThreadCtx *lThread = m_pThreadPool->GetThread(currentuser->m_threadno);
                kxTraceLog(0, "Client Epoll:client sock(%d) currentuser=%x dispatch to Thread[%d] clients=%d\n",
                    currentuser->GetSockFD(), currentuser, lThread->m_ord_id - 1, lThread->m_clientnum);

                KX_QUEUE_ITEM qitem;
                qitem.mask = KXQUEUE_MASK_USE_EVENTS;
                qitem.events.epollevent = m_pEpollEvents[i];
                if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
                    pQueue = (CKxQueue *)lThread;
                }
                else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
                    pQueue = this->GetQueue();
                }
                pQueue->PushSignal(qitem);
            }
        }
        //TODO::Send pending data
    }
    kxTraceLog(0, "ServiceRunning Stopped.\n");
    rc = OnServiceEnd();
    return true;
}

int CKxServerService::OnThreadLoop(int threadid)
{
    int rc;
    int index,start;
    int count, count1;

    CKxThreadCtx * lThread;
    CKxService *pService;
    KX_QUEUE_ITEM qitem; //struct epoll_event ev;
    CKxQueue *pQueue = NULL;

    if (m_pThreadPool)
        lThread = m_pThreadPool->GetThreadByID(threadid);

    start = OnThreadBegin(lThread->m_ord_id);

    pService = (CKxService *)lThread->m_pService;//same as this
    kxTraceLog(TL_LOW, "[SockThread %d] pthread_self() = %x thr_id = %x (%x) started\n",
        lThread->m_ord_id, pthread_self(), lThread->m_thr_id,threadid);

    //lThread = (CKxThreadCtx *)arg;
    //lThread->WatchJob();
    /*pthread_cond_wait()는 조건변수 cond로 시그널이 전달되기를 기다린다.
      이 함수는 호출되면 자동적으로 mutex잠금을 되돌려준다(UnLock m_mut internally)
      내부적으로 pthread_unlock_mutex()를 호출한다.
      이 함수를 호출한 쓰레드는 조건변수에 시그널이 전달될 때까지 기다리며,
      이때 CPU는 소비하지 않는다.
      시그널을 전달받아서 쓰레드가 깨어나면 자동적으로 mutex잠금을 얻는다(
      pthread_lock_mutex()를 사용)
      */
    //TODO::Check This!!
    //조건변수 m_cond가 모든쓰레드와 공용으로 사용하게 되면 
    //쓰레드 하나가 모든 이벤트를 모두 다 처리하는게 아닌가?
    //그렇다면 쓰레드풀의 나머지 쓰레드는 놀고있고 오직 한 쓰레드만 깨어나 작업하는거 아닌가??
    //따라서,쓰레드별로 조건변수를 두고 뮤텍스만 공용처리하여,이벤트큐에서 꺼내갈때,
    //충돌나지 않도록만 하게 한다?
    //그러면 먼저 시작된 쓰레드가 수신데이타 처리시간이 더 길면,나중에 깨어난 쓰레드의
    //처리시간이 더 짧으면 클라이언트로 응답하는 데이타가 순서가 바뀔 가능성이 있는것 같다.
    //이렇게 되면 쓰레드간 동기화를 별도로 처리해줘야 할 것같다.
    //즉,소켓이 현재 처리하는 것과 같은 경우만 처리토록 하고 이벤트큐를 끝까지 단독처리 하지 않게?
    //http://www.joinc.co.kr/modules/moniwiki/wiki.php/Site/Thread 참고해서 재분석 및 고려할것

    while (start == 0 && m_bServiceRunning)
    {
        //Pop Queue Data
        //Subscribe mode or workqueue mode
        //fanout 방식(각 쓰레드별 자체 queue 처리-같은데이터)
        //  => 모든 쓰레드가 각각 자신의 큐에 들어온 같은 데이터를 꺼내와서 동일한 데이터를 분배처리
        //work queue 방식(각 쓰레드별 공용 queue사용-각기다른 데이터처리)
        //  => 모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
        
        //Subscribe mode or workqueue mode
        //workqueue:모든 쓰레드중 1개가 각각 서로다른 소켓데이터의 큐를 꺼내와서 처리
        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            pQueue = (CKxQueue *)lThread;
        }
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            pQueue = pService->GetQueue();
        }
        //kxTraceLog(TL_LOW, "[SockThread] WaitCondSignal() threadid=%x pQueue=%x(mode=%x)\n",threadid,pQueue,m_queuemode);
        pQueue->WaitCondSignal();
        while (1){
            if (lThread->m_idlecnt % 100 == 0)
                kxTraceLog(TL_LOW, "[Thread %d] idle(%ld) threadid=%x pQueue=%x(mode=%x)\n",
                    lThread->m_ord_id,lThread->m_idlecnt,threadid,pQueue,m_queuemode);

            pQueue->Lock();
            count = pQueue->GetQueueSize();
            if (pQueue->IsEmpty() || count < 1){
                //kxTraceLog(TL_LOW,"[CKxThreadCtx %d] queue count=%d suspended to wait signal.\n",m_ord_id,count);
                pQueue->UnLock();
                lThread->m_idlecnt++;
                if (lThread->m_idlecnt >= 0X7FFF)
                    lThread->m_idlecnt = 0;
                /*usleep(1);*/
                break;
            }
			
            //kxTraceLog(TL_LOW, "[SockThread] PopData()\n");
			qitem = pQueue->PopData(0);
			count1 = pQueue->GetQueueSize();
			pQueue->UnLock();
            //kxTraceLog(TL_LOW, "[Thread %d] thr_id = %x popup ev,pService=%x\n", lThread->m_ord_id, lThread->m_thr_id, pService);
            pService->OnThreadProc(lThread->m_ord_id,(void*)&qitem);
            pQueue->Free(&qitem);//2018.09.30
        }
    }
    kxTraceLog(TL_LOW, "[Thread %d] pthread_self() = %x thr_id = %x stop\n", lThread->m_ord_id, pthread_self(), lThread->m_thr_id);
    rc = OnThreadEnd(lThread->m_ord_id);
    return 0;
}

bool CKxServerService::StopService()
{
    SetServiceRun(false);
    // closing sock server
    DestroySockServer();
    return CKxService::StopService();
}

//느림
CKxSockSession * CKxServerService::AddClientSock(int sockfd, sockaddr_in *pClientAddr)
{
    bool rc = false;
    CKxSockSession *currentuser = new CKxSockSession();
    if (currentuser != NULL)
    {
        currentuser->SetSockFD(sockfd, pClientAddr);
        currentuser->InitContext(MAX_IOBUFFSIZE);
        m_ClientSockList.Add(currentuser);
        return currentuser;
    }
    return NULL;
}

//가벼운 소켓등록
CKxSockSession * CKxServerService::SetClientSock(int sockfd)
{
    bool rc = false;
    CKxSockSession *currentuser = new CKxSockSession();
    if (currentuser != NULL)
    {
        currentuser->SetSockFD(sockfd);
        pclient->InitContext(MAX_IOBUFFSIZE);
        m_ClientSockList.Add(pclient);
        return currentuser;
    }
    return NULL;
}

//실제 소켓풀이아님,제대로 하려면 윈도우NT의 IOCP AcceptEx() 처럼 Pre할당하고 대기타야한다.
CKxSockSession * CKxServerService::AddSocketPool(CKxSockSession *pclient, sockaddr_in *pClientAddr)
{
    bool rc = false;
    if (pclient != NULL)
    {
        pclient->SetSockFD(pclient->GetSockFD(), pClientAddr);
        return pclient;
    }
    return NULL;
}

bool CKxServerService::RemoveClientSock(int sockfd)
{
    return m_ClientSockList.Remove(sockfd);
}

bool CKxServerService::RemoveClientSock(CKxSockSession *pUser)
{
    return m_ClientSockList.Remove(pUser);
}


//void substr_rtn(char *data,int pos_start,int pos_length,char *data_return)
//{
//	char bufx[BUFSIZE];
//	int rof,end_pos,mem_rof;
//
//	strcpy(bufx,"");
//
//	end_pos=pos_start+pos_length;
//	if(end_pos>strlen(data))	end_pos=strlen(data);
//
//	mem_rof=0;
//	for(rof=pos_start;rof<end_pos;rof++)
//	{
//		bufx[mem_rof]=data[rof];
//		mem_rof++;
//	}
//	bufx[mem_rof]=0;
//	strcpy(data_return,bufx);
//}

int CKxServerService::OnServicePolling(TIPARAM *polltime)
{
	return CKxService::OnServicePolling(polltime);
}

int CKxServerService::OnThreadQueue(int nID/*orderid*/,void *param/*popped queue data*/)
{
    return CKxService::OnThreadQueue(nID,param);
}

//http://vpj.posterous.com/tcp-echo-server-example-in-c-using-epoll
//int CKxServerService::OnThreadProc(struct epoll_event ev)
int CKxServerService::OnThreadProc(int nID/*orderid*/,void *param/*popped queue data*/)
{
    // read from the cfd
    int rc = 0;
    int rc2;
    KX_QUEUE_ITEM *qitem = (KX_QUEUE_ITEM*) param;
    struct epoll_event ev;
    ev = qitem->events.epollevent;
    int clientfd = ev.data.fd;
    CKxSockSession *currentuser = (CKxSockSession *)ev.data.ptr;//m_ClientSockList.Find(clientfd);
    CKxThreadCtx * lThread;
    CKxService *pService;

    /*lThread = m_pThreadPool->GetThreadByOrderID(nID);*/
    lThread = m_pThreadPool->GetThread(currentuser->m_threadno);
    pService = (CKxService *)lThread->m_pService;//same as this

    clientfd = currentuser->GetSockFD();
    kxTraceLog(0, "[KxServerService] OnThreadProc:client sock(%d) currentuser=%x event=%x\n", clientfd, currentuser,ev.events);

    if (currentuser == NULL)
    {
        kxTraceLog(0, "error,client sock_context is null\n");
        return -1000;
    }

    if (ev.events & EPOLLHUP){
        kxTraceLog(0, "client(%d) EPOLLHUP detected.close socket.Thread[%d] clients= %d\n",
            clientfd, currentuser->m_threadno, lThread->m_clientnum);
        CloseClient(currentuser);
        return -1002;
    }

    if (ev.events & EPOLLERR){
        kxTraceLog(0, "client(%d) EPOLLERR detected.so skip.\n", clientfd);
        return -1003;
    }

    if (ev.events & EPOLLIN){
        kxTraceLog(0, "client(%d) EPOLLIN status detected.available to recv incoming data now.\n", clientfd);
        currentuser->Lock();
        rc = currentuser->OnRecvPacket();
        currentuser->UnLock();
        if (rc == SOCK_TRY_CLOSE){
            CloseClient(currentuser);
            //int trid = 123;
            //int duringsec = 60;
            //char pkstr[1031];
            //system hangup
            //sprintf(pkstr, "HUP %d %d\n", cfd,trid,duringsec);
            //senddatatoall(pkstr);
            return -1004;
        }
        // re add to epoll on ET mode (in oneshot)
#ifndef EPOLL_NEW
        rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_WRITE|EPEVT_ET|EPEVT_ONESHOT);
#else
        rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
#endif
        if (rc2 < 0)
        {
            kxTraceLog(0, "client(%d) ERROR:modify_epollfd to recv.\n", clientfd);
        }
    }
    if (ev.events & EPOLLOUT){
        kxTraceLog(0, "client(%d) EPOLLOUT status detected.available to send now.\n", clientfd);
        currentuser->Lock();
        rc = currentuser->OnSendPacket();
        currentuser->UnLock();
        if (rc == SOCK_TRY_CLOSE){
            CloseClient(currentuser);
            kxTraceLog(0, "client(%d) EPOLLOUT finish DATA,close socket.Thread[%d] clients= %d\n",
                clientfd, currentuser->m_threadno, lThread->m_clientnum);
        }
        else if (rc == SOCK_DATA_PENDING_SHIFT_SENDBUFFER){
            kxTraceLog(0, "client(%d) send buffer is remain,so modify_epollfd to send at nexttime.\n", clientfd);
#ifndef EPOLL_NEW
            rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_WRITE|EPEVT_ET|EPEVT_ONESHOT);
#else			
            rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
#endif			
            if (rc2 < 0)
            {
                kxTraceLog(0, "client(%d) ERROR:modify_epollfd to send.\n", clientfd);
            }
        }
        else{//to read event only
            kxTraceLog(0, "client(%d) send buffer is flush,so modify_epollfd to recv at nexttime.\n", clientfd);
#ifndef EPOLL_NEW
            rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_ET|EPEVT_ONESHOT);
#else
            rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_ET | EPEVT_ONESHOT);
#endif
            if (rc2 < 0)
            {
                kxTraceLog(0, "client(%d) ERROR:modify_epollfd to recv after send.\n", clientfd);
            }
        }
    }
    return rc;
}

bool CKxServerService::senddatatoall(const char *chatmsg, int len)
{
    int sendflags = 0;
    CKxSockSession *currentuser = NULL;
    list < CKxSockSession* >::iterator iter;
    m_ClientSockList.Lock();
    for (iter = m_ClientSockList.GetClientSockListPtr()->begin();
        iter != m_ClientSockList.GetClientSockListPtr()->end(); ++iter)
    {
        currentuser = *iter;
        // sending only / no need to lock
        send((*iter)->GetSockFD(), chatmsg, len, sendflags);
    }
    m_ClientSockList.UnLock();
    return true;
}

int CKxServerService::PostSendToAll(char *pData, int nLen,int bNow)
{
    int sendflags = 0;
    CKxSockSession *currentuser = NULL;
    list < CKxSockSession* >::iterator iter;
    m_ClientSockList.Lock();
    for (iter = m_ClientSockList.GetClientSockListPtr()->begin();
        iter != m_ClientSockList.GetClientSockListPtr()->end(); ++iter)
    {
        currentuser = *iter;
        currentuser->PostSend(pData, nLen);
        if (bNow)
            currentuser->SendPacket();
        //kxTraceLog(0, "PostSendToAll:currentuser=%x\n",currentuser);
    }
    m_ClientSockList.UnLock();
    return 0;
}

int CKxServerService::PostSendToDataChannel(char *pData, int nLen,int bNow)
{
    return 0;
}

//void *CKxServerService::epollthread_function(void *arg)
//{
//    int rc;
//    CKxServerService *pServer = (CKxServerService *)arg;
//    //pthread_detach(pthread_self());
//    // 쓰레드에 대한 취소요청을 활성화한다.
//    rc = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
//    if (rc != 0)
//    {
//        kxTraceLog(0, "Thread pthread_setcancelstate failed\n");
//        exit(EXIT_FAILURE);
//    }
//
//    // 취소요청시 바로 종료한다.
//    // Using here for PTHREAD_CANCEL_ASYNCHRONOUS instread of PTHREAD_CANCEL_DEFERRED
//    // Because epoll_wait is not supporting canncellation point
//    rc = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
//    // res = pthread_setcanceltype (PTHREAD_CANCEL_DEFERRED, NULL);
//    if (rc != 0)
//    {
//        kxTraceLog(0, "Thread pthread_setcanceltype failed\n");
//        exit(EXIT_FAILURE);
//    }
//    pServer->serverthreadloop(pthread_self());
//    pthread_exit(NULL);
//}


/**
 *
 * @brief   해당 클라이언트소켓을 닫는다.
 * @details
 * @param   CKxSockSession pclient 클라이언트 소켓 클래스
 * @date    2018-09-09
 * @version 0.0.1
 * @return  0 -> 성공, -1 -> 실패
 * @todo
 */
int CKxServerService::CloseClient(CKxSockSession *pclient)
{
    int clientfd;
    clientfd = pclient->GetSockFD();
    kxTraceLog(TL_LOW, "CloseClient(): sock(%x) pclient=%x\n", clientfd,pclient);
    if (clientfd == INVALID_SOCKET){
        RemoveClientSock(clientfd);
        return 0;
    }
    pclient->OnDisconnected();
#ifndef EPOLL_NEW
    del_epollfd(m_eventfd, clientfd);
#else
    Epoll_Del(m_eventfd, clientfd);
    CKxThreadCtx * lThread = m_pThreadPool->GetThread(pclient->m_threadno);
    lThread->Lock();
    lThread->m_clientnum--;
    lThread->UnLock();
#endif
    RemoveClientSock(clientfd);
    shutdown(clientfd, 2);
    close(clientfd);
    /*delete pclient;*/
    return 0;
}

//쓰레드간 이벤트큐 동기화 문제
//TODO::쓰레드풀중 노는쓰레드에 대한 flag를 두어,
//노는놈을 깨운다.이때 이벤트큐는 공용 뮤텍스로 추가,제거시 잠금을 얻어
//쓰레드간 이벤트큐를 잠금한다.
//쓰레드당 커넥션풀 관리(소켓리스트)

//
//void *workerthread_function(void *arg)
//{
//	// do not use THREAD CANCELLATION
//	//CKxServerService *pServer = (CKxServerService *)arg;
//	CKxThreadCtx *lThread = (CKxThreadCtx *)arg; 		
//	lThread->Job();
//	//pServer->workerthreadloop(pthread_self());
//	pthread_exit(NULL);
//}
