c7db_analys.c

unxi下共享内存的使用

#include "./common/common.h"
#include "./common/sybfunc.c"
#include "./common/commonfunc.c"

int		logflag, dbgflag,nenum_l,nenum_p;
DBPROCESS	*grs_svr;
RETCODE		result_code;		
LOGINREC	*login;
char		errtxt[1024];
long		dbtime = 0;
int		msu_counter = 0;
struct	shmem	*shm_p;		//share memory pointer
char		tempstr[1024],sqlstr[1024];
char		KEY;
char		magfile[100];
time_t		tempt,ltime;
struct tm 	*tmp;
int		sock_fd;
struct		sockaddr_in address;
int		address_len;
int n;
int sendlimit;			//发送次数限制
jmp_buf 	tmout_jump;
struct variant  *isms_ptr;
struct isms_buf *isms_buf_ptr;
int		isms_buf_cnt=100;

void tmout(int sig)
{
   signal(SIGALRM, SIG_IGN);
   //alarm(0);
   siglongjmp(tmout_jump,1);
}

struct	shmem	*GetShmPtr(char *KEYPATH,char KEYID){
	key_t	key;
	int	shmid;
	struct	shmem	*shm;

	shm = NULL;
	key	= ftok(KEYPATH, KEYID);
	printf("\nKEYID:%c,key:%d\n",KEYID,key);
	shmid	= shmget(key, sizeof(struct shmem), 0);
	if(shmid < 0) {
		printf("shmid < 0");
		return(shm);
	}
	shm = shmat(shmid, 0, 0);
	if(shm < 0) {
		printf("shm_p < 0");
		return(shm);
	}
	return(shm);
}


void release_shm(struct shmem* shm_pTMP,char *KEYPATH,char KEYID)
{
	struct 	shmid_ds	shm_desc;
	key_t			key;
	int			shmid;

	key	= ftok(KEYPATH, KEYID);
	shmid	= shmget(key, sizeof(struct shmem), 0);
	if(shmid < 0) {
		perror("release_shm: shmid:");
	}
	if (shmdt(shm_pTMP) == -1){
		perror("release_shm: shmdt:");
	}	
}

void house_keeper(int sig_num)
{
	printf("\nClearing share memory and malloced memory.\n");
	if(grs_svr>0)db_disconnect(grs_svr);			
	release_shm(shm_p,"/usr/mbcp/",KEY);
	printf("\nNow exit safely.\n");
	exit (0);
}

int nation(char *msisdn){
	char temp[10];

	return(0);
	bzero(temp,10);
	strncpy(temp,msisdn,4);	
	temp[4] = 0;
	if (dbgflag) printf("\nnation():temp:%s,msisdn:%s",temp,msisdn);
	if (strcmp(temp,"8613")==0) return(1);
	else return(0);
}

