stream_server.c

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

									if( NET_TIME >= MEM_TIME ) snd_cnt=GOP_STR_NUM;
									
								}		
								break;
								
							case 1:								
								if( (str_data_sav.frame_gop == str_data_cur.frame_gop) && 
									(str_data_cur.frame_gop_num-str_data_sav.frame_gop_num==1) )	/*continusly*/
								{
									str_data_sav.frame_gop_num = str_data_cur.frame_gop_num;	
									str_data_sav.frame_gop = str_data_cur.frame_gop;

#if 0//test code
									SEND_P_FLAG = 1;
#endif

									if( SEND_P_FLAG )
									{
										str_data_snd.frame_id = str_data_cur.frame_id;
										str_data_snd.frame_number = str_data_cur.frame_number;
										str_data_snd.frame_size = str_data_cur.frame_size;
										str_data_snd.frame_type = str_data_cur.frame_type;
										str_data_snd.frame_gop = str_data_cur.frame_gop;
										str_data_snd.frame_gop_num++;
										
										gettimeofday( (struct timeval *)&s1_timeout, (struct timezone *)&rok_zone);
										DEV_LEVEL2 ("P:%d:%d\n",  str_data_snd.frame_gop, str_data_snd.frame_number);
										if ( (write_slice_bytes=write_slice(client_sock, (STRDATA *)&str_data_snd, 
											send_buff_ipb[snd_cnt]+sizeof(STRDATA))) <=0 ) 
										{
											stop_flag=1;
											for(mem_cnt=0;mem_cnt<MAX_FRAME_IPB;mem_cnt++) 
												free((unsigned char *)send_buff_ipb[mem_cnt]);
											DEV_LEVEL0("[Client Thread ID %d:%d STOPED]\n", ch_id, client_sock);
											close(client_sock);
											pthread_exit((void*)p_value);
										} 
										gettimeofday( (struct timeval *)&s2_timeout, (struct timezone *)&rok_zone);
										SND_TIME=1000000*(s2_timeout.tv_sec-s1_timeout.tv_sec)+(s2_timeout.tv_usec-s1_timeout.tv_usec);									
										NET_TIME+=(SND_TIME/1000000);
										(long)NET_BYTE+=(long)write_slice_bytes;	

										if( NET_TIME >= MEM_TIME ) snd_cnt=GOP_STR_NUM;
									}
								}
								break;
								
#if 	0
							case 2:
								if( (str_data_sav.frame_gop == str_data_cur.frame_gop) && 
									(str_data_cur.frame_gop_num-str_data_sav.frame_gop_num==1) )
								{
									str_data_sav.frame_gop_num = str_data_cur.frame_gop_num;	
									str_data_sav.frame_gop = str_data_cur.frame_gop;

									if( SEND_P_FLAG && SEND_B_FLAG )
									{
										str_data_snd.frame_id = str_data_cur.frame_id;
										str_data_snd.frame_number = str_data_cur.frame_number;
										str_data_snd.frame_size = str_data_cur.frame_size;
										str_data_snd.frame_type = str_data_cur.frame_type;
										str_data_snd.frame_gop = str_data_cur.frame_gop;
										str_data_snd.frame_gop_num++;

										gettimeofday( (struct timeval *)&s1_timeout, (struct timezone *)&rok_zone);
										if ( (write_slice_bytes=write_slice(client_sock, (STRDATA *)&str_data_snd, 
											send_buff_ipb[snd_cnt]+sizeof(STRDATA))) <=0 ) 
										{
											stop_flag=1;
											for(mem_cnt=0;mem_cnt<MAX_FRAME_IPB;mem_cnt++) 
												free((unsigned char *)send_buff_ipb[mem_cnt]);
											DEV_LEVEL0("[Client Thread ID %d:%d STOPED]\n", ch_id, client_sock);
											close(client_sock);
											pthread_exit((void*)p_value);
										} 
										gettimeofday( (struct timeval *)&s2_timeout, (struct timezone *)&rok_zone);
										SND_TIME=1000000*(s2_timeout.tv_sec-s1_timeout.tv_sec)+(s2_timeout.tv_usec-s1_timeout.tv_usec);									
										NET_TIME+=(SND_TIME/1000000);
										(long)NET_BYTE+=(long)write_slice_bytes;

										if( NET_TIME >= MEM_TIME ) snd_cnt=GOP_STR_NUM;
									 }								
								}	
								break;
