sender.c

这个程序实现了FLUTE协议

/* $Author: peltotas $ $Date: 2006/03/16 11:05:00 $ $Revision: 1.17 $ */
/*
 *   MAD-FLUTELIB: Implementation of FLUTE protocol.
 *   Copyright (c) 2003-2006 TUT - Tampere University of Technology
 *   main authors/contacts: [e-mail] and [e-mail]
 *
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#include "flute_inc.h"

/*
 * This function sends one FDT Instance.
 *
 * Params:	char *fdt_instance: Pointer to buffer which contains FDT Instance,
 *			ULONGLONG/unsigned long long fdt_inst_len:,
 *			int *s_id: Pointer to Session identifier,
 *			int *ch_id: Pointer to Channel identifier (if NULL use queue mode),
 *			unsigned short eslen: Encoding Symbol Length,
 *			unsigned int max_sblen: Maximum-Size Source Block Length,
 *			unsigned char fec_enc_id: FEC Encoding ID,
 *			unsigned short fec_inst_id: FEC Instance ID,
 *                      int verbosity:.
 *
 * Return:	int: 0 in success, -1 otherwise.
 *
 */
                                                                                                                                              
int send_fdt_instance(char *fdt_instance,
#ifdef WIN32
		      ULONGLONG fdt_inst_len,
#else
		      unsigned long long fdt_inst_len,
#endif
		      int *s_id, int *ch_id, unsigned short eslen,
		      unsigned int max_sblen, unsigned char fec_enc_id,
		      unsigned short fec_inst_id, int verbosity) {

#ifdef WIN32
  ULONGLONG toilen;
#else
  unsigned long long toilen;
#endif
	
  int sentbytes = 0;
  int bytes_left;
  int nbytes;
  int sent = 0;
  unsigned int sbn = 0;
  char *buf = NULL;

  blocking_struct_t *bs;                
                                                                                                                          
  toilen = fdt_inst_len;
  bytes_left = (int)toilen;
   
  if(ch_id != NULL) {

    if(verbosity == 4) {
      printf("Sending FDT Instance (ID=%i)\n", get_fdt_instance_id(*s_id));
      fflush(stdout);
    }
    
    set_object_sent_bytes(*s_id, 0);
  }
  else {
    if(verbosity == 4) {
      printf("Adding FDT Instance to Tx queue (ID=%i)\n", get_fdt_instance_id(*s_id));
      fflush(stdout);
    }
  }

  /* Let's compute the blocking structure */
  
  bs = compute_blocking_structure(toilen, max_sblen, eslen);

  while(sbn < bs->N) {

    if(get_session_state(*s_id) == SExiting) {
      free(buf);
      return -2;
    }

    if(sbn < bs->I) {
      nbytes = eslen*(bs->A_large);
    }
    else {
      nbytes = eslen*(bs->A_small);
    }

    nbytes = bytes_left < nbytes ? bytes_left : nbytes;

    buf = fdt_instance + sentbytes;

    sent = alc_send(s_id, ch_id, buf, nbytes, FDT_TOI, toilen, eslen, max_sblen, sbn,
		    fec_enc_id, fec_inst_id);

    if(sent == -1) {
      free(bs);
      return -1;
    }
    else if(sent == -2) {
      free(bs);
      return -2;
    }
    
    bytes_left -= nbytes;
    sentbytes += sent;
    sbn++;
  }

  free(bs);

  if(ch_id != NULL) {
    if(verbosity == 4) {
      printf("FDT Instance sent (ID=%i)\n", get_fdt_instance_id(*s_id));
      fflush(stdout);
    }
  }
  else {
    if(verbosity == 4) {
      printf("FDT Instance added to Tx queue (ID=%i)\n", get_fdt_instance_id(*s_id));
      fflush(stdout);
    }
  }

  return 0;
}

