libgfaservices/datalogger/datalogger.cpp

#include <string.h>
#include <cmath>
#include <malloc.h>
#include <limits.h>
#include <signal.h>
#include <sys/stat.h>
#include "fileutil.h"
#include "strutil.h"
#include "datalogger.h"
#include "processclock.h"
#include "debug.h"

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

#define _USE_MODIFIED_INDEX					1
#define _TRACK_TIME							1
#define _FILE_SIZE_DELETE_MARGIN_PERCENT	10
#define _SIZE_GUARD_TIMESTAMP_FILE_NAME		"sguard.ts"
#define _INVALID_TIMESTAMP_VALUE			((time_t)-1)

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

inline static bool _IsLoggable(double d)
{
	int c = std::fpclassify(d);
	return (c == FP_NORMAL) || (c == FP_ZERO);
}

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

CDataLogger::CDataLogger(LPCDLPARAMS pdlp, CLogfile &rlf) : m_lf(rlf),
															m_tidSGThread(0),
															m_bSGHasSizeLimitPrerequisites(false),
															m_bSGInProgress(false),
															m_bSGConfigured(false),
															m_nSGCurPassUTC(0),
															m_nSGLastPassUTC(0),
															m_condmtx1(PTHREAD_MUTEX_INITIALIZER),
															m_cond1(PTHREAD_COND_INITIALIZER),
															m_nLastLogTimestamp(0),
															m_bBadDateLogsDetected(false)
{
	memset(&m_dlp, 0, sizeof(m_dlp));
	memset(&m_gv, 0, sizeof(m_gv));
	memset(&m_mtx, 0, sizeof(m_mtx));
	memset(&m_mutexAttr, 0, sizeof(m_mutexAttr));

	if(pdlp)
		memcpy(&m_dlp, pdlp, sizeof(m_dlp));

	if(!(m_bSGConfigured = !!m_dlp.nMaxSize || !!m_dlp.nMaxAge))
		m_lf.Error("Size guard not configured.\n");

	if(m_dlp.pszBaseDir)
		strncpy(m_szAppDir, m_dlp.pszBaseDir, sizeof(m_szAppDir) - 1);
	else
		::GetAppDirectory(m_szAppDir, sizeof(m_szAppDir));

	::pthread_mutexattr_init(&m_mutexAttr);
	::pthread_mutexattr_settype(&m_mutexAttr, PTHREAD_MUTEX_RECURSIVE);
	::pthread_mutex_init(&m_mtx, &m_mutexAttr);
}

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

CDataLogger::~CDataLogger(void)
{
	Release();
	::pthread_mutex_destroy(&m_mtx);
	::pthread_mutexattr_destroy(&m_mutexAttr);
	::pthread_cond_destroy(&m_cond1);
}

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

void CDataLogger::Release(void)
{
	if(m_tidSGThread)
	{
		::pthread_mutex_lock(&m_condmtx1);
		bool bSgInProg = m_bSGInProgress;
		::pthread_mutex_unlock(&m_condmtx1);
		if(bSgInProg)
			CLogfile::StdErr("\nDatalogger is terminating. This may take some time ...\nPlease wait, before powering off the device!\n");
		Lock();
		m_lf.Lock();
		::pthread_cancel(m_tidSGThread);
		m_lf.Unlock();
		Unlock();
		::pthread_join(m_tidSGThread, NULL);
		m_tidSGThread = 0;
	}
}

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

unsigned long CDataLogger::GetTagID(const char *pszVarPath, int nDataType, int nLogType)
{
	CMySqlDB db;
	const char *pszID = NULL, *pszDataType = NULL, *pszLogType = 0;
	char *pszEndptr;
	unsigned long nID = ULONG_MAX;
	int nDT, nLT;

	std::string sSql;
//	sSql = formatString("select `tagid` from `%s` where `path` = '%s' and `dataType` = %d and `logType` = %d", m_dlp.szTagsTable, pszVarPath, nDataType, nLogType);
	sSql = formatString("select `tagid`, `dataType`+0, `logType`+0 from `%s` where `path` = '%s'", m_dlp.szTagsTable, pszVarPath);

	if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, m_dlp.szDBName))
	{
		m_lf.Error("CDataLogger::GetTagID: DB Error: %s\n", db.LastError().c_str());
		return nID;
	}

	CMySqlResult res = db.Query(sSql.c_str());
	bool bError = res.error();

	if(bError)
		m_lf.Error("CDataLogger::GetTagID: DB Error: %s\n", db.LastError().c_str());
	else
	{
		my_ulonglong nCount = res.RowCount();

		if(nCount > 0)
		{
			MYSQL_ROW pRow = res.FetchRow();
			pszID = pRow[0];
			pszDataType = pRow[1];
			pszLogType = pRow[2];

			nID = strtoul(pszID, &pszEndptr, 10);
			nDT = strtol(pszDataType, &pszEndptr, 10);
			nLT = strtol(pszLogType, &pszEndptr, 10);
			
			if((nDT != nDataType) || (nLT != nLogType))
			{
				sSql = formatString("update `%s` set `dataType` = %d, `logType` = %d where `tagid` = %u", m_dlp.szTagsTable, nDataType, nLogType, nID);
//				TRACE("%s\n", sSql.c_str());
				res = db.Query(sSql.c_str());
				bError = res.error();

				if(bError)
					m_lf.Error("Error updating Tags-Table: %s\n", db.LastError().c_str());
			}
		}
		else
		{
			sSql = formatString("insert into `%s` (`dataType`, `logType`, `path`) values(%d, %d, '%s')", m_dlp.szTagsTable, nDataType, nLogType, pszVarPath);
			res = db.Query(sSql.c_str());
			bError = res.error();

			if(bError)
				m_lf.Error("DB Error: %s\n", db.LastError().c_str());
			else
			{
				sSql = formatString("select `tagid` from `%s` where `path`='%s'", m_dlp.szTagsTable, pszVarPath);
				CMySqlResult res = db.Query(sSql.c_str());
				bool bError = res.error();

				if(bError)
					m_lf.Error("DB Error: %s\n", db.LastError().c_str());
				else if(res.RowCount() > 0)
				{
					MYSQL_ROW pRow = res.FetchRow();
					pszID = pRow[0];
					nID = strtoul(pszID, &pszEndptr, 10);
				}
			}
		}
	}

	return nID;
}

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

