/*
 KX Base Object and Service Class
 ---------------------------------------------------------------------------
 $Id: $ $Revision: $
 $HeadURL: $
 */

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

/*
 * History:
 * ===========================================================================
 * 2004.08.13  KRKIM   Initialized.
 *                     Relay Server Manager (epoll edge-triggered / worker thread)
 *                     KRKIM(Kyoung Rae Kim) durumul@gmail.com
 *
 */

/*
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 "kxThread.h"
#include "iocq.h"
#include "tokenizer.h"
#include <kxService.h>
#include "kxtracelog.h"

using namespace std;

/////////////////////////////////////////////////////////////////////////////////////////////////////////////
CKxService::CKxService()
{
    Init();
}

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

bool CKxService::Init()
{
    m_epolltimeout = 5000;
    m_polltime = 5000;//밀리초 단위
    m_npollcount = 0;
    m_bServiceRunning = false;
    m_servicethread = 0;
    m_queuemode = QUEUE_MODE_WORKQUEUE;/*ENUM_QUEUE_MODE*/
    m_nMaxThreadPool = 1;
    m_pThreadPool = NULL;
    m_nSendTRID = 0;
    m_idlecnt = 0;
    return true;
}

int CKxService::getNextSendTRID()
{
	m_nSendTRID++;
	if (m_nSendTRID > MAX_TRID_LIMIT)
		m_nSendTRID = 1;
	return m_nSendTRID;
}

bool CKxService::InitService()
{
    return true;
}

bool CKxService::PauseService()
{
    return true;
}

void CKxService::SetMaxThreadPool(int maxthreadpool)
{
    m_nMaxThreadPool = maxthreadpool;
}

void CKxService::SetServiceRun(bool flag)
{
    m_bServiceRunning = flag;
}

bool CKxService::GetServiceRun()
{
    return m_bServiceRunning;
}

CKxQueue * CKxService::GetQueue(int index)
{
    if (index >= sizeof(m_queue) / sizeof(m_queue[0]))//num
        return NULL;
    return &m_queue[index];
}

bool CKxService::InitThreadPool(int maxthreadpool)
{
    SetServiceRun(true);
    SetMaxThreadPool(maxthreadpool);

    kxTraceLog(0,"CKxService::InitThreadPool maxthreadpool=%d\n",maxthreadpool);
    m_pThreadPool = new CKxThreadPool(m_nMaxThreadPool);
    if (NULL == m_pThreadPool) {
        return false;
    }
    m_pThreadPool->setthreadfuncptr(workerthread_function);
    m_pThreadPool->setthreadparam(this);
    return true;
}

bool CKxService::InitDBPool(int maxdbpool,char * host,char * dbname,char * user,char * pass,int port)
{
    int rc;
#ifdef USE_INTERDBPOOL
    kxTraceLog(0,"CKxService::InitDBPool:maxdbpool=%d host=%s dbname=%s user=%s pass=%s port=%d\n",
        maxdbpool,host,dbname,user,pass,port);
    CDBPool *dbpool = CDBPool::CreateInstance(maxdbpool);
    dbpool->SetConnectionInfo(host,dbname,user,pass,port);
    kxTraceLog(0,"try connect db\n");
    if((rc = dbpool->ConnectDB()) < 0){
        kxTraceLog(0,"InitDBPool:ConnectDB error rc=%d\n",rc);
    }
    else
        kxTraceLog(0,"InitDBPool:ConnectDB ok!\n");
#else
    kxTraceLog(0,"CKxService::InitDBPool:Empty function Use override your code\n",
        maxdbpool,host,dbname,user,pass,port);
#endif
    return true;
}

bool CKxService::Destroy()
{
    // delete users
    kxTraceLog(0,"CKxService::Destroy\n");
    if(m_pThreadPool != NULL)
    {
        delete m_pThreadPool;
        m_pThreadPool = NULL;
    }
#ifdef USE_INTERDBPOOL
    CDBPool::DestroyInstance();
#endif
    return true;
}

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

int CKxService::WaitService()
{
    void *thread_result;
    int rc;
    rc = pthread_join(m_servicethread, &thread_result);
    if (rc == 0)
    {
        m_servicethread = 0;//if thread already exit,and try twice join, cause illegal error detected with SIGILL signal
        kxTraceLog(0, "WaitService:picked up thread\n");
    }
    else
    {
        kxTraceLog(0, "WaitService:pthread join fail\n");
    }
    return rc;
}

