sendfile.c

开源的组播程序

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <sys/time.h>
#include <time.h>
#include <libgen.h>

#ifdef DMALLOC
#include "dmalloc.h"
#endif

#include <rs.h>

#include "pmtp.h"

uint32_t process_upload (struct pmtp_upload *upload);

int get_uploadsocket (struct pmtp_upload *upload);
 
uint8_t send_upload_announcement (struct pmtp_upload *upload);
uint8_t create_pmsp_header (struct pmtp_upload *upload, unsigned char *header);

int main(int argc, char *argv[]) {
	uint32_t bytes_read, segmentsize = (NN - NROOTS) * PMTP_PAYLOAD;
	struct pmtp_upload upload;
	struct pmtp_segment segment;
	unsigned char sha_printable[60];
	struct in_addr in_addr;
	struct timespec sleeper; 

	if (argc != 2) {
		fprintf (stderr, "Usage: %s <filename>\n", argv[0]);
		exit (1);
	}
	if (strlen (argv[1]) > 127) {
		fprintf (stderr, "%s: Filename too long.\n", argv[0]);
	 	exit (1);
	}
	upload.previous_invocation.tv_sec = 0;
	upload.previous_invocation.tv_usec = 0;
	upload.bytes_sent = 0;
	upload.packets_sent = 0;
	upload.bytes_sent_interval = 0;
	upload.session.version = 1;
	upload.session.codingid = 1;
	upload.session.sessionid = 1;
	upload.session.payload = PMTP_PAYLOAD;
	upload.session.start = 5;
	upload.session.family = AF_INET;
	upload.session.protocol = SOCK_DGRAM;
	inet_aton (PMTP_GROUP, &in_addr);
	upload.session.group = in_addr.s_addr;
	upload.session.port = htons (PMTP_PORT);
	strncpy (upload.session.filename, argv[1], 127);
	upload.session.filename[127] = 0;
	upload.session.filesize = get_filesize (upload.session.filename);
	upload.session.partoffset = 0;
	upload.session.partlength = upload.session.filesize;
	upload.session.bitrate = PMTP_BITRATE / 8000;
	upload.credits = 0;

	get_tigertree (upload.session.file_tigertree, upload.session.filename,
		0, upload.session.filesize);
	get_tigertree (upload.session.part_tigertree, upload.session.filename,
		upload.session.partoffset, upload.session.partlength);

	if (get_uploadsocket(&upload) == -1) {
		fprintf (stderr, "Couldn't open upload socket.\n");
		exit (1);
	}

	srandom(time(NULL));
	if (DEBUG & DEBUG_CODING)
		fprintf (stderr, "Code (%d / %d)\n", NN, NROOTS);

	sleeper.tv_sec = 1;
	sleeper.tv_nsec = 0;
	while (upload.session.start > 0) {
		send_upload_announcement (&upload); 
		upload.session.start--;
		nanosleep(&sleeper,NULL);
	}

	if ((upload.fd = open(upload.session.filename, O_RDONLY)) < 0) {
		fprintf (stderr, "Error opening %s for reading\n",
			upload.session.filename);
		exit (1);
	}

	init_segment (&segment, NULL);
	segment.segmentid = 0;
	if((segment.codec=init_rs_char(SYMSIZE,GENPOLY,FCS,PRIM,NROOTS)) == NULL) {
		fprintf(stderr, "init_rs_int failed!\n");
		exit (1);
	}
	init_packet_buffer (&(upload.buffer));
	if ((segment.packets = (uint8_t *) malloc (PMTP_PAYLOAD*NN)) == NULL) {
		fprintf (stderr, "Couldn't malloc %d bytes for segment\n",
			PMTP_PAYLOAD * NN);
		exit (1);
	}

	sleeper.tv_sec = 0;
	sleeper.tv_nsec = 100000000;
	while ((bytes_read = read(upload.fd, segment.packets, segmentsize)) != 0) {
		while (buffer_packets_free(&(upload.buffer)) < NN) {
			process_upload (&upload);
		nanosleep(&sleeper,NULL);
		}
		if (bytes_read == segmentsize) {
			segment.payload_size = upload.session.payload;
		} else {
			segment.payload_size = (uint16_t) (bytes_read/(NN-NROOTS))+1;
		}
		if (DEBUG & DEBUG_SEGMENTS) 
			fprintf (stderr, "Segment: %d, Payload %d, bytes %d\n",
				segment.segmentid, segment.payload_size, bytes_read);
		if (encode_segment (&segment, NROOTS) == NULL) {
			fprintf (stderr, "Error encoding segment\n");
			exit (1);
		}
		if (DEBUG & DEBUG_SEGMENTS_SHA)
			fprintf (stderr, "Segment %d, Packets %d, Payload %d, SHA 0x%s\n",
				segment.segmentid, segment.packets_received,
				segment.payload_size, get_printable_sha (sha_printable,
					segment.packets, NN * segment.payload_size));
		if (buffer_segment (&upload, &segment) == 0) {
			fprintf (stderr, "Error buffering segment %d.\n",segment.segmentid);
			exit (1);
		}
		segment.segmentid++;
	}
	while (upload.buffer.size != 0) {
		process_upload (&upload);
		nanosleep(&sleeper,NULL);
	}

	printf ("Packets sent: %u, Bytes sent: %u.\n", upload.packets_sent,
		upload.bytes_sent);
	free_packet_buffer (&(upload.buffer));
	free_segment (&segment);
	close (upload.fd);
	return 0;
}