bool CDataLogger::TableFileExists(const char *pszTableName)
{
	char szDataFile[PATH_MAX];
	sprintf(szDataFile, "%s%s/%s.ibd", m_gv.szDataDir, m_dlp.szDBName, pszTableName);
	return !!FileExist(szDataFile);
}

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

int64_t CDataLogger::TableFileSize(const char *pszTableName)
{
	struct stat statbuf;
	char szDataFile[PATH_MAX];
	sprintf(szDataFile, "%s%s/%s.ibd", m_gv.szDataDir, m_dlp.szDBName, pszTableName);
	if(stat(szDataFile, &statbuf) == -1)
		return 0;
	return (int64_t)statbuf.st_size;
}

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

bool CDataLogger::QueryServerVariable(CMySqlDB &rdb, const char *pszVarname, CMySqlVar &val)
{
	if(!pszVarname || !*pszVarname)
	{
		m_lf.Error("CDataLogger::QueryServerVariable: No Variable name!\n");
		return false;
	}

	bool bError;
	std::string sSql;
	sSql = formatString("select @@%s;", pszVarname);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bError = res.error();

	if(!bError)
	{
		bError = true;

		do
		{
			my_ulonglong nRowCount		= res.RowCount();
			unsigned int nFldCount		= res.FieldCount();
			const MYSQL_FIELD *pFields	= res.FetchFields();

			if(nRowCount != 1 || nFldCount != 1 || !pFields)
				break;

			MYSQL_ROW pRow = res.FetchRow();

			if(!pRow || !*pRow || !**pRow)
				break;

			bError = !val.FromField(pFields[0], pRow[0]);
		}
		while(false);
	}

	return !bError;
}

/////////////////////////////////////////////////////////////////////////////
#if 0
bool CDataLogger::QueryStatusVariable(CMySqlDB &rdb, const char *pszVarname, CMySqlVar &val)
{
	if(!pszVarname || !*pszVarname)
	{
		m_lf.Error("CDataLogger::QueryStatusVariable: No Variable name!\n");
		return false;
	}

	bool bError;
	std::string sSql;
	sSql = formatString("show session status like '%s'", pszVarname);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bError = res.error();

	if(!bError)
	{
		bError = true;

		do
		{
			my_ulonglong nRowCount		= res.RowCount();
			unsigned int nFldCount		= res.FieldCount();
			const MYSQL_FIELD *pFields	= res.FetchFields();

			if(nRowCount != 1 || nFldCount != 2 || !pFields)
				break;

			MYSQL_ROW pRow = res.FetchRow();

			if(!pRow || !pRow[1] || !*pRow[1])
				break;

			bError = !val.FromField(pFields[1], pRow[1]);
		}
		while(false);
	}

	return !bError;
}
#endif
/////////////////////////////////////////////////////////////////////////////

bool CDataLogger::ReadGlobalOptions(CMySqlDB &rdb)
{
	CMySqlVar val;
	memset(&m_gv, 0, sizeof(m_gv));

	if(QueryServerVariable(rdb, "datadir", val))
	{
		memset(m_gv.szDataDir, 0, sizeof(m_gv.szDataDir));
		if(!val.IsNull())
			val.CopyStrVal(m_gv.szDataDir, sizeof(m_gv.szDataDir) - 1, 0);
	}

	if(QueryServerVariable(rdb, "innodb_file_per_table", val))
	{
		if(!val.IsNull())
			m_gv.bInnoDbFilePerTable = val;
	}

	if(QueryServerVariable(rdb, "innodb_strict_mode", val))
	{
		if(!val.IsNull())
			m_gv.bInnoDbIsStrictMode = val;
	}

	if(QueryServerVariable(rdb, "innodb_file_format", val))
	{
		memset(m_gv.szInnoDbFileFormat, 0, sizeof(m_gv.szInnoDbFileFormat));
		if(!val.IsNull())
		{
			val.CopyStrVal(m_gv.szInnoDbFileFormat, sizeof(m_gv.szInnoDbFileFormat) - 1, 0);
			m_gv.bInnoDbIsBarracuda = !strcasecmp(m_gv.szInnoDbFileFormat, "barracuda");
		}
	}

	return true;
}

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

void CDataLogger::SizeGuardTrigger(time_t ts)
{
	if(m_bSGConfigured && m_tidSGThread)
	{
		::pthread_mutex_lock(&m_condmtx1);

		if(	!m_bSGInProgress &&
			!SizeGuardDayWorkDone(ts))
		{
			unsigned long long nMnUTC = (unsigned long long)_MIDNIGHT_TIMESTAMP_UTC(ts);

			if(SizeGuardLastPassRead() != nMnUTC)
			{
				m_nSGCurPassUTC = ts;
				::pthread_cond_signal(&m_cond1);
			}
			else
				m_nSGLastPassUTC = nMnUTC;
		}

		::pthread_mutex_unlock(&m_condmtx1);
	}
}

/////////////////////////////////////////////////////////////////////////////
#if 0
bool CDataLogger::QueryTableSizes(CMySqlDB &rdb, const char *pszTableName)
{
	bool bError;
	char *pszEndptr;
	MYSQL_ROW pRow;
    uint64_t nDataLength = 0, nDataFree = 0, nIndexLength = 0, nFreeExtents, nTotalExtents;
	std::string sSql;
	sSql = formatString("select `DATA_LENGTH`, `INDEX_LENGTH`, `DATA_FREE` from `information_schema`.`TABLES` where (`TABLE_SCHEMA` = '%s') AND (`TABLE_NAME` = '%s');", m_dlp.szDBName, pszTableName);

	CMySqlResult res = rdb.Query(sSql.c_str());
	bError = res.error();

	if(bError)
	{
		m_lf.Error("CDataLogger::QueryTableSizes: DB Error: %s\n", rdb.LastError().c_str());
		return false;
	}

	pRow = res.FetchRow();
	nDataLength = strtoull(pRow[0], &pszEndptr, 10);
	nIndexLength = strtoull(pRow[1], &pszEndptr, 10);
	nDataFree = strtoull(pRow[2], &pszEndptr, 10);

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

	sSql = formatString("select `FREE_EXTENTS`, `TOTAL_EXTENTS`, `DATA_FREE` from `information_schema`.`FILES` where `FILE_NAME` like '%%demo/%s.ibd';", pszTableName);
	CMySqlResult res2 = rdb.Query(sSql.c_str());
	bError = res2.error();

	if(bError)
	{
		m_lf.Error("CDataLogger::QueryTableSizes: DB Error: %s\n", rdb.LastError().c_str());
		return false;
	}

	pRow = res2.FetchRow();
	nFreeExtents = strtoull(pRow[0], &pszEndptr, 10);
	nTotalExtents = strtoull(pRow[1], &pszEndptr, 10);
	int64_t nFileSize = TableFileSize(pszTableName);

    m_lf.Info("Data length: %ju, Data free: %ju, Total extents: %ju, Free extents: %ju, File size: %ju.\n", nDataLength + nIndexLength, nDataFree, nTotalExtents, nFreeExtents, nFileSize);
    TRACE("Data length: %ju, Data free: %ju, Total extents: %ju, Free extents: %ju, File size: %ju.\n", nDataLength + nIndexLength, nDataFree, nTotalExtents, nFreeExtents, nFileSize);

	return true;
}
#endif
/////////////////////////////////////////////////////////////////////////////