/*
 * This function sends one file.
 *
 * Params:	char *tx_file: Pointer to buffer containing filepath,
 *			int *s_id: Pointer to Session identifier,
 *			int *ch_id: Pointer to Channel identifier (if NULL use queue mode),
 *			ULONGLONG/unsigned long long toi: Transport Object Identifier,
 *			unsigned short eslen: Encoding Symbol Length,
 *			unsigned int max_sblen: Maximum-Size Source Block Length,
 *			unsigned char fec_enc_id: FEC Encoding ID,
 *			unsigned short fec_inst_id: FEC Instance ID
 *			char* fdt_inst_buf: Pointer to buffer containing FDT Instance,
 *			ULONGLONG/unsigned long long fdt_inst_len: FDT Instance length,
 *                      int verbosity:.
 *
 * Return:	int: 0 in success, -1 otherwise
 *
 */

int send_file(char *tx_file, int *s_id, int *ch_id,
			  unsigned short es_len, unsigned int max_sb_len,
			  unsigned char fec_enc_id, unsigned short fec_inst_id,
			  file_t *file,
#ifdef WIN32
			  ULONGLONG fdt_inst_len,
#else
			  unsigned long long fdt_inst_len,
#endif
			  char *fdt_inst_buf, int verbosity) {

#ifdef WIN32
	ULONGLONG toilen;
	ULONGLONG sent = 0;
#else
	unsigned long long toilen;
	unsigned long long sent = 0;
#endif

	int nbytes;
	int read_bytes;
	double print_percent;

	FILE *fp;	
	/*struct stat file_stats;*/
#ifdef WIN32
	struct __stat64 file_stats;
#else
	struct stat64 file_stats;
#endif
	char *buf = NULL;

	unsigned int sbn = 0;
	blocking_struct_t *bs;

	unsigned int Mbytes_of_data = 0;
	int retval = 0;
	alc_session_t *s;

	s = get_alc_session(*s_id);

	set_session_tx_toi(*s_id, file->toi);

#ifdef WIN32
	if(_stat64(tx_file, &file_stats) == -1) {
#else
	if(stat64(tx_file, &file_stats) == -1) {
#endif
		printf("Error: %s is not valid file name\n", tx_file);
		fflush(stdout);
		return -1;
	}

	if(file->encoding == NULL) {
		toilen = file->file_len;
	}
	else {
		toilen = file->toi_len;
	}

/*
#ifdef WIN32
	toilen = (ULONGLONG)file_stats.st_size;
#else
	toilen = (unsigned long long)file_stats.st_size;
#endif
*/
	
	if(toilen == 0) {
		printf("Tx_file %s size = 0\n", tx_file);
		fflush(stdout);
		return -1;
	}

#ifdef WIN32
    if(toilen > (ULONGLONG)0xFFFFFFFFFFFF) {
#else
	if(toilen > 0xFFFFFFFFFFFFULL) {
#endif
		printf("Tx_file %s too big!!\n", tx_file);
		fflush(stdout);
		return -1;
	}
	
	/* Allocate memory for buf */
	if(!(buf = (char*)calloc((unsigned int)(es_len*max_sb_len), sizeof(char)))) {
		printf("Could not alloc memory for buf!\n");
		fflush(stdout);
		return -1;
	}

	/* File to send */
#ifdef WIN32
	if((fp = fopen(tx_file, "rb")) == NULL) {
#else
	if((fp = fopen64(tx_file, "rb")) == NULL) {
#endif
	  printf("Error: unable to open tx_file %s\n", tx_file);
	  fflush(stdout);
	  free(buf);
	  return -1;
	}

	if(ch_id != NULL) {
	  if(verbosity > 0) {
#ifdef WIN32
            printf("Sending file: %s (TOI=%I64u)\n", tx_file, file->toi);
#else
            printf("Sending file: %s (TOI=%llu)\n", tx_file, file->toi);
#endif
	    fflush(stdout);
	  }
	  set_object_sent_bytes(*s_id, 0);
	  set_object_last_print_tx_percent(*s_id, 0);
	}
	else {
	  if(verbosity > 0) {
#ifdef WIN32
            printf("Adding file to Tx queue: %s (TOI=%I64u)\n", tx_file, file->toi);
#else
            printf("Adding file to Tx queue: %s (TOI=%llu)\n", tx_file, file->toi);
#endif
            fflush(stdout);
          }
	}
	
	/* Let's compute the blocking structure */
	bs = compute_blocking_structure(toilen, max_sb_len, es_len);

	while(sbn < bs->N) {

        if(sbn < bs->I) {
			nbytes = es_len*(bs->A_large);
        }
        else {
			nbytes = es_len*(bs->A_small);
        }

        memset(buf, 0, (es_len * max_sb_len));
        read_bytes = fread(buf, 1, (unsigned int)nbytes, fp);

		if(ferror(fp)) {
			printf("fread error, file: %s\n", tx_file);
			fflush(stdout);
			free(buf);
			fclose(fp);
			return -1;
		}

		if(file->encoding == NULL || strcmp(file->encoding, "gzip") == 0) {
			retval = alc_send(s_id, ch_id, buf, read_bytes, file->toi, toilen, es_len, max_sb_len, sbn, fec_enc_id,
							fec_inst_id);
		}
		else if (strcmp(file->encoding, "pad") == 0) {
			retval = alc_send(s_id, ch_id, buf, nbytes, file->toi, toilen, es_len, max_sb_len, sbn, fec_enc_id,
							fec_inst_id);
		}

		if(ch_id != NULL) {
#ifdef WIN32
		  if(verbosity > 2) {
		    printf("%u/%u Source Blocks sent (TOI=%I64u SBN=%u)\n", (sbn+1), bs->N, file->toi, sbn);
		    fflush(stdout);
		  }
#else
		  if(verbosity > 2) {
		    printf("%u/%u Source Blocks sent (TOI=%llu SBN=%u)\n", (sbn+1), bs->N, file->toi, sbn);
		    fflush(stdout);
		  }
#endif
		}
		
		if(retval == -1) {
		  fclose(fp);
		  free(buf);
		  free(bs);
		  return -1;
		}
		else if(retval == -2) {
		  fclose(fp);
		  free(buf);
		  free(bs);
		  return -2;
		}

		sent = get_object_sent_bytes(*s_id);
		print_percent = get_object_last_print_tx_percent(*s_id);

		if((sent/FDT_INTERVAL) > Mbytes_of_data) {

			Mbytes_of_data = (unsigned int)(sent/FDT_INTERVAL);
			retval = send_fdt_instance(fdt_inst_buf, fdt_inst_len, s_id, ch_id, s->def_eslen,
						   s->def_max_sblen, s->def_fec_enc_id, s->def_fec_inst_id,
						   verbosity);
				
			if(retval == -1) {
				free(bs);
				fclose(fp);
				free(buf);
				return -1;
			}
			else if(retval == -2) {
				free(bs);
				fclose(fp);
				free(buf);
				return -2;
			}

			set_object_sent_bytes(*s_id, sent);
			set_object_last_print_tx_percent(*s_id, print_percent);
		}

        sbn++;
	}

	free(bs);
	fclose(fp);
	free(buf);

	if(ch_id != NULL) {
	  if(verbosity > 0) {
#ifdef WIN32
	    printf("File sent: %s (TOI=%I64u)\n", tx_file, file->toi);
#else 
	    printf("File sent: %s (TOI=%llu)\n", tx_file, file->toi);
#endif
	    fflush(stdout);
	  }
	}
	else {
	  if(verbosity > 0) {
#ifdef WIN32
            printf("File added to Tx queue: %s (TOI=%I64u)\n", tx_file, file->toi);
#else
            printf("File added to Tx queue: %s (TOI=%llu)\n", tx_file, file->toi);
#endif
	    fflush(stdout);
	  }
	}
	
	return 0;
	}
	
/*
 * This function creates FDT Instance string buffer from file structure(s).
 *
 * Params:	file_t *file: Pointer to first file to be defined in an FDT Instance,
 *			int nb_of_files: Number of files to be defined in an FDT Instance,
 *			fdt_t *fdt:	Pointer to Complete FDT to be splitted to FDT Instances,
 *			int *s_id: Pointer to Session identifier,
 *			ULONGLONG/unsigned long long *fdt_inst_len: Pointer to length of created FDT Instance.
 *
 * Return:	char*: Pointer to created FDT Instance string buffer.
 *
 */

char *create_fdt_instance(file_t *file, int nb_of_files, fdt_t *fdt, int *s_id, 
#ifdef WIN32
			  ULONGLONG *fdt_inst_len
#else
			  unsigned long long *fdt_inst_len
#endif
			  ) {
  file_t *tmp_file;
  char *fdt_inst_payload = NULL;
  char tmp_line[MAX_PATH];
  int i;
  int print_help = 0;
  
#ifdef WIN32
  ULONGLONG size = 0;
  ULONGLONG position = 0;
#else
  unsigned long long size = 0;
  unsigned long long position = 0;
#endif
  
  alc_session_t *s;
  
  s = get_alc_session(*s_id);
  
  tmp_file = file;
  
  memset(tmp_line, 0, MAX_PATH);
  sprintf(tmp_line, "<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>\n");
  size = strlen(tmp_line);
  
  /* Allocate memory for fdt_inst_payload */
  if(!(fdt_inst_payload = (char*)realloc(fdt_inst_payload, ((unsigned int)size * sizeof(char))))) {
#ifdef WIN32
    printf("Could not (re)alloc memory for fdt_inst_payload, size: %I64u!\n", size);
#else
    printf("Could not (re)alloc memory for fdt_inst_payload, size: %llu!\n", size);
#endif
    return NULL;
  }
  
  memcpy((fdt_inst_payload + (unsigned int)position), tmp_line, strlen(tmp_line));
  position += strlen(tmp_line);
  
  memset(tmp_line, 0, MAX_PATH);
  
#ifdef WIN32
  sprintf(tmp_line, "<FDT-Instance Expires=\"%I64u\"", fdt->expires);
#else
  sprintf(tmp_line, "<FDT-Instance Expires=\"%llu\"", fdt->expires);
#endif
  size += strlen(tmp_line);
  
  /* Reallocate memory for fdt_inst_payload */
  if(!(fdt_inst_payload = (char*)realloc(fdt_inst_payload, ((unsigned int)size * sizeof(char))))) {
#ifdef WIN32
    printf("Could not (re)alloc memory for fdt_inst_payload, size: %I64u!\n", size);
#else
    printf("Could not (re)alloc memory for fdt_inst_payload, size: %llu!\n", size);
#endif
    return NULL;
  }
  
  memcpy((fdt_inst_payload + (unsigned int)position), tmp_line, strlen(tmp_line));
  position += strlen(tmp_line);
  
  if(fdt->complete) {
    if(nb_of_files == fdt->nb_of_files) {
      memset(tmp_line, 0, MAX_PATH);
      sprintf(tmp_line, "\n\tComplete=\"true\"");
      size += strlen(tmp_line);
      
      /* Reallocate memory for fdt_inst_payload */
      if(!(fdt_inst_payload = (char*)realloc(fdt_inst_payload, ((unsigned int)size * sizeof(char))))) {
#ifdef WIN32
	printf("Could not (re)alloc memory for fdt_inst_payload, size: %I64u!\n", size);
#else
	printf("Could not (re)alloc memory for fdt_inst_payload, size: %llu!\n", size);
#endif
	return NULL;
      }
      
      memcpy((fdt_inst_payload + (unsigned int)position), tmp_line, strlen(tmp_line));
      position += strlen(tmp_line);
    }
  }
  if(fdt->type != NULL) {
    memset(tmp_line, 0, MAX_PATH);
    sprintf(tmp_line, "\n\tContent-Type=\"%s\"", fdt->type);
    size += strlen(tmp_line);
    
    /* Reallocate memory for fdt_inst_payload */
    if(!(fdt_inst_payload = (char*)realloc(fdt_inst_payload, ((unsigned int)size * sizeof(char))))) {
#ifdef WIN32
      printf("Could not (re)alloc memory for fdt_inst_payload, size: %I64u!\n", size);
#else
      printf("Could not (re)alloc memory for fdt_inst_payload, size: %llu!\n", size);
#endif