server.c

接收端的程序

/******************************************************************************

                                   Server beta
                                   version 1.0

								@ author:  WISCOM 

******************************************************************************/




#include        <stdio.h>
#include        <string.h>
#include        <stdlib.h>
#include		<stdarg.h>
#include        <sys/socket.h>
#include        <sys/types.h>
#include        <netinet/in.h>
#include        <errno.h>
#include        <sys/select.h>
#include		<fcntl.h>
#include		<mysql.h>
#include		<signal.h>
#include		<stdarg.h>
#include		<pthread.h>

#include 		<net-snmp/net-snmp-config.h>
#include		 <net-snmp/net-snmp-includes.h>
#include 		<net-snmp/library/snmp.h>
#include		<net-snmp/library/snmp_logging.h>

#include		"queue.h"
#include 		"devicelog.h"
#include		"server.h"
#include		"logging.h"

#define MAXCONN 		 20    //定义最大线程数，本服务器程序最多支持20个客户端同时连接
#define F_CONNECTING 	1		//定义线程的状态，表示线程已经创建，正与客户端连接
#define F_JOINED 		0		//表示线程已经被回收
#define F_DONE			2		//表示线程已经完成工作，可以被回收

static char client_ip[16];		//用来存储客户端的ip地址
char msg[5000];					//用来存储客户端发来的日志信息

extern int sem;					//信号量，用来服务器消息接收线程和数据库存储线程通信
extern pthread_cond_t ndone_cond;   //条件变量，用来服务器消息接收线程和数据库存储线程通信
extern pthread_mutex_t ndone_mutex;	//线程互斥锁，用来服务器消息接收线程和数据库存储线程通信
extern LinkQueue * Q;



int server_sem=0;				//信号量，用来服务器与客户端建立多个线程时，线程间通信
pthread_cond_t server_cond=PTHREAD_COND_INITIALIZER;     //条件变量，用来服务器与客户端建立多个线程时，线程间通信
pthread_mutex_t server_mutex=PTHREAD_MUTEX_INITIALIZER;	//线程互斥锁，用来服务器与客户端建立多个线程时，线程间通信

struct file{
	int flags;
	int sockid;
	pthread_t tid;
}files[MAXCONN];              //线程创建时传递的参数

int nconn=0;
int nleftconn=MAXCONN;
int nlefttoread=MAXCONN;




/*struct eventinfo log;	*/

/**************************************************
			返回a,b中的最大值
****************************************************/
int  max(int a,int b)
{
        if(a>b)
                return a;
        else
                return b;
}








int my_asynch_response(int operation, struct snmp_session *sp, int reqid, 
                    struct snmp_pdu *pdu, void *magic) 
{ 
    

	
	char trap_facility[50], buffer[10240], temp_buffer[1024], severity[2];
	
	

	
	memset(severity, 0 ,sizeof(severity));
	memset(trap_facility, 0, sizeof(trap_facility));
	
	memset(buffer, 0, sizeof(buffer));
	
//	printf("\nGot the message!!\n"); 
    
    netsnmp_transport *t = (netsnmp_transport*) magic;
    
    struct variable_list *vars;
   
