stream_server.c

1、基于韩国at2041芯片开发的Linux环境的DVR代码。 2、以原来代码进行了修改。 3、主要修改网络通讯出现异常问题处理 4、硬盘覆盖录像不起作用

/***
**   HJT
***/

#include "stream_server.h"
#include <fcntl.h>

#include <time.h>
#include <sys/types.h>
#include <utime.h>

#include <syslog.h>	//syslog

#include <net/if.h> 
#include <linux/sockios.h> 

unsigned short QUIT_STREAM;

#define __SELECT__	1

#define MAX_SERVER_NUM 4
pthread_t server_thread[MAX_SERVER_NUM];
int BASE_PORT;
int server_number;

pthread_mutex_t main_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t main_cond = PTHREAD_COND_INITIALIZER;

#define MIN_DELAY 100000

static
void SIG_INT_STREAM()
{		
	QUIT_STREAM=1;
	exit(0);
}

static 
void sig_chld()
{	
	pid_t pid;
	int stat;

	while( (pid = waitpid(-1, &stat ,WNOHANG) ) > 0) {}
	
	return;
}

static int create_shm(int vid_port) //Create Shared Memory Segment Video Display and Video Data
{

	int cnt = 0;
	
////////////////////// Video Shared Memory Segment //////////////////////////
	switch(vid_port)
	{
		case 0:
			for( cnt = 0 ; cnt < MAX_FRAME_IPB ; cnt++)
			{
				shm_key_0_ipb[cnt] = ftok(".",'0') + cnt;
				if( (shm_vid_0_ipb[cnt] = shmget(shm_key_0_ipb[cnt], BUFFER_LEN, IPC_CREAT|IPC_EXCL|0666)) == -1)	
				{
					DEV_LEVEL2("[CH-%d Shared buffer client 0x%X]\n", vid_port, shm_key_0_ipb[cnt]);
					// Segment probably already exists - try as a client
					if( (shm_vid_0_ipb[cnt] = shmget(shm_key_0_ipb[cnt], BUFFER_LEN, 0)) == -1)
					{
						perror("shmget");
						exit(1);
					}
				}
				else DEV_LEVEL2("[CH-%d Shared buffer NEW 0x%X]\n",  vid_port, shm_key_0_ipb[cnt]);
				if( (ch_buff_0_ipb[cnt] = shmat(shm_vid_0_ipb[cnt], 0, 0)) <0 )
				{
					perror("shmat");
					exit(1);
				}
			}			
			break;
			
		case 1:
			for( cnt = 0 ; cnt < MAX_FRAME_IPB ; cnt++)
			{
				shm_key_1_ipb[cnt] = ftok(".",'1') + cnt;
				if( (shm_vid_1_ipb[cnt] = shmget(shm_key_1_ipb[cnt], BUFFER_LEN, IPC_CREAT|IPC_EXCL|0666)) == -1)	
				{
					DEV_LEVEL2("[CH-%d Shared buffer client 0x%X]\n", vid_port, shm_key_1_ipb[cnt]);
					// Segment probably already exists - try as a client
					if( (shm_vid_1_ipb[cnt] = shmget(shm_key_1_ipb[cnt], BUFFER_LEN, 0)) == -1)
					{
						perror("shmget");
						exit(1);
					}
				}
				else DEV_LEVEL2("[CH-%d Shared buffer NEW 0x%X]\n",  vid_port, shm_key_1_ipb[cnt]);
				if( (ch_buff_1_ipb[cnt] = shmat(shm_vid_1_ipb[cnt], 0, 0)) <0 )
				{
					perror("shmat");
					exit(1);
				}
			}
			break;
			
		case 2:
			for( cnt = 0 ; cnt < MAX_FRAME_IPB ; cnt++)
			{
				shm_key_2_ipb[cnt] = ftok(".",'2') + cnt;
				if( (shm_vid_2_ipb[cnt] = shmget(shm_key_2_ipb[cnt], BUFFER_LEN, IPC_CREAT|IPC_EXCL|0666)) == -1)	
				{
					DEV_LEVEL2("[CH-%d Shared buffer client 0x%X]\n", vid_port, shm_key_2_ipb[cnt]);
					// Segment probably already exists - try as a client
					if( (shm_vid_2_ipb[cnt] = shmget(shm_key_2_ipb[cnt], BUFFER_LEN, 0)) == -1)
					{
						perror("shmget");
						exit(1);
					}
				}
				else DEV_LEVEL2("[CH-%d Shared buffer NEW 0x%X]\n",  vid_port, shm_key_2_ipb[cnt]);
				if( (ch_buff_2_ipb[cnt] = shmat(shm_vid_2_ipb[cnt], 0, 0)) <0 )
				{
					perror("shmat");
					exit(1);
				}
			}
			break;

		case 3:
			for( cnt = 0 ; cnt < MAX_FRAME_IPB ; cnt++)
			{
				shm_key_3_ipb[cnt] = ftok(".",'3') + cnt;
				if( (shm_vid_3_ipb[cnt] = shmget(shm_key_3_ipb[cnt], BUFFER_LEN, IPC_CREAT|IPC_EXCL|0666)) == -1)	
				{
					DEV_LEVEL2("[CH-%d Shared buffer client 0x%X]\n", vid_port, shm_key_3_ipb[cnt]);
					// Segment probably already exists - try as a client
					if( (shm_vid_3_ipb[cnt] = shmget(shm_key_3_ipb[cnt], BUFFER_LEN, 0)) == -1)
					{
						perror("shmget");
						exit(1);
					}
				}
				else DEV_LEVEL2("[CH-%d Shared buffer NEW 0x%X]\n",  vid_port, shm_key_3_ipb[cnt]);
				if( (ch_buff_3_ipb[cnt] = shmat(shm_vid_3_ipb[cnt], 0, 0)) <0 )
				{
					perror("shmat");
					exit(1);
				}
			}
			break;

		default:
			break;
	}
	
	return (cnt);
	
}

