ps_client_dbfsms.cpp

这个程序是用来操作Berkeley DB来使用

#include "PS_client_dbfsms.h"
#include "PS_client_conf.h"
#include <iostream>
#include <fstream>
#include <time.h>

#ifdef WIN32
#include <process.h>   
#else
#include <unistd.h>
#include <pthread.h>
#endif

//CPSDBFSMS CPSDBFSMS::m_dbms;

bool CPSDBFSMS::m_bStopFlag = false;
MMutex * CPSDBFSMS::m_pmutexStat = NULL;
MMutex * CPSDBFSMS::m_pmutexTmp = NULL;
std::vector<int> CPSDBFSMS::m_vecTime;

CPSDBFSMS::CPSDBFSMS() : m_pdbfs(NULL), m_pEnv(NULL), m_strStatFileName(NULL), m_pdbfsTmp(NULL)
{	
	memset(m_strTmpStatFile, 0, 256);

	m_vecStat.clear();

	m_vecstatTmp.clear();

	m_vecTime.clear();

	m_pmutexStat = NULL;

	m_pmutexTmp = NULL;
}

CPSDBFSMS::~CPSDBFSMS()
{
/*
#ifdef __TEMP__
	m_bStopFlag = true;
#ifdef WIN32
	if (WAIT_FAILED == ::WaitForSingleObject((HANDLE)m_pid, INFINITE))
	{
		GetLastError();
	}
	CloseHandle((HANDLE)m_pid);
#else
	if (0 !=pthread_join(m_pid, NULL) )
	{
		printf("error joining thread.");
	}
#endif
#endif
*/

	if (m_pmutexStat != NULL)
	{
		delete []m_pmutexStat;
		m_pmutexStat = NULL;
	}

	if (m_pmutexTmp != NULL)
	{
		delete []m_pmutexTmp;
		m_pmutexTmp = NULL;
	}

	m_vecStat.clear();

	m_vecstatTmp.clear();

	m_vecTime.clear();

	if (m_pdbfs != NULL)
	{
		delete []m_pdbfs;
		m_pdbfs = NULL;
	}

	if (m_pdbfsTmp != NULL)
	{
		delete []m_pdbfsTmp;
		m_pdbfsTmp = NULL;
	}
	
	if (m_strStatFileName != NULL)
	{
		delete []m_strStatFileName;
		m_strStatFileName = NULL;
	}

	if (m_pEnv != NULL)
	{
//		m_pEnv->mutex_free(m_pEnv, m_mutex);

		// Close the environment
		int ret_c = m_pEnv->close(m_pEnv, 0);
		if (ret_c != 0) 
		{
			printf("environment close failed: %s\n",
				db_strerror(ret_c));
		}
	}
}