bool CKxService::StopService()
{
    int index = 0;
    // set to run status to false
    kxTraceLog(0, "StopService now\n");
    SetServiceRun(false);
    //m_queue.PushBroadCastSignal();
    for (index = 0; index < sizeof(m_queue) / sizeof(m_queue[0]); index++){
        m_queue[index].PushBroadCastSignal();
    }
    DestoryThreadPool();
    DestroyService();
    return true;
}

bool CKxService::DestroyService()
{
    // send to cancellation to workerthread
    // stop epoll loop
    int rc;
    void *thread_result;
    kxTraceLog(0, "DestroyService:thread = %x\n", m_servicethread);
    if (m_servicethread != 0){
        rc = pthread_cancel(m_servicethread);
        if (rc != 0)
        {
            kxTraceLog(0, "DestroyService:thread cancel failed\n");
        }
        else
        {
            kxTraceLog(0, "DestroyService:pthread_join thread = %x\n", m_servicethread);
            rc = pthread_join(m_servicethread, &thread_result);
            if (rc == 0)
            {
                m_servicethread = 0;//if thread already exit,and try twice join, cause illegal error detected with SIGILL signal
                kxTraceLog(0, "DestroyService:picked up thread\n");
            }
            else
            {
                kxTraceLog(0, "DestroyService:pthread join fail\n");
            }
        }
    }
    return true;
}

bool CKxService::DestoryThreadPool()
{
    if(NULL != m_pThreadPool){
        m_pThreadPool->Stop();
        delete m_pThreadPool;
        m_pThreadPool = NULL;
        return true;
    }
    return false;
}
//쓰레드간 이벤트큐 동기화 문제
//TODO::쓰레드풀중 노는쓰레드에 대한 flag를 두어,
//노는놈을 깨운다.이때 이벤트큐는 공용 뮤텍스로 추가,제거시 잠금을 얻어
//쓰레드간 이벤트큐를 잠금한다.
//쓰레드당 커넥션풀 관리(소켓리스트)

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

void *CKxService::servicethread_function(void *arg)
{
    int rc;
    CKxService *pServer = (CKxService *)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->OnServiceLoop(pthread_self());
    pthread_exit(NULL);
}

void *CKxService::workerthread_function(void *arg)
{
    // do not use THREAD CANCELLATION
     CKxThreadCtx *lThread = (CKxThreadCtx *)arg;
    CKxService *pService = (CKxService *)lThread->GetService();
    pService->OnThreadLoop(pthread_self());
    pthread_exit(NULL);
}

int CKxService::OnServiceLoop(int threadid)
{
    int	i;
    int rc,start;
    start = OnServiceBegin();
    while(start == 0 && m_bServiceRunning)
    {
        OnServiceProc(NULL);
    }
    kxTraceLog(0,"ServiceRunning Stopped.\n");
    rc = OnServiceEnd();
    return true;
}

//쓰레드풀의 작업쓰레드의 실행
int CKxService::OnThreadLoop(int threadid)
{
    int index;
    int rc,start;
    CKxThreadCtx * lThread;
    CKxService *pService;

    lThread = m_pThreadPool->GetThreadByID(threadid);
    pService = (CKxService *)lThread->m_pService;//same as this

    start = OnThreadBegin(lThread->m_ord_id);
    while (start == 0 && m_bServiceRunning)
    {
        //Pop Queue Data

        //Subscribe mode or workqueue mode
        if (m_queuemode == QUEUE_MODE_FANOUT){
            //push service push to dispatch to each thread's queue for distribution
            //and pop each thread 's queue
            //모든 쓰레드가 각각 자신의 큐에 들어온 같은 데이터를 꺼내와서 동일한 데이터를 분배처리
            //TODO::
        }
        else if(m_queuemode == QUEUE_MODE_WORKQUEUE){
            //pop service 's queue and dispatch distribution
            //모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
            //TODO::
        }
        pService->OnThreadProc(lThread->m_ord_id,NULL);
        //for (index = 0; index < m_nMaxThreadPool; index++){
        //    lThread = m_pThreadPool->GetThread(index);
        //    //lThread->WatchJob();
        //}
        //pQueue->Free(&qitem);
    }
    kxTraceLog(0, "Thread[%d] is end\n",lThread->m_ord_id - 1);
    rc = OnThreadEnd(lThread->m_ord_id);
    return 0;
}

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

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

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

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