bool CDataLogger::CheckTable(CMySqlDB &rdb, const char *pszTableName, bool &bExists, bool &bUseResortTable, bool &bIsInnoDB)
{
	bool bError;
	std::string sSql;
	const char *pszEngine = NULL;
	uint64_t nDataLength = 0, nIndexLength = 0;
	int64_t nFileSize = -1;
	bExists = bUseResortTable = false;
	bIsInnoDB = false;

	sSql = formatString("select `ENGINE`, `DATA_LENGTH`, `INDEX_LENGTH` from `information_schema`.`TABLES` where (`TABLE_SCHEMA` = '%s') AND (`TABLE_NAME` = '%s')", m_dlp.szDBName, pszTableName);
	CMySqlResult res1 = rdb.Query(sSql.c_str());
	bError = res1.error();

	if(bError)
	{
		m_lf.Error("CDataLogger::CheckTables: DB Error: %s\n", rdb.LastError().c_str());
		return false;
	}
	else
	{
		my_ulonglong nCount = res1.RowCount();

		if(nCount > 0)
		{
			MYSQL_ROW pRow = res1.FetchRow();
			unsigned int nFc = res1.FieldCount();
			const MYSQL_FIELD *pFields = res1.FetchFields();
			CMySqlVar vals[3];

			bExists = true;

			for(unsigned int i = 0; i < nFc; i++)
			{
				vals[i].FromField(pFields[i], pRow[i]);

				if(!vals[i].FieldnameCmp("ENGINE"))
				{
					pszEngine = vals[i].StrVal();
					bIsInnoDB = !strcasecmp(pszEngine, "InnoDB");
				}
				else if(!vals[i].FieldnameCmp("DATA_LENGTH"))
					nDataLength = (uint64_t)vals[i];
				else if(!vals[i].FieldnameCmp("INDEX_LENGTH"))
					nIndexLength = (uint64_t)vals[i];
			}

			if(bIsInnoDB && m_gv.bInnoDbFilePerTable)
			{
				if(!TableFileExists(pszTableName))
					m_lf.Warning("'innodb_file_per_table' configured, but table '%s' does not have a data file!\n", pszTableName);
				else
					nFileSize = TableFileSize(pszTableName);
			}

			if(nFileSize >= 0)
				m_lf.Info("Found info on table '%s': Engine: '%s', Data size: %s, Index size: %s, File size: %s\n", pszTableName, pszEngine ? pszEngine : "unknown", strFormatByteSize(nDataLength, 1).c_str(), strFormatByteSize(nIndexLength, 1).c_str(), strFormatByteSize(nFileSize, 1).c_str());
			else
				m_lf.Info("Found info on table '%s': Engine: '%s', Data size: %s, Index size: %s\n", pszTableName, pszEngine ? pszEngine : "unknown", strFormatByteSize(nDataLength, 1).c_str(), strFormatByteSize(nIndexLength, 1).c_str());

			if(!pszEngine)
				m_lf.Warning("Failed to determine the Database engine of Table '%s'!\n", pszTableName);
			else if(strcmp(pszEngine, _DL_DATABSE_ENGINE_NAME))
				m_lf.Warning("Current Database engine of Table '%s' is '%s'! Please consider to migrate to '%s'!\n", pszTableName, pszEngine, _DL_DATABSE_ENGINE_NAME);

			sSql = formatString("select * from `%s` limit 0, 10", pszTableName);
			CMySqlResult res2 = rdb.Query(sSql.c_str()); // try to query table

			if(res2.error())	// if the query results in an error, the table may be corrupt!
			{
				char szTblName[256];
				m_lf.Error("Failed to query table '%s': %s\n", pszTableName, rdb.LastError().c_str());

				// if the table is corrupt, try to rename it and create a new instane of the original table.
				sprintf(szTblName, "%s_corrupt_%08zX", pszTableName, (size_t)time(NULL));
				sSql = formatString("rename table `%s` to `%s`", pszTableName, szTblName);
				CMySqlResult res3 = rdb.Query(sSql.c_str()); // try to rename the corrupt table

				if(res3.error())
				{
					// if the renaming fails, try to create a new working table with a different name
					m_lf.Error("Failed to rename table '%s': %s\n", pszTableName, rdb.LastError().c_str());
					bExists = false;
					bUseResortTable = true;
				}
				else
				{
					m_lf.Warning("Renamed table '%s' to '%s'\n", pszTableName, szTblName);
					bExists = false;
				}
			}
		}
	}

	return !bError;
}

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

