libgfaservices/datalogger/main.cpp

////////////////////////////////////////////////////////////////////////////////////////////////////
//

#include <signal.h>
#include <unistd.h>
#include <linux/limits.h>
#include "projal.h"
#include "logfile.h"
#include "fileutil.h"
#include "strutil.h"
#include "instance.h"
#include "processclock.h"
#include "debug.h"

////////////////////////////////////////////////////////////////////////////////////////////////////
// app control

#define _APPID						GFA_APPCTRL_APPID_DATALOGGER
#define _APPNAME					"Datalogger"
#define _DEPENDENCIES				((appid_t)(GFA_APPCTRL_APPID_REMANENT))

////////////////////////////////////////////////////////////////////////////////////////////////////

#define _TRACK_TIME					1
#define _WRITE_PAST_NULL_AT_STARTUP	1
#define _LOGFILE_NAME				"datalogger.log"

#define _SIG_BLOCK(s)				sigprocmask(SIG_BLOCK, (s), NULL)
#define _SIG_UNBLOCK(s)				sigprocmask(SIG_UNBLOCK, (s), NULL)

#define BYTES_FROM_MiB(mib)			((mib) * 1024 * 1024)

////////////////////////////////////////////////////////////////////////////////////////////////////

static volatile bool				g_fRun			= false;
static volatile bool				g_fPauseImp		= false;
static volatile bool				g_fPauseCmd		= false;
static volatile bool				g_fZombie		= false;
static appid_t						g_nDepRunning	= 0;
static sigset_t						g_set;
static CLogfile						g_lf;
int									g_nLastSig		= -1;

////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////
//

static const char* _GetBaseDir(std::string &rstrBaseDir)
{
	char szBaseDir[PATH_MAX];
	const char *pszBaseDir = NULL;

#ifdef _LOG_BASE_DIR
	pszBaseDir = _LOG_BASE_DIR;
	if(!pszBaseDir || !*pszBaseDir || !::DirectoryExist(pszBaseDir))
	{
		CLogfile::StdErr("Invalid base directory config! Using app directory!\n");
		pszBaseDir = ::GetAppDirectory(szBaseDir, sizeof(szBaseDir));
	}
	rstrBaseDir = pszBaseDir;
#else	//	_LOG_BASE_DIR
    UNUSED(pszBaseDir);
	rstrBaseDir = ::GetAppDirectory(szBaseDir, sizeof(szBaseDir));
#endif	//	_LOG_BASE_DIR

	rtrim(rstrBaseDir, "/");
	return rstrBaseDir.c_str();
}

static void _SigHandler(int sig)
{
	g_nLastSig = sig;
	g_fRun = g_fPauseImp = g_fPauseCmd = g_fZombie = false;
}

////////////////////////////////////////////////////////////////////////////////////////////////////

static void _ProcessCtrlMessages(HAPPCTRL hAC, HAPPINFO hAI)
{
    ctrlmsg_t nCtrlMsg;

	while(g_fRun && (nCtrlMsg = ::GfaIpcAppCtrlGetNextCtrlMsg(hAI)))
	{
		switch(nCtrlMsg)
		{
		case GFA_APPCTRL_CTRLMSG_STOP:
			g_fRun = false;
			g_fPauseImp = false;
			g_fPauseCmd = false;
			g_fZombie = false;
			g_lf.Info("Received Control Message 'Stop'\n");
			return;
		case GFA_APPCTRL_CTRLMSG_PAUSE:
			if(!g_fPauseCmd)
			{
				g_fPauseCmd = true;
				if(!g_fPauseImp)
				{
					::GfaIpcAppCtrlSetState(hAC, GIAS_Paused);
					g_lf.Info("Received Control Message 'Pause'\n");
					g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(GIAS_Paused));
					TRACE("%-8s: State: %s\n", "Me", ::GfaIpcAppCtrlGetStateText(GIAS_Paused));
				}
			}
			break;
		case GFA_APPCTRL_CTRLMSG_RESUME:
			if(g_fPauseCmd)
			{
				g_fPauseCmd = false;
				if(!g_fPauseImp)
				{
					g_lf.Info("Received Control Message 'Resume'\n");
					g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(GIAS_Running));
					::GfaIpcAppCtrlSetState(hAC, GIAS_Running);
					TRACE("%-8s: State: %s\n", "Me", ::GfaIpcAppCtrlGetStateText(GIAS_Running));
				}
			}
			break;
		default:
			break;
		}
	}
}