int CKxService::OnServicePolling(TIPARAM *polltime)
{
	int rc = 0;
	return rc;
}

int CKxService::OnThreadQueue(int nID/*orderid*/,void *param/*popped queue data*/)
{
	int rc = 0;
    KX_QUEUE_ITEM *qitem = (KX_QUEUE_ITEM*) param;
	return rc;
}

int CKxService::OnThreadProc(int nID/*orderid*/,void *param/*popped queue data*/)
{
    int rc = 0;
    CKxThreadCtx * lThread;
    CKxService *pService;

    KX_QUEUE_ITEM *qitem = (KX_QUEUE_ITEM*) param;
    lThread = m_pThreadPool->GetThreadByOrderID(nID);
    pService = (CKxService *)lThread->m_pService;//same as this
    return rc;
}

int CKxService::OnServiceProc(void *param/*popped queue data*/)
{
    int rc = 0;
    KX_QUEUE_ITEM *qitem = (KX_QUEUE_ITEM*) param;
    return rc;
}

//////////////////////////////////////////////////////////////////////////////////////////
//Queue functons
//조합의 종류
//m_queuemode = QUEUE_MODE_WORKQUEUE | QUEUE_MODE_USE_CLIENT;
//m_queuemode = QUEUE_MODE_WORKQUEUE | QUEUE_MODE_USE_SERVER;
//m_queuemode = QUEUE_MODE_FANOUT | QUEUE_MODE_USE_CLIENT; 같은데이터를 브로드캐스트
//fanout 방식(각 쓰레드별 자체 queue 처리-같은데이터)
// => 모든 쓰레드가 각각 자신의 큐에 들어온 같은 데이터를 꺼내와서 동일한 데이터를 분배처리
//work queue 방식(각 쓰레드별 공용 queue사용-각기다른 데이터처리)
// => 모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
//2018.09.30

int CKxService::Publish(KX_QUEUE_ITEM qitem, int bLockNow,int mainqueueindex,int threadindex)
{
    int rc = 0;
    int index = 0;
    int sent = 0;
    CKxQueue *pQueue = NULL;
    CKxThreadCtx *lThread = NULL;
    int count1, count2;

    if (((m_queuemode & QUEUE_MODE_WORKQUEUE) & QUEUE_MODE_MASK) == QUEUE_MODE_WORKQUEUE){
        //특정 쓰레드에게만 1개의 qitem 일감을 부여
        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            lThread = m_pThreadPool->GetThread(threadindex);
            pQueue = (CKxQueue *)lThread;
        }
        //메인큐에 1개의 일감을 push하고 쓰레드풀에서 어떤 한놈이 가져가길 바람
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            pQueue = this->GetQueue(mainqueueindex);
        }
        if (pQueue){
            count1 = pQueue->GetQueueSize();
            pQueue->PushData(qitem,bLockNow);
            sent++;
            //임의의 1개의 쓰레드만 깨운다.
            pQueue->PushSignalNow(bLockNow);
            count2 = pQueue->GetQueueSize();
        }
    }
    //1개의 일감에 대해 모든 쓰레드가 동일한 큐데이터로 병렬작업을 원함
    //즉,Exchange 가 각 쓰레드의 큐에 동일데이터를 분배해야 한다.

    else if (((m_queuemode & QUEUE_MODE_FANOUT) & QUEUE_MODE_MASK) == QUEUE_MODE_FANOUT){
        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            for (index = 0; index < m_nMaxThreadPool; index++){
                lThread = m_pThreadPool->GetThread(index);
                pQueue = (CKxQueue *)lThread;
                if (pQueue){
                    count1 = pQueue->GetQueueSize();
                    pQueue->PushData(qitem,bLockNow);
                    sent++;
                }
            }
            //모든쓰레드를 깨운다.
            if (pQueue){
                pQueue->PushBroadCastSignal(bLockNow);
                count2 = pQueue->GetQueueSize();
            }
        }
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            kxTraceLog(TL_LOW, "[CKxService] Publish:잘못된 큐설정(m_queuemode = %x),원작자에게 사용법을 문의하시오.\n",m_queuemode);
            rc = -1;
        }
    }
    kxTraceLog(TL_LOW, "[CKxService] Publish: pQueue=%x(mode=%x,mask=%x) sent = %d queue count %d -> %d\n",
        pQueue,pQueue->GetQueueMask(),m_queuemode,sent,count1,count2);
    return rc;
}