bool CDataLogger::InitDatabase(bool bEnforceCreate)
{
	CMySqlDB db;
	time_t nMaxLogTimestamp = m_nLastLogTimestamp = 0;
	bool bExists, bUseResortTable, bDummy;

	m_lf.Info("Connecting to Database server @ 'localhost'.\n");

	if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, NULL))
	{
		m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", db.LastError().c_str());
		return false;
	}

	m_lf.Info("Success!\n");

	if(!ReadGlobalOptions(db))
		return false;

	if(!m_gv.bInnoDbIsBarracuda || !m_gv.bInnoDbFilePerTable || !m_gv.bInnoDbIsStrictMode)
		m_lf.Warning("InnoDB file format: '%s', InnoDB table space: '%s', InnoDB strict mode: %s.\n", m_gv.szInnoDbFileFormat, m_gv.bInnoDbFilePerTable ? "File per table" : "System", m_gv.bInnoDbIsStrictMode ? "yes" : "no");
	else
		m_lf.Info("InnoDB file format: '%s', InnoDB table space: '%s', InnoDB strict mode: %s.\n", m_gv.szInnoDbFileFormat, m_gv.bInnoDbFilePerTable ? "File per table" : "System", m_gv.bInnoDbIsStrictMode ? "yes" : "no");

	if(!CreateDatabase(db, bEnforceCreate))
		return false;

	m_lf.Info("Opening Database '%s'.\n", m_dlp.szDBName);

	if(db.SelectDB(m_dlp.szDBName))
	{
		m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", db.LastError().c_str());
		return false;
	}

	m_lf.Info("Success!\n");

	if(!CheckTable(db, m_dlp.szTagsTable, bExists, bUseResortTable, bDummy))
		return false;

	if(!bExists)
	{
		if(bUseResortTable)
		{
			char szOrigTable[_DL_MAX_TABLE_NAME_LENGTH];
			memcpy(szOrigTable, m_dlp.szTagsTable, sizeof(szOrigTable));
			sprintf(m_dlp.szTagsTable, "%s_resort_%08zX", szOrigTable, (size_t)time(NULL));
			m_lf.Warning("Resorting to work-around table '%s'.\n", m_dlp.szTagsTable);
		}

		if(!CreateTagsTable(db))
			return false;
	}
	
	if(!AlterTagsTable(db)) // 23.10.2020, extend logtype enum if not up to date
		return false;

	if(!CheckTable(db, m_dlp.szLogsTable, bExists, bUseResortTable, m_gv.bLogsTblIsInnoDB))
		return false;

	if(!bExists)
	{
		if(bUseResortTable)
		{
			char szOrigTable[_DL_MAX_TABLE_NAME_LENGTH];
			memcpy(szOrigTable, m_dlp.szLogsTable, sizeof(szOrigTable));
			sprintf(m_dlp.szLogsTable, "%s_resort_%08zX", szOrigTable, (size_t)time(NULL));
			m_lf.Warning("Resorting to work-around table '%s'.\n", m_dlp.szLogsTable);
		}

		if(!CreateLogsTable(db))
			return false;

#if _DL_DATABSE_ENGINE == _DL_DATABASE_ENGINE_INNODB
		m_gv.bLogsTblIsInnoDB = true;
#endif	//	_DL_DATABSE_ENGINE == _DL_DATABASE_ENGINE_INNODB
	}

	if(m_bSGConfigured)
	{
		if(!(m_bSGHasSizeLimitPrerequisites = m_gv.bLogsTblIsInnoDB && m_gv.bInnoDbFilePerTable && TableFileExists(m_dlp.szLogsTable)))
		{
			m_lf.Warning("SG: Database configuration: InnoDB: %s, File-per-table tablespace: %s\n", m_gv.bLogsTblIsInnoDB ? "yes" : "no", m_gv.bInnoDbFilePerTable ? "yes" : "no");
			m_lf.Warning("SG: The current database configuration does not support monitoring of size limits. Size guard will operate on age limits only!\n");
		}

		::pthread_mutex_lock(&m_condmtx1);
		if(::pthread_create(&m_tidSGThread, NULL, &CDataLogger::SizeGuardWorker, reinterpret_cast<void*>(this)))
		{
			m_tidSGThread = 0;
			m_lf.Error("SG: Failed to start size monitoring thread!\n");
			::pthread_mutex_unlock(&m_condmtx1);
			return false;
		}

		::pthread_cond_wait(&m_cond1, &m_condmtx1);
		::pthread_mutex_unlock(&m_condmtx1);
	}

	if(!CreateLogsBDTable(db))
		return false;

	if(((nMaxLogTimestamp = GetLastLogTimestamp(db)) != _INVALID_TIMESTAMP_VALUE))
		m_nLastLogTimestamp = nMaxLogTimestamp;

	m_lf.Info("Database initialization complete.\n");

	return true;
}

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

bool CDataLogger::CreateDatabase(CMySqlDB &rdb, bool bEnforceCreate)
{
	bool bError = false;
	std::string sSql;

	if(bEnforceCreate)
	{
		sSql = formatString("drop database if exists `%s`", m_dlp.szDBName);
		CMySqlResult res = rdb.Query(sSql.c_str());
		bError = res.error();

		if(bError)
			m_lf.Error("CDataLogger::CreateDatabase: DB Error: %s\n", rdb.LastError().c_str());
	}

	if(!bError)
	{
		sSql = formatString("create database if not exists `%s`", m_dlp.szDBName);
		CMySqlResult res = rdb.Query(sSql.c_str());
		bError = res.error();

		if(bError)
			m_lf.Error("CDataLogger::CreateDatabase: DB Error: %s\n", rdb.LastError().c_str());
	}

	return !bError;
}

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

bool CDataLogger::CreateTagsTable(CMySqlDB &rdb)
{
	std::string sSql;

	const char *pszFormat =
	"CREATE TABLE IF NOT EXISTS `%s` (" \
	" `tagid` smallint(6) unsigned NOT NULL AUTO_INCREMENT," \
	" `dataType` enum('bool', 'I1', 'UI1', 'I2', 'UI2', 'I4', 'UI4', 'I8', 'UI8', 'float', 'double', 'string') NOT NULL," \
	" `logType` enum('IC', 'IU', 'VC', 'VU', 'ICR', 'IUR', 'VCR', 'VUR') NOT NULL," \
	" `path` varchar(%u) NOT NULL," \
	" PRIMARY KEY (`tagid`, `dataType`, `logType`)," \
	" KEY `path` (`path`)" \
	") ENGINE=%s DEFAULT CHARSET=ascii";

	m_lf.Info("Creating Table '%s' (if not exists) using engine '%s'.\n", m_dlp.szTagsTable, _DL_DATABSE_ENGINE_NAME);

	sSql = formatString(pszFormat, m_dlp.szTagsTable, _DL_MAX_VARPATH_LENGTH, _DL_DATABSE_ENGINE_NAME);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bool bError = res.error();

	if(bError)
		m_lf.Error("CDataLogger::CreateTagsTable: DB Error: %s\n", rdb.LastError().c_str());
	else
		m_lf.Info("Success!\n");

	return !bError;
}