////////////////////////////////////////////////////////////////////////////////////////////////////

static void _ProcessStateEvents(HAPPCTRL hAC, HAPPINFO hAI)
{
    appid_t nAppIdSrc;
    bool fOldPaused = g_fPauseImp;
	char szDispName[128];

	while(g_fRun && (nAppIdSrc = ::GfaIpcAppCtrlGetNextStateEvtSrc(hAI)))
	{
		GfaIpcAppStates state = ::GfaIpcAppCtrlGetState(hAC, nAppIdSrc);
		GfaIpcAppCtrlGetDisplayName(hAC, nAppIdSrc, szDispName, sizeof(szDispName));
		TRACE("%-8s: State: %s\n", szDispName, ::GfaIpcAppCtrlGetStateText(state));
		
		if(nAppIdSrc & _DEPENDENCIES)
		{
			if(state == GIAS_Running)
			{
				g_lf.Info("%s -> %s.\n", szDispName, ::GfaIpcAppCtrlGetStateText(state));
				g_nDepRunning |= nAppIdSrc;
			}
			else
			{
				g_lf.Warning("%s -> %s.\n", szDispName, ::GfaIpcAppCtrlGetStateText(state));
				g_nDepRunning &= ~nAppIdSrc;
			}
		}
	}

	if(g_fRun)
	{
		g_fPauseImp = (g_nDepRunning != _DEPENDENCIES);

		if(!g_fPauseCmd && (fOldPaused != g_fPauseImp))
		{
			fOldPaused = g_fPauseImp;
			GfaIpcAppStates newState = g_fPauseImp ? GIAS_Paused : GIAS_Running;
			::GfaIpcAppCtrlSetState(hAC, newState);
			if(g_fPauseImp)
				g_lf.Warning("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(newState));
			else
				g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(newState));
		}
	}
}

////////////////////////////////////////////////////////////////////////////////////////////////////
//