uint8_t send_upload_announcement (struct pmtp_upload *upload) {
	uint8_t headerlength;
	uint8_t header[212];
	memset (header, 0, 212);

	headerlength = create_pmsp_header (upload, header);

	if (DEBUG & DEBUG_PMSP) 
		fprintf (stderr, "Sending PMSP packet with length %u: %s\n",
			headerlength, &header[PMSP_HEADER_FILENAME]);

	if (sendto (upload->s, header, headerlength, 0, (struct sockaddr *)
			&(upload->s_in), sizeof (struct sockaddr_in)) < 0) {
		perror ("sendto");
		exit (1);
	}
	return headerlength;
}

uint8_t create_pmsp_header (struct pmtp_upload *upload, uint8_t *header) {
	unsigned char *base;
	unsigned char filename[128];

	if (header == NULL) {
		fprintf (stderr, "Header is a NULL pointer.\n");
		exit (1);
	}
	encode_header_16 (header,
		upload->session.version << 12 | upload->session.codingid );
	encode_header_16 (header+PMSP_HEADER_SESSIONID, upload->session.sessionid);
	encode_header_16 (header+PMSP_HEADER_PAYLOAD, upload->session.payload);
	encode_header_16 (header+PMSP_HEADER_START, upload->session.start << 4);
	/*
	 * Source, multicast group and port are already in network byte order
	 */
	*((uint32_t *) (header + PMSP_HEADER_SOURCE)) = upload->session.source;
	*((uint32_t *) (header + PMSP_HEADER_GROUP)) = upload->session.group;
	*((uint16_t *) (header + PMSP_HEADER_PORT)) = upload->session.port;

	memcpy (header + PMSP_HEADER_FILETIGERTREE, upload->session.file_tigertree,
		24);
	encode_header_16 (header+PMSP_HEADER_BITRATE, upload->session.bitrate);
	encode_header_32 (header+PMSP_HEADER_FILESIZE+4, upload->session.filesize);
	encode_header_32 (header+PMSP_HEADER_PARTOFFSET,upload->session.partoffset);
	encode_header_32 (header+PMSP_HEADER_PARTLENGTH,upload->session.partlength);
	memcpy (header + PMSP_HEADER_PARTTIGERTREE, upload->session.part_tigertree,
		24);

	strncpy (filename, upload->session.filename, 128);
	filename[127] = 0;
	base = basename (filename);
	strncpy (header+PMSP_HEADER_FILENAME, base, 128);
	return PMSP_HEADER_FILENAME + strlen(base);
}