    if(operation == NETSNMP_CALLBACK_OP_RECEIVED_MESSAGE)
    {
   		for(vars = pdu->variables; vars; vars = vars->next_variable)
   		{
   		 	memset(temp_buffer, 0, sizeof(temp_buffer));
   			snprint_variable(temp_buffer, sizeof(temp_buffer), vars->name, vars->name_length, vars);
   		 	
   		 	//printf("%s\n",buffer);
   		 	strcat(buffer, temp_buffer);
   		 	strcat(buffer, "  ");
   		 
   		 
   		 	
   		 	
   		 
   		 	
   	
   		 	
   		 	
 
   		 	
   		 
   		 	
   		 }	
   		
   		
  		buffer[strlen(buffer) - 1] = '\0';
   		
   		
   		memset(trap_facility, 0, sizeof(trap_facility));
   		if(pdu->trap_type == 0)
   		{
   			strcpy(trap_facility,"COLDSTART\0");
   			strcpy(severity,"0\0");
   		}	
   		else
   			if(pdu->trap_type == 1)
   			{
   				strcpy(trap_facility, "WARMSTART\0");
   				strcpy(severity,"1\0");
   			}
   			else
   				if(pdu->trap_type == 2)
   				{
   					
   					strcpy(trap_facility, "LINKDOWN\0");
   					strcpy(severity,"2\0");
   				}
   				else
   					if(pdu->trap_type == 3)
   					{	
   						strcpy(trap_facility, "LINKUP\0");
   						strcpy(severity,"3\0");
   					}
   					else
   						if(pdu->trap_type ==4)
   						{
   							strcpy(trap_facility, "AUTH FAIL\0");
   							strcpy(severity,"4\0");
   						}
   						else 
   					 		if(pdu->trap_type ==5)
   					 		{
   					 		
   					 			strcpy(trap_facility, "EGPNBORLOSS\0");
   					 			strcpy(severity,"5\0");
   					 		}
   					 		else
   					 			if(pdu->trap_type ==6)
   					 			{
   					 				strcpy(trap_facility, "EPRISE SPEC\0");
   					 				strcpy(severity,"6\0");
   					 			}
   					 			else
   					 			{
   					 				strcpy(trap_facility, "Unkown\0");
   					 				strcpy(severity,"7\0");
   					 			}
   					
   			
    }
    
   
   
  
    

	char addr[16];
	   
	char *tstr = t->f_fmtaddr(t, pdu->transport_data, pdu->transport_data_length);
 	
   	char *count;
   	int j, i;
  
   	i = strcspn(tstr, "["); i++;
   	j = strcspn(tstr, "]");
   	count = tstr + i;
   	
   
   	
   	strncpy(addr, count, j - i);
   	
   	addr[j-i] = '\0';

	pthread_mutex_lock(&ndone_mutex);
  	
   	struct eventinfo trap_log ;
  	
  	
  	
  //	trap_log = (struct eventinfo*) malloc(sizeof(struct eventinfo));
  	
  	bzero(&trap_log, sizeof(struct eventinfo));
  	
  	sprintf(trap_log.time_stamp, "%d-%d-%d %d:%d:%d",getyear(),getmon(),getday(),gethour(),getmin(),getsec());
  	
  	
  	
  	strcpy(trap_log.facility, trap_facility);
  	
  	strcpy(trap_log.content, buffer);
  	
  	strcpy(trap_log.serverity, severity);
  	
  	strcpy(trap_log.ip, addr);
  	
  	strcpy(trap_log.source, "2\0");
   	
  	
  	log_debug("TRAP_TIME", trap_log.time_stamp);
  	log_debug("TRAP_FACILITY", trap_log.facility);
  	log_debug("TRAP_SEVERITY", trap_log.serverity);
  	log_debug("TRAP_CONTENT", trap_log.content);
  	log_debug("AGENT_IP", trap_log.ip);
  	log_debug("SOURCE", trap_log.source);
 	
  	//加入队列
  	
  	
  	
  	Enqueue(Q, trap_log);
  	
  	sem++;
  	

  	
  	pthread_cond_signal(&ndone_cond);
  	

  	pthread_mutex_unlock(&ndone_mutex);
  	
  
  	

//	snmp_free_pdu(pdu);