#endif

							default:
								break;
						}//end of switch									

						str_data_sav.frame_number = str_data_cur.frame_number;
						
					}//end of if( (str_data_sav.frame_number < str_data_cur.frame_number) )
				}// end of memcpy
				
			}//end of for(buf_count=0;buf_count<MAX_FRAME;buf_count++)

#if 0
			if (NET_RATE ==0) {
				NET_RATE = (long double)NET_BYTE/NET_TIME;
			}
			else {
				NET_RATE = ((long double)NET_BYTE/NET_TIME + NET_RATE)/2;			
			}
#endif			
			NET_RATE = (long double)NET_BYTE/NET_TIME;
#if 0
			switch(ch_id) {
				case 0:
					DEV_LEVEL0 ("\rCH0:%07.2f", (long double)NET_RATE);
					break;
				case 1:
					DEV_LEVEL0 ("\r\t\tCH1:%07.2f", (long double)NET_RATE);
					break;
				case 2:
					DEV_LEVEL0 ("\r\t\t\t\tCH2:%07.2f", (long double)NET_RATE);
					break;
				case 3:
					DEV_LEVEL0 ("\r\t\t\t\t\t\tCH3:%07.2f", (long double)NET_RATE);
					break;
				default :
					break;
			}
#endif			
			if( NET_TIME >= MEM_TIME ) 
			{
				DEV_LEVEL2 ("[%07d Usec][%07d Usec][%07d Bytes][%07.2f Bytes/Sec]\n",
					MEM_TIME, NET_TIME, NET_BYTE, NET_RATE);						
			}			
		}	
		else if (frame_on_off != CH_ON)
		{
			stop_flag=1;
			for(mem_cnt=0;mem_cnt<MAX_FRAME_IPB;mem_cnt++) 
				free((unsigned char *)send_buff_ipb[mem_cnt]);
			DEV_LEVEL0 ("[Client Thread ID %d:%d STOPED]\n", ch_id, client_sock);
			close(client_sock);
			pthread_exit((void*)p_value);
		}
		
	}//end of while( (!quit) && !(stop_flag) ) 


	for(mem_cnt=0;mem_cnt<MAX_FRAME_IPB;mem_cnt++) {
		free((unsigned char *)send_buff_ipb[mem_cnt]);
	}
	DEV_LEVEL0 ("[Client Thread ID %d:%d STOPED]\n", ch_id, client_sock);
	close(client_sock);
	pthread_exit((void*)p_value);
	
}

static 
void *server_loop0(void *args)//Client Thread
{
	pthread_t client_thread;
	int sockserver;
	int sockclient;

	int addr_len = sizeof(struct sockaddr);
	struct sockaddr_in client_addr; // connector's address information 
		
	_streamparams_ stream_params;
	
	int p_value=-1;
	int svr_ch;
	_serverparams_* svr_params = (struct serverparams*)args ;

	svr_ch=svr_params->server_ch;
	sockserver=svr_params->server_sock;

#ifndef __DETACH_THREAD__
	if( pthread_detach(pthread_self()) != 0) perror("pthread_detatch");
	else {
		DEV_LEVEL0("[Server Thread ID %d DETATCHED]\n", svr_ch);	
	}
#endif

	signal(SIGINT, SIG_INT_STREAM);
	//fork();setsid();
	signal(SIGCHLD, sig_chld);
	
	signal(SIGPIPE, SIG_IGN);

 	while( !QUIT_STREAM ) 
 	{
		if((sockclient = accept(sockserver, (struct sockaddr *) &client_addr, &addr_len)) <0) { 
			perror("accept");
			continue;
		}
		DEV_LEVEL0("\n[CH-%d Connection From %s:%d]\n", svr_ch, 	(char *)inet_ntoa(client_addr.sin_addr),ntohs(client_addr.sin_port));

		stream_params.client_sock = sockclient;
		stream_params.client_addr = svr_ch;		
		memcpy(stream_params.client_ip , (char *)inet_ntoa(client_addr.sin_addr) , sizeof(stream_params.client_ip));
		stream_params.client_port = ntohs(client_addr.sin_port);
		
		if(pthread_create(&client_thread , NULL, (void * (*)(void *)) client_loop0, &stream_params) !=0){
			perror ("pthread_create");
		}		
	}//End of while(!Quit)
	
	close(sockclient);
	close(sockserver);
	
	SIG_INT_STREAM();
	
	DEV_LEVEL0("[Server Thread ID %d STOPED]\n", svr_ch);	

	pthread_exit((void*)p_value);
	
}