int C7dbproc()
{	
	char	curtime[40],imsi[18],ci[6],isms[80],msisdn[18],lasttime[40],send_counter[10];
	char	id[20],smid[20],smcontent[1024],stime[40],etime[40];
	char 	send_buf[2048],recv_buf[2048];
	int	sqlflag;
	long	lasttime_mi,stime_mi,etime_mi;
	int	counter;		//the cursor is the position in ne table
	int	send_cnt;		//同一时段发送次数
	int 	receive_b;
	int	retry,n,m,l;
	
	//msu_counter ++;     	
    	sqlflag = 0;   		
	getcurtime(curtime);
	bzero(imsi,18);
	bzero(ci,6);
	strcpy(imsi,shm_p->buffer[shm_p->rp].imsi);
	strcpy(ci,shm_p->buffer[shm_p->rp].ci);
	ci[4]=0;
	/*
	//首先查MBCP_ISMS表,看看能否找到MSISDN.如果没有则发送数据到OSS
	
	*/
	if ((grs_svr==NULL)||(grs_svr<0)) grs_svr=db_connect("MBCP","sa","");
	bzero(send_buf,2048);
	for(n=0;n<isms_buf_cnt;n++){
		strcpy(imsi,isms_buf_ptr->buffer[n].imsi);
		bzero(sqlstr,1024);
		sprintf(sqlstr,"select MSISDN from MBCP_ISMS WHERE IMSI = '%s'",imsi);
		if(dbgflag) breakpoint("sqlstr",sqlstr);
		db_operation(grs_svr, sqlstr);
		bzero(msisdn,18);		
		//sqlflag = 0;
		while((result_code = dbresults(grs_svr))!=NO_MORE_RESULTS){ 
			if(result_code == SUCCEED){
				dbbind(grs_svr, 1, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)msisdn);
				while(dbnextrow(grs_svr)!=NO_MORE_ROWS){
					//if(msisdn[0]!=0) sqlflag = 1;
				}	
			}
		}
		if(msisdn[0]!=0){
			strcpy(isms_buf_ptr->buffer[n].msisdn,msisdn);
			isms_buf_ptr->buffer[n].new_flag=0;
		}
		if(msisdn[0]==0){
			isms_buf_ptr->buffer[n].new_flag=1;
			if(n==0){
				strcat(send_buf,isms_buf_ptr->buffer[n].imsi);
			}else{
				strcat(send_buf,"|");
				strcat(send_buf,isms_buf_ptr->buffer[n].imsi);
			}
		}
	}
	
	if (send_buf[0]!=0){
		retry = 0;
redo:
		//发送数据
		sendto(sock_fd, send_buf, strlen(send_buf)+1, 0,(struct sockaddr *)&address, sizeof(address)); 
		if(dbgflag) printf("\nsend to OSS : %s\n",send_buf);
		fflush(stdout);
		//printf("\nci is %s , imsi is %s\n",ci,imsi);
		//fflush(stdout);
		//接收数据
		signal(SIGALRM, tmout);
		alarm(20); 
		if(sigsetjmp(tmout_jump, 1)){
			retry++;
			if(retry>=2){
				retry = 0;
				goto next;
			}
			else{
				if(dbgflag) printf("\nresend to OSS : %s\n",send_buf);
				fflush(stdout);
				//printf("\nresending ci is %s , imsi is %s\n",ci,imsi);
				goto redo;	
			}
		}
		bzero(recv_buf,2048);
		receive_b = recvfrom(sock_fd, recv_buf, 2048, 0, NULL, NULL);
//next:
		alarm(0);		//取消alarm
		if (dbgflag) printf("received %d:%s", receive_b, recv_buf);
		fflush(stdout);
		split(isms_ptr,recv_buf,'|');
		for(m=1;m<=isms_ptr->eleno;m++){
			bzero(imsi,18);
			bzero(msisdn,18);
			//if(isms_ptr->element[m].length<=17) continue;
			l=gettoken(isms_ptr->element[m].var,':');
			if (l<=0){
				continue;	
			}
			strncpy(imsi,isms_ptr->element[m].var,l);
			imsi[l]=0;
			strcpy(msisdn,isms_ptr->element[m].var+l+1);
			if(msisdn[0]==0) continue;
			for (n=0;n<isms_buf_cnt;n++){
				if(strcmp(imsi,isms_buf_ptr->buffer[n].imsi)==0){
					if(msisdn[0]!=0){
						strcpy(isms_buf_ptr->buffer[n].msisdn,msisdn);	
					}
				}
			}
		}
		for(n=0;n<isms_buf_cnt;n++){
			//if(receive_b<=17) return -1;	
			//strcpy(msisdn,isms+16);
			if ((isms_buf_ptr->buffer[n].new_flag==1)&&(isms_buf_ptr->buffer[n].msisdn[0]!=0)){
				bzero(sqlstr,1024);
				bzero(imsi,18);
				sprintf(sqlstr,"select IMSI from MBCP_ISMS WHERE IMSI = '%s'",isms_buf_ptr->buffer[n].imsi);
				if(dbgflag) breakpoint("sqlstr",sqlstr);
				db_operation(grs_svr, sqlstr);
				while((result_code = dbresults(grs_svr))!=NO_MORE_RESULTS){ 
					if(result_code == SUCCEED){
						dbbind(grs_svr, 1, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)imsi);
						while(dbnextrow(grs_svr)!=NO_MORE_ROWS){
						}	
					}
				}
				bzero(sqlstr,1024);
				if (imsi[0]==0){
					sprintf(sqlstr,"insert into MBCP_ISMS(IMSI,MSISDN) values ('%s','%s')",isms_buf_ptr->buffer[n].imsi,isms_buf_ptr->buffer[n].msisdn);
				}else{
					sprintf(sqlstr,"update MBCP_ISMS set MSISDN='%s' where IMSI='%s'",isms_buf_ptr->buffer[n].msisdn,isms_buf_ptr->buffer[n].imsi);
				}
				if(dbgflag) breakpoint("sqlstr",sqlstr);
				fflush(stdout);
				db_operation(grs_svr, sqlstr);
			}
		}
	}
	/*
	//拿到MSISDN后查MBCP_MSISDNCOMP表,得出LASTTIME(最后发送时间)和counter(发送计数器)
	
	//根据系统发送策略参数send_limit(同一个时段内发送次数)  
	
	*/
