/*
Server Monitoring Watcher
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
*/

#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 <sys/time.h>
#include <list>
#include <queue>
#include <vector>
#include <pthread.h>

#ifdef USE_INTERDBPOOL
#include "dbpool.h"
#endif
#include "kxQueue.h"
#include "kxHashMap.h"
#include "kxThread.h"
#include "kxSockSession.h"
#include "kxServerService.h"
#include "kxPollService.h"
#include "kxtracelog.h"

using namespace std;

CKxPollService::CKxPollService() : CKxService()
{
	//pthread_cond_init(&m_timer_cond, NULL);
	//pthread_mutex_init(&m_timer_mutex, NULL);
	m_pThreadPool = NULL;
	m_polltime = 10;//초단위
	m_pServerService = NULL;
    m_queuemode = QUEUE_MODE_WORKQUEUE | QUEUE_MODE_USE_SERVER;
}

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

void CKxPollService::SetSockService(CKxServerService *pSockService)
{
	m_pServerService = pSockService;
}

CKxHashMap *CKxPollService::GetHashMap()
{
	return &m_hashmap;
}

bool CKxPollService::Init(int maxthreadpool, int polltimeout_milisec)
{
	SetMaxThreadPool(maxthreadpool);
	m_polltime = polltimeout_milisec;
	if(m_polltime <= 0)
		m_polltime = 1000;

	kxTraceLog(TL_LOW,"ServerWatch::Init:m_polltime=%d miliseconds maxthreadpool=%d\n",
		m_polltime,maxthreadpool);
#if 0
	if(m_nMaxThreadPool > 0){
		m_pThreadPool = new CKxThreadPool(m_nMaxThreadPool);
		if (NULL == m_pThreadPool) {
			return false;
		}
		m_pThreadPool->setthreadfuncptr(workerthread_function);
		m_pThreadPool->setthreadparam(this);
	}
#else
    if (m_nMaxThreadPool > 0){
        InitThreadPool(maxthreadpool);
    }
#endif
	return true;
}

bool CKxPollService::Destroy()
{
	// delete users
	kxTraceLog(TL_LOW,"ServerWatch::Destroy\n");
	//pthread_cond_destroy(&m_timer_cond);
	//pthread_mutex_destroy(&m_timer_mutex);
	return CKxService::Destroy();
}

bool CKxPollService::StartService()
{
	int rc;
#if 0
	// run thread pool
	if(m_pThreadPool && !m_pThreadPool->Start((void*)this))
	{
		return false;
	}

	// create server loop thread
	rc = pthread_create(&m_serverthread, NULL, serverthread_function,this);
	if(rc != 0)
	{
		kxTraceLog(TL_LOW,"error:create server thread\n");
		return false;
	}

	return true;
#else
    return CKxService::StartService();
#endif
}

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

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

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

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

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

