alc_tx.c

这个程序实现了FLUTE协议

	}

	def_lct_hdr->hdr_len = hdrlen >> 2; /* Header length in 32-bit words */	
	*(unsigned short*)def_lct_hdr = htons(*(unsigned short*)def_lct_hdr);

	sendlen = hdrlen;

	if(ch->s->cc_id == RLC) {
		retval = mad_rlc_fill_header(ch->s, (rlc_hdr_t*)(sendbuf + 4), ch->ch_id);
	}
                	
	retval = sendto(ch->tx_sock, (char *)sendbuf, sendlen, 0,				
		ch->addrinfo->ai_addr, ch->addrinfo->ai_addrlen);

	add_session_sent_bytes(s_id, sendlen);

        /* Should we take UDP/IP headers into account? */
        /* UDP */
        add_session_sent_bytes(s_id, 8);
        /* IPv4 or IPv6 */
        if(ch->s->addr_family == PF_INET) {
          add_session_sent_bytes(s_id, 20);
        }
        else {
          add_session_sent_bytes(s_id, 40);
        }

	return retval;
}

/*  
 * This function.
 * 
 * Params:	void *s: Pointer to session.	
 *
 * Return:	void
 *
 */

void* tx_thread(void *s) {

  alc_session_t *session;
  alc_channel_t *channel;
  tx_queue_t *tmp_ptr;
  
  tx_queue_t *next_pkt;
  tx_queue_t *pkt;
  
  int i, j;
  int retcode;
  int packet_length;
  
  double interval;
  double packetpersec;
  double currenttime;
  double lasttime;
  
  double loss_prob;
  
  session = (alc_session_t *)s;
  
  packet_length = calculate_packet_length(session);
  
  packetpersec = ((double)(session->def_tx_rate * 1000) / (double)(packet_length * 8));   
  interval = ((double)1 / packetpersec);
  
  /* interval is too small for FDT Instance, because in packet length calculation FDT Instance's
     extra header fields are not counted */
  
  while(session->state == SActive) {	
    
    if(session->tx_queue_begin == NULL) {
#ifdef WIN32
      Sleep(1);
#else
      usleep(1000);
#endif		
      continue;
    }
    else {
      break;
    }
  }
  
  lasttime = sec();
  
  while(session->state == SActive) {
    
    currenttime = sec();
    
	if(session->calculate_session_size == false) {

		while(currenttime < (lasttime + interval)) {
      
			if(session->optimize_tx_rate) {
#ifdef WIN32
				Sleep(0);
#else
				usleep(0);
#endif
			}
			else {
#ifdef WIN32
				Sleep(1);
#else
				usleep(1000);
#endif
			}
			currenttime = sec();
		}
	}
    
    for(i = 0; i < session->nb_channel; i++) {
      
      channel = session->ch_list[i];
      
      if(session->cc_id == RLC) {
	
	if(channel->start_sending == false) {
	  continue;
	}
	
	if(channel->wait_after_sp > 0) {
	  channel->wait_after_sp--;
	  continue;
	}
      }
      
      for(j = 0; j < channel->nb_tx_units; j++) {
	
	if(channel->queue_ptr == NULL) {
	  
	  if(channel->ready == false) {
	    session->nb_ready_channel++;
	    channel->ready = true;
	  }
	  
	  if(session->nb_ready_channel == session->nb_sending_channel) {
	    
	    pkt = session->tx_queue_begin;
	    
	    while(pkt != NULL) {
	      next_pkt = pkt->next;
	      free(pkt->data);
	      free(pkt);
	      pkt = next_pkt;
	    }	
	    
	    session->tx_queue_begin = NULL;
	    session->tx_queue_size = 0;
	  }
	  
	  break;
	}
	
	if(session->first_unit_in_loop) {
	  
	  if(session->cc_id == RLC) {
	    mad_rlc_reset_tx_sp(session);
	  }
	  
	  session->first_unit_in_loop = false;
	}
	if(session->cc_id == RLC) {
	  
	  retcode = mad_rlc_fill_header(session, (rlc_hdr_t*)(channel->queue_ptr->data + 4),
					channel->ch_id);
	  
	  if(retcode < 0) {
	  }
	}
	
	loss_prob = 0;
	
	if(session->simul_losses) {
	  if(channel->previous_lost == true) {
	    loss_prob = channel->s->loss_ratio2; /*P_LOSS_WHEN_LOSS;*/
	  }
	  else {
	    loss_prob = channel->s->loss_ratio1; /*P_LOSS_WHEN_OK;*/
	  }
	}
	
	if(!randomloss(loss_prob)) {
  	
		if(session->calculate_session_size == false) {
			retcode = sendto(channel->tx_sock, (char*)channel->queue_ptr->data,	
				channel->queue_ptr->datalen, 0, channel->addrinfo->ai_addr,
				channel->addrinfo->ai_addrlen);

			if(retcode < 0) {
	    
#ifdef WIN32
				printf("sendto failed with: %d\n", WSAGetLastError());
#else
				printf("sendto failed with: %d\n", h_errno);
#endif
				break;
			}
		}
	  
	  add_session_sent_bytes(session->s_id, channel->queue_ptr->datalen);

	  /* Should we take UDP/IP headers into account? */
	  /* UDP */
	  add_session_sent_bytes(session->s_id, 8);
	  /* IPv4 or IPv6 */
	  if(session->addr_family == PF_INET) {
	    add_session_sent_bytes(session->s_id, 20);
	  }
	  else {
	    add_session_sent_bytes(session->s_id, 40);
	  }

	  channel->previous_lost = false;
	}
	else {
	  channel->previous_lost = true;
	}
	
	channel->queue_ptr->nb_tx_ch++;
	
	if(channel->queue_ptr->nb_tx_ch == (unsigned int)session->nb_channel) {
	  
	  tmp_ptr = channel->queue_ptr->next;
	  free(channel->queue_ptr->data);
	  free(channel->queue_ptr);
	  
	  channel->queue_ptr = tmp_ptr;
	  session->tx_queue_begin = tmp_ptr;
	  session->tx_queue_size--;
	}
	else {
	  channel->queue_ptr = channel->queue_ptr->next;
	}
      }
    }
    
    lasttime += interval;
  }

#ifdef WIN32
  _endthread();
#else
  pthread_exit(0);
#endif

  return NULL;
}