static 
int create_sock(int port) //Create Listen Socket
{
  	int enable = 1L;
	int server_sock;
	struct sockaddr_in server_addr; /* connector's address information */

	if ((server_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) 
	//if ((server_sock = socket(AF_INET, SOCK_DGRAM, 0)) == -1) 
	{
		DEV_LEVEL0 ("[CH-%d socket error]", port-BASE_PORT);
		perror("socket");
		exit(1);
	}

  	//Allow multiple instance of the client to share the same address and port 
	if(setsockopt(server_sock, SOL_SOCKET, SO_REUSEADDR, (char *) &enable, sizeof(unsigned long int)) < 0) 
		return(0);
	
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(port);
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	bzero(&(server_addr.sin_zero), 8);
	
	if (bind(server_sock, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -1) 
	{
		DEV_LEVEL0 ("[CH-%d binding error]", port-BASE_PORT);
		perror("bind");
		exit(1);
	}

	if(listen(server_sock, LISTENQ) == -1)
	{
		DEV_LEVEL0 ("[CH-%d listening error]", port-BASE_PORT);
		perror("listen");
		exit(1);
	}

	return server_sock;
	
}

//////////////////////// Certification /////////////////////
static 
int stream_certify(int channel, struct loginfo log_info)
{
	FILE *fp1;

	char file_info[MAX_LOG_SIZE*3+2];
	char user_id[MAX_LOG_SIZE];
	char user_pass[MAX_LOG_SIZE];
	char user_group[MAX_LOG_SIZE];

	int i ;
	int j ;
	int val = -1;
	
	if( (fp1 = fopen("/usr/.id_pass","r")) == NULL) {
		return(-1);
		DEV_LEVEL0 ("[Password File Open Error]\n");
	}
	
	DEV_LEVEL2("[CH-%d UserID:%s(%dBytes) Pass:%s(%dBytes)]\n", channel,
		log_info.user_id, strlen(log_info.user_id), log_info.user_pass, strlen(log_info.user_pass));			

	while( fgets(file_info, MAX_LOG_SIZE*3+2, fp1) != NULL) {

		DEV_LEVEL2( "[File Info\t%s]\n",file_info);
	
		j = 0;
		
		for( i = 0  ; i < MAX_LOG_SIZE ; i++) {
			if(file_info[i] != ';') {
				user_id[i] = file_info[i+j];
			}
			else if(file_info[i] == ';') {
				j = i+j+1;
				break;			
			}
		}// UserID
		user_id[i] = '\0';
		DEV_LEVEL2 ("[%s][%s][%d]\n","ID", user_id, strlen(user_id));			
		
		for( i = 0 ; i < MAX_LOG_SIZE ; i++) {
			if(file_info[i+j] != ';') {
				user_pass[i] = file_info[i+j];
			}
			else if(file_info[i+j] == ';') {
				j = i+j+1;
				break;			
			}
		}// UserPass
		user_pass[i] = '\0';
		DEV_LEVEL2("[%s][%s][%d]\n","PASS", user_pass, strlen(user_pass));			
		
		for( i = 0 ; i < MAX_LOG_SIZE ; i++) {
			if(file_info[i+j] != '\n' && file_info[i+j] != '\t' && file_info[i+j] != ' ') {
				user_group[i] = file_info[i+j];
			}
			else if(file_info[i+j] == '\n' || file_info[i+j] == '\t' || file_info[i+j] == ' ') {
				j = i+j+1;
				break;			
			}
		}// UserGroup
		user_group[i] = '\0';		
		DEV_LEVEL2("[%s][%s][%d]\n","GROUP", user_group, strlen(user_group));
		
		if(strlen(user_id) == strlen(log_info.user_id) && strlen(user_pass) == strlen(log_info.user_pass) ) {
			if( !strncmp(user_id, log_info.user_id, strlen(user_id)) && !strncmp(user_pass, log_info.user_pass, strlen(user_pass)) ) {
				if( strlen(SUPER_GROUP) == strlen(user_group) && !strncmp(user_group, SUPER_GROUP, strlen(user_group)) ) {
					val = 0;
					break;
				}
				else if( strlen(NORMAL_GROUP) == strlen(user_group) && !strncmp(user_group, NORMAL_GROUP, strlen(user_group)) ) {
					val = 1;
					break;
				}
			}
			else continue;
		}
		else continue;
	}//End of while(fgets)
	
	fclose(fp1);

	return (val);
			
}
////////////////////////////////////////////////////////////////////////////////////

/*
 * Write to OUTPUT by packets of MAX_SLICE length
 */
static 
int write_slice(int fd, struct strdata *stream_str, unsigned char *data)
{
	pthread_mutex_t write_slice_mutex = PTHREAD_MUTEX_INITIALIZER;
	pthread_cond_t write_slice_cond = PTHREAD_COND_INITIALIZER;

	int WriteByte;
	int r, w;
	unsigned short seq_sz;
	unsigned short seq_nr = 0;
#if 1//xchannel test
	unsigned long t_long;
	unsigned short t_short;
#endif	
	
	STRHDR str_h;

	struct sockaddr *addr;
	size_t *length_ptr;
	int s;

	unsigned short crc_checksum=0;
	int crc_cnt=0;
	
#ifdef __SELECT__	
	fd_set rr_set, ww_set ;
	struct  timeval rtimeout;	
	struct  timeval wtimeout;	

	rtimeout.tv_sec = 0L;    
	rtimeout.tv_usec = 1000L;

	wtimeout.tv_sec = 100L;    
	wtimeout.tv_usec = 1000L;
#endif

	seq_sz = stream_str->frame_size;
	w = 0;		
    	pthread_mutex_lock(&write_slice_mutex);

	memset((struct strhdr *)&str_h, '\0', sizeof(struct strhdr));
	
	while( seq_sz > MAX_STREAM_SLICE)	
	{		
#ifdef __SELECT__	
		FD_ZERO(&rr_set);
		FD_SET(fd, &rr_set);	
		FD_ZERO(&ww_set);
		FD_SET(fd, &ww_set);	
#endif				
		//memset((struct strhdr *)&str_h, '\0', sizeof(struct strhdr));
		memcpy((unsigned char *)str_h.start_code, "penta" , sizeof("penta"));

#if 1//xchannel modify ( big endian -> little endian )
		t_long = ((stream_str->frame_number & 0xff0000) >> 8) | ((stream_str->frame_number & 0xff000000) >> 24) 
			   | ((stream_str->frame_number & 0xff00) << 8) | ((stream_str->frame_number & 0xff) << 24);
		str_h.frame_num = t_long;

		t_short = ((stream_str->frame_size & 0xff00) >> 8) | ((stream_str->frame_size & 0xff) << 8);
		str_h.frame_size = t_short;

		t_short = ((seq_nr & 0xff00) >> 8) | ((seq_nr & 0xff) << 8);
		str_h.slice_num = t_short;
		seq_nr++;

		t_short = ((MAX_STREAM_SLICE & 0xff00) >> 8) | ((MAX_STREAM_SLICE & 0xff) << 8);
		str_h.slice_size = t_short;

		t_short = ((stream_str->frame_id & 0xff00) >> 8) | ((stream_str->frame_id & 0xff) << 8);
		str_h.frame_id = t_short;
		
		//str_h.frame_id = stream_str->frame_crc;

		t_long = ((stream_str->frame_gop & 0xff0000) >> 8) | ((stream_str->frame_gop & 0xff000000) >> 24) 
			   | ((stream_str->frame_gop & 0xff00) << 8) | ((stream_str->frame_gop & 0xff) << 24);
		str_h.frame_gop = t_long;

		t_short = ((stream_str->frame_gop_num & 0xff00) >> 8) | ((stream_str->frame_gop_num & 0xff) << 8);
		str_h.frame_gop_num = t_short;
		

		t_short = ((stream_str->frame_type & 0xff00) >> 8) | ((stream_str->frame_type & 0xff) << 8);