dgrambuf: fix slots init, _add() error checking, detect dups on read

Duplicates have been counted on write() because of qsort().

2 files changed, 30 insertions, 18 deletions

diff --git a/mcastseed/src/dgrambuf.c b/mcastseed/src/dgrambuf.c
index 9feda89..03f9ea0 100644
--- a/mcastseed/src/dgrambuf.c
+++ b/mcastseed/src/dgrambuf.c
@@ -95,10 +95,10 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) {
 	ssize_t recv_byte;
 	size_t i, dgram_index, recv_msg_count, free_count;
 	int res;
-	bool bres;
 	unsigned int seq, dgram_len;
 	struct sigaction sa_old;
 	struct indexed_uint *active_slot;
+	gl_list_node_t pos;
 
 
 	/* Info ptr is mandatory */
@@ -185,14 +185,21 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) {
 					dbuf->stats.dgram_future++;
 					*info |= DGRAMBUF_RECV_FUTURE_DGRAM;
 				} else {
-					/*fprintf(stderr, "dgrambuf_recvmmsg(): #%zu valid (%u)\n", i, seq);*/
 					active_slot->value = seq;
-					bres = gl_sortedlist_nx_add(dbuf->dgram_used_slots, _compare_indexed_uint, active_slot);
-					if ( !bres ) /*TODO: better oom handling */
-						return -4;
-					dbuf->dgram_len[active_slot->index] = dgram_len;
-					*info |= DGRAMBUF_RECV_VALID_DGRAM;
-					continue;
+					pos = gl_sortedlist_search(dbuf->dgram_used_slots, _compare_indexed_uint, active_slot);
+					if ( pos != NULL ) {
+						fprintf(stderr, "dgrambuf_recvmmsg(): #%zu duplicate (%u)\n", i, seq);
+						dbuf->stats.dgram_dup++;
+						*info |= DGRAMBUF_RECV_DUPLICATE_DGRAM;
+					} else {
+						/*fprintf(stderr, "dgrambuf_recvmmsg(): #%zu valid (%u)\n", i, seq);*/
+						pos = gl_sortedlist_nx_add(dbuf->dgram_used_slots, _compare_indexed_uint, active_slot);
+						if ( pos == NULL ) /*TODO: better oom handling */
+							return -4;
+						dbuf->dgram_len[active_slot->index] = dgram_len;
+						*info |= DGRAMBUF_RECV_VALID_DGRAM;
+						continue;
+					}
 				}
 				break;
 			case 2:
@@ -207,16 +214,16 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) {
 				break;
 		}
 		/* In all invalid dgram cases, put back active_slot in dgram_free_slots */
-		bres = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
-		if ( !bres ) /*TODO: better oom handling */
+		pos = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
+		if ( !pos ) /*TODO: better oom handling */
 			return -4;
 	}
 
 	/* Push remaining active slots in dgram_empty_slots */
 	for (/*next i*/; i < dbuf->dgram_read_active_slots_count; i++) {
 		active_slot = dbuf->dgram_read_active_slots[i];
-		bres = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
-		if ( !bres ) /*TODO: better oom handling */
+		pos = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
+		if ( !pos ) /*TODO: better oom handling */
 			return -4;
 	}
 
@@ -259,7 +266,7 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) {
 	ssize_t nwrite, total, remain, len;
 	struct iovec *iov;
 	struct indexed_uint *active_slot;
-	bool bres;
+	bool pos;
 
 	/* FIXME Info ptr is mandatory */
 	*info = 0;
@@ -295,7 +302,7 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) {
 			}
 			/* Skip if current dgram is a dup of the previous */
 			if ( curr_seq == prev_seq ) {
-				dbuf->stats.dgram_dup++;
+				fprintf(stderr, "Oops : found duplicated dgram in buffer (%u)\n", curr_seq);
 				continue;
 			}
 			/* Skip dgram comming from the past */
@@ -392,8 +399,8 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) {
 		for (i=0; i<dbuf->dgram_write_active_slots_count; i++) {
 			active_slot = (struct indexed_uint *) dbuf->dgram_write_active_slots[i];
 			active_slot->value = 0;
-			bres = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
-			if ( !bres ) /*TODO: better oom handling */
+			pos = gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot);
+			if ( !pos ) /*TODO: better oom handling */
 				return -4;
 		}
 		dbuf->dgram_write_active_slots_count = 0;
@@ -484,7 +491,8 @@ dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_
 
 	dgram_slot_seq_ptrs = calloc(dgram_slots, sizeof(void *));
 	for (i=0; i<dgram_slots; i++) {
-		dgram_slot_seq_ptrs[i] = dbuf->dgram_slot_seq + i;
+		dbuf->dgram_slot_seq[i].index = i;
+		dgram_slot_seq_ptrs[i] = &(dbuf->dgram_slot_seq[i]);
 	}
 	if (!dgram_slot_seq_ptrs) goto fail7;
 
@@ -547,16 +555,19 @@ int _compare_indexed_uint(const void *pa, const void *pb) {
 	else if ( a->value > b->value )
 		return 1;
 	else
+		return 0;
+/*
 		if (a->index < b->index)
 			return -1;
 		else if (a->index > b->index)
 			return 1;
 		else
 			return 0;
+*/
 }
 
 bool _equals_indexed_uint(const void *pa, const void *pb) {
 	const struct indexed_uint *a = pa;
 	const struct indexed_uint *b = pb;
-	return (a->value == b->value) && (a->index == b->index);
+	return (a->value == b->value) /*&& (a->index == b->index)*/;
 }
diff --git a/mcastseed/src/dgrambuf.h b/mcastseed/src/dgrambuf.h
index 4d712c6..a83647b 100644
--- a/mcastseed/src/dgrambuf.h
+++ b/mcastseed/src/dgrambuf.h
@@ -12,6 +12,7 @@
 #define DGRAMBUF_RECV_IOVEC_FULL 1 << 3
 #define DGRAMBUF_RECV_FINALIZE 1 << 4
 #define DGRAMBUF_RECV_FUTURE_DGRAM 1 << 5
+#define DGRAMBUF_RECV_DUPLICATE_DGRAM 1 << 6
 #define DGRAMBUF_RECV_VALID_DGRAM 1 << 6
 
 #define DGRAMBUF_WRITE_PARTIAL 1 << 1