vodfile.cpp

组播服务端程序

		pshared=NULL; 
	}
		
// printf("main normal Quit\n");
    
    	return 0;
}


static  void *sendThread(void *p)
{
    STREAMPACKETHEADER *pPackHeader;
    double startPTS, currPTS;
    int index=0;

    int buflen;	

    sigset_t sigset;
    
    sigemptyset(&sigset);
    sigaddset(&sigset, SIGKILL);
    sigaddset(&sigset, SIGQUIT);
    sigaddset(&sigset, SIGTERM);
    sigaddset(&sigset, SIGINT);
    sigaddset(&sigset, SIGSTOP);
    sigaddset(&sigset, SIGCONT);
    sigprocmask(SIG_BLOCK, &sigset, NULL);

//    set_sched(getpid(),SCHED_RR, 1, -20);	

    dbg_print1(COMMON_LOG,"Enter sendThread %d\n", sendpid); 
//    signal(MAIN_BROADCAST_SIGNAL, SIG_IGN);
    signal(CHILD_QUIT_SIGNAL, child_Quit);


    switch(StreamFileType)
    {
    	
    	case MPEG2_TS_NEW:
    	{
		s64 last_timestamp;
		s64 current_timestamp;
		s64	wait_time;
		s64 iSendDate;
		s64 m_iDeltaClock = 0;
		struct timeval last_time;
		struct timeval current_time;
	
	 	dbg_print0(COMMON_LOG,"Enter (StreamFileType == MPEG2_TS_NEW) %d\n"); 
	
		while(1)
		{
			sem_getvalue(&(pshared->nempty),&buflen);
	
			if(buflen>MAXBUFFLEN/5)
				break;
			else
				usleep(1000);
		}
	
	    	index = getDataSpace();
	    	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	
		current_timestamp = (s64)(pPackHeader->start_timestamp);
	
		GETTIMEOFDAY(&current_time, NULL);
	    				
		// Compute the delta between the clock of the PC and the one of the encoder
		m_iDeltaClock =(s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec) - current_timestamp;
		
	    	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	      		dbg_print0(COMMON_LOG, "send error\n");
	      		sendpid = 0;
		  
	      		return NULL;
	    	} 
		
	    	putEmptySpace();
	
	   	while ( 1 ){
			
	        	index = getDataSpace(); 
				
	        	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
			
	        	if ( pPackHeader->length<=0 ) { // occur error
	          		putEmptySpace();
	          		dbg_print1(COMMON_LOG, "pPackHeader->length return %d\n", pPackHeader->length);
			  
	          		break;
	        	}
	
	
			current_timestamp = (s64)(pPackHeader->start_timestamp);
	
			iSendDate =   current_timestamp+ m_iDeltaClock ;
	
			GETTIMEOFDAY(&current_time, NULL);
	
			wait_time = iSendDate - (s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec);
	
			if(wait_time > 0)
			{
				usleep((long) wait_time);
	
			}
	
	        	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	            		dbg_print0(COMMON_LOG, "send error\n");
	            		putEmptySpace();
				
	           		 break;
	          	}
	
	
	        	putEmptySpace();
			
		}
    		
    	}
     	
    	break;

   	case MPEG1_NEW:
    	{
		s64 last_timestamp;
		s64 current_timestamp;
		s64	wait_time;
		s64 iSendDate;
		s64 m_iDeltaClock = 0;
		struct timeval last_time;
		struct timeval current_time;
	
	 	dbg_print0(COMMON_LOG,"Enter (StreamFileType == MPEG1_NEW) %d\n"); 
	
		while(1)
		{
			sem_getvalue(&(pshared->nempty),&buflen);
	
			if(buflen>MAXBUFFLEN/10)
				break;
			else
				usleep(1000);
		}
	
	    	index = getDataSpace();
	    	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	
		current_timestamp = (s64)(pPackHeader->start_timestamp );
	
		GETTIMEOFDAY(&current_time, NULL);
	    				
		// Compute the delta between the clock of the PC and the one of the encoder
		m_iDeltaClock =(s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec) - current_timestamp;
		
	    	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	      		dbg_print0(COMMON_LOG, "send error\n");
	      		sendpid = 0;
		  
	      		return NULL;
	    	} 
		
	    	putEmptySpace();
	
	   	while ( 1 ){
			
	        	index = getDataSpace(); 
				
	        	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
			
	        	if ( pPackHeader->length<=0 ) { // occur error
	          		putEmptySpace();
	          		dbg_print1(COMMON_LOG, "pPackHeader->length return %d\n", pPackHeader->length);
			  
	          		break;
	        	}
	
	
			current_timestamp = (s64)(pPackHeader->start_timestamp);
	
			iSendDate =   current_timestamp+ m_iDeltaClock ;
	
			GETTIMEOFDAY(&current_time, NULL);
	
			wait_time = iSendDate - (s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec);
	
			if(wait_time > 0)
			{
				usleep((long) wait_time);
	
			}
	
	        	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	            		dbg_print0(COMMON_LOG, "send error\n");
	            		putEmptySpace();
				
	           		 break;
	          	}
	
	
	        	putEmptySpace();
			
		}
    		
    	}
     	
    	break;

   	case MPEG1:
    	{
		s64 last_timestamp;
		s64 current_timestamp;
		s64	wait_time;
		s64 iSendDate;
		s64 m_iDeltaClock = 0;
		struct timeval last_time;
		struct timeval current_time;
	
	 	dbg_print0(COMMON_LOG,"Enter (StreamFileType == MPEG1_NEW) %d\n"); 
	
		while(1)
		{
			sem_getvalue(&(pshared->nempty),&buflen);
	
			if(buflen>MAXBUFFLEN/10)
				break;
			else
				usleep(1000);
		}
	
	    	index = getDataSpace();
	    	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	
		current_timestamp = (s64)((pPackHeader->start_timestamp) * 1000.0 );
	
		GETTIMEOFDAY(&current_time, NULL);
	    				
		// Compute the delta between the clock of the PC and the one of the encoder
		m_iDeltaClock =(s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec) - current_timestamp;
		
	    	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	      		dbg_print0(COMMON_LOG, "send error\n");
	      		sendpid = 0;
		  
	      		return NULL;
	    	} 
		
	    	putEmptySpace();
	
	   	while ( 1 ){
			
	        	index = getDataSpace(); 
				
	        	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
			
	        	if ( pPackHeader->length<=0 ) { // occur error
	          		putEmptySpace();
	          		dbg_print1(COMMON_LOG, "pPackHeader->length return %d\n", pPackHeader->length);
			  
	          		break;
	        	}
	
	
			current_timestamp = (s64)((pPackHeader->start_timestamp) * 1000.0);
	
			iSendDate =   current_timestamp+ m_iDeltaClock ;
	
			GETTIMEOFDAY(&current_time, NULL);
	
			wait_time = iSendDate - (s64)( current_time.tv_sec * 1000000.0 + current_time.tv_usec);
	
			if(wait_time > 0)
			{
				usleep((long) wait_time);
	
			}
	
	        	if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
	            		dbg_print0(COMMON_LOG, "send error\n");
	            		putEmptySpace();
				
	           		 break;
	          	}
	
	
	        	putEmptySpace();
			
		}
    		
    	}
     	
    	break;

    	
    	case MPEG2_WS:
//    	case MPEG1:
    	{
    	   index = getDataSpace(); //wyj: place it before while
    	   pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	
    	   nextPTS = pPackHeader->start_timestamp*1000;
    	   startPTS   = nextPTS;
	
    	   GETTIMEOFDAY(&firstTime, NULL);
    	   secondTime = firstTime;
	
    	   if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
      		dbg_print0(COMMON_LOG, "send error\n");
      		sendpid = 0;
      		
      		return NULL;
    	   } 
	
    	   putEmptySpace();
	
           // add rate limit
    	   if (global_type==MPEG1) {
      		one_peak_token = ONE_SHOT_TIME*MAX_PEAK_RATE/8000000;
      		peak_token = 0;
      		createTimer();
    	   }
        
    	   while ( 1 ){
		
           	index = getDataSpace(); 
           	// wait until time is expired
           	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
		
           	if ( pPackHeader->length<=0 ) { // occur error
          		putEmptySpace();
          		dbg_print1(COMMON_LOG, "pPackHeader->length return %d\n", pPackHeader->length);
          	
          		break;
           	}
		
           	nextPTS = pPackHeader->start_timestamp*1000;
           	GETTIMEOFDAY(&secondTime, NULL);
           	currPTS = startPTS+(secondTime.tv_sec-firstTime.tv_sec)*1000000.0+secondTime.tv_usec-firstTime.tv_usec;
           	firstTime = secondTime;
           	startPTS = currPTS;
           	currPTS = nextPTS - currPTS;
		
           	if ( currPTS>0 ) {
			
          		if ( currPTS>30000000 )
		  		currPTS = 30000000;
		  
          		usleep((long)currPTS); 
           	}
           	else { // update base PTS
          		startPTS = nextPTS;
        	}
/*
        	if ( currPTS<nextPTS ) {
        		//  printf("token %f, nextPTS %f, usleep %ld\n", currPTS, nextPTS, (long)(nextPTS-currPTS));
          		usleep((long)(nextPTS-currPTS)); 
        	}
        	else { // update base PTS
          		startPTS = nextPTS;
        	}
*/        
        	if ( nooutput==0 ) {
			
          		if ( global_type==MPEG1) {
		  	
            			while ( peak_token<MAXPACKETLEN )
              				sigsuspend(&sigset);
			
            			peak_token = 0;
          		}
		  
          		if ( !t_sock.SendMsg((char*)&(pPackHeader->packet), pPackHeader->length) ) {
            			dbg_print0(COMMON_LOG, "send error\n");
            			putEmptySpace();
			
            			break;
          		}
		  
           	}
		
           	putEmptySpace();
		
   	   }// end while (1)

    	   if (global_type==MPEG1) 
   	   	destroyTimer();  
    		
    	}
    	
   	break;
    
        default:
        {
      	   dbg_print1(COMMON_LOG, "unknown file type 0x%x\n", StreamFileType);  
        } 
      	break;  	
    }

   sendpid=0;
   
   dbg_print0(COMMON_LOG , "Leave sendThread\n");
   
   return NULL;
}

