From 817d7e5d3cf722ce5a1b72cc3d7826bd551ec16f Mon Sep 17 00:00:00 2001 From: Ludovic Pouzenc Date: Sun, 24 Jul 2016 23:54:46 +0200 Subject: Drop qsort(), use GnuLib's gl_rbtree_list Performances came back, bugs are likely : many changes. --- mcastseed/src/dgrambuf.c | 242 ++++++++++++++++++++++++++++------------------- 1 file changed, 145 insertions(+), 97 deletions(-) (limited to 'mcastseed/src') diff --git a/mcastseed/src/dgrambuf.c b/mcastseed/src/dgrambuf.c index 4863bb7..4296f89 100644 --- a/mcastseed/src/dgrambuf.c +++ b/mcastseed/src/dgrambuf.c @@ -9,7 +9,7 @@ #include "config.h" #include /* recvmmsg() _GNU_SOURCE */ -#include /* calloc(), free(), qsort() */ +#include /* calloc(), free() */ #include /* perror() */ #include /* errno */ #include /* memset() */ @@ -18,10 +18,10 @@ #include /* sigaction() */ #include /* alarm() */ #include /* SSIZE_MAX */ -#include "gl_rbtree_oset.h" /* Red-Black Tree Ordered Set, gnulib-tool --import rbtree-oset */ +#include "gl_rbtree_list.h" /* Red-Black Tree backed Sorted list, gnulib-tool --import rbtree-list */ -struct uint_pair { - unsigned int index; +struct indexed_uint { + size_t index; unsigned int value; }; @@ -42,7 +42,6 @@ struct dgrambuf_t { struct sigaction sa_sigalrm; size_t dgram_slots; - size_t dgram_free_count; size_t dgram_max_size; size_t dgram_header_size; @@ -58,15 +57,22 @@ struct dgrambuf_t { unsigned int dgram_seq_last; unsigned int dgram_seq_base; unsigned int *dgram_len; - unsigned int *dgram_seq_numbers; /* Decoded datagram sequence number for each dgram_slot of buf */ - int dgram_ordered_seq_numbers_is_dirty; - struct uint_pair *dgram_ordered_seq_numbers; /* Pairs to track original items ranks after qsort() */ + + struct indexed_uint *dgram_slot_seq; /* malloc'ed array */ + struct indexed_uint **dgram_read_active_slots; /* malloc'd array of pointers to items of dgram_slot_seq[] */ + size_t dgram_read_active_slots_count; + struct indexed_uint **dgram_write_active_slots; /* malloc'd array of pointers to items of dgram_slot_seq[] */ + size_t dgram_write_active_slots_count; + + gl_list_t dgram_empty_slots; + gl_list_t dgram_used_slots; uint8_t *buf; /* malloc-ed 2d byte array : buf[dgram_slots][dgram_max_size] */ }; void _sigalrm_handler(int signum); -int _compare_uint_pair(const void *pa, const void *pb); +int _compare_indexed_uint(const void *pa, const void *pb); +bool _equals_indexed_uint(const void *pa, const void *pb); void _update_ordered_seq_numbers(dgrambuf_t dbuf); #ifndef HAVE_MIN_SIZE_T @@ -78,21 +84,22 @@ void dgrambuf_set_validate_func(dgrambuf_t dbuf, int (*validate_func)(unsigned i } size_t dgrambuf_get_free_count(const dgrambuf_t dbuf) { - return dbuf->dgram_free_count; + return gl_list_size(dbuf->dgram_empty_slots); } -int dgrambuf_everything_was_received(dgrambuf_t dbuf) { - /*TODO really implement this */ - return dbuf->dgram_seq_last && ( dbuf->dgram_seq_base - 1 == dbuf->dgram_seq_last ); +size_t dgrambuf_get_used_count(const dgrambuf_t dbuf) { + return gl_list_size(dbuf->dgram_used_slots); } ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { uint8_t *dgram_base; ssize_t recv_byte; - size_t i, vlen, dgram_index, recv_msg_count; + size_t i, vlen, dgram_index, recv_msg_count, free_count, free_iovecs_count; int res; unsigned int seq, dgram_len; struct sigaction sa_old; + struct indexed_uint *active_slot; + /* Info ptr is mandatory */ *info = 0; @@ -103,27 +110,33 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { } /* Buffer is full, can't receive */ - if ( dbuf->dgram_free_count == 0 ) { + if ( dgrambuf_get_free_count(dbuf) == 0 ) { dbuf->stats.dgrambuf_read_on_full++; *info |= DGRAMBUF_RECV_OVERWRITE; /*FIXME : this locks everything if buf full + next seq missing*/ return 0; } - /* Initialize recvmmsg() syscall arguments */ - for (i=0, vlen=0; i < dbuf->dgram_slots; i++) { - /*XXX linear search is not optimal, notably if is_dirty == 0*/ - if ( dbuf->dgram_seq_numbers[i] == 0 ) { - 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->iov_recv + vlen; - dbuf->msgs[vlen].msg_hdr.msg_iovlen = 1; - vlen++; - if ( vlen == dbuf->iovec_slots ) - break; - } + /* Initialize recvmmsg() syscall arguments and keep track of active slots */ + free_count = dgrambuf_get_free_count(dbuf); + free_iovecs_count = gl_list_size(dbuf->dgram_empty_slots); + + /* XXX condition is strange */ + for (i=0; i < dbuf->iovec_slots && i < free_count && i < free_iovecs_count; i++) { + /* Pop a free slot, ignoring const modifier from gl_list_get_at() */ + dbuf->dgram_read_active_slots[i] = (struct indexed_uint *) gl_list_get_at(dbuf->dgram_empty_slots, 0); + gl_list_remove_at(dbuf->dgram_empty_slots, 0); + dbuf->dgram_read_active_slots_count++; + + dgram_index = dbuf->dgram_read_active_slots[i]->index; + dbuf->iov_recv[i].iov_base = dbuf->buf + dgram_index * dbuf->dgram_max_size; + dbuf->iov_recv[i].iov_len = dbuf->dgram_max_size; + + memset(dbuf->msgs + i, 0, sizeof(struct mmsghdr)); + dbuf->msgs[i].msg_hdr.msg_iov = dbuf->iov_recv + i; + dbuf->msgs[i].msg_hdr.msg_iovlen = 1; } + vlen = i; /* Do the syscall with alarm() to circumvent bad behavior in recvmmsg(2) timeout */ if (timeout) { @@ -150,7 +163,6 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { } if (recv_msg_count > 0) { - dbuf->dgram_ordered_seq_numbers_is_dirty = 1; dbuf->stats.recv_dgrams += recv_msg_count; if ( recv_msg_count == vlen ) { *info |= DGRAMBUF_RECV_IOVEC_FULL; @@ -159,8 +171,8 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { /* Check all received messages */ for (i=0, recv_byte=0; idgram_read_active_slots[i]; 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; /* dgrambuf_new() adjust iovec_len to prevent overflows on ssize_t*/ @@ -179,10 +191,10 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { } else { /*fprintf(stderr, "dgrambuf_recvmmsg(): #%zu valid (%u)\n", i, seq);*/ *info |= DGRAMBUF_RECV_VALID_DGRAM; - dbuf->dgram_seq_numbers[dgram_index] = seq; - dbuf->dgram_ordered_seq_numbers_is_dirty = 1; - dbuf->dgram_len[dgram_index] = dgram_len; - dbuf->dgram_free_count--; + active_slot->value = seq; + gl_sortedlist_nx_add(dbuf->dgram_used_slots, _compare_indexed_uint, active_slot); + dbuf->dgram_len[active_slot->index] = dgram_len; + continue; } break; case 2: @@ -196,8 +208,17 @@ ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info) { dbuf->stats.dgram_invalid++; break; } + /* In all invalid dgram cases, put back active_slot in dgram_free_slots */ + gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot); + } + + /* 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]; + gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot); } + dbuf->dgram_read_active_slots_count = 0; dbuf->stats.recv_byte += recv_byte; return recv_byte; @@ -211,18 +232,13 @@ int dgrambuf_have_data_ready_to_write(dgrambuf_t dbuf) { return 1; } - /* Buffer is empty, nothing to write */ - if ( dbuf->dgram_free_count == dbuf->dgram_slots ) { + /* dgram_used_slots is empty, nothing to write */ + if ( dgrambuf_get_used_count(dbuf) == 0 ) { return 0; } - /* Other test cases needs ordered_seq_numbers */ - if ( dbuf->dgram_ordered_seq_numbers_is_dirty ) { - _update_ordered_seq_numbers(dbuf); - } - /* Nothing to write if next dgram is not in buffer at all */ - next_dgram_seq = dbuf->dgram_ordered_seq_numbers[dbuf->dgram_free_count].value; + next_dgram_seq = ((struct indexed_uint *) gl_list_get_at(dbuf->dgram_used_slots, 0))->value; if ( next_dgram_seq != dbuf->dgram_seq_base ) { return 0; } @@ -231,15 +247,16 @@ int dgrambuf_have_data_ready_to_write(dgrambuf_t dbuf) { } int dgrambuf_have_received_everything(dgrambuf_t dbuf) { - if (dbuf) {}; - return 0; /*FIXME to be implemented*/ + /*FIXME : Really implement this */ + return dbuf->dgram_seq_last && ( dbuf->dgram_seq_base - 1 == dbuf->dgram_seq_last ); } ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) { - size_t dgram_index, i, vlen; + size_t dgram_index, i, vlen, used_count; unsigned int curr_seq, prev_seq, dgram_len; ssize_t nwrite, total, remain, len; struct iovec *iov; + struct indexed_uint *active_slot; /* FIXME Info ptr is mandatory */ *info = 0; @@ -256,14 +273,17 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) { iov = dbuf->iov_write; vlen = 0; total = 0; - /* Write needs up to date ordered_seq_numbers (dgrams could be unsorted or some are lost)*/ - if ( dbuf->dgram_ordered_seq_numbers_is_dirty ) { - _update_ordered_seq_numbers(dbuf); - } + /* Initialize iovecs for writev, take dgram payloads following the sequence numbers */ prev_seq = 0; - for (i = dbuf->dgram_free_count; i < dbuf->dgram_slots && vlen < dbuf->iovec_slots; i++) { - curr_seq = dbuf->dgram_ordered_seq_numbers[i].value; + used_count = dgrambuf_get_used_count(dbuf); + for (i = 0, vlen = 0; vlen < dbuf->iovec_slots && i < used_count; i++) { + /* Pop a used slot */ + dbuf->dgram_write_active_slots[i] = (struct indexed_uint *) gl_list_get_at(dbuf->dgram_used_slots, 0); + gl_list_remove_at(dbuf->dgram_used_slots, 0); + dbuf->dgram_write_active_slots_count++; + + curr_seq = dbuf->dgram_write_active_slots[i]->value; /* Skip empty dgram slot */ if ( curr_seq == 0 ) { @@ -273,25 +293,25 @@ 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++; - goto mark_empty; + continue; } /* Skip dgram comming from the past */ if ( curr_seq < dbuf->dgram_seq_base ) { fprintf(stderr, "Oops : found dgram from past in buffer (%u)\n", curr_seq); - goto mark_empty; + continue; + } + /* Stop if current seq dgram is missing */ + if ( ( vlen > 0 ) && (curr_seq > prev_seq+1 ) ) { + break; } /* 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; } - /* Stop if current seq dgram is missing */ - if ( ( vlen > 0 ) && (curr_seq > prev_seq+1 ) ) { - break; - } /* Normal case : curr_seq is the next dgram to write */ - dgram_index = dbuf->dgram_ordered_seq_numbers[i].index; + dgram_index = dbuf->dgram_write_active_slots[i]->index; dgram_len = dbuf->dgram_len[dgram_index] - dbuf->dgram_header_size; /* Setup iovecs */ @@ -303,15 +323,7 @@ ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) { total += dgram_len; prev_seq = curr_seq; vlen++; - - /* Mark dgram slot about to be written out as empty for next read */ - /*FIXME These cause harm if writev() is incomplete*/ dbuf->dgram_seq_base = curr_seq + 1; -mark_empty: - /* Mark slot as empty */ - dgram_index = dbuf->dgram_ordered_seq_numbers[i].index; - dbuf->dgram_seq_numbers[dgram_index] = 0; - dbuf->dgram_free_count++; } /* Nothing valid to write out (but buffer not empty, missing the next dgram) */ @@ -368,13 +380,19 @@ mark_empty: remain -= len; } if ( i == vlen ) { - /* FIXME : this happens */ fprintf(stderr, "Fatal : failed to find partial iov after partial write\n"); return -3; } } else { - /* Wipe outdated values for clarity in debug mode (only _bytes is use on branching) */ + /* Full write has happened */ + for (i=0; idgram_write_active_slots_count; i++) { + active_slot = (struct indexed_uint *) dbuf->dgram_write_active_slots[i]; + active_slot->value = 0; + gl_sortedlist_nx_add(dbuf->dgram_empty_slots, _compare_indexed_uint, active_slot); + } + dbuf->dgram_write_active_slots_count = 0; + /* Wipe outdated partial_* values for clarity in debug mode (only _bytes is use on branching) */ dbuf->partial_write_iov = NULL; dbuf->partial_write_remaining_iovcnt = 0; } @@ -384,7 +402,11 @@ mark_empty: 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)); + const void **dgram_slot_seq_ptrs = NULL; + dgrambuf_t dbuf; + size_t i; + + dbuf = calloc(1, sizeof(struct dgrambuf_t)); if (!dbuf) goto fail0; dbuf->validate_func = NULL; @@ -395,7 +417,6 @@ dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_ dbuf->sa_sigalrm.sa_handler = _sigalrm_handler; 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; @@ -423,20 +444,48 @@ dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_ dbuf->dgram_len = calloc(dgram_slots, sizeof(unsigned int)); if (!dbuf->dgram_len) goto fail4; - dbuf->dgram_seq_numbers = calloc(dgram_slots, sizeof(unsigned int)); - if (!dbuf->dgram_seq_numbers) goto fail5; + dbuf->dgram_slot_seq = calloc(dgram_slots, sizeof(struct indexed_uint)); + if (!dbuf->dgram_slot_seq) goto fail5; + + /* Implicit with dbuf = calloc(...) + dbuf->dgram_read_active_slots_count = 0; + */ + dbuf->dgram_read_active_slots = calloc(iovec_slots, sizeof(struct indexed_uint *)); + if (!dbuf->dgram_read_active_slots) goto fail6; + + /* Implicit with dbuf = calloc(...) + dbuf->dgram_write_active_slots_count = 0; + */ + dbuf->dgram_write_active_slots = calloc(iovec_slots, sizeof(struct indexed_uint *)); + if (!dbuf->dgram_write_active_slots) goto fail7; + + dgram_slot_seq_ptrs = calloc(dgram_slots, sizeof(void *)); + for (i=0; idgram_slot_seq + i; + } + if (!dgram_slot_seq_ptrs) goto fail7; + + dbuf->dgram_empty_slots = gl_list_nx_create(GL_RBTREE_LIST, _equals_indexed_uint, + NULL, NULL, false, dgram_slots, dgram_slot_seq_ptrs); + if (!dbuf->dgram_empty_slots) goto fail8; + + free(dgram_slot_seq_ptrs); + dgram_slot_seq_ptrs=NULL; - dbuf->dgram_ordered_seq_numbers = calloc(dgram_slots, sizeof(struct uint_pair)); - if (!dbuf->dgram_ordered_seq_numbers) goto fail6; - dbuf->dgram_ordered_seq_numbers_is_dirty = 1; + dbuf->dgram_used_slots = gl_list_nx_create_empty(GL_RBTREE_LIST, _equals_indexed_uint, + NULL, NULL, false); + if (!dbuf->dgram_used_slots) goto fail9; dbuf->buf = calloc(dgram_slots, dgram_max_size); - if (!dbuf->buf) goto fail7; + if (!dbuf->buf) goto fail10; return dbuf; -fail7: free(dbuf->dgram_ordered_seq_numbers); -fail6: free(dbuf->dgram_seq_numbers); +fail10: gl_list_free(dbuf->dgram_used_slots); +fail9: gl_list_free(dbuf->dgram_empty_slots); +fail8: free(dbuf->dgram_write_active_slots); +fail7: free(dbuf->dgram_read_active_slots); +fail6: free(dbuf->dgram_slot_seq); fail5: free(dbuf->dgram_len); fail4: free(dbuf->iov_write); fail3: free(dbuf->iov_recv); @@ -448,8 +497,11 @@ fail0: return NULL; void dgrambuf_free(dgrambuf_t *dbuf) { if (dbuf && *dbuf) { free((*dbuf)->buf); - free((*dbuf)->dgram_ordered_seq_numbers); - free((*dbuf)->dgram_seq_numbers); + gl_list_free((*dbuf)->dgram_used_slots); + gl_list_free((*dbuf)->dgram_empty_slots); + free((*dbuf)->dgram_write_active_slots); + free((*dbuf)->dgram_read_active_slots); + free((*dbuf)->dgram_slot_seq); free((*dbuf)->dgram_len); free((*dbuf)->iov_write); free((*dbuf)->iov_recv); @@ -464,28 +516,24 @@ void _sigalrm_handler(int signum) { if (signum) {} /* Avoid compiler warning */ } -int _compare_uint_pair(const void *pa, const void *pb) { - const struct uint_pair *a = pa; - const struct uint_pair *b = pb; +int _compare_indexed_uint(const void *pa, const void *pb) { + const struct indexed_uint *a = pa; + const struct indexed_uint *b = pb; if (a->value < b->value) return -1; 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; } -void _update_ordered_seq_numbers(dgrambuf_t dbuf) { - size_t i; - /* Initialize dgram_ordered_seq_numbers from dgram_seq_numbers */ - 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_slots, sizeof(struct uint_pair), _compare_uint_pair); - dbuf->stats.qsort_calls++; - - dbuf->dgram_ordered_seq_numbers_is_dirty = 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); } - -- cgit v1.2.3