bool CDataLogger::AlterTagsTable(CMySqlDB &rdb)
{
	std::string sSql = formatString("ALTER TABLE `%s` MODIFY COLUMN `logType` enum('IC','IU','VC','VU','ICR','IUR','VCR','VUR') NOT NULL", m_dlp.szTagsTable);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bool bError = res.error();
	if(bError)
		m_lf.Error("CDataLogger::AlterTagsTable: DB Error: %s\n", rdb.LastError().c_str());
	return !bError;
}

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

bool CDataLogger::CreateLogsTable(CMySqlDB &rdb)
{
	std::string sSql;

	const char *pszFormat =
#if _USE_MODIFIED_INDEX
	"CREATE TABLE IF NOT EXISTS `%s` (" \
	"	`tagid` smallint(6) unsigned NOT NULL," \
	"	`tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
	"	`value` double," \
	"	`valueMin` double DEFAULT NULL," \
	"	`valueMax` double DEFAULT NULL," \
	"	PRIMARY KEY (`tslog`, `tagid`)," \
	"	KEY `taglog` (`tagid`, `tslog`)" \
	") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";
#else //	_USE_MODIFIED_INDEX
	"CREATE TABLE IF NOT EXISTS `%s` (" \
	"	`tagid` smallint(6) unsigned NOT NULL," \
	"	`tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
	"	`value` double," \
	"	`valueMin` double DEFAULT NULL," \
	"	`valueMax` double DEFAULT NULL," \
	"	PRIMARY KEY (`tagid`, `tslog`)," \
	"	KEY `tslog` (`tslog`)" \
	") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";
#endif	//	_USE_MODIFIED_INDEX

	m_lf.Info("Creating Table '%s' if not exists using engine '%s'.\n", m_dlp.szLogsTable, _DL_DATABSE_ENGINE_NAME);

	sSql = formatString(pszFormat, m_dlp.szLogsTable, _DL_DATABSE_ENGINE_NAME);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bool bError = res.error();

	if(bError)
		m_lf.Error("CDataLogger::CreateLogsTable: DB Error: %s\n", rdb.LastError().c_str());
	else
		m_lf.Info("Success!\n");

	return !bError;
}

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

bool CDataLogger::CreateLogsBDTable(CMySqlDB &rdb)
{
	std::string sSql;

	const char *pszFormat =
	"CREATE TABLE IF NOT EXISTS `%s` (" \
	"	`id` int unsigned unsigned NOT NULL AUTO_INCREMENT," \
	"	`tagid` smallint(6) unsigned NOT NULL," \
	"	`tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
	"	`value` double," \
	"	`valueMin` double DEFAULT NULL," \
	"	`valueMax` double DEFAULT NULL," \
	"	PRIMARY KEY (`id`)," \
	"	KEY `tslog` (`tslog`)," \
	"	KEY `tagid` (`tagid`)" \
	") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";

	m_lf.Info("Creating Table '%s' if not exists using engine '%s'.\n", m_dlp.szLogsTableBD, _DL_DATABSE_ENGINE_NAME);

	sSql = formatString(pszFormat, m_dlp.szLogsTableBD, _DL_DATABSE_ENGINE_NAME);
	CMySqlResult res = rdb.Query(sSql.c_str());
	bool bError = res.error();

	if(bError)
		m_lf.Error("CDataLogger::CreateLogsBDTable: DB Error: %s\n", rdb.LastError().c_str());
	else
		m_lf.Info("Success!\n");

	return !bError;
}

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

bool CDataLogger::Log(unsigned long nTagID, double fValue, double fMin, double fMax, time_t nTimestamp, int nIndex, LogTypes lt, bool bNull, bool bNoBadDateCheck)
{
	bNull = bNull || !_IsLoggable(fValue) || !_IsLoggable(fMin) || !_IsLoggable(fMax);

	DL_LOG_ENTRY log;
	log.nTagID		= nTagID;
	log.nTimestamp	= nTimestamp;
	log.fValue		= fValue;
	log.fMin		= fMin;
	log.fMax		= fMax;
	log.nIndex		= nIndex;
	log.lt			= lt;
	log.bNull		= bNull;

	if(bNoBadDateCheck || (nTimestamp > m_nLastLogTimestamp))
		m_logs.push_back(log);
	else
		m_logsBD.push_back(log);

	return true;
}

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

time_t CDataLogger::GetLastLogTimestamp(CMySqlDB &rdb)
{
	time_t nTs = _INVALID_TIMESTAMP_VALUE;
	std::string sSql;

	do
	{
		sSql = formatString("select unix_timestamp(max(`tslog`)) from `%s`", m_dlp.szLogsTable);
		CMySqlResult res = rdb.Query(sSql.c_str());

		if(res.error())
		{
			m_lf.Error("CDataLogger::GetLastLogTimestamp: DB Error: %s\n", rdb.LastError().c_str());
			break;
		}
		else
		{
			CMySqlVar val;
			my_ulonglong nRowCount		= res.RowCount();
			unsigned int nFldCount		= res.FieldCount();
			const MYSQL_FIELD *pFields	= res.FetchFields();

			if(nRowCount != 1 || nFldCount != 1 || !pFields)
			{
//				m_lf.Error("CDataLogger::GetLastLogTimestamp: Unexpected error!\n");
				break;
			}

			MYSQL_ROW pRow = res.FetchRow();

			if(	!pRow ||
				!val.FromField(pFields[0], pRow[0]))
			{
//				m_lf.Error("CDataLogger::GetLastLogTimestamp: Unexpected error!\n");
				break;
			}
			
			nTs = (time_t)(uint64_t)val;
		}
	}
	while(false);

	return nTs;
}

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