next:
		alarm(0);		//取消alarm
		if (dbgflag) printf("received %d:%s", receive_b, recv_buf);

	for(n=0;n<isms_buf_cnt;n++){
		if(isms_buf_ptr->buffer[n].msisdn[0]==0) continue;
		strcpy(msisdn,isms_buf_ptr->buffer[n].msisdn);
		bzero(lasttime,40);
		bzero(curtime,40);
		bzero(send_counter,10);
		bzero(id,20);
		bzero(sqlstr,1024);
		sprintf(sqlstr,"select ID,LASTTIME,COUNTER from MBCP_MSISDNCOMP where MSISDN='%s'",msisdn);
		if(dbgflag) breakpoint("sqlstr",sqlstr);
		db_operation(grs_svr, sqlstr);
		sqlflag = 0;
		while((result_code = dbresults(grs_svr))!=NO_MORE_RESULTS){ 
			if(result_code == SUCCEED){
				dbbind(grs_svr, 1, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)id);
  				dbbind(grs_svr, 2, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)lasttime);
  				dbbind(grs_svr, 3, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)send_counter);
				while(dbnextrow(grs_svr)!=NO_MORE_ROWS){
					if(id[0]!=0) sqlflag = 1;
				}	
			}
		}					

		if(id[0]==0){			//新记录
			getcurtime(lasttime);	
			strcpy(send_counter,"0");
			bzero(sqlstr,1024);
			sprintf(sqlstr,"insert into MBCP_MSISDNCOMP (MSISDN,LASTTIME,COUNTER,STATE) values ('%s','%s',%s,'N')",msisdn,lasttime,send_counter);
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);
		}
		send_cnt = atoi(send_counter);
		
		bzero(sqlstr,1024);
		bzero(smid,20);
		bzero(stime,40);
		getcurtime(curtime);
		sprintf(sqlstr,"select ID,datediff(mi,STIME,'%s') from MBCP_SMCONTENT where ((STIME < '%s') and (ETIME > '%s') and (CELLS like '%%%s%%'))",lasttime,curtime,curtime,ci);
		//strcat(sqlstr,"%%'))");
		strcpy(errtxt,sqlstr);
		if(dbgflag) breakpoint("sqlstr",sqlstr);
		db_operation(grs_svr, sqlstr);
		sqlflag = 0;
		while((result_code = dbresults(grs_svr))!=NO_MORE_RESULTS){ 
			if(result_code == SUCCEED){
				dbbind(grs_svr, 1, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)smid);
  				dbbind(grs_svr, 2, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)stime);
				while(dbnextrow(grs_svr)!=NO_MORE_ROWS){
					if(smid[0]!=0) sqlflag = 1;
				}	
			}
		}
		if(sqlflag==0) return -1;
		
		stime_mi = atol(stime);
		if(dbgflag) printf("\nstime_mi:%d\n",stime_mi);
		if(stime_mi<0){
			bzero(sqlstr,1024);
			sprintf(sqlstr,"insert into MBCP_SMTMP (MSISDN,SMID,TIME,SEND) values ('%s',%s,getdate(),'N')",msisdn,smid);
			strcpy(errtxt,sqlstr);
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);
			sprintf(sqlstr,"update MBCP_MSISDNCOMP set LASTTIME=getdate(),COUNTER=1 where MSISDN='%s'",msisdn);
			strcpy(errtxt,sqlstr);
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);		
		}else if ((stime_mi>0)&&(send_cnt<=sendlimit)){
			bzero(sqlstr,1024);
			sprintf(sqlstr,"insert into MBCP_SMTMP (MSISDN,SMID,TIME,SEND) values ('%s',%s,getdate(),'N')",msisdn,smid);
			strcpy(errtxt,sqlstr);
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);
			if(id[0]!=0){
				sprintf(sqlstr,"update MBCP_MSISDNCOMP set LASTTIME=getdate(),COUNTER=%d where ID=%s",send_cnt+1,id);
			}else{
				sprintf(sqlstr,"update MBCP_MSISDNCOMP set LASTTIME=getdate(),COUNTER=%d where MSISDN='%s'",send_cnt+1,msisdn);		
			}
			strcpy(errtxt,sqlstr);
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);			
		}else{
			return -1;
		}
	}		
}