/*  
 * This function calculates packet length used in session.
 * 
 * Params:	alc_session_t *s: Pointer to session.		
 *
 * Return:	int: Packet length for this session.
 *
 */

int calculate_packet_length(alc_session_t *s) {

	int packet_length = 0;

	if(s->addr_family == PF_INET) {

		if(s->use_fec_oti_ext_hdr == 1) {

			/* eslen + DEF_LCT_HDR + TSI + TOI + EXT_FTI + FEC_PL_ID + UDP + IP */

			if(((s->def_fec_enc_id == COM_NO_C_FEC_ENC_ID) ||
				(s->def_fec_enc_id == COM_FEC_ENC_ID))) {	 	
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 16 + 4 + 8 + 20); 
			}
			else if(((s->def_fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_LB_E_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_SYS_FEC_ENC_ID))) {
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 16 + 8 + 8 + 20); 
			}
		}
		else if(s->use_fec_oti_ext_hdr == 0) {
			/* eslen + DEF_LCT_HDR + TSI + TOI + FEC_PL_ID + UDP + IP */ 

			if(((s->def_fec_enc_id == COM_NO_C_FEC_ENC_ID) ||
				(s->def_fec_enc_id == COM_FEC_ENC_ID))) {	
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 4 + 8 + 20); 
			}
			else if(((s->def_fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_LB_E_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_SYS_FEC_ENC_ID))) {
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 8 + 8 + 20); 
			}
		}
	}
	else if(s->addr_family == PF_INET6) {
		if(s->use_fec_oti_ext_hdr == 1) {
			/* eslen + DEF_LCT_HDR + TSI + TOI + EXT_FTI + FEC_PL_ID + UDP + IP */

			if(((s->def_fec_enc_id == COM_NO_C_FEC_ENC_ID) ||
				(s->def_fec_enc_id == COM_FEC_ENC_ID))) {	 	
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 16 + 4 + 8 + 40); 
			}
			else if(((s->def_fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_LB_E_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_SYS_FEC_ENC_ID))) {
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 16 + 8 + 8 + 40); 
			}
		}
		else if(s->use_fec_oti_ext_hdr == 0) {
			/* eslen + DEF_LCT_HDR + TSI + TOI + FEC_PL_ID + UDP + IP */

			if(((s->def_fec_enc_id == COM_NO_C_FEC_ENC_ID) ||
				(s->def_fec_enc_id == COM_FEC_ENC_ID))) {	 	
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 4 + 8 + 20); 
			}
			else if(((s->def_fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_LB_E_FEC_ENC_ID) ||
				 (s->def_fec_enc_id == SB_SYS_FEC_ENC_ID))) {
				packet_length =  s->def_eslen + (sizeof(def_lct_hdr_t) + 4 + 4 + 8 + 8 + 20); 
			}
		}
	}

	return packet_length;	
}

/*  
 * This function adds packet to tx queue.
 * 
 * Params:	alc_session_t *s: Pointer to session,
 *			unsigned char* sendbuf: Pointer to data to be added to queue,
 *			unsigned int sendlen: Lenght of data.
 *
 * Return:	int: > 0 in success, 0 or -1 in error cases.
 *
 */

int add_pkt_to_tx_queue(alc_session_t *s, unsigned char *sendbuf,  unsigned int sendlen) {
	int retval = 0;
	tx_queue_t *pkt;
	int i;
	alc_channel_t *ch;

	if(s->tx_queue_size == MAX_TX_QUEUE_SIZE) {
		return retval;
	}

	/* Allocate memory for pkt */
	if(!(pkt = (tx_queue_t*)calloc(1, sizeof(tx_queue_t)))) {
		printf("Could not alloc memory for tx_queue pkt!\n");
		return -1;
	}

	pkt->nb_tx_ch = 0;
	pkt->next = NULL;
	pkt->datalen = sendlen;
	pkt->data = sendbuf;

	if(s->tx_queue_begin == NULL) {

		s->tx_queue_begin = pkt;
		s->tx_queue_end = pkt;

		for(i = 0; i < s->nb_channel; i++) {
			ch = s->ch_list[i];
			ch->ready = false;
			ch->queue_ptr = s->tx_queue_begin;

			if(s->cc_id == RLC) {
				if(ch->ch_id != 0) {
					ch->wait_after_sp = RLC_WAIT_AFTER_SP;	
					ch->start_sending = false;
				}
			}
		}

		s->nb_ready_channel = 0;

		if(s->cc_id == RLC) {
			s->nb_sending_channel = 1;
		}
	}
	else {
		s->tx_queue_end->next = pkt;
		s->tx_queue_end = pkt;

		for(i = 0; i < s->nb_channel; i++) {
			ch = s->ch_list[i];
			
			if(ch->queue_ptr == NULL) {
				ch->queue_ptr = s->tx_queue_end;
			}
		}
	}

	s->tx_queue_size++;

	return sendlen;
}