From 3aec660e814f470cc9453e69231cbe3102fd2a4c Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Mon, 28 Sep 2020 20:04:57 +0200 Subject: lib: Add compiler configuration for FRCP This allows configuring some parameters for FRCP at compile time, such as default values for Delta-t and configuration of the timerwheel. The timerwheel will now reschedule when it fails to create a packet, instead of setting the flow down immediately. Some new things added are options to store packets for retransmission on the heap, and using non-blocking calls for retransmission. The defaults do not change the current behaviour. Signed-off-by: Dimitri Staessens Signed-off-by: Sander Vrijders --- src/lib/CMakeLists.txt | 32 ++++++++++- src/lib/config.h.in | 26 ++++++++- src/lib/dev.c | 3 +- src/lib/frct.c | 32 +++++------ src/lib/timerwheel.c | 153 ++++++++++++++++++++++++++----------------------- 5 files changed, 153 insertions(+), 93 deletions(-) diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt index 4aad3a11..5e60e375 100644 --- a/src/lib/CMakeLists.txt +++ b/src/lib/CMakeLists.txt @@ -180,14 +180,42 @@ set(SHM_RDRB_NAME "/${SHM_PREFIX}.rdrb" CACHE INTERNAL "Name for the main POSIX shared memory buffer") set(SHM_RDRB_BLOCK_SIZE "sysconf(_SC_PAGESIZE)" CACHE STRING "Packet buffer block size, multiple of pagesize for performance") -set(SHM_RDRB_MULTI_BLOCK true CACHE BOOL +set(SHM_RDRB_MULTI_BLOCK TRUE CACHE BOOL "Packet buffer multiblock packet support") -set(SHM_RBUFF_LOCKLESS 0 CACHE BOOL +set(SHM_RBUFF_LOCKLESS TRUE CACHE BOOL "Enable shared memory lockless rbuff support") set(QOS_DISABLE_CRC TRUE CACHE BOOL "Ignores ber setting on all QoS cubes") +set(DELTA_T_MPL 60 CACHE STRING + "Maximum packet lifetime (s)") +set(DELTA_T_ACK 10 CACHE STRING + "Maximum time to acknowledge a packet (s)") +set(DELTA_T_RTX 120 CACHE STRING + "Maximum time to retransmit a packet (s)") +set(DELTA_T_ACK_DELAY 10 CACHE STRING + "Maximum time to wait before acknowledging a packet (ms)") +set(FRCT_REORDER_QUEUE_SIZE 256 CACHE STRING + "Size of the reordering queue, must be a power of 2") set(FRCT_RTO_MIN 250 CACHE STRING "Minimum Retransmission Timeout (RTO) for FRCT (us)") +set(FRCT_TICK_TIME 500 CACHE STRING + "Tick time for FRCT activity (retransmission, acknowledgments) (us)") +set(RXM_BUFFER_ON_HEAP FALSE CACHE BOOL + "Store packets for retransmission on the heap instead of in packet buffer") +set(RXM_BLOCKING TRUE CACHE BOOL + "Use blocking writes for retransmission") +set(RXM_MIN_RESOLUTION 20 CACHE STRING + "Minimum retransmission delay (ns), as a power to 2") +set(RXM_WHEEL_MULTIPLIER 4 CACHE STRING + "Factor for retransmission wheel levels as a power to 2") +set(RXM_WHEEL_LEVELS 3 CACHE STRING + "Number of levels in the retransmission wheel") +set(RXM_WHEEL_SLOTS_PER_LEVEL 256 CACHE STRING + "Number of slots per level in the retransmission wheel, must be a power of 2") +set(ACK_WHEEL_SLOTS 128 CACHE STRING + "Number of slots in the acknowledgment wheel, must be a power of 2") +set(ACK_WHEEL_RESOLUTION 20 CACHE STRING + "Minimum acknowledgment delay (ns), as a power to 2") set(SOURCE_FILES_DEV # Add source files here diff --git a/src/lib/config.h.in b/src/lib/config.h.in index 7cac76a6..38c364c6 100644 --- a/src/lib/config.h.in +++ b/src/lib/config.h.in @@ -55,7 +55,7 @@ #cmakedefine HAVE_FUSE #ifdef HAVE_FUSE -#define FUSE_PREFIX "@FUSE_PREFIX@" +#define FUSE_PREFIX "@FUSE_PREFIX@" #endif #define PTHREAD_COND_CLOCK @PTHREAD_COND_CLOCK@ @@ -67,4 +67,26 @@ #define DU_BUFF_HEADSPACE @DU_BUFF_HEADSPACE@ #define DU_BUFF_TAILSPACE @DU_BUFF_TAILSPACE@ -#define RTO_MIN @FRCT_RTO_MIN@ +/* Default Delta-t parameters */ +#define DELT_MPL (@DELTA_T_MPL@ * BILLION) /* ns */ +#define DELT_A (@DELTA_T_ACK@ * BILLION) /* ns */ +#define DELT_R (@DELTA_T_RTX@ * BILLION) /* ns */ + +#define DELT_ACK (@DELTA_T_ACK_DELAY@ * MILLION) /* ns */ + +#define RQ_SIZE (@FRCT_REORDER_QUEUE_SIZE@) +#define RTO_MIN (@FRCT_RTO_MIN@ * 1000) + +#define TICTIME (@FRCT_TICK_TIME@ * 1000) /* ns */ + +/* Retransmission tuning */ +#cmakedefine RXM_BUFFER_ON_HEAP +#cmakedefine RXM_BLOCKING + +#define RXMQ_RES (@RXM_MIN_RESOLUTION@) /* 2^N ns */ +#define RXMQ_BUMP (@RXM_WHEEL_MULTIPLIER@) +#define RXMQ_LVLS (@RXM_WHEEL_LEVELS@) +#define RXMQ_SLOTS (@RXM_WHEEL_SLOTS_PER_LEVEL@) + +#define ACKQ_SLOTS (@ACK_WHEEL_SLOTS@) +#define ACKQ_RES (@ACK_WHEEL_RESOLUTION@) /* 2^N ns */ diff --git a/src/lib/dev.c b/src/lib/dev.c index 5cf23639..ed377367 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -63,7 +63,6 @@ #define SECMEMSZ 16384 #define SYMMKEYSZ 32 #define MSGBUFSZ 2048 -#define TICTIME 1000000 /* ns */ struct flow_set { size_t idx; @@ -801,7 +800,7 @@ int flow_dealloc(int fd) timeo = frcti_dealloc(f->frcti); - if (ret == -ETIMEDOUT && timeo < 0) + if ((ret == -ETIMEDOUT || ret == -EFLOWDOWN) && timeo < 0) timeo = -timeo; } diff --git a/src/lib/frct.c b/src/lib/frct.c index c26910fa..6ead72af 100644 --- a/src/lib/frct.c +++ b/src/lib/frct.c @@ -20,15 +20,6 @@ * Foundation, Inc., http://www.fsf.org/about/contact/. */ -/* Default Delta-t parameters */ -#define DELT_MPL (5 * BILLION) /* ns */ -#define DELT_A (1 * BILLION) /* ns */ -#define DELT_R (20 * BILLION) /* ns */ - -#define DELT_ACK (10 * MILLION) /* ns */ - -#define RQ_SIZE 256 - #define FRCT_PCILEN (sizeof(struct frct_pci)) struct frct_cr { @@ -96,7 +87,7 @@ static bool after(uint32_t seq1, return (int32_t)(seq2 - seq1) < 0; } -static void __send_ack(int fd, +static int __send_ack(int fd, int ackno) { struct shm_du_buff * sdb; @@ -105,9 +96,13 @@ static void __send_ack(int fd, struct flow * f; /* Raw calls needed to bypass frcti. */ +#ifdef RXM_BLOCKING idx = shm_rdrbuff_alloc_b(ai.rdrb, sizeof(*pci), NULL, &sdb, NULL); +#else + idx = shm_rdrbuff_alloc(ai.rdrb, sizeof(*pci), NULL, &sdb); +#endif if (idx < 0) - return; + return -ENOMEM; pci = (struct frct_pci *) shm_du_buff_head(sdb); memset(pci, 0, sizeof(*pci)); @@ -116,13 +111,18 @@ static void __send_ack(int fd, pci->ackno = hton32(ackno); f = &ai.flows[fd]; - +#ifdef RXM_BLOCKING if (shm_rbuff_write_b(f->tx_rb, idx, NULL)) { +#else + if (shm_rbuff_write(f->tx_rb, idx)) { +#endif ipcp_sdb_release(sdb); - return; + return -ENOMEM; } shm_flow_set_notify(f->set, f->flow_id, FLOW_PKT); + + return 0; } static void frct_send_ack(struct frcti * frcti) @@ -153,7 +153,8 @@ static void frct_send_ack(struct frcti * frcti) if (diff > frcti->a || diff < DELT_ACK) return; - __send_ack(fd, ackno); + if (__send_ack(fd, ackno) < 0) + return; pthread_rwlock_wrlock(&frcti->lock); @@ -439,8 +440,7 @@ static void rtt_estimator(struct frcti * frcti, frcti->srtt = MAX(1000U, srtt); frcti->mdev = MAX(100U, rttvar); - frcti->rto = MAX(RTO_MIN * 1000, - frcti->srtt + (frcti->mdev << 1)); + frcti->rto = MAX(RTO_MIN, frcti->srtt + (frcti->mdev << 1)); } static void __frcti_tick(void) diff --git a/src/lib/timerwheel.c b/src/lib/timerwheel.c index 33fcbc1c..4443832d 100644 --- a/src/lib/timerwheel.c +++ b/src/lib/timerwheel.c @@ -22,14 +22,6 @@ #include -#define RXMQ_SLOTS (1 << 8) /* #slots / level. */ -#define RXMQ_LVLS 3 /* #levels, bump for DTN. */ -#define RXMQ_BUMP 4 /* factor to bump lvl. */ -#define RXMQ_RES 20 /* res (ns) of lowest lvl. */ - -#define ACKQ_SLOTS (1 << 7) /* #slots for delayed ACK. */ -#define ACKQ_RES 20 /* resolution for dACK. */ - /* Overflow limits range to about 6 hours. */ #define ts_to_ns(ts) (ts.tv_sec * BILLION + ts.tv_nsec) #define ts_to_rxm_slot(ts) (ts_to_ns(ts) >> RXMQ_RES) @@ -38,9 +30,14 @@ struct rxm { struct list_head next; uint32_t seqno; +#ifdef RXM_BUFFER_ON_HEAP + uint8_t * pkt; + size_t pkt_len; +#else struct shm_du_buff * sdb; uint8_t * head; uint8_t * tail; +#endif time_t t0; /* Time when original was sent (us). */ size_t mul; /* RTO multiplier. */ struct frcti * frcti; @@ -85,8 +82,12 @@ static void timerwheel_fini(void) struct rxm * rxm; rxm = list_entry(p, struct rxm, next); list_del(&rxm->next); +#ifdef RXM_BUFFER_ON_HEAP + free(rxm->pkt); +#else shm_du_buff_ack(rxm->sdb); ipcp_sdb_release(rxm->sdb); +#endif free(rxm); } } @@ -156,8 +157,7 @@ static void timerwheel_move(void) j_max_slot += RXMQ_SLOTS; while (j++ < j_max_slot) { - list_for_each_safe(p, - h, + list_for_each_safe(p, h, &rw.rxms[i][j & (RXMQ_SLOTS - 1)]) { struct rxm * r; struct frct_cr * snd_cr; @@ -178,15 +178,12 @@ static void timerwheel_move(void) snd_cr = &r->frcti->snd_cr; rcv_cr = &r->frcti->rcv_cr; f = &ai.flows[r->fd]; - +#ifndef RXM_BUFFER_ON_HEAP shm_du_buff_ack(r->sdb); - +#endif if (f->frcti == NULL - || f->flow_id != r->flow_id) { - ipcp_sdb_release(r->sdb); - free(r); - continue; - } + || f->flow_id != r->flow_id) + goto cleanup; pthread_rwlock_wrlock(&r->frcti->lock); @@ -197,69 +194,57 @@ static void timerwheel_move(void) pthread_rwlock_unlock(&r->frcti->lock); /* Has been ack'd, remove. */ - if ((int) (r->seqno - snd_lwe) < 0) { - ipcp_sdb_release(r->sdb); - free(r); - continue; - } + if ((int) (r->seqno - snd_lwe) < 0) + goto cleanup; /* Check for r-timer expiry. */ - if (ts_to_ns(now) - r->t0 > r->frcti->r) { - ipcp_sdb_release(r->sdb); - free(r); - shm_rbuff_set_acl(f->rx_rb, - ACL_FLOWDOWN); - shm_rbuff_set_acl(f->tx_rb, - ACL_FLOWDOWN); - continue; - } + if (ts_to_ns(now) - r->t0 > r->frcti->r) + goto flow_down; if (r->frcti->probe && (r->frcti->rttseq + 1) == r->seqno) r->frcti->probe = false; - - /* Copy the data, safe rtx in other layers. */ - if (ipcp_sdb_reserve(&sdb, r->tail - r->head)) { - ipcp_sdb_release(r->sdb); - free(r); - shm_rbuff_set_acl(f->rx_rb, - ACL_FLOWDOWN); - shm_rbuff_set_acl(f->tx_rb, - ACL_FLOWDOWN); - continue; - } - +#ifdef RXM_BLOCKING + #ifdef RXM_BUFFER_ON_HEAP + if (ipcp_sdb_reserve(&sdb, r->pkt_len)) + #else + if (ipcp_sdb_reserve(&sdb, r->tail - r->head)) + #endif +#else + #ifdef RXM_BUFFER_ON_HEAP + if (shm_rdrbuff_alloc(ai.rdrb, r->pkt_len, NULL, + &sdb)) + #else + if (shm_rdrbuff_alloc(ai.rdrb, + r->tail - r->head, NULL, + &sdb)) + #endif +#endif + goto reschedule; /* rbuff full */ idx = shm_du_buff_get_idx(sdb); head = shm_du_buff_head(sdb); +#ifdef RXM_BUFFER_ON_HEAP + memcpy(head, r->pkt, r->pkt_len); +#else memcpy(head, r->head, r->tail - r->head); - ipcp_sdb_release(r->sdb); - - ((struct frct_pci *) head)->ackno = - hton32(rcv_lwe); - - /* Retransmit the copy. */ - if (shm_rbuff_write_b(f->tx_rb, idx, NULL)) { - ipcp_sdb_release(sdb); - free(r); - shm_rbuff_set_acl(f->rx_rb, - ACL_FLOWDOWN); - shm_rbuff_set_acl(f->tx_rb, - ACL_FLOWDOWN); - continue; - } - - /* Reschedule. */ - shm_du_buff_wait_ack(sdb); - - shm_flow_set_notify(f->set, - f->flow_id, - FLOW_PKT); - + r->sdb = sdb; r->head = head; r->tail = shm_du_buff_tail(sdb); - r->sdb = sdb; + shm_du_buff_wait_ack(sdb); +#endif + /* Retransmit the copy. */ + ((struct frct_pci *) head)->ackno = + hton32(rcv_lwe); +#ifdef RXM_BLOCKING + if (shm_rbuff_write_b(f->tx_rb, idx, NULL) == 0) +#else + if (shm_rbuff_write(f->tx_rb, idx) == 0) +#endif + shm_flow_set_notify(f->set, f->flow_id, + FLOW_PKT); + reschedule: r->mul++; /* Schedule at least in the next time slot. */ @@ -268,10 +253,24 @@ static void timerwheel_move(void) & (RXMQ_SLOTS - 1); list_add_tail(&r->next, &rw.rxms[i][rslot]); + + continue; + + flow_down: + shm_rbuff_set_acl(f->tx_rb, ACL_FLOWDOWN); + shm_rbuff_set_acl(f->rx_rb, ACL_FLOWDOWN); + cleanup: +#ifdef RXM_BUFFER_ON_HEAP + free(r->pkt); +#else + ipcp_sdb_release(r->sdb); +#endif + free(r); } } /* Move up a level in the wheel. */ rxm_slot >>= RXMQ_BUMP; + j >>= RXMQ_BUMP; } ack_slot = ts_to_ack_slot(now) & (ACKQ_SLOTS - 1) ; @@ -326,11 +325,20 @@ static int timerwheel_rxm(struct frcti * frcti, r->t0 = ts_to_ns(now); r->mul = 0; r->seqno = seqno; + r->frcti = frcti; +#ifdef RXM_BUFFER_ON_HEAP + r->pkt_len = shm_du_buff_tail(sdb) - shm_du_buff_head(sdb); + r->pkt = malloc(r->pkt_len); + if (r->pkt == NULL) { + free(r); + return -ENOMEM; + } + memcpy(r->pkt, shm_du_buff_head(sdb), r->pkt_len); +#else r->sdb = sdb; r->head = shm_du_buff_head(sdb); r->tail = shm_du_buff_tail(sdb); - r->frcti = frcti; - +#endif pthread_rwlock_rdlock(&r->frcti->lock); rto_slot = frcti->rto >> RXMQ_RES; @@ -348,6 +356,9 @@ static int timerwheel_rxm(struct frcti * frcti, } if (lvl >= RXMQ_LVLS) { /* Out of timerwheel range. */ +#ifdef RXM_BUFFER_ON_HEAP + free(r->pkt); +#endif free(r); return -EPERM; } @@ -357,9 +368,9 @@ static int timerwheel_rxm(struct frcti * frcti, pthread_mutex_lock(&rw.lock); list_add_tail(&r->next, &rw.rxms[lvl][slot]); - +#ifndef RXM_BUFFER_ON_HEAP shm_du_buff_wait_ack(sdb); - +#endif pthread_mutex_unlock(&rw.lock); return 0; -- cgit v1.2.3