//밀리세컨트단위로 폴링
int CKxPollService::OnServiceLoop(int threadid)
{
	int	i;
    int doit;
	int	max_got_events;
	int sec, usec;
	struct timeval now, start;
	struct tm *t;
	int init_cut;
	int rcwait;
	int rc = 0;
	int preday = 0;
	int daychanged = 0;
	//int mhor;
	int mmin;
	int msec;
	int micro;
	int mili;

    CKxQueue *pQueue = NULL;
	TIPARAM *ptip;

    doit = OnServiceBegin();
	//struct timespec {
	//	time_t	tv_sec;	       /* seconds */
	//	long	tv_nsec;       /* nanoseconds */
	//};
	//struct timeval {
	//	long    tv_sec;         /* seconds */
	//	long    tv_usec;        /* and microseconds */
	//};
    //while (1)//0초부터 시작하기로 한다
    //{
    //    gettimeofday(&now, NULL);
    //    t = localtime(&now.tv_sec);
    //    if (t->tm_sec == 0)
    //        break;
    //    //kxTraceLog(TL_LOW, "t->tm_sec = %ld\n",t->tm_sec);
    //}
    gettimeofday(&now, NULL);
    t = localtime(&now.tv_sec);
	m_timeout.tv_nsec = (now.tv_usec + (m_polltime % 1000) * 1000) * 1000;
	m_timeout.tv_sec = now.tv_sec + (m_polltime / 1000) + (m_timeout.tv_nsec / 1000000000);
	m_timeout.tv_nsec %= 1000000000;

	init_cut = (m_timeout.tv_nsec) % 1000000;
	m_timeout.tv_nsec -= init_cut;

	micro = now.tv_usec;
	mili = micro / 1000;
	msec = now.tv_sec % 60;
	mmin = (now.tv_sec / 60) % 60;
	//mhor = (now.tv_sec / 60 / 60) % 24;

#if 1 //def _DEBUG
	kxTraceLog(TL_LOW,"start time: %02d-%02d %02ld:%02ld:%02ld:%03ld:%04ld, timeval = %dmilisec\n",
		t->tm_mon + 1, t->tm_mday, t->tm_hour, mmin, msec, mili, micro % 1000, m_polltime);
#endif
	m_npollcount = 0;
	/* 
	 * 예약 가동 - cond에 signal을 기다린다.mutex를 unlock하고.. timeout되거나 signal들오면 mutex lock하고 빠져 나온다.
	 * 따라서 호출전후에 mutex를 lock,unlock 해줘야 함
	 * 여기서는 일부러 signal을 주지않고 타이머 효과를 얻기 위해 timeout을 유도한다 -krkim
	 */

	//pthread_mutex_lock(&m_timer_mutex);
    m_timer_cond.Lock();

    while (doit == 0 && m_bServiceRunning)
    {
#if 0
//		// 타이머 가동
//		//호출됨과 동시에 mutex를 내부적으로 unlock 하고 시그널이 전달되거나 timeout 되어
//		//쓰레드가 깨어나면 lock 한다(내부적으로 pthread_lock_mutex()를 사용)
//		rcwait = pthread_cond_timedwait(&m_timer_cond, &m_timer_mutex, &m_timeout);
//#if 1 //def _DEBUG /*krkim*/
//		if(rcwait != 0 && rcwait != ETIMEDOUT)
//			kxTraceLog(TL_LOW,"timer error: rc=[%d] ETIMEDOUT=%d EINTR=%d.\n", rcwait,ETIMEDOUT,EINTR);
//#endif
		/*rcwait = 0 OK,error = ETIMEDOUT,EINTR */
#else
        m_timer_cond.WaitCondSignalTimeout(&m_timeout,0);
#endif
        gettimeofday(&start, NULL);
		t = localtime(&start.tv_sec);
		daychanged = (preday == 0) ? 1 : (preday != t->tm_mday) ? 1 : 0;

		ptip = (TIPARAM *) malloc(sizeof(TIPARAM));
		//ptip->interval = *((int *) data);
		ptip->year = t->tm_year + 1900;
		ptip->month = t->tm_mon + 1;
		ptip->day = t->tm_mday;
		ptip->hour = t->tm_hour;
		ptip->minute = t->tm_min;
		ptip->second = t->tm_sec;
		ptip->daychanged = daychanged;
		//if((m_npollcount % 60) == 0)
		micro = start.tv_usec;
		mili = micro / 1000;
		msec = start.tv_sec % 60;
		mmin = (start.tv_sec / 60) % 60;
		//mhor = (start.tv_sec / 60 / 60) % 24;
        if ( m_npollcount  % 32 == 0)
		kxTraceLog(TL_MID, "polling(%d) %02d-%02d %02ld:%02ld:%02ld:%03ld:%04ld daychanged(%d)\n",
			m_npollcount, t->tm_mon + 1, t->tm_mday, t->tm_hour, mmin, msec, mili, micro % 1000, daychanged);
		OnServicePolling(ptip);
		m_npollcount++; 

		preday = t->tm_mday;
		//>>
		free(ptip);
		//<<
		gettimeofday(&now, NULL);

		sec = now.tv_sec - start.tv_sec;
		usec = now.tv_usec - start.tv_usec;//micro second 10^-6
		int miliss = sec * 1000 + (usec / 1000);
		int pollmili = m_polltime;

		if (miliss > m_polltime){
			init_cut = miliss % 100;
			miliss = miliss - init_cut;
			kxTraceLog(TL_MID, "처리시간이 타이머설정을 초과하였습니다.(%ld milisec) \n", miliss);
			miliss = miliss % m_polltime;
			pollmili = miliss;
		}
#if 0
		m_timeout.tv_nsec = (now.tv_usec + (pollmili % 1000) * 1000) * 1000;
		m_timeout.tv_sec = now.tv_sec + (pollmili / 1000) + (m_timeout.tv_nsec / 1000000000);
		m_timeout.tv_nsec %= 1000000000;
#else
		m_timeout.tv_nsec += ((pollmili / 1000) + (pollmili % 1000) * 1000) * 1000;
		m_timeout.tv_sec += (pollmili / 1000) + (m_timeout.tv_nsec / 1000000000);
		m_timeout.tv_nsec %= 1000000000;
#endif

    }
	//pthread_mutex_unlock(&m_timer_mutex);
    m_timer_cond.UnLock();
    kxTraceLog(0, "ServiceRunning Stopped.\n");
    rc = OnServiceEnd();
    return true;
}

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

    CKxThreadCtx * lThread;
    CKxService *pService;
    KX_QUEUE_ITEM qitem;
    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, "[TimeThread %d] pthread_self() = %x thr_id = %x (%x) started\n",
        lThread->m_ord_id, pthread_self(), lThread->m_thr_id,threadid);

 	/*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사용-각기다른 데이터처리)
        //  => 모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
#if 0    
        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();
        }
#else
        pQueue = pService->GetPopQueue(0, threadid);
#endif
        //kxTraceLog(TL_LOW, "[PollThread] WaitCondSignal() threadid=%x pQueue=%x(mode=%x)\n",threadid,pQueue,m_queuemode);
        
        pQueue->WaitCondSignal();
        while (1){
            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();
                break;
            }
			
            //kxTraceLog(TL_LOW, "[TimeThread] PopData()\n");
			qitem = pQueue->PopData(0);
			count1 = pQueue->GetQueueSize();
			pQueue->UnLock();
            //kxTraceLog(TL_LOW, "[PollThread %d] thr_id = %x popup 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 CKxPollService::StopService()
{
	// set to run status to false
	kxTraceLog(TL_LOW,"StopService now\n");
    SetServiceRun(false);
	m_WatchQue.PushBroadCastSignal();
	// closing sock server
	DestroyService();
	return CKxService::StopService();
}
 
int CKxPollService::OnServicePolling(TIPARAM *polltime)
{
	
	return CKxService::OnServicePolling(polltime);
}

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

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

	return rc;
}

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


    