bool CDataLogger::Flush(time_t nTimestamp)
{
	std::string strErr;
	::pthread_mutex_lock(&m_condmtx1);
	bool bGoodBadTransition = false, bSgInProg = m_bSGInProgress;
	::pthread_mutex_unlock(&m_condmtx1);

	time_t nMaxLogTimestamp = 0, nFirstGoodTimestamp = 0;
	size_t nCountValidDate = m_logs.size();
	size_t nCountBadDate = m_logsBD.size();
	
	if(!nCountValidDate && !nCountBadDate)
		return true;	// nothing to do
		
	TRACE("Trying to flush %zu logs ...\n", nCountValidDate + nCountBadDate);
	
	if(bSgInProg && nCountValidDate)
	{
		TRACE("Flush: SG in progress - flushing of %zu logs deferred!\n", nCountValidDate);
		m_lf.Info("Flush: SG in progress - flushing of %zu logs deferred!\n", nCountValidDate);
		nCountValidDate = 0; // size guard operates on the valid-date-logs-table only, so we can safely flush invalid-date-logs
		
		if(!nCountBadDate)
			return true; // no error
	}
	
	if(!m_bBadDateLogsDetected && nCountBadDate)
	{
		m_bBadDateLogsDetected = true;
		bGoodBadTransition = true;
	}
	else if(m_bBadDateLogsDetected && !nCountBadDate)
	{
		m_bBadDateLogsDetected = false;
		bGoodBadTransition = true;
	}

	CMySqlDB db;
	std::string strUnlock, strSaveFgnKey, strDisableFgnKey, strRestoreFgnKey, strSaveTZ, strSetTZ, strRestoreTZ;
	bool bError = false;
	std::string sSql;

	if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, m_dlp.szDBName))
	{
		m_lf.Error("CDataLogger::Flush: DB Error: %s\n", db.LastError().c_str());
		return false;
	}

	strUnlock = formatString("unlock tables");
	strSaveFgnKey = formatString("set @old_foreign_key_checks = @@foreign_key_checks");
	strDisableFgnKey = formatString("set foreign_key_checks = 0");
	strRestoreFgnKey = formatString("set foreign_key_checks = @old_foreign_key_checks");
	strSaveTZ = formatString("set @old_time_zone = @@time_zone");
	strSetTZ = formatString("set time_zone = '+00:00'");
	strRestoreTZ = formatString("set time_zone = @old_time_zone");

	db.Query(strSaveTZ.c_str());
	db.Query(strSetTZ.c_str());
	db.Query(strSaveFgnKey.c_str());
	db.Query(strDisableFgnKey.c_str());

	if(nCountValidDate > 0)
	{
		std::string strSql, strLock;
		strLock = formatString("lock tables `%s` write", m_dlp.szLogsTable);
		strSql.reserve(130 + 111 * nCountValidDate);

		auto itFirst = m_logs.begin();
		const DL_LOG_ENTRY &rle0 = *itFirst;
		if(nMaxLogTimestamp < rle0.nTimestamp)
			nMaxLogTimestamp = rle0.nTimestamp;
		nFirstGoodTimestamp = rle0.nTimestamp;

		if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle0.lt))
		{
			if(!rle0.bNull)
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp, rle0.fValue, rle0.fMin, rle0.fMax);
			else
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp);
		}
		else
		{
			if(!rle0.bNull)
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp, rle0.fValue);
			else
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp);
		}
		strSql = sSql;

		for(++itFirst; itFirst < m_logs.end(); itFirst++)
		{
			const DL_LOG_ENTRY &rle = *itFirst;
			if(nMaxLogTimestamp < rle.nTimestamp)
				nMaxLogTimestamp = rle.nTimestamp;

			if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle.lt))
			{
				if(!rle.bNull)
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", rle.nTagID, rle.nTimestamp, rle.fValue, rle.fMin, rle.fMax);
				else
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
			}
			else
			{
				if(!rle.bNull)
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", rle.nTagID, rle.nTimestamp, rle.fValue);
				else
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
			}
			strSql += sSql;
		}

		if(TryLock())
		{
			db.Query(strLock.c_str());
			CMySqlResult res = db.Query(strSql.c_str());
			if((bError = res.error()))
				strErr = db.LastError();
			db.Query(strUnlock.c_str());
			Unlock();

			if(bError)
				m_lf.Error("CDataLogger::Flush: DB Error: %s\n", strErr.c_str());
			else
				m_nLastLogTimestamp = nMaxLogTimestamp;

			m_logs.clear();
		}
	}

	if(bGoodBadTransition)
	{
		if(m_bBadDateLogsDetected)
		{
			bool bErr;
			char szTs[64];
			unsigned long long nNextAIVal = 0;
			sSql = formatString("select auto_increment from `information_schema`.`TABLES` where `TABLE_SCHEMA` = '%s' and `TABLE_NAME` = '%s';", m_dlp.szDBName, m_dlp.szLogsTableBD);
			CMySqlResult res = db.Query(sSql.c_str());

			if(!(bErr = res.error()))
			{
				CMySqlVar val;
				my_ulonglong nRowCount		= res.RowCount();
				unsigned int nFldCount		= res.FieldCount();
				const MYSQL_FIELD *pFields	= res.FetchFields();

				do
				{
					if(nRowCount != 1 || nFldCount != 1 || !pFields)
						break;

					MYSQL_ROW pRow = res.FetchRow();

					if(!pRow || !*pRow || !**pRow)
						break;

					if(!val.FromField(pFields[0], pRow[0]))
						break;

					nNextAIVal = val;
				}
				while(false);
			}
			
			Timestamp2String(m_nLastLogTimestamp, szTs, sizeof(szTs));

			if(nNextAIVal)
			{
				TRACE("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s` starting with `id` %llu.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD, nNextAIVal);
				m_lf.Warning("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s` starting with `id` %llu.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD, nNextAIVal);
			}
			else
			{
				TRACE("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s`.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD);
				m_lf.Warning("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s`.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD);
			}
		}
		else
		{
			bool bErr;
			char szTs[64];
			unsigned long long nLastID = 0;
			sSql = formatString("select max(`id`) from `%s`.`%s`;", m_dlp.szDBName, m_dlp.szLogsTableBD);
			CMySqlResult res = db.Query(sSql.c_str());

			if(!(bErr = res.error()))
			{
				CMySqlVar val;
				my_ulonglong nRowCount		= res.RowCount();
				unsigned int nFldCount		= res.FieldCount();
				const MYSQL_FIELD *pFields	= res.FetchFields();

				do
				{
					if(nRowCount != 1 || nFldCount != 1 || !pFields)
						break;

					MYSQL_ROW pRow = res.FetchRow();

					if(!pRow || !*pRow || !**pRow)
						break;

					if(!val.FromField(pFields[0], pRow[0]))
						break;

					nLastID = val;
				}
				while(false);
			}
			
			Timestamp2String(nFirstGoodTimestamp, szTs, sizeof(szTs));

			if(nLastID)
			{
				TRACE("Flush: Transition invalid -> valid date detected! The last `id` in table `%s` was %llu. Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTableBD, nLastID, m_dlp.szLogsTable, szTs);
				m_lf.Warning("Flush: Transition invalid -> valid date detected! The last `id` in table `%s` was %llu. Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTableBD, nLastID, m_dlp.szLogsTable, szTs);
			}
			else
			{
				TRACE("Flush: Transition invalid -> valid date detected! Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTable, szTs);
				m_lf.Warning("Flush: Transition invalid -> valid date detected! Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTable, szTs);
			}
		}
	}

	if(nCountBadDate > 0)
	{
		std::string strSql, strLock;
		strLock = formatString("lock tables `%s` write", m_dlp.szLogsTableBD);
		strSql.reserve(130 + 111 * nCountBadDate);

		auto itFirst = m_logsBD.begin();
		const DL_LOG_ENTRY &rle0 = *itFirst;

		if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle0.lt))
		{
			if(!rle0.bNull)
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp, rle0.fValue, rle0.fMin, rle0.fMax);
			else
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp);
		}
		else
		{
			if(!rle0.bNull)
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp, rle0.fValue);
			else
				sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp);
		}
		strSql = sSql;

		for(++itFirst; itFirst < m_logsBD.end(); itFirst++)
		{
			const DL_LOG_ENTRY &rle = *itFirst;

			if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle.lt))
			{
				if(!rle.bNull)
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", rle.nTagID, rle.nTimestamp, rle.fValue, rle.fMin, rle.fMax);
				else
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
			}
			else
			{
				if(!rle.bNull)
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", rle.nTagID, rle.nTimestamp, rle.fValue);
				else
					sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
			}
			strSql += sSql;
		}

		db.Query(strLock.c_str());
		CMySqlResult res = db.Query(strSql.c_str());
		if((bError = res.error()))
			strErr = db.LastError();
		db.Query(strUnlock.c_str());

		if(bError)
			m_lf.Error("CDataLogger::Flush: DB Error: %s\n", strErr.c_str());

		m_logsBD.clear();
	}

	db.Query(strRestoreFgnKey.c_str());
	db.Query(strRestoreTZ.c_str());

	return !bError;
}

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