int main(int argc, char **argv)
{
	int nRet = -1;
	HSHM hShm = NULL;
	HAPPCTRL hAC = NULL;
	HAPPINFO hAI;
	void *pShm = NULL;
	DLPARAMS dlp;
    CProcessInstance pi;
    unsigned long nSamplesPerLog, nLogsPerFlush;
	char szLogFile[PATH_MAX];
	std::string strBaseDir;
	const char *pszBaseDir = NULL;
#if _TRACK_TIME
	unsigned long long nElapsed;
	double fTime;
	CProcessClock pc;
#endif	//	_TRACK_TIME
	unsigned long long nUsecWorkTime = 0;
	CProcessClock pcWork;

    ////////////////////////////////////////////////////////////////////////////////////////////////
    // check for multiple instances

    if(!pi.LockInstance(UUID_SHM))
    {
		CLogfile::StdErr("Failed to start instance!\n");
        return -1;
    }

    ////////////////////////////////////////////////////////////////////////////////////////////////
	// configure signal handling

	struct sigaction sa;
	::sigfillset(&g_set);
	sigaddset(&g_set, SIGUSR1);
	memset(&sa, 0, sizeof(sa));

	sa.sa_handler = _SigHandler;
    sigaction(SIGHUP, &sa, NULL);	// handles user's terminal disconnect
    sigaction(SIGQUIT, &sa, NULL);	// handles Ctrl + '\'
	sigaction(SIGTERM, &sa, NULL);	// handles normal termination
	sigaction(SIGABRT, &sa, NULL);	// handles abnormal termination (i.e. abort())
	sigaction(SIGINT, &sa, NULL);	// handles Ctrl + 'C'

	sa.sa_handler = SIG_IGN;
    sigaction(SIGTSTP, &sa, NULL);	// ignores Ctrl + 'Z'
    sigaction(SIGSTOP, &sa, NULL);	// ignores Stop
    sigaction(SIGCONT, &sa, NULL);	// ignores Continue
    sigaction(SIGCHLD, &sa, NULL);	// ignores child process termination
    sigaction(0, &sa, NULL);		// ignores shell termination

	do
	{
		g_fZombie = true;

		////////////////////////////////////////////////////////////////////////////////////////////
		// get the base directory for output files

		if(!pszBaseDir)
			pszBaseDir = _GetBaseDir(strBaseDir);

		CLogfile::StdOut("Using base directory \"%s\".\n", pszBaseDir);

		////////////////////////////////////////////////////////////////////////////////////////////
		// initialize log file

		sprintf(szLogFile, "%s/%s", pszBaseDir, _LOGFILE_NAME);

		if(!g_lf.Open(szLogFile, true, CLogfile::VB_Inf))
		{
			CLogfile::StdErr("Failed to create/open log file!\n");
			break;
		}

		g_lf.Info("Process enter.\n");

		////////////////////////////////////////////////////////////////////////////////////////////
		// initialize app control

		g_lf.Info("Acquire AppCtrl-Handle.\n");

		if(!(hAC = ::GfaIpcAppCtrlAcquire(_APPID, _APPNAME, _LOG_INTV_SAMPLE * 1000, _LOG_INTV_SAMPLE * 3000)))
		{
			g_lf.Error("Failed to acquire AppCtrl-Handle!\n");
			break;
		}

		::GfaIpcAppCtrlSetState(hAC, GIAS_Initializing);
		g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(GIAS_Initializing));

		if(!::GfaIpcAppCtrlSubscribeStateEvents(hAC, _DEPENDENCIES))
		{
			g_lf.Error("Failed to subscribe state event notifications!\n");
			break;
		}

		////////////////////////////////////////////////////////////////////////////////////////////
		// validate config parameters

		if(strlen(_LOG_DB_NAME) >= _DL_MAX_DB_NAME_LENGTH)
		{
	        g_lf.Error("Database name too long!\n");
			break;
		}

		if(strlen(_LOG_DB_USER) >= _DL_MAX_DB_USER_LENGTH)
		{
	        g_lf.Error("Database username too long!\n");
			break;
		}

		if(strlen(_LOG_DB_PASS) >= _DL_MAX_DB_PASS_LENGTH)
		{
	        g_lf.Error("Database password too long!\n");
			break;
		}

		if(strlen(_LOG_TAGS_TABLE) >= _DL_MAX_TABLE_NAME_LENGTH)
		{
	        g_lf.Error("Tag table name too long!\n");
			break;
		}

		if(strlen(_LOG_LOGS_TABLE) >= _DL_MAX_TABLE_NAME_LENGTH)
		{
	        g_lf.Error("Log table name too long!\n");
			break;
		}

		if((_LOG_INTV_LOG < _LOG_INTV_SAMPLE) || (_LOG_INTV_LOG % _LOG_INTV_SAMPLE))
		{
	        g_lf.Error("The log interval must be an integer multiple of the sample interval!\n");
			break;
		}

		if((_LOG_INTV_WRITE < _LOG_INTV_LOG) || (_LOG_INTV_WRITE % _LOG_INTV_LOG))
		{
	        g_lf.Error("The flush interval must be an integer multiple of the log interval!\n");
			break;
		}

		nSamplesPerLog	= _LOG_INTV_LOG / _LOG_INTV_SAMPLE;
		nLogsPerFlush	= _LOG_INTV_WRITE / _LOG_INTV_LOG;

		////////////////////////////////////////////////////////////////////////////////////////////
		// configure data logger

		memset(&dlp, 0, sizeof(dlp));
		strncpy(dlp.szDBName, _LOG_DB_NAME, _DL_MAX_DB_NAME_LENGTH - 1);
		strncpy(dlp.szDBUser, _LOG_DB_USER, _DL_MAX_DB_USER_LENGTH - 1);
		strncpy(dlp.szDBPass, _LOG_DB_PASS, _DL_MAX_DB_PASS_LENGTH - 1);
		strncpy(dlp.szTagsTable, _LOG_TAGS_TABLE, _DL_MAX_TABLE_NAME_LENGTH - 1);
		strncpy(dlp.szLogsTable, _LOG_LOGS_TABLE, _DL_MAX_TABLE_NAME_LENGTH - 1);
		snprintf(dlp.szLogsTableBD, _DL_MAX_TABLE_NAME_LENGTH - 1, "%s_bad_date", _LOG_LOGS_TABLE);
		dlp.nIntvSample	= _LOG_INTV_SAMPLE;
		dlp.nIntvLog	= _LOG_INTV_LOG;
		dlp.nIntvFlush	= _LOG_INTV_WRITE;
		dlp.bMinMax		= _LOG_INTV_MIN_MAX;
		dlp.pszBaseDir	= pszBaseDir;

#if _LOG_MAX_AGE
		dlp.nMaxAge = _LOG_MAX_AGE;
		g_lf.Info("Max. age of log entries: %lu days.\n", dlp.nMaxAge);
#else	//	_LOG_MAX_AGE
		dlp.nMaxAge = 0;
		g_lf.Info("Max. age of log entries not defined.\n");
#endif	//	_LOG_MAX_AGE

#if _LOG_MAX_SIZE
		dlp.nMaxSize = BYTES_FROM_MiB(_LOG_MAX_SIZE);
		g_lf.Info("Max. size for logs table: %u MiB.\n", _LOG_MAX_SIZE);
#else	//	_LOG_MAX_SIZE
		dlp.nMaxSize = 0;
		g_lf.Info("Max. size for logs table not defined.\n");
#endif	//	_LOG_MAX_SIZE

		////////////////////////////////////////////////////////////////////////////////////////////
		// acquire SHM

		if(!(hShm = ::acquire_shm(sizeof(shm_t), 1)))
		{
	        g_lf.Error("Unable to acquire SHM Handle!\n");
	        break;
		}

		g_lf.Info("Acquired SHM Handle.\n");

        if(!(pShm = ::GfaIpcAcquirePointer(hShm)))
        {
			g_lf.Error("Unable to acquire SHM Pointer!\n");
	        break;
        }

		g_lf.Info("Acquired SHM Pointer.\n");
        ::GfaIpcDumpSHMROT();

		////////////////////////////////////////////////////////////////////////////////////////////
		// init datalogger and SHM

		CDataLoggerClock dlc(_LOG_INTV_SAMPLE * 1000000);
		CDataLogger dl(&dlp, g_lf);
    	CShm_t shm(pShm, hShm);

		if(!dl.InitDatabase(false))
	        break;

        shm.InitPath(NULL, NULL);
        shm.InitTagID(dl);

		bool bTimerUnderrun = false;
		unsigned long nSamples = 0, nLogs = 0;
		time_t t = time(NULL);

		_SIG_BLOCK(&g_set);
#if _WRITE_PAST_NULL_AT_STARTUP
		time_t t2 = dl.LastLogTimestamp();
		if(t2)
			shm.LogValueChanged(t2 + 1, dl, true, true);
#endif	//	_WRITE_PAST_NULL_AT_STARTUP
		shm.LogValueChanged(t, dl, true, false);
#if _TRACK_TIME
		pc.ClockTrigger();
#endif	//	_TRACK_TIME
		dl.Flush(t);
		_SIG_UNBLOCK(&g_set);
#if _TRACK_TIME
		nElapsed = pc.ClockGetElapsed();
		fTime	= (double)nElapsed / 1000000.0;
		TRACE("Flush (%.2f ms).\n", fTime);
#endif	//	_TRACK_TIME

		////////////////////////////////////////////////////////////////////////////////////////////
		// enter running loop

		g_fZombie = false;
		g_fRun = true;
		::GfaIpcAppCtrlSetState(hAC, GIAS_Running);
		g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(GIAS_Running));

		while(g_fRun)
		{
			////////////////////////////////////////////////////////////////////////////////////////
			// update app control info

			if((hAI = ::GfaIpcAppCtrlInfoUpdate(hAC, nUsecWorkTime)))
			{
				_ProcessCtrlMessages(hAC, hAI);
				if(!g_fRun)
					break;
				_ProcessStateEvents(hAC, hAI);
			}

			////////////////////////////////////////////////////////////////////////////////////////

			if(!dlc.Sleep(bTimerUnderrun, true))
			{
				g_fRun = g_fPauseImp = g_fPauseCmd = false;
				g_fZombie = true;
				g_lf.Error("CDataLoggerClock::Sleep Error!\n", t);
				break;
			}

			////////////////////////////////////////////////////////////////////////////////////////

			t = time(NULL);
			pcWork.ClockTrigger();

			if(!bTimerUnderrun && !g_fPauseImp && !g_fPauseCmd)
			{
				_SIG_BLOCK(&g_set);

				shm.LogValueChanged(t, dl, false, false);
	            shm.Sample();
				++nSamples;

				if(nSamples == nSamplesPerLog)
				{
#if _TRACK_TIME
					pc.ClockTrigger();
#endif	//	_TRACK_TIME
                    shm.LogInterval(t, dl);
#if _TRACK_TIME
					nElapsed = pc.ClockGetElapsed();
					fTime	= (double)nElapsed / 1000000.0;
					TRACE("Log (%.2f ms).\n", fTime);
#endif	//	_TRACK_TIME
					++nLogs;
					nSamples = 0;
				}

				if(nLogs == nLogsPerFlush)
				{
#if _TRACK_TIME
					pc.ClockTrigger();
#endif	//	_TRACK_TIME
					dl.Flush(time(NULL));
#if _TRACK_TIME
					nElapsed = pc.ClockGetElapsed();
					fTime	= (double)nElapsed / 1000000.0;
					TRACE("Flush (%.2f ms).\n", fTime);
#endif	//	_TRACK_TIME
					nLogs = 0;
					dl.SizeGuardTrigger(t);
				}

				_SIG_UNBLOCK(&g_set);
			}
			else if(bTimerUnderrun)
			{
				g_lf.Warning("Timer underrun (%ld)!\n", t);
				bTimerUnderrun = false;
			}

			nUsecWorkTime = pcWork.ClockGetElapsed() / 1000;
		}

		dl.Flush(time(NULL));
	}
	while(false);

	if(g_nLastSig >= 0)
	{
		g_lf.Info("Received signal '%s'.\n", strsignal(g_nLastSig));
		g_nLastSig = -1;
	}

	if(hShm)
	{
		if(pShm)
		{
			g_lf.Info("Releasing SHM Pointer ...\n");
			::GfaIpcReleasePointer(hShm, pShm);
		}

		g_lf.Info("Releasing SHM Handle ...\n");
    	::GfaIpcReleaseSHM(hShm);
    }
	
	if(g_fZombie)
	{
		if(hAC)
			::GfaIpcAppCtrlSetState(hAC, GIAS_Zombie);
		TRACE("Enter Zombie state ...\n");
		g_lf.Warning("Enter Zombie state ...\n");
		g_lf.Flush();
		pause();

		if(g_nLastSig >= 0)
			g_lf.Info("Received signal '%s'.\n", strsignal(g_nLastSig));
	}

	if(hAC)
	{
		g_lf.Info("Enter state %s ...\n", ::GfaIpcAppCtrlGetStateText(GIAS_Terminating));
		::GfaIpcAppCtrlSetState(hAC, GIAS_Terminating);
		g_lf.Info("Releasing App Control ...\n");
		::GfaIpcAppCtrlRelease(hAC);
	}

	g_lf.Info("Process exit.\n\n");
	g_lf.Close();
	CLogfile::StdErr("Datalogger exit.\n");
	return nRet;
}