    return 1; 


} 


 
void trap_receive()
{ 

    netsnmp_session     session, *ss = &session; 
 //   netsnmp_pdu            *pdu         = NULL, *response; 
    netsnmp_transport     *transport     = NULL; 
 //   int                 status; 
 //   int                 command; 


    init_snmp_logging();
    snmp_enable_syslog(); 
	init_mib();

    SOCK_STARTUP; 

	netsnmp_ds_set_boolean(NETSNMP_DS_LIBRARY_ID, 
                           NETSNMP_DS_LIB_DUMP_PACKET, 1); 
	
	
	
	
	struct sockaddr_in server_addr;
//	socklen_t length;
	bzero(&server_addr, sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(162);
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	
	transport = netsnmp_udp_transport(&server_addr, 1);

  //  transport = netsnmp_tdomain_transport("udp:162", 1, "udp"); 
    if (transport == NULL) 
   { 
		printf("Can not open the port!");
   } 


    snmp_sess_init(ss); 
    ss->peername = SNMP_DEFAULT_PEERNAME;  /* Original code had NULL here */ 
    ss->local_port = 162; 
    ss->localname = "172.18.31.242"; 
    ss->version = SNMP_DEFAULT_VERSION; 
    ss->community = strdup("wisocm"); 
    ss->community_len = strlen(ss->community);             
//SNMP_DEFAULT_COMMUNITY_LEN; 
    ss->retries = SNMP_DEFAULT_RETRIES; 
    ss->timeout = SNMP_DEFAULT_TIMEOUT; 
    ss->callback = my_asynch_response; 
    ss->callback_magic = (void *) transport; 
    //ss->authenticator = NULL; 
    //session.isAuthoritative = SNMP_SESS_UNKNOWNAUTH;     



    ss = snmp_add(ss, transport, NULL,  NULL); 
    if (ss == NULL) 
    { 
            log_error("snmptrapd","snmpadd error!");
    } 


  //  printf("\nPort: %d\n", ss->local_port); 
     


    int fds = 0, block = 1; 
    fd_set fdset; 
    struct timeval timeout; 


    FD_ZERO(&fdset); 
    
   // char msg[1024];
   
  //  socklen_t len;
    while(1)
    { 
	 	snmp_select_info(&fds, &fdset, &timeout, &block); 
	
	
	    fds = select(fds, &fdset, 0, 0, block ? NULL : &timeout); 
	    if (fds < 0) 
	    { 
	            perror("select failed"); 
	            exit(1); 
	    } 
	    if (fds) 
	   { 
	    /*	
	    	recvfrom(transport->sock, msg, sizeof(msg), 0, (struct sockaddr *)&client_addr, &len);
	    	
	    	memset(agent_addr, sizeof(agent_addr), 0);
	  		log_debug("IP",(char *)inet_ntoa(client_addr.sin_addr));
	  		
	  	*/	
	  		
	  	
	        snmp_read(&fdset); 
	        
	        
	        
	   } 
	   else 
	   { 
	       
	            snmp_timeout(); 
	   }       

    }
    
   

} 













/*********************************************************************************************************
		初始化服务器，使服务器在514端口监听
		@ socketid1，socketid2:  socket的文件表述符指针
	
	
***********************************************************************************************************/

void initserver(int *socketid1,int *socketid2)
{
	int yes=1;
	
	struct sockaddr_in server_addr;
	*socketid1=socket(AF_INET,SOCK_DGRAM,0);
	if(*socketid1<0)
	{
		log_fatal("SERVER","Can not open the socket for port1!");
		exit(1);
	}
	setsockopt(*socketid1,SOL_SOCKET,SO_REUSEADDR,(char *)&yes,sizeof(yes));
	bzero(&server_addr,sizeof(server_addr));
	server_addr.sin_family=AF_INET;
	server_addr.sin_port=htons(DEFAULT_SYSLOG_PORT);
	server_addr.sin_addr.s_addr=htonl(INADDR_ANY);
	if(bind(*socketid1,(struct sockaddr *)&server_addr,sizeof(server_addr))<0)
	{
		log_fatal("SERVER","Bind the port1 failure!");
		exit(1);
	}


}

/***********************************************************************************************************
		函数功能： 接收客户端发来的消息，同时通知回收本线程
		

********************************************************************************************************/
void recmsg(struct file *fptr)
{
        socklen_t len;
        struct eventinfo * log=NULL;
        struct sockaddr_in client_addr;
        len=sizeof(client_addr);
        
        pthread_mutex_lock(&server_mutex);       /*互斥锁，防止其他线程访问共享变量msg*/
        memset(msg,0,5000);
        recvfrom(fptr->sockid,msg,sizeof(msg),0,(struct sockaddr *)&client_addr,&len);
        memset(client_ip,0,16);
        strcpy(client_ip,(char*) inet_ntoa(client_addr.sin_addr));
        fptr->flags=F_DONE;        				/*更改线程的标识值，表示本线程可以被回收*/
        nconn--;	
        
        pthread_mutex_lock(&ndone_mutex);		/*锁住，防止数据库存取线程访问共享变量*/
		log=(eventinfo *)malloc(sizeof(struct eventinfo));		
        
        if(!log)
        {
        	log_warn("SERVER","The log can not malloced!");
        	return ;
        }
		getmsg(msg,log);						/*解析日志消息*/
			
		strncpy(log->ip,client_ip, sizeof(client_ip));	
		Enqueue(Q,*log);						/*存入缓冲区队列*/
		
		printf("Enqueu: %s\n", msg);
		
		log_debug("SERVER","insert a msg into Queue!");	
		sem++;										
		free(log);	
		log=NULL;
		pthread_cond_signal(&ndone_cond);		/*通知数据库存取线程从缓冲区队列中去数据*/
		pthread_mutex_unlock(&ndone_mutex);
        
		
		server_sem++;							/*更改信号量*/
        pthread_cond_signal(&server_cond);		/*通知回收本线程*/
        pthread_mutex_unlock(&server_mutex);	/*释放互斥锁*/
        
		
		pthread_exit(NULL);
                                                         
}


void startserver()
{
	fd_set rset;
	int socketid1,socketid2,k, maxsockid,i=0;
	

	initserver(&socketid1,&socketid2); 			/*初始化服务器*/

	
	

	FD_ZERO(&rset);
	maxsockid=max(socketid1,socketid2)+1;
	
	for(k = 0;k < MAXCONN;k ++ ) files[k].flags=0;    /*初始化标志*/
	
	FD_SET(socketid1,&rset);
	FD_SET(socketid2,&rset);
	
	
	pthread_t trap_thread;
	
	pthread_create(&trap_thread, NULL, (void *) trap_receive, NULL);
	pthread_detach(trap_thread);
	
	
	
	while(1)
	{
		
		while(nconn<MAXCONN && nleftconn>0)
		{
			select(maxsockid,&rset,NULL,NULL,NULL);		/*等待连接*/
			if(FD_ISSET(socketid1,&rset))				/*SYSLOG转发过来的消息*/
			{
			
				for(i = 0;i < MAXCONN;i ++)
					if(files[i].flags==0)   	break; /*查看是否有空闲的pthread_t变量可以创建线程*/
				
				if(i==MAXCONN)							/*如果现在已有MAXCONN个线程在运行，则不能建立连接*/
				{
				
					log_error("SERVER","cannot create any more thread!");
					break;
				}
			
				files[i].sockid=socketid1;				
				files[i].flags=F_CONNECTING ; 			/*更改标志，表示改线程已经处于连接状态*/
				pthread_create((&files[i].tid),NULL,(void *)recmsg,(void *)&files[i]);	/*创建线程*/
			//	pthread_detach(files[i].tid);
				
				pthread_join(files[i].tid, NULL);
				
				nconn++;
				nleftconn--;
			
			}	
			if(FD_ISSET(socketid2,&rset))
			{
			
			}
		}//while
			
			pthread_mutex_lock(&server_mutex);										/*互斥锁*/
			while(server_sem==0)
				pthread_cond_wait(&server_cond,&server_mutex);						/*等待通知有线程可以回收*/
			
			for(i=0;i<MAXCONN;i++)
					if(files[i].flags==F_DONE)										/*搜索处于可回收状态的线程*/
					{
						pthread_join(files[i].tid,NULL);							/*回收线程*/
						files[i].flags=F_JOINED;									/*该线程以回收*/
						
						nleftconn++;
						server_sem--;													/*更改信号量*/
					}
			pthread_mutex_unlock(&server_mutex);									/*释放互斥锁*/
		
	
	}//while(1)
		
}