c_shm_mem.cpp

HP共享内存实现,只要将sql语句配置好,装载共享内存即可

/***********************************************************************
 * Module:  c_shm_mem.cpp
 * Author:  Administrator
 * Modified: 2005年9月15日 9:03:22
 * Purpose: Implementation of the class c_shm_mem
 * Comment: 共享内存基本类
 ***********************************************************************/


#include "c_shm_mem.h"

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::c_shm_mem()
// Purpose:		构造函数
// Comment:		创建c_share_info对象
// Return:     
////////////////////////////////////////////////////////////////////////
c_shm_mem::c_shm_mem()
{
	p_share_info = new c_share_info();
	p_share_info->loadAllInfo();
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::~c_shm_mem()
// Purpose:		析构函数
// Comment:		删除c_share_info对象
// Return:     
////////////////////////////////////////////////////////////////////////
c_shm_mem::~c_shm_mem()
{
	delete p_share_info;
}

/*void c_shm_mem::init( char* mem_name )
{
	SHARE_MEM_INFO* tmp = p_share_info->getMemInfo( mem_name);
	if( tmp == NULL )
	{
		printf("不存在%s的共享内存\n", mem_name );fflush(stdout);
		exit(1);
	}
	c_shm newshm(tmp->share_area_id);
	newshm.getshm();
	c_shm_mem* newmem = newshm.get_mem(mem_name);	

	p_share_mem_info = newmem->p_share_mem_info;
	p_stShareInfo = newmem->p_stShareInfo;
	p_field_info = newmem->p_field_info;
	head_point = newmem->head_point;
	append_mutex._mutex = newmem->append_mutex._mutex;
	record_count_mutex._mutex = newmem->record_count_mutex._mutex;
	v_update_mutex.clear();
	for( int i = 0 ; i < newmem->p_stShareInfo->block_count ; i ++ )
	{
		c_mutex t_mutex;
		t_mutex._mutex = &(newmem->p_stShareInfo->update_lock[i]);
		v_update_mutex.push_back( t_mutex );
	}
	append_key_mutex._mutex = newmem->append_key_mutex._mutex;

	//拷贝路径
	strcpy(file_path, newmem->file_path );
}*/

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::loadSQL()
// Purpose:		加载SQL信息
// Comment:		从c_share_info对象中加载SQL信息
// Return:     int
////////////////////////////////////////////////////////////////////////
int c_shm_mem::loadSQL()
{
	//theLog<<"mem_area_name:"<<p_share_mem_info->mem_name<<endi;

	p_share_info->getSqlInfo( p_share_mem_info->mem_name , v_sql_info );
	
	return 0;
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::loadFieldInfo()
// Purpose:		加载字段信息
// Comment:		从c_share_info对象中加载字段信息
// Return:     int
////////////////////////////////////////////////////////////////////////
int c_shm_mem::loadFieldInfo( )
{
	if( v_sql_info.size() <= 0 )
	{
		theLog<<"未加载SQL信息!"<<endi;
		return errNotloadSQL;
	}

	//取第一个的sql语句
	int select_no = v_sql_info[0].select_no;
	
	//取得字段信息
	p_share_info->getFieldInfo( select_no, v_mem_field_info );
	//theLog<<"select_no:"<<select_no<<"\tv_mem_field_info.size():"<<v_mem_field_info.size()<<endi;
	for( int i = 0 ; i < MAX_ATTR_COUNT ; i ++ )
	{
		field_attr_id[i] = -1 ;
	}
	int pos = 0;
	for( i = 0 ; i < v_mem_field_info.size() ; i ++ )
	{
		field_attr_id[v_mem_field_info[i].event_attr_id] = i;
		v_mem_field_info[i].field_pos = pos;
		
		//20060609 加载事件属性名称
		EVENT_ATTR tmp_event_attr;
		tmp_event_attr.attr_id = v_mem_field_info[i].event_attr_id;
		//////////2005-11-23
		//pos += v_mem_field_info[i].field_size;
		if( v_mem_field_info[i].field_size == -1 )
		{
			pos += 4;
		}
		else
		{
			pos += v_mem_field_info[i].field_size;
		}
		///////////////////////

		p_field_info[i].event_attr_id = v_mem_field_info[i].event_attr_id ;
		p_field_info[i].field_pos = v_mem_field_info[i].field_pos ;
		p_field_info[i].field_size = v_mem_field_info[i].field_size ;
		p_field_info[i].field_type = v_mem_field_info[i].field_type ;
		p_field_info[i].key_count = 0;
	}
	
	p_share_mem_info->field_count = v_mem_field_info.size();
	return 0;
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::buildSQLString()
// Purpose:		组装SQL语句
// Comment:		组装select count(*) from 语句与select field语句，分别放入v_sqlstring与v_countsqlstring中
// Return:     int 成功：0 失败：errNotloadSQL
////////////////////////////////////////////////////////////////////////
int c_shm_mem::buildSQLString()
{
	if( v_sql_info.size() <= 0 )
	{
		theLog<<"未加载SQL信息!"<<endi;
		return errNotloadSQL;
	}

	int i, j;
	v_sqlstring.reserve( 10 );
	v_countsqlstring.reserve( 10 );

	for( i = 0 ; i < v_sql_info.size() ; i ++ )
	{		
		vector <MEM_FIELD_INFO> memField;
		p_share_info->getFieldInfo( v_sql_info[i].select_no, memField );
		//theLog<<"select_no:"<<v_sql_info[i].select_no<<"\tmemfield:"<<memField.size()<<endi;

		if( memField.size() > 0 )
		{
			string sql = "select ";
			for( j = 0 ; j < memField.size() ; j ++ )
			{
				sql += memField[j].field_sql ;
				if( j < ( memField.size() - 1 ) )
					sql += ", ";			
			}
			sql += " from ";
			sql += v_sql_info[i].from_sql;
			
			if( strcmp( v_sql_info[i].condition , "" ) != 0 )
			{
				sql += " ";
				sql += v_sql_info[i].condition;
			}
			//theLog<<"ddd:"<<sql<<"|"<<endi;
			v_sqlstring.push_back( sql );

			string count_sql = "select count(*) from ";
			count_sql += v_sql_info[i].from_sql;
			if( strcmp( v_sql_info[i].condition , "" ) != 0 )
			{
				count_sql += " ";
				count_sql += v_sql_info[i].condition;
			}
			v_countsqlstring.push_back( count_sql );
		}
	}

	return 0;
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::loadData()
// Purpose:     加载数据，继承类可以重写，在初始化后一次调用
// Return:     int 成功：0 失败：errMemDelete，errAppenElement
////////////////////////////////////////////////////////////////////////
int c_shm_mem::loadData(const int sql_seq)
{
	if( p_stShareInfo->valid_flag != 1 )
	{
		theLog<<"共享内存块不可用！"<<endi;
		return errMemDelete;
	}

	int i, j = 0;
	char field[MAX_FIELD_SIZE];

	//从数据库中加载数据
	//loadSQL();
	//loadFieldInfo();
	//buildSQLString();

	Statement statement = conn.createStatement();
	statement.setSQLString( "alter session set NLS_DATE_FORMAT = yyyymmddhh24miss" );
	statement.execute();

	
	//把记录组合成一个结构，并写入共享内存中
	char buffer[ MAX_RECORD_SIZE ];
	int int_field = 0;
	int* p_int_field;
	double double_field = 0;
	long long_field = 0;
	double* p_double_field;
	long* p_long_field;
	char kk[10];
	char temp[MAX_RECORD_SIZE];
	long load_record_count = 0;
	long record_counts = 0;
	for( i = 0 ; i < v_sqlstring.size() ; i ++ )
	{
		if( sql_seq != -1 )
		{
			if( sql_seq != i )
				continue;
		}

		try
		{
			statement.setSQLString( v_countsqlstring[i].c_str() );
			statement.execute();
			statement>>load_record_count;
			theLog<<"begin load "<<load_record_count<<" records!"<<endi;
		}
		catch(Exception& ex1)
		{
			theLog<<"get count error!"<<endi<<ex1.what();
			return -1;
		}

		theLog<<" exec sql:"<<v_sqlstring[i].c_str()<<endi;
		statement.setSQLString( v_sqlstring[i].c_str() );
		try
		{
			statement.execute();
		}
		catch(Exception& ex)
		{
			theLog<<"execute sql error!"<<endi<<ex.what();
			return -1;
		}
		try
		{
			while( statement>>field )
			{
				//theLog<<i<<":"<<j<<":"<<field<<":"<<p_field_info[j].field_type<<endi;
				//theLog<<"p_field_info[j].field_pos:"<<p_field_info[j].field_pos<<endi;
				//theLog<<"p_field_info[j].field_size:"<<p_field_info[j].field_size<<endi;
				
				switch( p_field_info[j].field_type )
				{
				case STRING_TYPE_D:
					//theLog<<p_field_info[j].field_type<<"afe";
					//memcpy( &(buffer[p_field_info[j].field_pos]), field, p_field_info[j].field_size );
					if(SetData(j,field)<0)
					{
						setRecordCount();
						theLog<<"load count3:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					break;
				case INT_TYPE_D:
					int_field =  atoi( field );
					//p_int_field = (int*) temp;
					//memcpy( temp, &int_field, sizeof(int) );
					//memcpy( &(buffer[p_field_info[j].field_pos]), temp, sizeof(int) );
					
					if(SetData(j, &int_field)<0)
					{
						setRecordCount();
						theLog<<"load count3:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					break;
				case CHAR_TYPE_D:
					//memcpy( &(buffer[p_field_info[j].field_pos]), field, p_field_info[j].field_size );
					if(SetData(j,field)<0)
					{
						setRecordCount();
						theLog<<"load count3:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					break;
				case DOUBLE_TYPE_D :
					double_field = atof(field);
					//p_double_field = (double*) temp;
					//memcpy( temp, &double_field, sizeof(double) );
					//memcpy( &(buffer[p_field_info[j].field_pos]), temp, sizeof(double) );
					if(SetData(j,&double_field)<0)
					{
						setRecordCount();
						theLog<<"load count3:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					break;
				case LONG_TYPE_D :
					long_field = atol(field);
					//p_double_field = (double*) temp;
					//memcpy( temp, &double_field, sizeof(double) );
					//memcpy( &(buffer[p_field_info[j].field_pos]), temp, sizeof(double) );
					if(SetData(j,&long_field)<0)
					{
						setRecordCount();
						theLog<<"load count3:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					break;
				default:
					//memcpy( &(buffer[p_field_info[j].field_pos]), field, p_field_info[j].field_size );
					SetData(j,field);
					break;
				};
				
				j ++;
				if( j >= p_share_mem_info->field_count )
				{		
					if( singal_terminate.isTerminated() )
					{
						theLog<<"程序终止!"<<endi;
						singal_terminate.kill();
					}
					j = 0;
					if( appendElement( ) < 0 )
					{
						setRecordCount();
						theLog<<"装载失败1,已装载记录数:"<<getRecordCount()<<endi;
						return errAppenElement;
					}
					//theLog<<"addok1"<<endi;
					//theLog<<"buffer:"<<&buffer[30]<<"|"<<endi;
					memset( buff, 0, p_share_mem_info->element_size );
					record_counts++;

					if( record_counts % 200000 == 0 )
					{
						theLog<<"已装载记录数:"<<record_counts<<endi;
					}
				}
			}
		}
		catch(Exception& ex)
		{
			theLog<<"装载第"<<record_counts<<"条记录第"<<j<<"个字段出错!"<<field<<endi;
			setRecordCount();
			return -2;
		}
		if( j > 0 )
		{
			//theLog<<"buffer:"<<&buffer[30]<<"|"<<endi;
			j = 0;
			if( appendElement(  ) < 0 )
			{
				setRecordCount();
				theLog<<"装载失败2,已装载记录数:"<<getRecordCount()<<endi;
				return errAppenElement;
			}
			memset( buff, 0, p_share_mem_info->element_size );
			record_counts++;
		}
	}

	setRecordCount();
	theLog<<"已装载记录数:"<<getRecordCount()<<endi;
	if( singal_terminate.isTerminated() )
	{
		theLog<<"程序终止!"<<endi;
		singal_terminate.kill();
	}
	return 0;
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::unloadData()
// Purpose:		卸载数据
// Comment:		修改标志，同时，将记录数修改为0
// Return:     int，成功：0，失败：errMemUnload
////////////////////////////////////////////////////////////////////////
int c_shm_mem::unloadData(void)
{
	// TODO : implement
	if( p_stShareInfo->valid_flag == 0 )
	{
		theLog<<"数据已经被卸载！"<<endi;
		return errMemUnload;
	}

	record_count_mutex.enterMutex();
	append_mutex.enterMutex();
	p_stShareInfo->real_record_count = 0;
	p_stShareInfo->append_record_count = 0;
	append_mutex.leaveMutex();
	record_count_mutex.leaveMutex();

	p_stShareInfo->valid_flag = 0;

	return 0;
}

////////////////////////////////////////////////////////////////////////
// Name:       c_shm_mem::getElement( int seq )
// Purpose:    取得指定序号的记录，返回头指针。
// Comment:		将共享内存中的记录拷贝到临时buffer中，返回buffer的头指针
// Return:     void*
////////////////////////////////////////////////////////////////////////
char* c_shm_mem::getElement( int seq )
{
	//theLog<<"seq:"<<seq<<endi;
	//theLog<<"p_stShareInfo->real_record_count:"<<p_stShareInfo->real_record_count<<endi;
   // TODO : implement