server.c

linux写的基于epoll技术的通信服务器

{
    char ver_in_client[50]="",client_id[12]="",client_build_id[12]="";
    char ver_in_server[50]="";
    char refuse_list[500]="";
    sprintf(ver_in_client,"%s",msg->ver);
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(client_build_id,"%s",msg->head.build_id);
    if(0!=(query_client_refuse_list_ver(p_mysql,ver_in_server,client_id)))
    {
	lprintf(logd,ERROR,"client:'%s' query refuse list fault !");
	return (-1);
    }
    if (-1!=vercmp(ver_in_client,ver_in_server))
    {
#ifdef DEBUG
	lprintf(logd,DEBUG,"client:'%s' not need update refuse list !");
#endif
	return (-1);
    }
    if(0!=(query_client_refuse_list(p_mysql,refuse_list,client_id)))
    {
	lprintf(logd,ERROR,"client:'%s' query refuse list path fault !");
	return (-1);
    }
    memcpy(&p_to_client_msg->head,&msg->head,sizeof(struct MSG_HEAD));
    p_to_client_msg->need_upt_refuse_list=1;
    sprintf(p_to_client_msg->ver,"%s",ver_in_server);
    sprintf(p_to_client_msg->refuse_list,"%s",refuse_list);
    *need_sendto_client=1;
    return (0); 
}
int do_check_cmd_shell(MYSQL *p_mysql,const msg_client * msg,struct MSG_SERVER * p_to_client_msg,int* need_sendto_client)
{
    char ver_in_client[50]="",client_id[12]="",client_build_id[12]="";
    char ver_in_server[50]="";
    char path[255]="";
    sprintf(ver_in_client,"%s",msg->ver);
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(client_build_id,"%s",msg->head.build_id);
    if (0!=(query_client_cmd_shell_ver(p_mysql,ver_in_server,client_id)))
    {
	lprintf(logd,ERROR,"client:'%s' query cmd shell ver fault !");
	return (-1);
    }
    if (-1!=vercmp(ver_in_client,ver_in_server))
    {
#ifdef DEBUG
	lprintf(logd,DEBUG,"client:'%s' not need exec a new cmd shell !");
#endif
	return (-1);
    }
    if(0!=(query_client_cmd_shell_path(p_mysql,path,client_id)))
    {
	lprintf(logd,ERROR,"client:'%s' query cmd shell file path fault !");
	return (-1);
    }
    memcpy(&p_to_client_msg->head,&msg->head,sizeof(struct MSG_HEAD));
    p_to_client_msg->need_exec_cmd_shell=1;
    sprintf(p_to_client_msg->path,"%s",path);
    *need_sendto_client=1;
    return (0);
}
int do_report_online_status(MYSQL *p_mysql,const msg_client * msg,int  fd)
{
    char client_id[12]="",client_build_id[12]="";
    struct sockaddr_in client_addr;
    socklen_t len;
    char client_ip[15]="";
    int client_port;
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(client_build_id,"%s",msg->head.build_id);
    len=sizeof(struct sockaddr_in);
    getpeername(fd,(struct sockaddr *) &client_addr,&len);
    sprintf(client_ip,"%s",inet_ntoa(client_addr.sin_addr));
    client_port=ntohs(client_addr.sin_port);
    return set_client_online_ip_port_and_time(p_mysql,client_id,client_ip,client_port);
}
int do_report_upt_soft_success(MYSQL *p_mysql,const msg_client * msg)
{
    char client_id[12]="";
    char ver[50]="";
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(ver,"%s",msg->ver);
    return set_curr_client_soft_ver(p_mysql,ver,client_id);
}
int do_report_upt_mat_success(MYSQL *p_mysql,const msg_client * msg)
{
    char client_id[12]="";
    char ver[50]="";
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(ver,"%s",msg->ver);
    return set_curr_client_mat_ver(p_mysql,ver,client_id);
}
int do_report_upt_refuse_list_success(MYSQL *p_mysql,const msg_client * msg)
{
    char client_id[12]="";
    char ver[50]="";
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(ver,"%s",msg->ver);
    return set_curr_client_refuse_list_ver(p_mysql,ver,client_id);
}
int do_report_exec_cmd_shell_success(MYSQL *p_mysql,const msg_client * msg)
{
    char client_id[12]="";
    char ver[50]="";
    sprintf(client_id,"%s",msg->head.terminal_id);
    sprintf(ver,"%s",msg->ver);
    return set_curr_client_cmd_shell_ver(p_mysql,ver,client_id);
}
void * handle_message(unsigned int thread_para[])
{
    int pool_index;  
    int sock_cli;   
    MYSQL * p_mysql=NULL;
    
    int  len;
    int mysql_index;
    msg_client msg;
    msg_server to_client_msg;
    int need_sendto_client=0;
    
    pthread_detach(pthread_self());
    pool_index = thread_para[7];

wait_unlock:
    
    pthread_mutex_lock(s_mutex + pool_index); //

    sock_cli= thread_para[1];
    
    mysql_index= thread_para[7];
    p_mysql= &s_mysql[mysql_index];

    memset(&msg,0,sizeof(struct MSG_CLIENT));
    len = recv(sock_cli, &msg, sizeof(msg_client), MSG_NOSIGNAL);

    memset(&to_client_msg,0,sizeof(msg_server));

    switch (msg.head.msg_type)
    {
	    case MSG_BEG_CHECK_SOFT_VER:
		do_check_soft_ver(&msg, &to_client_msg, &need_sendto_client);
		break;

	    case MSG_BEG_CHECK_SRC_VER:
		do_check_mat_ver(p_mysql,&msg, &to_client_msg, &need_sendto_client);
		break;

	    case MSG_BEG_CHECK_REFUSE_LIST:
		do_check_refuse_list(p_mysql, &msg, &to_client_msg, &need_sendto_client);
		break;

	    case MSG_BEG_CHECK_CMD_SHELL:
		do_check_cmd_shell(p_mysql, &msg, &to_client_msg, &need_sendto_client);
		break;

	    case MSG_REPORT_ONLINE_STATUS:
		do_report_online_status(p_mysql, &msg, sock_cli);
		break;

	    case MSG_REPORT_UPT_SOFT_SUCCESS:
		do_report_upt_soft_success(p_mysql, &msg);
		break;

	    case MSG_REPORT_UPT_MAT_SUCCESS:
		do_report_upt_mat_success(p_mysql, &msg);
		break;

	    case MSG_REPORT_UPT_REFUSE_LIST_SUCCESS:
		do_report_upt_refuse_list_success(p_mysql, &msg);
		break;

	    case MSG_REPORT_EXEC_CMD_SHELL_SUCCESS:
		do_report_exec_cmd_shell_success(p_mysql, &msg);
		break; 
    }
    if (1 == need_sendto_client)
    {
	 send(sock_cli,&to_client_msg,sizeof(struct MSG_SERVER), MSG_NOSIGNAL);
    }

    shutdown( sock_cli, SHUT_RDWR);
    close( sock_cli );

    thread_para[0] = 0;
    goto wait_unlock;

    pthread_exit(NULL);
}