CKxQueue * CKxService::GetPopQueue(int mainqueueindex,int threadid)
{
    int rc = 0;
    CKxQueue *pQueue = NULL;
    CKxThreadCtx *lThread = NULL;

    //Pop Queue Data
    //Subscribe mode or workqueue mode
    //fanout 방식(각 쓰레드별 자체 queue 처리-같은데이터)
    //  => 모든 쓰레드가 각각 자신의 큐에 들어온 같은 데이터를 꺼내와서 동일한 데이터를 분배처리
    //work queue 방식(각 쓰레드별 공용 queue사용-각기다른 데이터처리)
    //  => 모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
    if (((m_queuemode & QUEUE_MODE_WORKQUEUE) & QUEUE_MODE_MASK) == QUEUE_MODE_WORKQUEUE){
            //특정 쓰레드에게만 1개의 qitem 일감을 부여

        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            lThread = m_pThreadPool->GetThread(threadid);
            pQueue = (CKxQueue *)lThread;
        }
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            pQueue = this->GetQueue(mainqueueindex);
        }
    }
    //1개의 일감에 대해 모든 쓰레드가 동일한 큐데이터로 병렬작업을 원함
    //즉,Exchange 가 각 쓰레드의 큐에 동일데이터를 분배해야 한다.
    else if (((m_queuemode & QUEUE_MODE_FANOUT) & QUEUE_MODE_MASK) == QUEUE_MODE_FANOUT){
        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            lThread = m_pThreadPool->GetThread(threadid);
            pQueue = (CKxQueue *)lThread;
        }
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            kxTraceLog(TL_LOW, "GetQueue:잘못된 큐설정(m_queuemode = %x),원작자에게 사용법을 문의하시오.\n",m_queuemode);
            rc = -1;
        }
    }
    //kxTraceLog(TL_LOW, "[CKxService] GetPopQueue: m_queuemode=%x pQueue=%x\n",m_queuemode,pQueue);
    return pQueue;
}

KX_QUEUE_ITEM *CKxService::Subscribe(int bLockNow,int mainqueueindex)
{
    int rc = 0;
    int index = 0;
    int sent = 0;
    CKxQueue *pQueue = NULL;
    CKxThreadCtx *lThread = NULL;
    KX_QUEUE_ITEM *qitem = NULL;
    //if (((m_queuemode & QUEUE_MODE_WORKQUEUE) & QUEUE_MODE_MASK) == QUEUE_MODE_WORKQUEUE){
    //    //특정 쓰레드에게만 1개의 qitem 일감을 부여
    //    if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
    //        lThread = m_pThreadPool->GetThread(threadindex);
    //        pQueue = (CKxQueue *)lThread;
    //    }
    //    //메인큐에 1개의 일감을 push하고 쓰레드풀에서 어떤 한놈이 가져가길 바람
    //    else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
    //        pQueue = this->GetQueue(mainqueueindex);
    //    }
    //    if (pQueue){
    //        pQueue->PushData(qitem,bLockNow);
    //        sent++;
    //    }
    //}
    ////1개의 일감에 대해 모든 쓰레드가 동일한 큐데이터로 병렬작업을 원함
    ////즉,Exchange 가 각 쓰레드의 큐에 동일데이터를 분배해야 한다.

    //else if (((m_queuemode & QUEUE_MODE_FANOUT) & QUEUE_MODE_MASK) == QUEUE_MODE_FANOUT){
    //    if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
    //        for (index = 0; index < m_nMaxThreadPool; index++){
    //            lThread = m_pThreadPool->GetThread(index);
    //            pQueue = (CKxQueue *)lThread;
    //            if (pQueue){
    //                pQueue->PushData(qitem,bLockNow);
    //                sent++;
    //            }
    //        }
    //        pQueue->PushBroadCastSignal(bLockNow);
    //    }
    //    else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
    //        kxTraceLog(TL_LOW, "잘못된 큐설정(m_queuemode = %x),원작자에게 사용법을 문의하시오.\n",m_queuemode);
    //        rc = -1;
    //    }
    //}
    //kxTraceLog(TL_LOW, "CKxService::Publish: m_queuemode=%x,sent = %d\n",m_queuemode,sent);
    return qitem;
}