main(int argc,char **argv)
{
	unsigned char	curtime[256];
	int		overflag,emptyflag,sqlflag,i,j;
	long		overflow,counter;
	FILE*		fd;
	float		msucnt;
	char		send_limit_str[20];
		
	logflag = 0;
	dbgflag = 0;
	if(argc==2){
		strcpy(tempstr,argv[1]);
		if(strcmp(tempstr,"debug")==0)
			dbgflag=1;
	}else{
		printf("\nusage:c7dbjschild <debug,free>\n");
		exit(0);
	}
	
	//sleep(10);
	bzero(magfile,100);

	KEY = 'e';
	strcpy(magfile,"/usr/mbcp/c7db.mag");
	shm_p = NULL;
	while((shm_p==NULL) || (shm_p<0)){
		shm_p=GetShmPtr("/usr/mbcp/",KEY);
		sleep(1);
	}
	signal(SIGINT, house_keeper);
	db_init();
		//建立套接口
	sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
		//联接
	bzero(&address, sizeof(address));
	address.sin_family = AF_INET;
	address.sin_addr.s_addr = inet_addr("132.108.19.8");
	address.sin_port = htons(6200);
	address_len = sizeof(address);

	if (dbgflag) printf("\ncheck point 2\n");
	dbtime = 499;
	tempt = time(&tempt);
	ltime = tempt;

	if (dbgflag) printf("\ncheck point 3\n");
	isms_buf_ptr=malloc(sizeof(struct isms_buf));
	bzero(isms_buf_ptr,sizeof(struct isms_buf));
	isms_ptr=malloc(sizeof(struct variant));
	bzero(isms_ptr,sizeof(struct variant));
	i=0;	
	for(;;){
		dbtime++;
		if ((dbtime==100)||(dbtime==200)||(dbtime==300)||(dbtime==400)||(dbtime==500)){
			if(grs_svr>0)db_disconnect(grs_svr);			
			if(!(grs_svr=db_connect("MBCP","sa",""))){
				printf("\n connect sysbase error!!!\n");
				fflush(stdout);
				dbtime = 499;
				continue;
			}
			bzero(sqlstr,1024);
			strcpy(sqlstr,"select PARAM from MBCP_SYSCONFIG WHERE ITEM = 'sendlimit'");
			if(dbgflag) breakpoint("sqlstr",sqlstr);
			db_operation(grs_svr, sqlstr);
			bzero(send_limit_str,20);		
			while((result_code = dbresults(grs_svr))!=NO_MORE_RESULTS){ 
				if(result_code == SUCCEED){
					dbbind(grs_svr, 1, NTBSTRINGBIND, (DBINT)0, (BYTE DBFAR*)send_limit_str);
					while(dbnextrow(grs_svr)!=NO_MORE_ROWS){
						if(send_limit_str[0]!=0) sqlflag = 1;
					}	
				}
			}			
			if (sqlflag==1){
				sendlimit=atoi(send_limit_str);
			}else{
				sendlimit=1;
			}
		}		
		printf(" %d(%d)  ", shm_p->rp,shm_p->wp);
		fflush(stdout);
		if (i>99){
			C7dbproc();
			bzero(isms_buf_ptr,sizeof(struct isms_buf));
			i=0;
		}else{
			strcpy(isms_buf_ptr->buffer[i].imsi,shm_p->buffer[shm_p->rp].imsi);		
			bzero(isms_buf_ptr->buffer[i].msisdn,20);
			isms_buf_ptr->buffer[i].new_flag=0;
			i++;
		}
		//////////////////////////
		/*	for test only	*/
		/*	printf("\n shm_p->buffer[shm_p->rp].imsi:%s\n",shm_p->buffer[shm_p->rp].imsi);
			printf("\n shm_p->buffer[shm_p->rp].msisdn:%s\n",shm_p->buffer[shm_p->rp].msisdn);
			printf("\n shm_p->buffer[shm_p->rp].callingaddr:%s\n",shm_p->buffer[shm_p->rp].callingaddr);
			printf("\n shm_p->buffer[shm_p->rp].calledaddr:%s\n",shm_p->buffer[shm_p->rp].calledaddr);*/
		/////////////////////////
		emptyflag=0;
		counter=shm_p->rp;
		counter++;
		while(counter==shm_p->wp){
			if (emptyflag >= 120){					
				fd = fopen(magfile, "wb");
				getcurtime(curtime);
				fwrite(curtime, sizeof(char), strlen(curtime), fd);
				fclose(fd);
				emptyflag = 0;
			}
			emptyflag ++;
			if (emptyflag==1) printf("..buffer..is..empty...");
			fflush(stdout);
			sleep(1);
			if (emptyflag >= 5){
				if(grs_svr > 0){
					db_disconnect(grs_svr);
					grs_svr=NULL;
				}
			}			
		}
		shm_p->rp++;
		if (shm_p->rp >= MAXSLOT) shm_p->rp = 0;						
		if (shm_p->rp==shm_p->wp) sleep(2);
	}
	free(nePtr);
}