alc_rx.c

这个程序实现了FLUTE协议

	    return HDR_ERROR;
	  }
	}
	
	if(((fec_enc_id == COM_NO_C_FEC_ENC_ID) || (fec_enc_id == SB_LB_E_FEC_ENC_ID)
	    || (fec_enc_id == COM_FEC_ENC_ID) || (fec_enc_id == SIMPLE_XOR_FEC_ENC_ID))) {
	  
	  es_len = (word & 0x0000FFFF);
	  
	  max_sb_len = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
	  hdrlen += 4;
	  exthdrlen--;
	}
	else if(fec_enc_id == SB_SYS_FEC_ENC_ID) {
	  
	  es_len = (word & 0x0000FFFF);
	  
	  word = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
	  
	  max_sb_len = ((word & 0xFFFF0000) >> 16);
	  max_nb_of_es = (word & 0x0000FFFF);
	  
	  hdrlen += 4;
	  exthdrlen--;
	  
	}
	break;
	
      case EXT_AUTH:
	/* ignore */
	hdrlen += (hel-1) << 2;
	exthdrlen -= (hel-1);
	break;
	
      case EXT_NOP:
	/* ignore */
	hdrlen += (hel-1) << 2;
	exthdrlen -= (hel-1);
	break;
	
      default:
	
	printf("Unknown LCT Extension header, het: %i\n", het);
	return HDR_ERROR;
	break;
      }
    }
  }
  
  if((hdrlen >> 2) != def_lct_hdr->hdr_len) {
    /* Wrong header length */
    printf("analyze_packet: packet header length %d, should be %d\n", (hdrlen >> 2),
	   def_lct_hdr->hdr_len);
    return HDR_ERROR;
  }
  
  /* Check if we have an empty packet without FEC Payload ID */
  if(hdrlen == len) {
    return EMPTY_PACKET;		
  }
  
  if(((toi == 0) && (is_received_instance(ch->s, fdt_instance_id)))) {
    return DUP_PACKET;
  }
  
  if((fec_enc_id == COM_NO_C_FEC_ENC_ID) || (fec_enc_id ==  COM_FEC_ENC_ID)) {
    
    if(len < hdrlen + 4) {
      printf("analyze_packet: packet too short %d\n", len);
      return HDR_ERROR;
    }
    
    word = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
    sbn = (word >> 16);
    esi = (word & 0xFFFF);
    hdrlen += 4;
  }
  else if(((fec_enc_id == SB_LB_E_FEC_ENC_ID) || (fec_enc_id == SIMPLE_XOR_FEC_ENC_ID))) {
    if (len < hdrlen + 8) {
      printf("analyze_packet: packet too short %d\n", len);
      return HDR_ERROR;
    }
    
    sbn = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
    hdrlen += 4;
    esi = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
    hdrlen += 4;
    
  }
  else if(fec_enc_id == SB_SYS_FEC_ENC_ID) {
    if (len < hdrlen + 8) {
      printf("analyze_packet: packet too short %d\n", len);
      return HDR_ERROR;
    }
    
    sbn = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
    
    hdrlen += 4;
    word = ntohl(*(unsigned int*)((char*)def_lct_hdr + hdrlen));
    sb_len = (word >> 16);
    esi = (word & 0xFFFF);
    hdrlen += 4;
  }
  
  /* TODO: check if instance_id is set --> EXT_FDT header exists in packet */
  
  if(len - hdrlen != 0) {
    
    /* check if we have enough information */
    
    if(((toi_len == 0) || (fec_enc_id == -1) || ((fec_enc_id > 127) && (fec_inst_id == -1)) ||
	(es_len == 0) || (max_sb_len == 0))) {
#ifdef WIN32
      printf("Not enough information to create Transport Object, TOI: %I64u\n", toi);
#else
      printf("Not enough information to create Transport Object, TOI: %llu\n", toi);
#endif
      fflush(stdout);
      return HDR_ERROR;
    }
    
    /* Create transport unit */
    trans_unit = create_units(1);
    
    if(trans_unit == NULL) {
      return MEM_ERROR;
    }
    
    trans_unit->esi = esi;
    trans_unit->len = len - hdrlen; /* Data length */
    
    /* Alloc memory for incoming TU data */
    if(!(trans_unit->data = (char*)calloc(es_len, sizeof(char)))) { /* rs_fec&xor_fec: trans_unit->len --> eslen */
      printf("Could not alloc memory for transport unit's data!\n");
      return MEM_ERROR;
    }
    
    memcpy(trans_unit->data, (data + hdrlen), trans_unit->len);
    
    /* Check if object already exist */
    if(toi == FDT_TOI) {
      trans_obj = object_exist(fdt_instance_id, ch->s, 0);
    }
    else {
      trans_obj = object_exist(toi, ch->s, 1);
    }
    
    if(trans_obj == NULL) {
      
      trans_obj = create_object();
      
      if(trans_obj == NULL) {
	return MEM_ERROR;
      }
      
      if(toi == FDT_TOI) {
	trans_obj->toi = fdt_instance_id;
	trans_obj->content_enc_algo = content_enc_algo;
      }
      else {
	trans_obj->toi = toi;
	
	if(ch->s->rx_memory_mode == 1 || ch->s->rx_memory_mode == 2) {
	  
	  memset(filename, 0, MAX_PATH);
	  
	  if(content_enc_algo == 0) {
#ifdef WIN32
	    sprintf(filename, "%s\\object%I64u", ch->s->base_dir, toi);
#else
	    sprintf(filename, "%s/object%llu", ch->s->base_dir, toi);
#endif
	  }
#ifdef USE_ZLIB
	  else if(content_enc_algo == GZIP) {
#ifdef WIN32
	    sprintf(filename, "%s\\object%I64u%s", ch->s->base_dir, toi, GZ_SUFFIX);
#else
	    sprintf(filename, "%s/object%llu%s", ch->s->base_dir, toi, GZ_SUFFIX);
#endif
	  }
#endif
	  else if(content_enc_algo == PAD) {
#ifdef WIN32
	    sprintf(filename, "%s\\object%I64u%s", ch->s->base_dir, toi, PAD_SUFFIX);
#else
	    sprintf(filename, "%s/object%llu%s", ch->s->base_dir, toi, PAD_SUFFIX);
#endif
	  }
	  
	  /* Alloc memory for tmp_filename */
	  if(!(trans_obj->tmp_filename = (char*)calloc(strlen(filename)+1, sizeof(char)))) {
	    printf("Could not alloc memory for tmp_filename!\n");
	    return MEM_ERROR;
	  }
	  
	  memcpy(trans_obj->tmp_filename, filename, strlen(filename));
	  
#ifdef WIN32
	  if((trans_obj->fd = open((const char*)trans_obj->tmp_filename,
				   _O_WRONLY | _O_CREAT | _O_BINARY | _O_TRUNC , _S_IWRITE)) < 0) {
#else
	    if((trans_obj->fd = open64(trans_obj->tmp_filename,
				     O_WRONLY | O_CREAT | O_TRUNC , S_IRWXU)) < 0) {
#endif
	      printf("Error: unable to open file %s\n", trans_obj->tmp_filename);
	      fflush(stdout);
	      return MEM_ERROR;
	    }
	  }

	  if(ch->s->rx_memory_mode == 2) {
	    
	    /* when receiver is in large file mode a tmp file is used to store the data symbols */

	    memset(filename, 0, MAX_PATH);
#ifdef WIN32
	    sprintf(filename, "%s\\st%I64u", ch->s->base_dir, toi);
#else
	    sprintf(filename, "%s/st%llu", ch->s->base_dir, toi);
#endif
	    
	    /* Alloc memory for tmp_st_filename */
	    if(!(trans_obj->tmp_st_filename = (char*)calloc(strlen(filename)+1, sizeof(char)))) {
	      printf("Could not alloc memory for tmp_st_filename!\n");
	      return MEM_ERROR;
	    }
	    
	    memcpy(trans_obj->tmp_st_filename, filename, strlen(filename));
	    
#ifdef WIN32
	    if((trans_obj->fd_st = open((const char*)trans_obj->tmp_st_filename,
					_O_RDWR | _O_CREAT | _O_BINARY | _O_TRUNC , _S_IREAD | _S_IWRITE)) < 0) {
#else
	    if((trans_obj->fd_st = open64(trans_obj->tmp_st_filename,
					O_RDWR | O_CREAT | O_TRUNC , S_IRWXU)) < 0) {
#endif
	      printf("Error: unable to open file %s\n", trans_obj->tmp_st_filename);
	      fflush(stdout);
	      return MEM_ERROR;
	    }  
	  }
	}  
	
	trans_obj->len = toi_len;
	trans_obj->fec_enc_id = (unsigned char)fec_enc_id;
	trans_obj->fec_inst_id = (unsigned short)fec_inst_id;
	trans_obj->es_len = es_len;
	trans_obj->max_sb_len = max_sb_len;
	
	/* Let's calculate the blocking structure for this object */
	
	trans_obj->bs = compute_blocking_structure(toi_len, max_sb_len, es_len);

	if (!(trans_obj->block_list = (trans_block_t*)calloc(trans_obj->bs->N, sizeof(trans_block_t)))) {
	  printf("Could not alloc memory for transport block list!\n");
	  return MEM_ERROR;
        }
	
	for(i=0; i < trans_obj->bs->N; i++) {
          trans_block = trans_obj->block_list+i;
          trans_block->nb_of_rx_units = 0;
        }

	if(toi == FDT_TOI) {
	  insert_object(trans_obj, ch->s, 0);
	}
	else {
	  insert_object(trans_obj, ch->s, 1);
	}
  }

      /* Check if block already exist */

      /*

      trans_block = block_exist(sbn, trans_obj);

      if(trans_block == NULL) {
	
	trans_block = create_block();
	
	if(trans_block == NULL) {
	  return MEM_ERROR;
	}

      */

  trans_block = trans_obj->block_list+sbn;

  if(trans_block->nb_of_rx_units == 0) {

    trans_block->sbn = sbn;
    trans_block->nb_of_rx_units = 0;
	
    if(fec_enc_id == COM_NO_C_FEC_ENC_ID) { 
      
      if(sbn < trans_obj->bs->I) {
	trans_block->k = trans_obj->bs->A_large;
      }
      else {
	trans_block->k = trans_obj->bs->A_small;
      }
    }
    else if(fec_enc_id == SB_SYS_FEC_ENC_ID) {
      
      trans_block->k = sb_len;
      trans_block->max_k = max_sb_len;
      trans_block->max_n = max_nb_of_es;
    }
    else if(fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) {
      
      if(sbn < trans_obj->bs->I) {
	trans_block->k = trans_obj->bs->A_large;
      }
      else {
	trans_block->k = trans_obj->bs->A_small;
      }
      trans_block->max_k = max_sb_len;
    }
    /*
      insert_block(trans_block, trans_obj);
    */
  }
      
  if(!block_ready_to_decode(trans_block)) {
	
    if(insert_unit(trans_unit, trans_block, trans_obj) != 1) {

      /* if large file mode data symbol is stored in the tmp file */
      
      if(toi == FDT_TOI || ch->s->rx_memory_mode == 0) { 
	
	if(block_ready_to_decode(trans_block)) {
	  trans_obj->nb_of_ready_blocks++;
	}
      }

      if(toi != FDT_TOI && ch->s->rx_memory_mode == 2) {

#ifdef WIN32
	trans_unit->offset = _lseeki64(trans_obj->fd_st, 0, SEEK_END);
#else
	trans_unit->offset = lseek64(trans_obj->fd_st, 0, SEEK_END);
#endif
	if(trans_unit->offset == -1) {
#ifdef WIN32
	  printf("lseek error, toi: %I64u\n", toi);
#else
	  printf("alc_rx.c line 951 lseek error, toi: %llu\n", toi);
#endif
	  fflush(stdout);
	  set_session_state(ch->s->s_id, SExiting);
	  return MEM_ERROR;
	}
	
	if(write(trans_obj->fd_st, trans_unit->data, (unsigned int)trans_unit->len) == -1) {
#ifdef WIN32
	  printf("write error, toi: %I64u, sbn: %i\n", toi, sbn);
#else
	  printf("write error, toi: %llu, sbn: %i\n", toi, sbn);
#endif
	  fflush(stdout);
	  set_session_state(ch->s->s_id, SExiting);
	  return MEM_ERROR;
	}
	
	free(trans_unit->data);
	trans_unit->data = NULL;
      }
      
      if(((toi == FDT_TOI && ch->s->verbosity == 4) || (toi != FDT_TOI && ch->s->verbosity > 1))) {
#ifdef WIN32
	rx_percent = (double)((double)100 *
			      ((double)(LONGLONG)trans_obj->rx_bytes/(double)(LONGLONG)trans_obj->len)); 
	
	if(((rx_percent >= (trans_obj->last_print_rx_percent + 1)) || (rx_percent == 100))) {
	  trans_obj->last_print_rx_percent = rx_percent;
	  printf("%.2f%% of object received (TOI=%I64u LAYERS=%i)\n", rx_percent,
		 toi, ch->s->nb_channel);
	  fflush(stdout);
	}
#else
	rx_percent = (double)((double)100 *
			      ((double)(long long)trans_obj->rx_bytes/(double)(long long)trans_obj->len));
	
	if(rx_percent >= (trans_obj->last_print_rx_percent + 1)) {
	  trans_obj->last_print_rx_percent = rx_percent;
     	  printf("%.2f%% of object received (TOI=%llu LAYERS=%i)\n", rx_percent,
		 toi, ch->s->nb_channel);
	  fflush(stdout);
	}
#endif
      }
    }
    else {
      free(trans_unit->data);
      free(trans_unit);
      return DUP_PACKET;
    }
  }
  else {
    free(trans_unit->data);
    free(trans_unit);
    return DUP_PACKET;
  }
  
  if(toi != FDT_TOI) {
    
    if(ch->s->rx_memory_mode == 1 || ch->s->rx_memory_mode == 2) {	
      
      if(block_ready_to_decode(trans_block)) {
	
	if(ch->s->rx_memory_mode == 2){
	  
	  /* We have to restore the data symbols to trans_units from the symbol store tmp file */
	  
	  next_tu = trans_block->unit_list;
	  
	  while(next_tu != NULL) {
	    
	    tu = next_tu;
	    
#ifdef WIN32
	    if(_lseeki64(trans_obj->fd_st, tu->offset, SEEK_SET) == -1) {
#else
	      if(lseek64(trans_obj->fd_st, tu->offset, SEEK_SET) == -1) {
#endif
		
#ifdef WIN32
		printf("lseek error, toi: %I64u\n", toi);
#else
		printf("alc_rx.c line 1035 lseek error, toi: %llu\n", toi);
#endif
		fflush(stdout);
		set_session_state(ch->s->s_id, SExiting);
		return MEM_ERROR;
	      }
	      
	      /* let's copy the data symbols from the tmp file to the memory */
	      
	      /* Alloc memory for restoring data symbol */
	      
	      if(!(tu->data = (char*)calloc(es_len, sizeof(char)))) { /* rs_fec&xor_fec: trans_unit->len --> eslen */
		printf("Could not alloc memory for transport unit's data!\n");
		return MEM_ERROR;
	      }
	      
	      if(read(trans_obj->fd_st, tu->data, tu->len) == -1) {
#ifdef WIN32
		printf("read error, toi: %I64u, sbn: %i\n", toi, sbn);
#else
		printf("read error, toi: %llu, sbn: %i\n", toi, sbn);
#endif
		fflush(stdout);
		set_session_state(ch->s->s_id, SExiting);
		return MEM_ERROR;
	      }
	      
	      next_tu = tu->next;
	    }
	  }
	  
	  /* decode the block and save data to the tmp file */
	  
	  if(fec_enc_id == COM_NO_C_FEC_ENC_ID) {
	    buf = null_fec_decode_src_block(trans_block, &block_len, es_len);
	  }
	  else if(fec_enc_id == SIMPLE_XOR_FEC_ENC_ID) {
	    buf = xor_fec_decode_src_block(trans_block, &block_len, es_len);
	  }
	  
	  else if(fec_enc_id == SB_SYS_FEC_ENC_ID && fec_inst_id == REED_SOL_FEC_INST_ID) {		
	    buf = rs_fec_decode_src_block(trans_block, &block_len, es_len);
	  }
	  if(buf == NULL) {
	    return MEM_ERROR;
	  }
#ifdef WIN32
	  if(trans_block->sbn < trans_obj->bs->I) {
	    pos = ( (ULONGLONG)trans_block->sbn * (ULONGLONG)trans_obj->bs->A_large * (ULONGLONG)es_len );
	  }
	  else {
	    pos = ( ( ( (ULONGLONG)trans_obj->bs->I * (ULONGLONG)trans_obj->bs->A_large ) + 
		      ( (ULONGLONG)trans_block->sbn - (ULONGLONG)trans_obj->bs->I )  * (ULONGLONG)trans_obj->bs->A_small ) * (ULONGLONG)es_len );
	  }
#else
	  if(trans_block->sbn < trans_obj->bs->I) {
	    pos = ( (unsigned long long)trans_block->sbn * (unsigned long long)trans_obj->bs->A_large * (unsigned long long)es_len );
	  }
	  else {
	    pos = ( ( ( (unsigned long long)trans_obj->bs->I * (unsigned long long)trans_obj->bs->A_large ) + 
		      ( (unsigned long long)trans_block->sbn - (unsigned long long)trans_obj->bs->I )  * (unsigned long long)trans_obj->bs->A_small ) * (unsigned long long)es_len );
	  }
#endif
	  /* We have to check if there is padding in the last source symbol of the last source block */
	  
	  if(trans_block->sbn == ((trans_obj->bs->N) - 1)) {
	    block_len = (trans_obj->len - (es_len * (trans_obj->bs->I * trans_obj->bs->A_large +
						     (trans_obj->bs->N - trans_obj->bs->I -1) * trans_obj->bs->A_small)));
	  }
	  /*else {
	    block_len = trans_block->len * es_len;
	    }*/
	  
	  /* set correct position */
	  
#ifdef WIN32
	  if(_lseeki64(trans_obj->fd, pos, SEEK_SET) == -1) {
#else
	    if(lseek64(trans_obj->fd, pos, SEEK_SET) == -1) {
#endif
	      
#ifdef WIN32
	      printf("lseek error, toi: %I64u\n", toi);
#else
	      printf("alc_rx.c line 1111 lseek error, toi: %llu\n", toi);
#endif
	      fflush(stdout);
	      free(buf);
	      set_session_state(ch->s->s_id, SExiting);
	      return MEM_ERROR;
	    }
	    
	    if(write(trans_obj->fd, buf, (unsigned int)block_len) == -1) {
#ifdef WIN32
	      printf("write error, toi: %I64u, sbn: %i\n", toi, sbn);
#else
	      printf("write error, toi: %llu, sbn: %i\n", toi, sbn);
#endif
	      fflush(stdout);
	      free(buf);
	      set_session_state(ch->s->s_id, SExiting);
	      return MEM_ERROR;
	    }
	    
	    free(buf);

	    free_units(trans_block);
	    trans_obj->nb_of_ready_blocks++;

	    if(ch->s->verbosity > 2) {	
#ifdef WIN32
	      printf("%u/%u Source Blocks decoded (TOI=%I64u SBN=%u)\n", trans_obj->nb_of_ready_blocks, trans_obj->bs->N, toi, sbn);
	      fflush(stdout);
#else
	      printf("%u/%u Source Blocks decoded (TOI=%llu SBN=%u)\n", trans_obj->nb_of_ready_blocks, trans_obj->bs->N, toi, sbn);
	      fflush(stdout);
#endif
	    }
	  }
	}
      }
    }
    
    else { /* We have an empty packet with FEC Payload ID */
      return EMPTY_PACKET;	
    }
      
    return OK;
  }
  
/* 
 * This function gives an object to application, when it is completely received.
 *
 * Params:	int s_id: Session identifier,
 *			ULONGLONG/unsigned long long toi: Transport Object Identifier,
 *			ULONGLONG/unsigned long long *datalen: Pointer to object length,
 *			int *retval: Return value in error cases/stopping situations.
 *
 * Return:	char*: Pointer to buffer which contains object's data,
 *			NULL: In errors/stopping situations
 *
 */

char* alc_recv(int s_id, 
#ifdef WIN32			   
			   ULONGLONG toi,
			   ULONGLONG *data_len,
#else
			   unsigned long long toi,
			   unsigned long long *data_len,
#endif
			   int *retval) {

	bool obj_completed = false;
	alc_session_t *s;
	char *buf = NULL; /* Buffer where to construct the object from data units */
	trans_obj_t *to;
	int object_exists = 0;
	
	s = get_alc_session(s_id);

	while(!obj_completed) {

		if(s->state == SExiting) {
			/*printf("alc_recv() SExiting\n");
			fflush(stdout);*/
			*retval = -2;
			return NULL;	
		}
		else if(s->state == SClosed) {
			/*printf("alc_recv() SClosed\n");
			fflush(stdout);*/
			*retval = 0;
			return NULL;	
		}

		to = s->obj_list;