int main(int argc, char **argv)
{
    int 	i, j, rc;
    int 	sock_listen;
    int		sock_cli;
    int		listener;

    int		epfd;
    int		nfds;
    struct epoll_event ev;
    struct epoll_event events[MAXEPOLLSIZE];
    struct rlimit rt;
    socklen_t 	addrlen;
    struct sockaddr_in	 addr4;

    rt.rlim_max = rt.rlim_cur = MAXEPOLLSIZE;

    if (setrlimit(RLIMIT_NOFILE, &rt) == -1) {
        perror("setrlimit");
        exit(1);
    }
    else 
	printf("set system evn source success！\n");

    rc= init_thread_pool();
    if (0 != rc) exit(-1);

    rc=init_mysql_conn_pool();
    if (0 != rc) exit(-1);

    sprintf(s_listen.ip4,"%s", SERVER_IP);
    s_listen.port = SERVER_PORT;
    rc=init_listen4(s_listen.ip4, s_listen.port, 64);
    if (0 > rc){
	fprintf(stderr,"无法创建服务监听于%s: %d\r\n",SERVER_IP,SERVER_PORT);
	exit(-1);
    }
    listener = rc;
    epfd = epoll_create(MAXEPOLLSIZE);
    ev.events = EPOLLIN;
    ev.data.fd = listener;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listener, &ev) < 0 ){
	fprintf(stderr, "向epoll加入socket失败(fd= %d)\r\n",rc);
	exit(-1);
    } 


    for( ; ; ) {
	nfds = epoll_wait(epfd, events, MAXEPOLLSIZE, -1);
	for(i = 0; i< nfds; i++){
	    sock_listen = events[i].data.fd;
	    addrlen = sizeof(struct sockaddr_in);
	    bzero(&addr4, addrlen);
	    sock_cli = accept(sock_listen, (struct sockaddr *)&addr4, &addrlen);
	    if(0 > sock_cli){
		fprintf(stderr, "接收客户端连接失败\n");
		continue;
	    }

	    for(j = 0; j < THREAD_MAX; j++) {
		if (0 == s_thread_para[j][0]) break;
	    }
	    if (j >= THREAD_MAX) {
		if (0 == s_thread_para[j][0]) break;
	    }
	    if (j >= THREAD_MAX){
		fprintf(stderr, "线程池满,连接被放弃\r\n");
		shutdown(sock_cli, SHUT_RDWR);
		close(sock_cli);
		continue;
	    }

	    s_thread_para[j][0] = 1;
	    s_thread_para[j][1] = sock_cli;

	    pthread_mutex_unlock(s_mutex + j);

	}
    }
    exit(0);
}
static int init_mysql_conn_pool(void)
{
    int i;
    for( i = 0; i< THREAD_MAX; i++){
	mysql_init(&s_mysql[i]);
	if (! (mysql_real_connect (&s_mysql[i],DB_SERVER,DB_USER,DB_PASSWD,DB_NAME,0,NULL,0)))
	{
	    fprintf(stderr, "ERROR in connection : %s\n",mysql_error(&s_mysql[i]));
	    return (-1);
	}
    }
    return (0);

}
static int init_thread_pool(void)
{
    int i,j, rc;

    for( i = 0; i< THREAD_MAX; i++){
	mysql_init(&s_mysql[i]);
	if (! (mysql_real_connect (&s_mysql[i],DB_SERVER,DB_USER,DB_PASSWD,DB_NAME,0,NULL,0)))
	{
	    fprintf(stderr, "ERROR in connection : %s\n",mysql_error(&s_mysql[i]));
	    return (-1);
	}
    }
    for(j = 0; j < THREAD_MAX; j++){
	s_thread_para[i][0] = 0;
	s_thread_para[j][7] = j;
	pthread_mutex_lock(s_mutex +j);
    }

    for (i = 0; i< THREAD_MAX; i++) {
	rc = pthread_create(s_tid + i , 0, (void *) handle_message, (void *)(s_thread_para[i]));
	if (0 != rc){
	    fprintf(stderr, "线程创建失败\n");
	    return(-1);
	}
    }

    return(0);
}

static int init_listen4(char *ip4, int port, int max_link)
{
    int		sock_listen4;
    struct sockaddr_in	 addr4;
    unsigned int	optval;
    struct linger      optvall;

    bzero(&addr4, sizeof(addr4));
    inet_pton(AF_INET, ip4, &(addr4.sin_addr));
    addr4.sin_family = AF_INET;
    addr4.sin_port = htons(port);

    sock_listen4 = socket(AF_INET, SOCK_STREAM, 0);
    if (0 > sock_listen4) return(-1);

    optval =0x1;
    optvall.l_linger =60;
    setsockopt(sock_listen4, SOL_SOCKET, SO_LINGER, &optvall, sizeof(struct linger));

    if (0 > bind(sock_listen4, (struct sockaddr *)&addr4, sizeof(addr4))){
	close(sock_listen4);
	return(-1);
    }
    
    if (0 > listen(sock_listen4, max_link)){
	close(sock_listen4);
	return(-1);
    }
    return(sock_listen4);
}