Refactor, keep tracing on stderr, corrections for iovec size and dup dgram handling.

1 files changed, 98 insertions, 71 deletions

diff --git a/mcastseed/src/dgrambuf.c b/mcastseed/src/dgrambuf.c
index b07ba1f..b19b698 100644
--- a/mcastseed/src/dgrambuf.c
+++ b/mcastseed/src/dgrambuf.c
@@ -12,6 +12,7 @@
 #include <stdio.h> /* perror() */
 #include <string.h> /* memset() */
 #include <sys/uio.h> /* writev() */
+#include <sys/param.h> /* MIN() */
 
 struct uint_pair {
 	unsigned int index;
@@ -19,17 +20,19 @@ struct uint_pair {
 };
 
 struct dgrambuf_t {
-	size_t dgram_count;
+	size_t dgram_slots;
+	size_t dgram_free_count;
 	size_t dgram_max_size;
 	size_t dgram_header_size;
 
-	struct iovec *recv_iovecs;
-	struct iovec *write_iovecs;
+	size_t iovec_slots;
+	struct iovec *iov_recv;
+	struct iovec *iov_write;
 	struct mmsghdr *msgs;
 
-	unsigned int win_base;
-	unsigned int *dgram_seq_numbers; /* Stores the decoded datagram sequence number for each dgram slot of buf */
+	unsigned int dgram_seq_base;
 	unsigned int *dgram_len;
+	unsigned int *dgram_seq_numbers; /* Stores the decoded datagram sequence number for each dgram slot of buf */
 	struct uint_pair *dgram_ordered_seq_numbers;
 
 	void *buf;
@@ -44,33 +47,40 @@ void dgrambuf_set_validate_func(dgrambuf_t dbuf, unsigned int (*func)(unsigned i
 	dbuf->validate_func = func;
 }
 
+size_t dgrambuf_free_count(const dgrambuf_t dbuf) {
+	return dbuf->dgram_free_count;
+}
+
 int dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd) {
   void *dgram_base;
 	size_t vlen, i, dgram_index;
 	int recv_msg_count, res;
 	unsigned int seq, dgram_len;
 
-	if ( !dbuf->validate_func ) {
+	/* Buffer is full, can't receive */
+	if ( dbuf->dgram_free_count == 0 ) {
 		return -1;
 	}
 
+	/* Validate function is mandatory */
+	if ( !dbuf->validate_func ) {
+		return -2;
+	}
+
 	/* Initialize recvmmsg() syscall arguments */
-	for (i=0, vlen=0; i < dbuf->dgram_count; i++) {
+	for (i=0, vlen=0; i < dbuf->dgram_slots; i++) {
 		if ( dbuf->dgram_seq_numbers[i] == 0 ) {
-			dbuf->recv_iovecs[vlen].iov_base = dbuf->buf + i*dbuf->dgram_max_size;
-			dbuf->recv_iovecs[vlen].iov_len = dbuf->dgram_max_size;
+			dbuf->iov_recv[vlen].iov_base = dbuf->buf + i*dbuf->dgram_max_size;
+			dbuf->iov_recv[vlen].iov_len = dbuf->dgram_max_size;
 			memset(dbuf->msgs + vlen, 0, sizeof(struct mmsghdr));
-			dbuf->msgs[vlen].msg_hdr.msg_iov = dbuf->recv_iovecs + vlen;
+			dbuf->msgs[vlen].msg_hdr.msg_iov = dbuf->iov_recv + vlen;
 			dbuf->msgs[vlen].msg_hdr.msg_iovlen = 1;
 			vlen++;
+			if ( vlen == dbuf->iovec_slots )
+				break;
 		}
 	}
 
-	/* Buffer is full, can't receive */
-	if ( vlen==0 ) {
-		return -2;
-	}
-
 	/* Do the syscall */
 	recv_msg_count = recvmmsg(sockfd, dbuf->msgs, vlen, MSG_WAITFORONE, NULL);
 	if (recv_msg_count < 0) {
@@ -81,28 +91,29 @@ int dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd) {
 	/* Check all received messages */
 	res = 1;
 	for (i=0; i<recv_msg_count; i++) {
-		dgram_base = dbuf->recv_iovecs[i].iov_base;
+		dgram_base = dbuf->iov_recv[i].iov_base;
 		dgram_index = (dgram_base - dbuf->buf) / dbuf->dgram_max_size;
 		dgram_len = dbuf->msgs[i].msg_len;
 		seq = dbuf->validate_func(dgram_len, dgram_base);
 		// TODO better feedback
 		if ( seq == 0 ) {
-			printf("#%zi invalid (%u)\n", i, seq);
+			fprintf(stderr, "dgrambuf_recvmmsg(): #%zi invalid (%u)\n", i, seq);
 			dbuf->dgram_seq_numbers[dgram_index] = 0;
 		} else if ( seq == -1 ) {
-			printf("#%zi end\n", i);
+			fprintf(stderr, "dgrambuf_recvmmsg(): #%zi end\n", i);
 			dbuf->dgram_seq_numbers[dgram_index] = 0;
 			res = 0;
-		} else if ( seq < dbuf->win_base ) {
-			printf("#%zi past (%u)\n", i, seq);
+		} else if ( seq < dbuf->dgram_seq_base ) {
+			fprintf(stderr, "dgrambuf_recvmmsg(): #%zi past (%u)\n", i, seq);
 			dbuf->dgram_seq_numbers[dgram_index] = 0;
-		} else if ( seq >= dbuf->win_base + dbuf->dgram_count ) {
-			printf("#%zi future (%u)\n", i, seq);
+		} else if ( seq >= dbuf->dgram_seq_base + dbuf->dgram_slots ) {
+			fprintf(stderr, "dgrambuf_recvmmsg(): #%zi future (%u)\n", i, seq);
 			dbuf->dgram_seq_numbers[dgram_index] = 0;
 		} else {
-			printf("#%zi valid (%u)\n", i, seq);
+			//fprintf(stderr, "dgrambuf_recvmmsg(): #%zi valid (%u)\n", i, seq);
 			dbuf->dgram_seq_numbers[dgram_index] = seq;
 			dbuf->dgram_len[dgram_index] = dgram_len;
+			dbuf->dgram_free_count--;
 		}
 	}
 
@@ -114,40 +125,52 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd) {
 	unsigned int curr_seq, prev_seq, dgram_len;
 	ssize_t nwrite, total;
 
+	/* Buffer is empty, nothing to write */
+	if ( dbuf->dgram_free_count == dbuf->dgram_slots ) {
+		return -1;
+	}
+
 	/* Initialize dgram_ordered_seq_numbers from dgram_seq_numbers */
-	for (i=0; i < dbuf->dgram_count; i++) {
+	for (i=0; i < dbuf->dgram_slots; i++) {
 		dbuf->dgram_ordered_seq_numbers[i].index = i;
 		dbuf->dgram_ordered_seq_numbers[i].value = dbuf->dgram_seq_numbers[i];
 	}
 	/* Inplace sorting of dgram_ordered_seq_numbers */
-	qsort(dbuf->dgram_ordered_seq_numbers, dbuf->dgram_count, sizeof(struct uint_pair), _compare_uint_pair);
+	qsort(dbuf->dgram_ordered_seq_numbers, dbuf->dgram_slots, sizeof(struct uint_pair), _compare_uint_pair);
 	
 	/* Initialize iovecs for writev, take dgram payloads following the sequence numbers */
-	for (prev_seq=0, vlen=0, total=0, i=0; i< dbuf->dgram_count; i++) {
+	prev_seq=0, vlen=0, total=0;
+	for (i=dbuf->dgram_free_count; i < dbuf->dgram_slots; i++) {
 		curr_seq = dbuf->dgram_ordered_seq_numbers[i].value;
 
 		/* Skip empty dgram slot */
-		if ( curr_seq == 0 )
+		if ( curr_seq == 0 ) {
+			fprintf(stderr, "Oops : found empty slot (i==%zi)\n", i);
+			continue;
+		}
+
+		/* Skip if current dgram is a dup of the previous */
+		if ( curr_seq == prev_seq ) {
+			dgram_index = dbuf->dgram_ordered_seq_numbers[i].index;
+			/* Mark slot as empty */
+			dbuf->dgram_seq_numbers[dgram_index] = 0;
+			dbuf->dgram_free_count++;
 			continue;
+		}
 
 		/* Skip dgram comming from the past */
-		if ( curr_seq < dbuf->win_base ) {
+		if ( curr_seq < dbuf->dgram_seq_base ) {
 			fprintf(stderr, "Oops : found dgram from past in buffer (%u)\n", curr_seq);
 			continue;
 		}
 
-		/* Break if first dgram to write is not in buffer at all */
-		if ( ( vlen==0 ) && (curr_seq != dbuf->win_base) ) {
-			fprintf(stderr, "Oops : nothing to write, missing %u seq\n", dbuf->win_base);
+		/* Stop if first dgram to write is not in buffer at all */
+		if ( ( vlen==0 ) && (curr_seq != dbuf->dgram_seq_base) ) {
+			fprintf(stderr, "Oops : nothing to write, missing %u seq\n", dbuf->dgram_seq_base);
 			break;
 		}
 
-		/* Skip if next dgram is a dup */
-		if ( ( vlen > 0 ) && (curr_seq == prev_seq) ) {
-			continue;
-		}
-
-		/* Break if next seq dgram is missing */
+		/* Stop if current seq dgram is missing */
 		if ( ( vlen > 0 ) && (curr_seq > prev_seq+1 ) ) {
 			break;
 		}
@@ -156,87 +179,91 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd) {
 		dgram_index = dbuf->dgram_ordered_seq_numbers[i].index;
 		dgram_len = dbuf->dgram_len[dgram_index] - dbuf->dgram_header_size;
 
-		dbuf->write_iovecs[vlen].iov_len = dgram_len; /* Setup iovecs */
-		dbuf->write_iovecs[vlen].iov_base = dbuf->buf + dgram_index*dbuf->dgram_max_size + dbuf->dgram_header_size;
-		dbuf->dgram_seq_numbers[dgram_index] = 0; /* Mark dgram slots about to be written out as reusable */
+		dbuf->iov_write[vlen].iov_len = dgram_len; /* Setup iovecs */
+		dbuf->iov_write[vlen].iov_base = dbuf->buf + dgram_index*dbuf->dgram_max_size + dbuf->dgram_header_size;
+		dbuf->dgram_seq_numbers[dgram_index] = 0; /* Mark dgram slots about to be written out as empty for next read */
 
 		total += dgram_len; /* Update counters */
-		vlen++;
-		dbuf->win_base = curr_seq;
+		dbuf->dgram_free_count++;
+		dbuf->dgram_seq_base = curr_seq + 1;
 		prev_seq = curr_seq;
+		vlen++;
+		/* Don't plan to write more than iovec_slots slots */
+		if ( vlen == dbuf->iovec_slots )
+			break;
 	}
 
-	/* If nothing valid to write out */
+	/* Nothing valid to write out (but buffer not empty, missing the next dgram) */
 	if ( vlen == 0 ) {
 		return -1;
 	}
 
-	nwrite = writev(fd, dbuf->write_iovecs, vlen);
+	nwrite = writev(fd, dbuf->iov_write, vlen);
 	if ( nwrite < 0 ) {
 		perror("writev()");
-		return nwrite;
-	}
-
-	if ( nwrite != total ) {
+	} else if ( nwrite != total ) {
+		//FIXME : everything break if there because all non writed data will be overwritted at next read
+		// Make a loop here could make dgrambuf_writev() unbounded in run time
 		fprintf(stderr, "writev() short\n");
-		return nwrite;
 	}
 
 	return nwrite;
 }
 
-dgrambuf_t dgrambuf_new(size_t dgram_count, size_t dgram_max_size, size_t dgram_header_size) {
+dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_header_size, size_t iovec_slots) {
 
 	dgrambuf_t dbuf = calloc(1, sizeof(struct dgrambuf_t));
 	if (!dbuf) goto fail0;
 
-	dbuf->dgram_count = dgram_count;
+	dbuf->dgram_slots = dgram_slots;
+	dbuf->dgram_free_count = dgram_slots;
 	dbuf->dgram_max_size = dgram_max_size;
 	dbuf->dgram_header_size = dgram_header_size;
+	dbuf->iovec_slots = MIN(iovec_slots,dgram_slots);
 
-	dbuf->recv_iovecs = calloc(dgram_count, sizeof(struct iovec));
-	if (!dbuf->recv_iovecs) goto fail1;
+	dbuf->iov_recv = calloc(iovec_slots, sizeof(struct iovec));
+	if (!dbuf->iov_recv) goto fail1;
 
-	dbuf->write_iovecs = calloc(dgram_count, sizeof(struct iovec));
-	if (!dbuf->write_iovecs) goto fail2;
+	dbuf->iov_write = calloc(iovec_slots, sizeof(struct iovec));
+	if (!dbuf->iov_write) goto fail2;
 
-	dbuf->msgs = calloc(dgram_count, sizeof(struct mmsghdr));
+	dbuf->msgs = calloc(iovec_slots, sizeof(struct mmsghdr));
 	if (!dbuf->msgs) goto fail3;
 
-	dbuf->win_base = 1;
-	dbuf->dgram_seq_numbers = calloc(dgram_count, sizeof(unsigned int));
-	if (!dbuf->dgram_seq_numbers) goto fail4;
+	dbuf->dgram_seq_base = 1;
+	dbuf->dgram_len = calloc(dgram_slots, sizeof(unsigned int));
+	if (!dbuf->dgram_len) goto fail4;
 
-	dbuf->dgram_len = calloc(dgram_count, sizeof(ssize_t));
-	if (!dbuf->dgram_len) goto fail5;
+	dbuf->dgram_seq_numbers = calloc(dgram_slots, sizeof(unsigned int));
+	if (!dbuf->dgram_seq_numbers) goto fail5;
 
-	dbuf->dgram_ordered_seq_numbers = calloc(dgram_count, sizeof(struct uint_pair));
+	dbuf->dgram_ordered_seq_numbers = calloc(dgram_slots, sizeof(struct uint_pair));
 	if (!dbuf->dgram_ordered_seq_numbers) goto fail6;
 
-	dbuf->buf = calloc(dgram_count, dgram_max_size);
+	dbuf->buf = calloc(dgram_slots, dgram_max_size);
 	if (!dbuf->buf) goto fail7;
 
 	return dbuf;
 
 fail7:  free(dbuf->dgram_ordered_seq_numbers);
-fail6:	free(dbuf->dgram_len);
-fail5:	free(dbuf->dgram_seq_numbers);
+fail6:	free(dbuf->dgram_seq_numbers);
+fail5:	free(dbuf->dgram_len);
 fail4:	free(dbuf->msgs);
-fail3:	free(dbuf->write_iovecs);
-fail2:	free(dbuf->recv_iovecs);
+fail3:	free(dbuf->iov_write);
+fail2:	free(dbuf->iov_recv);
 fail1:	free(dbuf);
-fail0:	return 0;
+fail0:	return NULL;
 }
 
 void dgrambuf_free(dgrambuf_t *dbuf) {
 	if (dbuf && *dbuf) {
 		free((*dbuf)->buf);
 		free((*dbuf)->dgram_ordered_seq_numbers);
-		free((*dbuf)->dgram_len);
 		free((*dbuf)->dgram_seq_numbers);
+		free((*dbuf)->dgram_len);
 		free((*dbuf)->msgs);
-		free((*dbuf)->write_iovecs);
-		free((*dbuf)->recv_iovecs);
+		free((*dbuf)->iov_write);
+		free((*dbuf)->iov_recv);
 		free(*dbuf);
 	}
 	*dbuf = NULL;