unsigned long long CDataLogger::SizeGuardLastPassRead(void)
{
	char szPath[PATH_MAX];
	sprintf(szPath, "%s/%s", m_szAppDir, _SIZE_GUARD_TIMESTAMP_FILE_NAME);
	FILE *pf = fopen(szPath, "rb");
	unsigned long long ts = 0;

	if(pf)
	{
        if(fread(&ts, sizeof(ts), 1, pf) != 1)
			ts = 0;
		fclose(pf);
	}

	return ts;
}

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

void CDataLogger::SizeGuardLastPassWrite(unsigned long long ts)
{
	char szPath[PATH_MAX];
	sprintf(szPath, "%s/%s", m_szAppDir, _SIZE_GUARD_TIMESTAMP_FILE_NAME);
	FILE *pf = fopen(szPath, "wb");

	if(pf)
	{
		fwrite(&ts, sizeof(ts), 1, pf);
		fclose(pf);
	}
}

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

size_t CDataLogger::Timestamp2String(time_t t, char *pszBuffer, size_t nCbBuffer)
{
	const struct tm * ptm = gmtime(&t);
	return strftime(pszBuffer, nCbBuffer, "%F %T", ptm);
}

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

const char* CDataLogger::Ns2String(unsigned long long nNs, char *pszBuffer, size_t nCbBuffer)
{
	*pszBuffer = '\0';

	if(nNs < 1000)
		snprintf(pszBuffer, nCbBuffer, "%llu ns", nNs);
	else if(nNs < 1000000)
		snprintf(pszBuffer, nCbBuffer, "%.1f us", (double)nNs / 1000.0);
	else
		return Ms2String((double)nNs / 1000000.0, pszBuffer, nCbBuffer);

	return pszBuffer;
}

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

const char* CDataLogger::Ms2String(double fMs, char *pszBuffer, size_t nCbBuffer)
{
	*pszBuffer = '\0';

	if(fMs < 1000.0)
		snprintf(pszBuffer, nCbBuffer, "%.2f ms", fMs);
	else if(fMs < 60000.0)
		snprintf(pszBuffer, nCbBuffer, "%.2f sec", fMs / 1000.0);
	else if(fMs < 3600000.0)
		snprintf(pszBuffer, nCbBuffer, "%.2f min", fMs / 60000.0);
	else
		snprintf(pszBuffer, nCbBuffer, "%.2f h", fMs / 3600000.0);
	
	return pszBuffer;
}

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