int CPSDBFSMS::PS_dbfsms_init(char *normalConfFile, char *tmpConfFile, char *strEnv)
{
	if (NULL == normalConfFile)
	{
		return -1;
	}

	if (NULL == m_strStatFileName)
	{
		m_strStatFileName = new char[256];
		memset(m_strStatFileName, 0, 256);
	}

	strcpy(m_strStatFileName, normalConfFile);

	int ret, ret_c, i;
    u_int32_t env_flags;

	if (m_pEnv == NULL)
	{
		// Open the environment
		ret = db_env_create(&m_pEnv, 0);
		if (ret != 0)
		{
			printf("Error creating environment handle: %s\n",
				db_strerror(ret));

			m_pEnv = NULL;

			return -1;
		}

		// 删除环境文件有些系统下会出问题
/*		// 1
		m_pEnv->remove(m_pEnv, strEnv, DB_FORCE);

		// Open the environment
		ret = db_env_create(&m_pEnv, 0);
		if (ret != 0)
		{
			printf("Error creating environment handle: %s\n",
					db_strerror(ret));

			return -1;
		}

		// 2
		char strTemp[256] = {0};
		for (int i=1; i<5; i++)
		{
			memset(strTemp, 0, 256);
#ifdef WIN32
		sprintf(strTemp, "%s\\__db.00%d", strEnv, i);
#else
		sprintf(strTemp, "%s/__db.00%d", strEnv, i);
#endif
		unlink(strTemp);
		}
*/
	}
                                                                                                                     
    env_flags = DB_CREATE |    // Create the environment if it does not already exist.
		        DB_INIT_CDB |
//				DB_THREAD   |
//                DB_INIT_TXN  | // Initialize transactions
//                DB_INIT_LOCK | // Initialize locking. 
//                DB_INIT_LOG  | // Initialize logging 
                DB_INIT_MPOOL; // Initialize the in-memory cache.

    ret = m_pEnv->open(m_pEnv, strEnv, env_flags, 0);
    if (ret != 0)
	{
        printf("Error opening environment: %s\n",
            db_strerror(ret));

		 // Close the environment
		if (m_pEnv != NULL) 
		{
			ret_c = m_pEnv->close(m_pEnv, 0);
			if (ret_c != 0) 
			{
				printf("environment close failed: %s\n",
					db_strerror(ret_c));
				ret = ret_c;
			}
		}

		return -2;
    }

	if (-1 == mf_getStatFile())
	{
		return -3;
	}

	if (NULL == m_pdbfs)
	{
		m_pdbfs = new PSClientDBFS[m_vecStat.size()];
	}

	if (NULL == m_pmutexStat)
	{
		m_pmutexStat = new MMutex[m_vecStat.size()];
	}

	// set the current size
	for (i=0; i < m_vecStat.size(); i++)
	{
		m_pdbfs[i].PS_dbfs_setEnv(m_pEnv);
		if (-1 == m_pdbfs[i].PS_dbfs_init(m_vecStat[i].dbfile))
		{
			return -4;
		}

		PSClientDBFS::PSDBFSStat tmpDbfsStat;
		m_pdbfs[i].PS_dbfs_info(&tmpDbfsStat);
		m_vecStat[i].lCurSize = tmpDbfsStat.db_block_count;
	}

	if (NULL == tmpConfFile)
	{
		return 0;
	}

	strcpy(m_strTmpStatFile, tmpConfFile);

	if (-1 == mf_getTmpStatFile())
	{
		return -5;
	}

	if (NULL == m_pdbfsTmp)
	{
		m_pdbfsTmp = new PSClientDBFS[m_vecstatTmp.size()];
	}

	if (NULL == m_pmutexTmp)
	{
		m_pmutexTmp = new MMutex[m_vecstatTmp.size()];
	}

	for (i = 0; i < m_vecstatTmp.size(); i++)
	{
		m_pdbfsTmp[i].PS_dbfs_setEnv(m_pEnv);
		if (-1 == m_pdbfsTmp[i].PS_dbfs_init(m_vecstatTmp[i].dbfile))
		{
			return -6;
		}

		PSClientDBFS::PSDBFSStat tmpDbfsStat;
		m_pdbfsTmp[i].PS_dbfs_info(&tmpDbfsStat);
		
		m_vecstatTmp[i].lCurSize = tmpDbfsStat.db_block_count;
	}

//	m_pEnv->mutex_alloc(m_pEnv, DB_MUTEX_SELF_BLOCK, &m_mutex);

/*
#ifdef __TEMP__
#ifdef WIN32
	m_pid = _beginthread(PS_dbfsms_thread, 0, this);
#else
	pthread_create(&m_pid, NULL, (void *(*)(void *))PS_dbfsms_thread, this);
#endif
#endif
*/
	
	return 0;
}

void CPSDBFSMS::PS_dbfsms_thread(void *para)
{
	CPSDBFSMS *pdbfsms = (CPSDBFSMS *)para;

	int day(0), hour(0);
	while (!m_bStopFlag)
	{
		time_t ts = time(NULL);
		tm *today = localtime(&ts);

		if (day != today->tm_yday || hour != today->tm_hour)
		{
			day = today->tm_yday;
			hour = today->tm_hour;

			int n = m_vecTime.size();
			while ( n>=0 )
			{
				n--;
				if (today->tm_hour == m_vecTime[n])
				{
					m_vecTime[n]++;

					pdbfsms->PS_dbfsms_deletetmp();

//					pdbfsms->mf_getTmpStatFile();
					
					break;
				}
			}
		}
		
#ifdef WIN32
		Sleep(1000);
#else
		sleep(1);
#endif
	}

#ifdef WIN32
	_endthread();
#else
	pthread_exit(0);
#endif
}

int CPSDBFSMS::PS_dbfsms_set(CPSDBFSMSStat_vector &stat)
{
/*
	m_mutexStat.lock();

	m_vecStat = stat;

	m_mutexStat.unlock();

	mf_saveStatFile();
*/

	return 0;
}