#if 0
/* Attributes for threads.  */
typedef struct __pthread_attr_s
{
  int __detachstate;
  int __schedpolicy;
  struct __sched_param __schedparam;
  int __inheritsched;
  int __scope;
  size_t __guardsize;
  int __stackaddr_set;
  void *__stackaddr;
  size_t __stacksize;
} pthread_attr_t;

int __detachstate;
int __schedpolicy;
struct __sched_param __schedparam;
struct __sched_param
{
	int __sched_priority;
};
int __inheritsched;
int __scope;
size_t __guardsize;
int __stackaddr_set;
void *__stackaddr;
size_t __stacksize;
#endif

int main(int argc, char *argv[])
{
	_serverparams_ server_params;

	int p_value;

	int server_thread_num;
	int server_port;	
	int server_socket;

	pthread_attr_t *attr_stream;
	struct __sched_param *param_strm;
	int NEW_PRIORITY = 30;

	//void *stackbase;
	//int stack_size_s= PTHREAD_STACK_MIN + BUFFER_LEN;
	//stackbase = (void *) malloc(stack_size_s);	

	// Hook to the break signal
 	signal(SIGINT, SIG_INT_STREAM);

	openlog("STREAMING", LOG_PID, LOG_USER);	//Open LOG
	syslog(LOG_INFO,"STREAM SERVER START"); //Write LOG

	server_number=MAX_SERVER_NUM;
	
	if(argc==1) 
		BASE_PORT=SERVER_PORT;
	else if(argc>=2) 
		BASE_PORT=atoi(argv[1]);
	
////////////////////////// Server Thread /////////////////////////////		
	for(server_thread_num=0;server_thread_num<server_number;server_thread_num++)
	{
		server_port = BASE_PORT+server_thread_num;
		
	 	if( (server_socket = create_sock(server_port)) == -1 ) {
	 		exit(0);
	 	}
	 	else {
	 		if(create_shm(server_thread_num)>0) 
			{
	 			DEV_LEVEL0 ("[OK Video Shared Memory CH-%d]\n", server_thread_num);
	 		}
	 	}

	#if 0		
		//if( pthread_attr_init (attr_stream) != 0 ) perror ("pthread_attr_init");
		//if( pthread_attr_setdetachstate(attr_stream, PTHREAD_CREATE_JOINABLE) != 0) perror ("pthread_attr_setdetachstate");
		//if( pthread_attr_setscope(attr_stream, PTHREAD_SCOPE_SYSTEM) != 0 ) perror ("pthread_attr_setscope");
		//if( pthread_attr_setschedpolicy(attr_stream, SCHED_OTHER) != 0 ) perror ("pthread_attr_setschedpolicy");
		//if( pthread_attr_setinheritsched(attr_stream, PTHREAD_EXPLICIT_SCHED) != 0 ) perror ("pthread_attr_setinheritsched");
		////param_strm->sched_priority = NEW_PRIORITY;
		////if( pthread_attr_setschedparam (attr_stream, (struct __sched_param *)param_strm) != 0 ) perror ("pthread_attr_setschedparam");
		////if( pthread_attr_setstacksize(attr_stream, stack_size_s) != 0 ) perror ("pthread_attr_setstacksize");
		////if( pthread_attr_setstack(attr_stream, stackbase) != 0 ) perror ("pthread_attr_setstack");
		server_params.server_ch=server_thread_num; server_params.server_sock=server_socket;
		if(pthread_create(&server_thread[server_thread_num] , attr_stream, (void * (*)(void *)) server_loop0, &server_params) !=0)
	#endif
		server_params.server_ch=server_thread_num; server_params.server_sock=server_socket;
		if(pthread_create(&server_thread[server_thread_num] , NULL, (void * (*)(void *)) server_loop0, &server_params) !=0)
		{
			perror ("pthread_create");
		}
		
	#if 0
			DEV_LEVEL0("PTHREAD_CREATE_DETACHED %d\n",PTHREAD_CREATE_DETACHED);
			DEV_LEVEL0("PTHREAD_CREATE_JOINABLE %d\n",PTHREAD_CREATE_JOINABLE);
			DEV_LEVEL0("PTHREAD_SCOPE_SYSTEM %d\n",PTHREAD_SCOPE_SYSTEM);
			DEV_LEVEL0("PTHREAD_SCOPE_PROCESS %d\n",PTHREAD_SCOPE_PROCESS);
			DEV_LEVEL0("SCHED_OTHER %d\n",SCHED_OTHER);
			DEV_LEVEL0("PTHREAD_INHERIT_SCHED %d\n",PTHREAD_INHERIT_SCHED); 
			DEV_LEVEL0("PTHREAD_EXPLICIT_SCHED %d\n",PTHREAD_EXPLICIT_SCHED); 
			DEV_LEVEL0("PTHREAD_STACK_MIN %d\n",PTHREAD_STACK_MIN); 
			
			DEV_LEVEL0("\n[__detachstate %d]\n", attr_stream->__detachstate);
			DEV_LEVEL0("\n[__schedpolicy %d]\n", attr_stream->__schedpolicy);
			DEV_LEVEL0("\n[__sched_priority %d]\n", attr_stream->__schedparam.__sched_priority);
			DEV_LEVEL0("\n[__inheritsched %d]\n", attr_stream->__inheritsched);				
			DEV_LEVEL0("\n[__scope %d]\n", attr_stream->__scope);
			DEV_LEVEL0("\n[__guardsize %d]\n", attr_stream->__guardsize);
			DEV_LEVEL0("\n[__stackaddr_set %d]\n", attr_stream->__stackaddr_set);
			DEV_LEVEL0("\n[__stackaddr %d]\n", attr_stream->__stackaddr);
			DEV_LEVEL0("\n[__stacksize %d]\n", attr_stream->__stacksize);
	#endif		
		//if( pthread_attr_destroy (attr_stream) != 0 ) {perror ("pthread_attr_destroy");}
		
		usleep(MIN_DELAY*3); 		
	}
#if 0	
	if(pthread_create(&ftp_server_thread , NULL, (void * (*)(void *)) ftp_server_loop, &server_params) !=0)
	{
		perror ("pthread_create");
	}
#endif
///////////////////////////////////////////////////////////////////////////////////////

	QUIT_STREAM = 0;

#ifndef __DAEMON__
 	while( !QUIT_STREAM ) 
 	{
	 	DEV_LEVEL0("[Stream Main Control] pthread_cond_wait\n");
	  	pthread_mutex_lock(&main_mutex);
	  	pthread_cond_wait(&main_cond, &main_mutex);	  	
	  	pthread_mutex_unlock(&main_mutex);
	}
#else
	for(server_thread_num=0;server_thread_num<server_number;server_thread_num++)
	{
		close(server_socket[server_thread_num]);
	}
#endif

	SIG_INT_STREAM();
	
	DEV_LEVEL0("[OK Main Server closing %d]\n", QUIT_STREAM);
	
	return 0;

}