uint32_t process_upload (struct pmtp_upload *upload) {
	uint32_t bytes_budget, bytes_transmitted;
	uint32_t packets_transmitted;
	/* 
	 * bitrate is in multiples of 8 Kbit/s. So to get to bytes per milisecond
	 * we calculate bitrate * (1024 * 8)  bit/s / 8 bytes / 1000 miliseconds
	*/
	float byterate = 1.024 * upload->session.bitrate;

	struct timeval now;
	float delta; /* Time delta in miliseconds */
	if (gettimeofday (&now, NULL) == -1) {
		perror ("gettimeofday");
		exit (1);
	}
	bytes_budget = (uint32_t) (byterate * 100.0);

	if (upload->previous_invocation.tv_sec != 0) {
		delta = (now.tv_sec - upload->previous_invocation.tv_sec) * 1000
			+ (now.tv_usec - upload->previous_invocation.tv_usec) / 1000;
		if (delta > SMOOTHING_SECONDS * 1000) {
			fprintf (stderr, "Time delta %d > %d miliseconds.\n",
				(uint32_t) delta, (uint32_t) SMOOTHING_SECONDS * 1000);
			delta = SMOOTHING_SECONDS * 1000;
		}
		upload->credits += (uint32_t) (delta * byterate);

		if (upload->bytes_sent_interval > upload->credits) {
			upload->bytes_sent_interval -= upload->credits;
		} else {
			upload->bytes_sent_interval = 0;
		}

		bytes_budget = upload->credits < upload->bytes_sent_interval ?
			0 : upload->credits - upload->bytes_sent_interval;
		if (DEBUG & DEBUG_UPLOAD_PROCESSING) 
			fprintf (stderr, "Now: %u.%u, Delta: %ums, byterate: %dKbit/s, credits: %u, Bytes: %u, ",
				(uint32_t) now.tv_sec, (uint32_t) now.tv_usec / 1000,
				(uint32_t) delta, (uint32_t) (byterate * 1000.0 / 1024.0),
				(unsigned int) upload->credits, upload->bytes_sent_interval);


	}
	if (DEBUG & DEBUG_UPLOAD_PROCESSING) 
		fprintf (stderr, "Sending budget %d bytes.\n", bytes_budget);

	bytes_transmitted = send_buffer (upload->s, &(upload->s_in),
		&(upload->buffer), bytes_budget, &packets_transmitted, 
		&bytes_transmitted);
	upload->credits -= bytes_transmitted;
	upload->bytes_sent_interval += bytes_transmitted;
	upload->bytes_sent += bytes_transmitted;
	upload->previous_invocation = now;
	return bytes_transmitted;
}

int get_uploadsocket (struct pmtp_upload *upload) {
	char hostname[256];
	struct hostent *hp;
	int multicast_loop = 1;

	if (gethostname (hostname, 255) == -1) {
		perror ("gethostname");
		exit (1);
	}
	if ((hp = gethostbyname2(hostname, AF_INET)) == NULL) {
		fprintf (stderr, "%s: host unknown.\n", hostname);
		exit (1);
	}
	if (hp->h_addr_list[0] == NULL) {
		fprintf (stderr, "No network interfaces found.\n");
		exit (1);
	}
	memcpy (&upload->session.source, hp->h_addr_list[0], 4);

	if ((upload->s = socket (upload->session.family,
			upload->session.protocol, 0)) < 0) {
		perror ("socket");
		exit (1);
	}
	if (setsockopt (upload->s, SOL_IP, IP_MULTICAST_LOOP, &multicast_loop,
			sizeof(multicast_loop))) {
		perror ("setsockopt");
		exit (1);
	}
	upload->s_in.sin_family = upload->session.family;
	upload->s_in.sin_addr.s_addr = upload->session.group;
	upload->s_in.sin_port = upload->session.port;
	return 0;
}