int CPSDBFSMS::PS_dbfsms_write(const char *key, void *buf, unsigned long size, int mode)
{
	// find in normal area
	if (PS_dbfsms_find(key) >= 0)
	{
		return 0;
	}

	switch (mode) 
	{
	case 0:
		{
			static int nDB = 0;

			int ndbSize = m_vecStat.size();
			nDB %= ndbSize;

			int tmp = nDB;

			while (nDB - tmp < ndbSize)
			{
				int i = nDB % ndbSize;

				m_pmutexStat[i].lock();
				
				if ((m_vecStat[i].lSize > m_vecStat[i].lCurSize) && (m_vecStat[i].lSize - m_vecStat[i].lCurSize >= size))
				{
					if (m_pdbfs[i].PS_dbfs_write(key, buf, size) == size)
					{		
						m_vecStat[i].lCurSize += size;

						m_pmutexStat[i].unlock();

						nDB++;

						return 1;
					}
				}
								
				m_pmutexStat[i].unlock();

				nDB++;
			}
			
/*
			while (t > 0)
			{
				t--;
				
				m_pmutexStat[t].lock();
				
				if ((m_vecStat[t].lSize > m_vecStat[t].lCurSize) && (m_vecStat[t].lSize - m_vecStat[t].lCurSize >= size))
				{
					if (m_pdbfs[t].PS_dbfs_write(key, buf, size) == size)
					{		
						m_vecStat[t].lCurSize += size;

						m_pmutexStat[t].unlock();

						return 1;
					}
				}
								
				m_pmutexStat[t].unlock();
			}
*/
		}
		break;

	case 1:
		{
			// find in temp area
			if (PS_dbfsms_findtmp(key) >= 0)
			{
				return 0;
			}

			static int ntmpDB = 0;

			unsigned long l_time = sizeof(time_t);
			unsigned long ul_size = l_time + size;

			int ntmpDBSize = m_vecstatTmp.size();

			ntmpDB %= ntmpDBSize;

			int tmp = ntmpDB;

			while (ntmpDB - tmp < ntmpDBSize)
			{
				int i = ntmpDB%ntmpDBSize;
				m_pmutexTmp[i].lock();

				if (m_vecstatTmp[i].lSize > m_vecstatTmp[i].lCurSize && m_vecstatTmp[i].lSize - m_vecstatTmp[i].lCurSize >= ul_size)
				{
					time_t tnow = time(NULL);

					char *pbuf = new char[ul_size];

					memcpy(pbuf, buf, size);

					memcpy(pbuf+size, &tnow, l_time);
					
					unsigned long ret = m_pdbfsTmp[i].PS_dbfs_write(key, pbuf, ul_size);

					delete []pbuf;

					if (ul_size == ret)
					{		
						m_vecstatTmp[i].lCurSize += ul_size;

						m_pmutexTmp[i].unlock();

						ntmpDB++;

						return 1;
					}
				}

				m_pmutexTmp[i].unlock();

				ntmpDB++;
			}
			
/*
			while (t > 0)
			{
				t--;

				m_pmutexTmp[t].lock();

				if (m_vecstatTmp[t].lSize > m_vecstatTmp[t].lCurSize && m_vecstatTmp[t].lSize - m_vecstatTmp[t].lCurSize >= ul_size)
				{
					time_t tnow = time(NULL);

					char *pbuf = new char[ul_size];

					memcpy(pbuf, buf, size);

					memcpy(pbuf+size, &tnow, l_time);
					
					unsigned long ret = m_pdbfsTmp[t].PS_dbfs_write(key, pbuf, ul_size);

					delete []pbuf;

					if (ul_size == ret)
					{		
						m_vecstatTmp[t].lCurSize += ul_size;

						m_pmutexTmp[t].unlock();

						return 1;
					}
				}

				m_pmutexTmp[t].unlock();
			}
*/
		}
		break;

	default:
		break;
	}

	return -1;
}
	
int CPSDBFSMS::PS_dbfsms_delete(const char *key, int mode)
{
	switch (mode) {
	case 0:
		{
			int t = m_vecStat.size();
			while (t > 0)
			{
				t--;

				m_pmutexStat[t].lock();

				if (m_pdbfs[t].PS_dbfs_delete(key) == 0)
				{
					PSClientDBFS::PSDBFSStat tmp;
					m_pdbfs[t].PS_dbfs_info(&tmp);
					m_vecStat[t].lCurSize = tmp.db_block_count;

					m_pmutexStat[t].unlock();

					return 1;
				}
				
				m_pmutexStat[t].unlock();
			}
		}
		break;

	case 1:
		{
			int t = m_vecstatTmp.size();
			while (t > 0)
			{
				t--;
				
				m_pmutexTmp[t].lock();
				
				int ret = m_pdbfsTmp[t].PS_dbfs_delete(key);
				
				if (ret == 0)
				{
					PSClientDBFS::PSDBFSStat tmp;
					m_pdbfsTmp[t].PS_dbfs_info(&tmp);
					m_vecstatTmp[t].lCurSize = tmp.db_block_count;
					m_pmutexTmp[t].unlock();
					return 1;
				}