bool CDataLogger::DoSizeGuard(void)
{
	if(!m_bSGConfigured)
		return true;

	CMySqlDB db;
	bool bError;
	MYSQL_ROW pRow;
	unsigned long long nElapsed;
	char szT1[32], szT2[32], szMs[32];
	CProcessClock pc;
    CMySqlVar vMinTs, vCountDelete;
    int64_t nMinTs, nTsCurMidnightUTC, nTsDeleteUpTo;
	std::string sSql;

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

	nTsCurMidnightUTC = _MIDNIGHT_TIMESTAMP_UTC(m_nSGCurPassUTC);

	if(SizeGuardDayWorkDone(nTsCurMidnightUTC))
		return true;
	m_nSGLastPassUTC = nTsCurMidnightUTC;

	if(SizeGuardLastPassRead() == (unsigned long long)(nTsCurMidnightUTC))
		return true;
	SizeGuardLastPassWrite(nTsCurMidnightUTC);

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

	if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, NULL))
	{
		m_lf.Error("CDataLogger::DoSizeGuard(%d): - DB Error: %s\n", __LINE__, db.LastError().c_str());
		return false;
	}

	sSql = formatString("select unix_timestamp(min(`tslog`)) from `%s`.`%s`;", m_dlp.szDBName, m_dlp.szLogsTable);
	m_lf.Info("SG: \"%s\".\n", sSql.c_str());

	pc.ClockTrigger();
	CMySqlResult res = db.Query(sSql.c_str());
	nElapsed = pc.ClockGetElapsed();
	
	bError = res.error();

	if(bError)
	{
		m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
		return false;
	}

	if(!(pRow = res.FetchRow()))
	{
		m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
		return false;
	}

	my_ulonglong nRowCount		= res.RowCount();
	unsigned int nFldCount		= res.FieldCount();
	const MYSQL_FIELD *pFields	= res.FetchFields();

	if(nRowCount != 1 || nFldCount != 1)
	{
		m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
		return false;
	}

	if(!vMinTs.FromField(pFields[0], pRow[0]))
	{
		m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
		return false;
	}

	nMinTs = vMinTs;

	m_lf.Info("SG: operation completed in %s. Min. timestamp: %ju\n", Ns2String(nElapsed, szMs, sizeof(szMs)), nMinTs);

	/////////////////////////////////////////////////////////////////////////
	// process max. size

	if(m_bSGHasSizeLimitPrerequisites && m_dlp.nMaxSize)
	{
		unsigned long long nFileSize		= TableFileSize(m_dlp.szLogsTable);
		unsigned long long nGuardThreshold	= m_dlp.nMaxSize * (100 - _FILE_SIZE_DELETE_MARGIN_PERCENT) / 100;

		m_lf.Info("SG: File size: %s.\n", strFormatByteSize(nFileSize, 1).c_str());

		if(nFileSize > nGuardThreshold)
		{
			Timestamp2String(nTsCurMidnightUTC - _SECONDS_PER_DAY, szT1, sizeof(szT1));
			Timestamp2String(nTsCurMidnightUTC - 1, szT2, sizeof(szT2));
			sSql = formatString("select count(*) from `%s`.`%s` where `tslog` between '%s' and '%s';", m_dlp.szDBName, m_dlp.szLogsTable, szT1, szT2);
			m_lf.Info("SG: \"%s\"\n", sSql.c_str());

			pc.ClockTrigger();
			CMySqlResult res = db.Query(sSql.c_str());
			bError = res.error();

			if(bError)
			{
				m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
				return false;
			}
			else
			{
				nElapsed = pc.ClockGetElapsed();

				nRowCount	= res.RowCount();
				nFldCount	= res.FieldCount();
				pFields		= res.FetchFields();
				pRow		= res.FetchRow();

				if(pRow && pFields && nRowCount == 1 && nFldCount == 1 && vCountDelete.FromField(pFields[0], pRow[0]))
				{
					unsigned long long nCountDelete = (uint64_t)vCountDelete;
					m_lf.Info("SG: operation completed in %s. Estimated number of rows to delete: %llu.\n", Ns2String(nElapsed, szMs, sizeof(szMs)), nCountDelete);

					if(nCountDelete > 0)
					{
#if _USE_MODIFIED_INDEX
						sSql = formatString("delete from `%s`.`%s` limit %llu;", m_dlp.szDBName, m_dlp.szLogsTable, nCountDelete);
#else	//	_USE_MODIFIED_INDEX
						sSql = formatString("delete from `%s`.`%s` where `tslog` <= (select `tslog` from (select `tslog` from `demo`.`logs` order by `tslog` asc limit %llu, 1) x);", m_dlp.szDBName, m_dlp.szLogsTable, nCountDelete - 1);
#endif	//	_USE_MODIFIED_INDEX
						m_lf.Info("SG: triggered on size limit: \"%s\".\n", sSql.c_str());

						pc.ClockTrigger();
						if(db.Query(sSql.c_str()).error())
						{
							m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
							return false;
						}
						else
						{
							nElapsed = pc.ClockGetElapsed();
							m_lf.Info("SG: operation completed in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
						}
					}
				}
				else
				{
					m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
					return false;
				}
			}
		}
	}

	/////////////////////////////////////////////////////////////////////////
	// process max. age entries

	if(m_dlp.nMaxAge)
	{
		nTsDeleteUpTo = nTsCurMidnightUTC - m_dlp.nMaxAge * _SECONDS_PER_DAY;

		if(nMinTs <= nTsDeleteUpTo)
		{
			Timestamp2String(nTsDeleteUpTo, szT1, sizeof(szT1));
			sSql = formatString("delete from `%s`.`%s` where `tslog` < '%s';", m_dlp.szDBName, m_dlp.szLogsTable, szT1);
			m_lf.Info("SG: triggered on age limit: \"%s\".\n", sSql.c_str());

			pc.ClockTrigger();
			CMySqlResult res = db.Query(sSql.c_str());
			nElapsed = pc.ClockGetElapsed();
			bError = res.error();

			if(bError)
			{
				m_lf.Error("CDataLogger::DoSizeGuard(%d): DB Error: %s\n", __LINE__, db.LastError().c_str());
				return false;
			}
			else
			{
				m_lf.Info("SG: operation completed in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
			}
		}
	}

	return true;
}

/////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////
// size guard worker thread

void* CDataLogger::SizeGuardWorker(void* pParam)
{
	unsigned long long nElapsed;
	char szMs[32];
	CProcessClock pc;
	CDataLogger *pThis = reinterpret_cast<CDataLogger*>(pParam);

	::pthread_mutex_lock(&pThis->m_condmtx1);
	::pthread_cond_signal(&pThis->m_cond1);

	while(true)
	{
		pThis->m_bSGInProgress = false;
		::pthread_cond_wait(&pThis->m_cond1, &pThis->m_condmtx1);
		pThis->m_bSGInProgress = true;
		::pthread_mutex_unlock(&pThis->m_condmtx1);
		TRACE("Size guard start.\n");
		pThis->Lock();
		pc.ClockTrigger();
		pThis->DoSizeGuard();
		nElapsed = pc.ClockGetElapsed();
		pThis->Unlock();
        pThis->m_lf.Info("SG: finished in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
		TRACE("Size guard end - [%s].\n", szMs);
		::pthread_yield();
		::pthread_mutex_lock(&pThis->m_condmtx1);
	}

	return NULL;
}