static void *main_broadcast(void *p)
{
    sigset_t sigset;
    u_int8_t *start;
    STREAMFILEHEADER allocFileHeader;
    STREAMFILEHEADER *pFileHeader=&allocFileHeader;
    TSFILEHEADER *pTSFileHeader;
    char buffer[MAXPACKETLEN+4*STREAMPACKETHEADER_PREFIX];
    STREAMPACKETHEADER PackHeader, *pPackHeader;
    int ret=0;
    double lastPTS;
    int index=0;
    LIVE_PACKET *packet;
    
    u_int8_t filetype;

//    set_sched(getpid(),SCHED_RR, 1, -20);
//   set_sched(getpid(),SCHED_OTHER, 0, -5);	
    
    dbg_print0(COMMON_LOG ,"Enter main_broadcast\n");
    main_broadcast_err = ERR_NONE;
    
/*
    signal(SIGKILL, SIG_IGN);
    signal(SIGQUIT, SIG_IGN);
    signal(SIGTERM, SIG_IGN);
    signal(SIGSTOP, SIG_IGN);
    signal(SIGINT, SIG_IGN);
    signal(CHILD_QUIT_SIGNAL, SIG_IGN);
*/
    signal(MAIN_BROADCAST_SIGNAL, main_broadcast_Quit);

    dbg_print1(COMMON_LOG ,"Enter main_broadcast, open a file %s.\n", currentprog.filename);


    connfd = 0; 
	
    connfd = open( currentprog.filename, O_RDONLY  );
   
    if( connfd == -1 )
    {
      	main_broadcast_err = ERR_FILE;
      	goto broadcast_quit;
    }

    totalSize = GetFilesize(connfd);
	
    if ( totalSize<=0 )  {
      dbg_print1(FATAL_ERROR,"get %s file size error\n", currentprog.filename);
      main_broadcast_err = ERR_FILE;
	  
      goto broadcast_quit;
    }
    
    dbg_print2(COMMON_LOG,"open a socket %s.%d\n", channel_ip, channel_port);
    
    if (!t_sock.Create(channel_ip, channel_port, lip, lport, S_WYJ_SEND)){
      dbg_print0(FATAL_ERROR,"socket create failure\n");
      main_broadcast_err = ERR_SOCKET;
	  
      goto broadcast_quit;
    }
	
    struct itimerval timer_value;
	
    timer_value.it_interval.tv_sec = timer_value.it_value.tv_sec =0;
    timer_value.it_interval.tv_usec= timer_value.it_value.tv_usec=ONE_SHOT_TIME;
	
    one_token = ONE_SHOT_TIME;