From 3f0a442799955f56b2c77aabd6bc7aa4458718b4 Mon Sep 17 00:00:00 2001 From: Ludovic Pouzenc Date: Sun, 17 Jul 2016 14:21:26 +0200 Subject: API changes, pedandic fixes, dgrambuf stats & info field, recvmmsg() with alarm(), partial writev() support. --- mcastseed/src/dgrambuf.c | 425 ++++++++++++++++++++++++++++++------------ mcastseed/src/dgrambuf.h | 24 ++- mcastseed/src/dgrambuf_test.c | 16 +- mcastseed/src/mcastleech.c | 133 ++++++++++--- mcastseed/src/mcastseed.c | 86 ++++----- 5 files changed, 486 insertions(+), 198 deletions(-) (limited to 'mcastseed/src') diff --git a/mcastseed/src/dgrambuf.c b/mcastseed/src/dgrambuf.c index 41ebc8a..75b82a6 100644 --- a/mcastseed/src/dgrambuf.c +++ b/mcastseed/src/dgrambuf.c @@ -10,67 +10,107 @@ #include /* recvmmsg() _GNU_SOURCE */ #include /* calloc(), free(), qsort() */ #include /* perror() */ +#include /* errno */ #include /* memset() */ #include /* writev() */ -#include /* MIN() */ +#include /* uint8_t, uint64_t */ +#include /* sigaction() */ +#include /* alarm() */ +#include /* SSIZE_MAX */ struct uint_pair { unsigned int index; unsigned int value; }; +struct dgrambuf_stats_t { + uint64_t dgrambuf_read_on_full; + uint64_t recvmmsg_calls, recv_dgrams, recv_byte; + uint64_t dgram_invalid, dgram_past, dgram_future, dgram_dup, dgram_end_marker; + uint64_t qsort_calls; + uint64_t writev_calls, write_partial, write_byte; +}; + struct dgrambuf_t { + /* dgram validation after receive, takes dgram len and a pointer to the start of dgram data + Must returns dgram seq number or 0 if invalid dgram */ + int (*validate_func)(unsigned int, void *, unsigned int*); + + struct dgrambuf_stats_t stats; + struct sigaction sa_sigalrm; + size_t dgram_slots; size_t dgram_free_count; size_t dgram_max_size; size_t dgram_header_size; size_t iovec_slots; - struct iovec *iov_recv; - struct iovec *iov_write; struct mmsghdr *msgs; + struct iovec *iov_recv; + struct iovec *iov_write; /* malloc'ed array */ + + struct iovec *partial_write_iov; /* Pointer to an item of iov_write[] */ + size_t partial_write_remaining_iovcnt; + size_t partial_write_remaining_bytes; + unsigned int dgram_seq_last; 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 */ + 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; - - void *buf; + struct uint_pair *dgram_ordered_seq_numbers; /* Pairs to track original items ranks after qsort() */ - unsigned int (*validate_func)(unsigned int, void *); - //TODO pthread_mutex_lock + uint8_t *buf; /* malloc-ed 2d byte array : buf[dgram_slots][dgram_max_size] */ }; -int _compare_uint_pair(const void *pa, const void *pb); +void _sigalrm_handler(int signum); +int _compare_uint_pair(const void *pa, const void *pb); void _update_ordered_seq_numbers(dgrambuf_t dbuf); -void dgrambuf_set_validate_func(dgrambuf_t dbuf, unsigned int (*func)(unsigned int, void *) ) { - dbuf->validate_func = func; +#ifndef HAVE_MIN_SIZE_T +size_t min_size_t(size_t a, size_t b) { return (avalidate_func = validate_func; } -size_t dgrambuf_free_count(const dgrambuf_t dbuf) { +size_t dgrambuf_get_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; +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 ); +} + +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; + int res; unsigned int seq, dgram_len; + struct sigaction sa_old; - /* Buffer is full, can't receive */ - if ( dbuf->dgram_free_count == 0 ) { - return -1; - } + /* Info ptr is mandatory */ + *info = 0; /* Validate function is mandatory */ if ( !dbuf->validate_func ) { - return -2; + return -3; + } + + /* Buffer is full, can't receive */ + if ( dbuf->dgram_free_count == 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; @@ -83,48 +123,92 @@ int dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd) { } } - /* Do the syscall */ - recv_msg_count = recvmmsg(sockfd, dbuf->msgs, vlen, MSG_WAITFORONE, NULL); - if (recv_msg_count < 0) { - perror("recvmmsg()"); - return recv_msg_count; + /* Do the syscall with alarm() to circumvent bad behavior in recvmmsg(2) timeout */ + if (timeout) { + sigaction(SIGALRM, &(dbuf->sa_sigalrm), &sa_old); + alarm(timeout); + } + res = recvmmsg(sockfd, dbuf->msgs, vlen, MSG_WAITFORONE, NULL); + if (timeout) { + alarm(0); + sigaction(SIGALRM, &sa_old, NULL); + } + dbuf->stats.recvmmsg_calls++; + + if (res < 0) { + if ( errno == EINTR ) { + recv_msg_count = 0; + *info |= DGRAMBUF_RECV_EINTR; + } else { + perror("recvmmsg()"); + return -1; + } + } else { + recv_msg_count = res; } + 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 ) { /* XXX -Wsigncompare hints problems here and above */ + *info |= DGRAMBUF_RECV_IOVEC_FULL; + } } /* Check all received messages */ - res = 1; - for (i=0; iiov_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 == -1 ) { - fprintf(stderr, "dgrambuf_recvmmsg(): #%zi end\n", i); - res = 0; - } else if ( seq == 0 ) { - fprintf(stderr, "dgrambuf_recvmmsg(): #%zi invalid (%u)\n", i, seq); - } else if ( seq < dbuf->dgram_seq_base ) { - fprintf(stderr, "dgrambuf_recvmmsg(): #%zi past (%u)\n", i, seq); - } else if ( seq >= dbuf->dgram_seq_base + dbuf->dgram_slots ) { - fprintf(stderr, "dgrambuf_recvmmsg(): #%zi future (%u)\n", i, seq); - } else { - //fprintf(stderr, "dgrambuf_recvmmsg(): #%zi valid (%u)\n", i, seq); - 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--; + + /* dgrambuf_new() adjust iovec_len to prevent overflows on ssize_t*/ + recv_byte += dgram_len; + + res = dbuf->validate_func(dgram_len, dgram_base, &seq); + switch (res) { + case 1: + if ( seq < dbuf->dgram_seq_base ) { + fprintf(stderr, "dgrambuf_recvmmsg(): #%zu past (%u)\n", i, seq); + dbuf->stats.dgram_past++; + } else if ( seq >= dbuf->dgram_seq_base + dbuf->dgram_slots ) { + fprintf(stderr, "dgrambuf_recvmmsg(): #%zu future (%u)\n", i, seq); + dbuf->stats.dgram_future++; + *info |= DGRAMBUF_RECV_FUTURE_DGRAM; + } 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--; + } + break; + case 2: + fprintf(stderr, "dgrambuf_recvmmsg(): #%zu finalize (%u)\n", i, seq); + dbuf->stats.dgram_end_marker++; + dbuf->dgram_seq_last = seq; + *info |= DGRAMBUF_RECV_FINALIZE; + break; + default: + fprintf(stderr, "dgrambuf_recvmmsg(): #%zu invalid\n", i); + dbuf->stats.dgram_invalid++; + break; } } - return res; + dbuf->stats.recv_byte += recv_byte; + + return recv_byte; } int dgrambuf_have_data_ready_to_write(dgrambuf_t dbuf) { unsigned int next_dgram_seq; + + /* Last write was partial, so there is more to write */ + if ( dbuf->partial_write_remaining_bytes > 0 ) { + return 1; + } + /* Buffer is empty, nothing to write */ if ( dbuf->dgram_free_count == dbuf->dgram_slots ) { return 0; @@ -144,80 +228,158 @@ int dgrambuf_have_data_ready_to_write(dgrambuf_t dbuf) { return 1; } -ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd) { +int dgrambuf_have_received_everything(dgrambuf_t dbuf) { + if (dbuf) {}; + return 0; /*FIXME to be implemented*/ +} + +ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info) { size_t dgram_index, i, vlen; unsigned int curr_seq, prev_seq, dgram_len; - ssize_t nwrite, total; - - /* Write needs up to date ordered_seq_numbers */ - 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, total=0; - for (i=dbuf->dgram_free_count, vlen=0; i < dbuf->dgram_slots && vlen < dbuf->iovec_slots; i++) { - curr_seq = dbuf->dgram_ordered_seq_numbers[i].value; - - /* Skip empty dgram slot */ - 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 ) { - goto mark_empty; - } - /* 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; - } - /* 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; + ssize_t nwrite, total, remain, len; + struct iovec *iov; + + /* FIXME Info ptr is mandatory */ + *info = 0; + + if ( dbuf->partial_write_remaining_bytes > 0 ) { + /* Previous writev() was partial, continue it */ + iov = dbuf->partial_write_iov; + vlen = dbuf->partial_write_remaining_iovcnt; + total = dbuf->partial_write_remaining_bytes; + } else if ( ! dgrambuf_have_data_ready_to_write(dbuf) ) { + return 0; /* XXX Inline code ? */ + } else { + /* Prepare a write batch, buffer state is in dgram_seq_numbers */ + 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; + + /* Skip empty dgram slot */ + if ( curr_seq == 0 ) { + fprintf(stderr, "Oops : found empty slot (i==%zu)\n", i); + continue; + } + /* Skip if current dgram is a dup of the previous */ + if ( curr_seq == prev_seq ) { + dbuf->stats.dgram_dup++; + goto mark_empty; + } + /* 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; + } + /* 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_len = dbuf->dgram_len[dgram_index] - dbuf->dgram_header_size; + /* Normal case : curr_seq is the next dgram to write */ + dgram_index = dbuf->dgram_ordered_seq_numbers[i].index; + dgram_len = dbuf->dgram_len[dgram_index] - dbuf->dgram_header_size; - /* Setup iovecs */ - dbuf->iov_write[vlen].iov_len = dgram_len; - dbuf->iov_write[vlen].iov_base = dbuf->buf + dgram_index*dbuf->dgram_max_size + dbuf->dgram_header_size; + /* Setup iovecs */ + dbuf->iov_write[vlen].iov_len = dgram_len; + dbuf->iov_write[vlen].iov_base = dbuf->buf + + dgram_index*dbuf->dgram_max_size + dbuf->dgram_header_size; - /* Update counters */ - total += dgram_len; - dbuf->dgram_seq_base = curr_seq + 1; - prev_seq = curr_seq; - vlen++; + /* Update counters */ + total += dgram_len; + prev_seq = curr_seq; + vlen++; - /* Mark dgram slot about to be written out as empty for next read */ - //XXX These cause harm if writev() is incomplete + /* 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++; - } + /* 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) */ - if ( vlen == 0 ) { - fprintf(stderr, "Oops : nothing to write at all\n"); - return -1; + /* Nothing valid to write out (but buffer not empty, missing the next dgram) */ + if ( vlen == 0 ) { + fprintf(stderr, "Oops : nothing to write at all\n"); + return -2; + } + + if ( vlen == dbuf->iovec_slots ) { + *info |= DGRAMBUF_WRITE_IOVEC_FULL; + } } - nwrite = writev(fd, dbuf->iov_write, vlen); + nwrite = writev(fd, iov, vlen); + dbuf->stats.writev_calls++; if ( nwrite < 0 ) { + /* Treat non fatal errors */ + if ( errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + /* Keeps some state informations for retry */ + dbuf->partial_write_remaining_bytes = total; + dbuf->partial_write_remaining_iovcnt = vlen; + dbuf->partial_write_iov = iov; + *info |= DGRAMBUF_WRITE_EWOULDBLOCK_OR_EINTR; + return 0; + } + /* Print fatal errors and bail out */ perror("writev()"); - } 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 -1; + } + + /* XXX Remove me when code is correct */ + if ( nwrite > total ) { + fprintf(stderr, "Fatal bug : nwrite > total\n"); + return -3; + } + if ( nwrite > 0 ) { + dbuf->stats.write_byte += nwrite; + *info |= DGRAMBUF_WRITE_SUCCESS; + } + + /* Check if the write was partially done */ + dbuf->partial_write_remaining_bytes = total - nwrite; + if ( dbuf->partial_write_remaining_bytes > 0 ) { + *info |= DGRAMBUF_WRITE_PARTIAL; + dbuf->stats.write_partial++; + /* Find the partially written iov and update it */ + remain = nwrite; + for (i=0; iiov_write[i].iov_len; + if ( remain < len ) { + dbuf->partial_write_remaining_iovcnt = vlen - i; + dbuf->partial_write_iov = dbuf->iov_write + i; + + dbuf->iov_write[i].iov_base = + (uint8_t *) dbuf->iov_write[i].iov_base + remain; + dbuf->iov_write[i].iov_len -= remain; + break; + } + remain -= len; + } + if ( i == vlen ) { + /* FIXME : this happens */ + fprintf(stderr, "Fatal bug, 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) */ + dbuf->partial_write_iov = NULL; + dbuf->partial_write_remaining_iovcnt = 0; } return nwrite; @@ -228,21 +390,38 @@ dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_ dgrambuf_t dbuf = calloc(1, sizeof(struct dgrambuf_t)); if (!dbuf) goto fail0; + dbuf->validate_func = NULL; + /* Implicit with dbuf = calloc(...) + memset(&(dbuf->stats), 0, sizeof(struct dgrambuf_stats_t)); + memset(&(dbuf->sa_sigalrm), 0, sizeof(struct sigaction)); + */ + 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; - dbuf->iovec_slots = MIN(iovec_slots,dgram_slots); + + /* writev() and dgrambuf_recvmmsg accumulates read/write bytes in ssize_t */ + iovec_slots = min_size_t(iovec_slots, SSIZE_MAX/dgram_max_size); + dbuf->iovec_slots = iovec_slots; + + dbuf->msgs = calloc(iovec_slots, sizeof(struct mmsghdr)); + if (!dbuf->msgs) goto fail1; dbuf->iov_recv = calloc(iovec_slots, sizeof(struct iovec)); - if (!dbuf->iov_recv) goto fail1; + if (!dbuf->iov_recv) goto fail2; dbuf->iov_write = calloc(iovec_slots, sizeof(struct iovec)); - if (!dbuf->iov_write) goto fail2; + if (!dbuf->iov_write) goto fail3; - dbuf->msgs = calloc(iovec_slots, sizeof(struct mmsghdr)); - if (!dbuf->msgs) goto fail3; + /* Implicit with dbuf = calloc(...) + dbuf->partial_write_iov = NULL; + dbuf->partial_write_remaining_iovcnt = 0; + dbuf->partial_write_remaining_bytes = 0; + dbuf->dgram_seq_last = 0; + */ dbuf->dgram_seq_base = 1; dbuf->dgram_len = calloc(dgram_slots, sizeof(unsigned int)); if (!dbuf->dgram_len) goto fail4; @@ -262,9 +441,9 @@ dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_ fail7: free(dbuf->dgram_ordered_seq_numbers); fail6: free(dbuf->dgram_seq_numbers); fail5: free(dbuf->dgram_len); -fail4: free(dbuf->msgs); -fail3: free(dbuf->iov_write); -fail2: free(dbuf->iov_recv); +fail4: free(dbuf->iov_write); +fail3: free(dbuf->iov_recv); +fail2: free(dbuf->msgs); fail1: free(dbuf); fail0: return NULL; } @@ -275,12 +454,17 @@ void dgrambuf_free(dgrambuf_t *dbuf) { free((*dbuf)->dgram_ordered_seq_numbers); free((*dbuf)->dgram_seq_numbers); free((*dbuf)->dgram_len); - free((*dbuf)->msgs); free((*dbuf)->iov_write); free((*dbuf)->iov_recv); + free((*dbuf)->msgs); free(*dbuf); + *dbuf = NULL; } - *dbuf = NULL; +} + +void _sigalrm_handler(int signum) { + /* Nothing to do except interrupting the pending syscall */ + if (signum) {} /* Avoid compiler warning */ } int _compare_uint_pair(const void *pa, const void *pb) { @@ -295,7 +479,7 @@ int _compare_uint_pair(const void *pa, const void *pb) { } void _update_ordered_seq_numbers(dgrambuf_t dbuf) { - ssize_t i; + 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; @@ -303,6 +487,7 @@ void _update_ordered_seq_numbers(dgrambuf_t dbuf) { } /* 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; } diff --git a/mcastseed/src/dgrambuf.h b/mcastseed/src/dgrambuf.h index fab7649..f405757 100644 --- a/mcastseed/src/dgrambuf.h +++ b/mcastseed/src/dgrambuf.h @@ -7,17 +7,31 @@ */ #include /* size_t */ +#define DGRAMBUF_RECV_OVERWRITE 1 << 1 +#define DGRAMBUF_RECV_EINTR 1 << 2 +#define DGRAMBUF_RECV_IOVEC_FULL 1 << 3 +#define DGRAMBUF_RECV_FINALIZE 1 << 4 +#define DGRAMBUF_RECV_FUTURE_DGRAM 1 << 5 +#define DGRAMBUF_RECV_VALID_DGRAM 1 << 6 + +#define DGRAMBUF_WRITE_PARTIAL 1 << 1 +#define DGRAMBUF_WRITE_EWOULDBLOCK_OR_EINTR 1 << 2 +#define DGRAMBUF_WRITE_IOVEC_FULL 1 << 3 +#define DGRAMBUF_WRITE_SUCCESS 1 << 4 + typedef struct dgrambuf_t *dgrambuf_t; dgrambuf_t dgrambuf_new(size_t dgram_slots, size_t dgram_max_size, size_t dgram_header_size, size_t iovec_slots); void dgrambuf_free(dgrambuf_t *dbuf); -size_t dgrambuf_free_count(const dgrambuf_t); -void dgrambuf_set_validate_func(dgrambuf_t dbuf, unsigned int (*func)(unsigned int, void *) ); +size_t dgrambuf_get_free_count(const dgrambuf_t); +int dgrambuf_everything_was_received(dgrambuf_t dbuf); +void dgrambuf_set_validate_func(dgrambuf_t dbuf, int (*validate_func)(unsigned int, void *, unsigned int *)); int dgrambuf_have_data_ready_to_write(dgrambuf_t dbuf); +int dgrambuf_have_received_everything(dgrambuf_t dbuf); - -int dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd); -ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd); +/* Warning : dgrambuf_recvmmsg sets and restore SIGALRM handler if timeout != 0 */ +ssize_t dgrambuf_recvmmsg(dgrambuf_t dbuf, int sockfd, int timeout, int *info); +ssize_t dgrambuf_write(dgrambuf_t dbuf, int fd, int *info); #endif /* DGRAMBUF_H */ diff --git a/mcastseed/src/dgrambuf_test.c b/mcastseed/src/dgrambuf_test.c index 1b96e3d..6f9ef22 100644 --- a/mcastseed/src/dgrambuf_test.c +++ b/mcastseed/src/dgrambuf_test.c @@ -15,13 +15,17 @@ int open_test_socket(); */ int main() { - int res=1, sockfd=open_test_socket(); - dgrambuf_t dgb=dgrambuf_new(3, 50); - while (res > 0) { - res = dgrambuf_recvmmsg(dgb, sockfd); + int res, sockfd, info; + dgrambuf_t dgb; + + sockfd = open_test_socket(); + dgb = dgrambuf_new(3, 50, 8, 8); + + do { + res = dgrambuf_recvmmsg(dgb, sockfd, 1, &info); printf("dgrambuf_recvmmsg() => %i\n", res); - printf("dgrambuf_free_count => %zi\n", dgrambuf_free_count(dgb)); - } + printf("dgrambuf_free_count => %zu\n", dgrambuf_get_free_count(dgb)); + } while ( res > 0 ); return 0; } diff --git a/mcastseed/src/mcastleech.c b/mcastseed/src/mcastleech.c index cdd0d9c..df069ac 100644 --- a/mcastseed/src/mcastleech.c +++ b/mcastseed/src/mcastleech.c @@ -11,13 +11,14 @@ #include /* close() */ #include /* fprintf(), stderr */ #include /* EXIT_SUCCESS */ +#include /* fcntl() */ #include "msock.h" #include "dgrambuf.h" #define MTU 1500 #define MULTICAST_RECV_BUF (MTU-20-8) #define MULTICAST_SO_RCVBUF_WANTED 425984 -//XXX Make it dynamic, with the effective value of so_rcvbuf +/*XXX Make it dynamic, with the effective value of so_rcvbuf */ #define MAX_IOVEC (MULTICAST_SO_RCVBUF_WANTED / MULTICAST_RECV_BUF) #define DGRAM_HEADER_SIZE 8 @@ -54,9 +55,10 @@ void usage(char *msg); void arg_parse(int argc, char* argv[]); void fsm_trace(int state); int get_available_mem_kb(); +void set_O_NONBLOCK(int fd, int set); void dgrambuf_init(); -uint32_t validate_data_dgram(unsigned int nread, void *recvbuf); -void ack(uint32_t seq); +int validate_data_dgram(unsigned int nread, void *recvbuf, unsigned int *seq); +int send_status(int state, int info_r, int info_w); /* Parts of the "protocol", definitions are after main() */ int wait_hello_and_connect_back(); @@ -72,6 +74,12 @@ int main(int argc, char* argv[]) { arg_parse(argc, argv); dgrambuf_init(); + /*XXX Maybe elsewhere, when popen'ing target program */ + set_O_NONBLOCK(1, 1); + +/* XXX Dummy */ + fcntl(1, F_SETPIPE_SZ, 4096); + fprintf(stderr, "pipe_size==%i\n", fcntl(1, F_GETPIPE_SZ)); /* Finite state machine */ while ( state > 0 ) { fsm_trace(state); @@ -85,7 +93,7 @@ int main(int argc, char* argv[]) { else state = -1; break; case 4: state = (finalize_job() == 0)?5:-2; break; - case 5: state = (is_there_more_job() == 0)?2:0; break; + case 5: state = (is_there_more_job() == 0)?2:0; break; /* XXX Should retry recv ? */ } } fsm_trace(state); @@ -143,6 +151,7 @@ int wait_hello_and_connect_back() { if ( ucast_sock > 0 ) { close(ucast_sock); } + /* FIXME : ucast_client_socket() use DNS resolver and could block */ ucast_sock = ucast_client_socket(hbuf,port); if(ucast_sock < 0) { fprintf(stderr, "Could not setup unicast socket or connect to %s:%s\n", hbuf, port); @@ -179,30 +188,75 @@ int wait_start_and_start_job() { return 0; } - +/* +#define DGRAMBUF_RECV_OVERWRITE 1 << 1 +#define DGRAMBUF_RECV_EINTR 1 << 2 +#define DGRAMBUF_RECV_IOVEC_FULL 1 << 3 +#define DGRAMBUF_RECV_FINALIZE 1 << 4 +#define DGRAMBUF_RECV_VALID_DGRAM 1 << 5 + +#define DGRAMBUF_WRITE_PARTIAL 1 << 1 +#define DGRAMBUF_WRITE_EWOULDBLOCK_OR_EINTR 1 << 2 +#define DGRAMBUF_WRITE_IOVEC_FULL 1 << 3 +#define DGRAMBUF_WRITE_SUCCESS 1 << 4 +*/ int receive_data() { - ssize_t nwrite; - if ( dgrambuf_have_data_ready_to_write(dgrambuf) ) { - nwrite=dgrambuf_write(dgrambuf, 1); - fprintf(stderr, "dgrambuf_write => %zi\n", nwrite); + int info_r, info_w, res; + ssize_t nread, nwrite; + + /* Read (blocking, timeout = 1 sec) */ + nread = dgrambuf_recvmmsg(dgrambuf, mcast_sock, 1, &info_r); + if ( nread < 0 ) { + return nread; + } + + /* Write (non-blocking) */ + nwrite = dgrambuf_write(dgrambuf, 1, &info_w); + if ( nwrite < 0 ) { + return nwrite; } - return dgrambuf_recvmmsg(dgrambuf, mcast_sock); -} + fprintf(stderr, "receive_data(): nread == %zi, nwrite == %zi\n", nread, nwrite); + + /* Consider sending status back to seeder */ + res = send_status(1, info_r, info_w); + if ( res < 0 ) { + return res; + } + + if ( dgrambuf_everything_was_received(dgrambuf) ) { + return 0; + } -void ack(uint32_t seq) { - //TODO + return 1; } int finalize_job() { - //XXX Dummy test - ssize_t res; - while ( (res=dgrambuf_write(dgrambuf, 1)) > 0 ) { - fprintf(stderr, "dgrambuf_write => %zi\n", res); + ssize_t nwrite; + int info_w, res; + + /* Don't eat reources in a pooling fashion, blocking IO is fine when no more recv to do */ + set_O_NONBLOCK(1, 0); + + /* Flush the whole buffer */ + do { + nwrite = dgrambuf_write(dgrambuf, 1, &info_w); + if ( nwrite < 0 ) { + return nwrite; + } + fprintf(stderr, "finalize_job(): nwrite == %zi\n", nwrite); + } while ( nwrite > 0); + + /* Inform the seeder that have have finished */ + res = send_status(2, 0, info_w); + if ( res < 0 ) { + return res; } - return 0; + + return 0; } + int is_there_more_job() { return 1; } @@ -272,6 +326,23 @@ int get_available_mem_kb() { return 0; } +void set_O_NONBLOCK(int fd, int set) { + int res, flags; + + flags = fcntl(fd, F_GETFL); + if ( flags == -1 ) { + perror("fcntl(1, F_GETFL)"); + } + if ( set ) { + res = fcntl(fd, F_SETFL, flags | O_NONBLOCK); + } else { + res = fcntl(fd, F_SETFL, flags & !O_NONBLOCK); + } + if ( res == -1 ) { + perror("fcntl(1, F_SETFL)"); + } +} + void dgrambuf_init() { /* Guess dgrambuf size from global free memory */ size_t dgram_count; @@ -282,8 +353,9 @@ void dgrambuf_init() { } else { dgram_count = avail_mem / MULTICAST_RECV_BUF / 2 * 1024; } - //XXX Dummy + /* XXX Dummy dgram_count = 5; + */ /* Allocate dgrambuf */ dgrambuf = dgrambuf_new(dgram_count, MULTICAST_RECV_BUF, DGRAM_HEADER_SIZE, MAX_IOVEC); @@ -292,17 +364,28 @@ void dgrambuf_init() { exit(EXIT_FAILURE); } - fprintf(stderr, "dgrambuf_free_count() => %zi\n", dgrambuf_free_count(dgrambuf)); + fprintf(stderr, "dgrambuf_get_free_count() => %zu\n", dgrambuf_get_free_count(dgrambuf)); dgrambuf_set_validate_func(dgrambuf, validate_data_dgram); } -unsigned int validate_data_dgram(unsigned int nread, void *recvbuf ) { - if ( nread >= DGRAM_HEADER_SIZE && strncmp("data", recvbuf, 4) == 0 ) { - return ntohl( *( (uint32_t *) recvbuf+1 ) ); +int validate_data_dgram(unsigned int nread, void *recvbuf, unsigned int *seq) { + + if ( nread < DGRAM_HEADER_SIZE ) { + return 0; } - if ( nread >= 5 && strncmp("final", recvbuf, 5) == 0 ) { - return -1; + if ( strncmp("data", recvbuf, 4) == 0 ) { + *seq = ntohl( *( (uint32_t *) recvbuf+1 ) ); + return 1; + } + if ( strncmp("end:", recvbuf, 4) == 0 ) { + *seq = ntohl( *( (uint32_t *) recvbuf+1 ) ); + return 2; } + return 0; +} +int send_status(int state, int info_r, int info_w) { + if ( state && info_r && info_w ) {} + /* TODO Implement it */ return 0; } diff --git a/mcastseed/src/mcastseed.c b/mcastseed/src/mcastseed.c index 09cadac..6440fc6 100644 --- a/mcastseed/src/mcastseed.c +++ b/mcastseed/src/mcastseed.c @@ -82,31 +82,31 @@ int main(int argc, char *argv[]) { switch ( state ) { case 1: res = send_hello(); state = (res==0)?2:-1; break; case 2: res = accept_pending_clients_or_wait_a_bit(); - if (res==0) state = 2; // Some clients has just come in, try to get more - else if (res==1) state = 1; // Nothing new. Keep accepting clients after another hello - else if (res==2) state = 3; // Wanted clients are accepted - else state = -2; - break; + if (res==0) state = 2; /* Some clients has just come in, try to get more */ + else if (res==1) state = 1; /* Nothing new. Keep accepting clients after another hello */ + else if (res==2) state = 3; /* Wanted clients are accepted */ + else state = -2; + break; case 3: res = start_job(); - if (res==0) state = 3; // Keep trying to convince every client to start - else if (res==1) state = 4; // All clients have started the job pipe - else if (res==2) state = 4; // There is dead clients but all alive are ready to go - else state = -3; - break; + if (res==0) state = 3; /* Keep trying to convince every client to start */ + else if (res==1) state = 4; /* All clients have started the job pipe */ + else if (res==2) state = 4; /* There is dead clients but all alive are ready to go */ + else state = -3; + break; case 4: res = send_data(); - if (res==0) state = 4; - else if (res==1) state = 5; // All data sent - else state = -4; - break; + if (res==0) state = 4; + else if (res==1) state = 5; /* All data sent */ + else state = -4; + break; case 5: res = wait_all_finalize_job(); - if (res==0) state = 5; - else if (res==1) state = 6; - else state = -5; + if (res==0) state = 5; + else if (res==1) state = 6; + else state = -5; case 6: res = is_there_more_job(); - if (res==0) state = 0; - else if (res==1) state = 3; - else state = -6; - break; + if (res==0) state = 0; + else if (res==1) state = 3; + else state = -6; + break; } } fsm_trace(state); @@ -144,7 +144,7 @@ int accept_pending_clients_or_wait_a_bit() { FD_ZERO(&readfds); FD_ZERO(&exceptfds); - FD_SET(0,&readfds); // Read from stdin. Will never work as is on Windows, requires threads and so. + FD_SET(0,&readfds); FD_SET(ucast_sock,&readfds); FD_SET(ucast_sock,&exceptfds); timeout.tv_sec = 2; @@ -158,7 +158,7 @@ int accept_pending_clients_or_wait_a_bit() { if ( res > 0 ) { if (FD_ISSET(ucast_sock, &readfds)) { - //TODO : this assumes that the event is an accept() while ones could be send data there + /*TODO : this assumes that the event is an accept() while ones could be send data there */ if ( clients_next >= MAX_CLIENTS ) { fprintf(stderr, "%s\n", "Bouncing client, MAX_CLIENTS reached"); close(accept(ucast_sock, NULL, 0)); @@ -170,13 +170,13 @@ int accept_pending_clients_or_wait_a_bit() { clients_next++; } } - //TODO : drop this keybord read with accept(), this is not portable + /*TODO : drop this keybord read with accept(), this is not portable */ if ( FD_ISSET(0, &readfds)) { nread = read(0, readbuf, READ_BUF_LEN); if ( nread <= 0 ) { fprintf(stderr, "%s\n", "lost stdin"); } - // User wants to go now + /* User wants to go now */ return 2; } if (FD_ISSET(ucast_sock, &exceptfds)) { @@ -185,7 +185,7 @@ int accept_pending_clients_or_wait_a_bit() { } } if (res == 0 ) { - // nothing happened before timeout + /* Nothing happened before timeout */ return 1; } return 0; @@ -243,7 +243,7 @@ int start_job() { fprintf(stderr, "unexpected data from %i\n", i); clients[i].state = 2; } else { - // Received "ready" ack from client + /* Received "ready" ack from client */ clients[i].state = 1; } } @@ -256,7 +256,7 @@ int start_job() { all_non_dead_ready &= (clients[i].state == 1); } } - // (res == 0 ) nothing happened before timeout + /* (res == 0 ) nothing happened before timeout */ if ( all_ready ) return 1; @@ -270,33 +270,34 @@ void send_fake(char buf[], int paylen, int i) { *( (uint32_t *) buf+1 ) = htonl(i); snprintf(buf+29, 5, "%04i", i); *( (char *) buf+33 ) = ')'; - sendto(mcast_sock, buf, 8+paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); + sendto(mcast_sock, buf, paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); } int send_data() { ssize_t nwrite; char buf[] = "dataXXXXJe suis à la plage (XXXX).\n"; - int paylen = strlen(buf)-8; + int paylen = strlen(buf); int i; - //XXX Dummy + /* XXX Dummy */ send_fake(buf, paylen, 5); send_fake(buf, paylen, 4); -/* - for (i=6; i<=300; i+=2) { + send_fake(buf, paylen, 3); + + for (i=6; i<=300; i+=2) { send_fake(buf, paylen, i); } - for (i=7; i<=300; i+=2) { + for (i=7; i<=300; i+=2) { send_fake(buf, paylen, i); } -*/ + send_fake(buf, paylen, 1); send_fake(buf, paylen, 1); send_fake(buf, paylen, 2); *( (uint32_t *) buf+1 ) = htonl(3); - buf[22]='m', buf[23]='e', buf[24]='r'; buf[25]='.'; buf[26]='\n'; paylen = 19; - nwrite = sendto(mcast_sock, buf, 8+paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); + buf[22]='m', buf[23]='e', buf[24]='r'; buf[25]='.'; buf[26]='\n'; paylen = 27; + nwrite = sendto(mcast_sock, buf, paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); if ( nwrite < 0 ) { perror("sendto() failed"); return -1; @@ -316,10 +317,11 @@ int wait_all_finalize_job() { int all_non_dead_done; int i, res; SOCKET client_sock; - const char *payload = "final"; - int paylen = strlen(payload); + char buf[] = "end:XXXX"; + int paylen = strlen(buf); - nwrite = sendto(mcast_sock, payload, paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); + *( (uint32_t *) buf+1 ) = htonl(300); + nwrite = sendto(mcast_sock, buf, paylen, 0, mcast_addr->ai_addr, mcast_addr->ai_addrlen); if ( nwrite < 0 ) { perror("sendto() failed"); return -1; @@ -360,7 +362,7 @@ int wait_all_finalize_job() { fprintf(stderr, "unexpected data from %i\n", i); clients[i].state = 2; } else { - // Received "done." ack from client + /* Received "done." ack from client */ clients[i].state = 3; } } @@ -372,7 +374,7 @@ int wait_all_finalize_job() { all_non_dead_done &= (clients[i].state == 3); } } - // (res == 0 ) nothing happened before timeout + /* (res == 0 ) nothing happened before timeout */ if ( all_non_dead_done ) return 1; -- cgit v1.2.3