Carousel testing piyush without debt

README.md

@@ -1,3 +1,11 @@
+# Improving Connection Scalability in TAS
+
+[Link to report](https://drive.google.com/file/d/1m7WcUe_nFjVfJE7XbAvxkYd1Iilics2B/view?usp=sharing)
+
+For optimatizations related to short-lived connections, check the short-lived-optimizations branch.
+
+The remaining README is from TAS's master branch.
+
 # TCP Acceleration Service
 
 [![Build Status](https://travis-ci.org/tcp-acceleration-service/tas.svg?branch=master)](https://travis-ci.org/tcp-acceleration-service/tas)

lib/sockets/control.c

@@ -482,6 +482,11 @@ int tas_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen,
     newfd = sp->fd;
   }
 
+  if (ns->data.connection.status == SOC_FAILED) {
+      errno = EAGAIN;
+      ret = -1;
+      goto out;
+  }
   /* check if connection is still pending */
   if (ns->data.connection.status == SOC_CONNECTING) {
     flextcp_epoll_clear(s, EPOLLIN);
@@ -498,6 +503,8 @@ int tas_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen,
     }
   }
 
+  if(ns->data.connection.status != SOC_CONNECTED)
+    fprintf(stderr, "Piyush assertion failed %d\n", ns->data.connection.status);
   /* connection is opened now */
   assert(ns->data.connection.status == SOC_CONNECTED);
 

tas/config.c

@@ -513,8 +513,8 @@ int config_parse(struct configuration *c, int argc, char *argv[])
 static int config_defaults(struct configuration *c, char *progname)
 {
   c->ip = 0;
-  c->nic_rx_len = 16 * 1024;
-  c->nic_tx_len = 16 * 1024;
+  c->nic_rx_len = 3000 * 1024;
+  c->nic_tx_len = 3000 * 1024;
   c->app_kin_len = 1024 * 1024;
   c->app_kout_len = 1024 * 1024;
   c->arp_to = 500;
@@ -544,6 +544,10 @@ static int config_defaults(struct configuration *c, char *progname)
   c->cc_timely_beta = 0.8 * UINT32_MAX;
   c->cc_timely_min_rtt = 11;
   c->cc_timely_min_rate = 10000;
+  //c->scheduler = CONFIG_PS_CAROUSEL;
+  c->scheduler = CONFIG_PS_FQ;
+  c->ps_timewheel_resolution_us = 1;
+  c->ps_timewheel_max_us = 500 * 1000;
   c->fp_cores_max = 1;
   c->fp_interrupts = 1;
   c->fp_xsumoffload = 1;

tas/fast/fast_appctx.c

@@ -26,6 +26,7 @@
 #include <rte_config.h>
 
 #include <tas_memif.h>
+#include <utils_log.h>
 
 #include "internal.h"
 #include "fastemu.h"
@@ -87,6 +88,9 @@ int fast_appctx_poll_bump(struct dataplane_context *ctx, void *pqe,
   struct flextcp_pl_atx *atx = pqe;
   int ret;
 
+  //fprintf(stderr, "Calling fast_flows_bump for flow_id=%u\n",
+  //  atx->msg.connupdate.flow_id);
+  //TAS_LOG(ERR, MAIN, "Fast flows bump for flow_id=%u\n", atx->msg.connupdate.flow_id);
   ret = fast_flows_bump(ctx, atx->msg.connupdate.flow_id,
       atx->msg.connupdate.bump_seq, atx->msg.connupdate.rx_bump,
       atx->msg.connupdate.tx_bump, atx->msg.connupdate.flags, nbh, ts);

tas/fast/fast_flows.c

@@ -29,6 +29,7 @@
 
 #include <tas_memif.h>
 #include <utils_sync.h>
+#include <utils_log.h>
 
 #include "internal.h"
 #include "fastemu.h"
@@ -106,6 +107,7 @@ int fast_flows_qman(struct dataplane_context *ctx, uint32_t queue,
 {
   uint32_t flow_id = queue;
   struct flextcp_pl_flowst *fs = &fp_state->flowst[flow_id];
+  //TAS_LOG(ERR, MAIN, "fast_flow_qman: flow_id=%u flow=%p\n", flow_id, fs);
   uint32_t avail, len, tx_pos, tx_seq, ack, rx_wnd;
   uint16_t new_core;
   uint8_t fin;
@@ -126,6 +128,7 @@ int fast_flows_qman(struct dataplane_context *ctx, uint32_t queue,
     }
 
     /* clear queue manager queue */
+    //TAS_LOG(ERR, MAIN, "qman_set: flow_id=%u\n", flow_id);
     if (qman_set(&ctx->qman, flow_id, 0, 0, 0,
           QMAN_SET_RATE | QMAN_SET_MAXCHUNK | QMAN_SET_AVAIL) != 0)
     {
@@ -164,6 +167,7 @@ int fast_flows_qman(struct dataplane_context *ctx, uint32_t queue,
 
   /* if there is no data available, stop */
   if (avail == 0) {
+    //TAS_LOG(ERR, MAIN, "fast_flows_qman: No data available for flow=%p\n", fs);
     ret = -1;
     goto unlock;
   }
@@ -182,6 +186,7 @@ int fast_flows_qman(struct dataplane_context *ctx, uint32_t queue,
     fs->tx_next_pos -= fs->tx_len;
   }
   fs->tx_sent += len;
+  //TAS_LOG(ERR, MAIN, "fast_flows_qman: flow=%p Changing tx_avail from %d to %d\n", fs, fs->tx_avail, fs->tx_avail - len);
   fs->tx_avail -= len;
 
   fin = (fs->rx_base_sp & FLEXNIC_PL_FLOWST_TXFIN) == FLEXNIC_PL_FLOWST_TXFIN &&
@@ -214,6 +219,7 @@ int fast_flows_qman_fwd(struct dataplane_context *ctx,
   avail = tcp_txavail(fs, NULL);
 
   /* re-arm queue manager */
+  //TAS_LOG(ERR, MAIN, "qman_set: flow_id=%u\n", flow_id);
   if (qman_set(&ctx->qman, flow_id, fs->tx_rate, avail, TCP_MSS,
         QMAN_SET_RATE | QMAN_SET_MAXCHUNK | QMAN_SET_AVAIL) != 0)
   {
@@ -617,6 +623,7 @@ int fast_flows_packet(struct dataplane_context *ctx,
   new_avail = tcp_txavail(fs, NULL);
   if (new_avail > old_avail) {
     /* update qman queue */
+    //TAS_LOG(ERR, MAIN, "qman_set: flow_id=%u\n", flow_id);
     if (qman_set(&ctx->qman, flow_id, fs->tx_rate, new_avail -
           old_avail, TCP_MSS, QMAN_SET_RATE | QMAN_SET_MAXCHUNK
           | QMAN_ADD_AVAIL) != 0)
@@ -738,6 +745,7 @@ int fast_flows_bump(struct dataplane_context *ctx, uint32_t flow_id,
 
   /* update queue manager queue */
   if (old_avail < new_avail) {
+    //TAS_LOG(ERR, MAIN, "qman_set: flow_id=%u\n", flow_id);
     if (qman_set(&ctx->qman, flow_id, fs->tx_rate, new_avail -
           old_avail, TCP_MSS, QMAN_SET_RATE | QMAN_SET_MAXCHUNK
           | QMAN_ADD_AVAIL) != 0)
@@ -748,6 +756,7 @@ int fast_flows_bump(struct dataplane_context *ctx, uint32_t flow_id,
   }
 
   /* update flow state */
+  //TAS_LOG(ERR, MAIN, "fast_flows_bump: Updating tx_avail from %d to %d for flow=%p\n", fs->tx_avail, tx_avail, fs);
   fs->tx_avail = tx_avail;
   rx_avail_prev = fs->rx_avail;
   fs->rx_avail += rx_bump;
@@ -815,6 +824,7 @@ void fast_flows_retransmit(struct dataplane_context *ctx, uint32_t flow_id)
 
   /* update queue manager */
   if (new_avail > old_avail) {
+    //TAS_LOG(ERR, MAIN, "qman_set: flow_id=%u\n", flow_id);
     if (qman_set(&ctx->qman, flow_id, fs->tx_rate, new_avail - old_avail,
           TCP_MSS, QMAN_SET_RATE | QMAN_SET_MAXCHUNK | QMAN_ADD_AVAIL) != 0)
     {

tas/fast/fastemu.c

@@ -61,7 +61,7 @@ int dataplane_init(void)
 {
   if (FLEXNIC_INTERNAL_MEM_SIZE < sizeof(struct flextcp_pl_mem)) {
     fprintf(stderr, "dataplane_init: internal flexnic memory size not "
-        "sufficient (got %x, need %zx)\n", FLEXNIC_INTERNAL_MEM_SIZE,
+        "sufficient (got %llx, need %zx)\n", FLEXNIC_INTERNAL_MEM_SIZE,
         sizeof(struct flextcp_pl_mem));
     return -1;
   }
@@ -73,7 +73,7 @@ int dataplane_init(void)
   }
   if (FLEXNIC_PL_FLOWST_NUM > FLEXNIC_NUM_QMQUEUES) {
     fprintf(stderr, "dataplane_init: more flow states than queue manager queues"
-        "(%u > %u)\n", FLEXNIC_PL_FLOWST_NUM, FLEXNIC_NUM_QMQUEUES);
+        "(%u > %llu)\n", FLEXNIC_PL_FLOWST_NUM, FLEXNIC_NUM_QMQUEUES);
     return -1;
   }
 
@@ -131,6 +131,7 @@ void dataplane_loop(struct dataplane_context *ctx)
 
   while (!exited) {
     unsigned n = 0;
+    unsigned ret = 0;
 
     /* count cycles of previous iteration if it was busy */
     prev_cyc = cyc;
@@ -153,16 +154,31 @@ void dataplane_loop(struct dataplane_context *ctx)
     n += poll_qman_fwd(ctx, ts);
 
     STATS_TS(poll_qman_start);
-    n += poll_qman(ctx, ts);
+    ret = poll_qman(ctx, ts);
+    n+=ret;
     STATS_TS(poll_qman_end);
+
+    if(ret>0)
+    {
+        STATS_ATOMIC_ADD(ctx, cyc_qm_useful, poll_qman_end - poll_qman_start);
+    }
+
     STATS_ATOMIC_ADD(ctx, cyc_qm, poll_qman_end - poll_qman_start);
 
-    n += poll_queues(ctx, ts);
-    STATS_TS(qs);
-    STATS_ATOMIC_ADD(ctx, cyc_qs, qs - poll_qman_end);
+    //n += poll_queues(ctx, ts);
+    ret = poll_queues(ctx, ts);
+    n+=ret;
+    STATS_TS(qs_end);
+    STATS_ATOMIC_ADD(ctx, cyc_qs, qs_end - poll_qman_end);
+
+    if(ret>0)
+    {
+        STATS_ATOMIC_ADD(ctx, cyc_qs_useful,  qs_end - poll_qman_end);
+    }
+
     n += poll_kernel(ctx, ts);
     STATS_TS(sp);
-    STATS_ATOMIC_ADD(ctx, cyc_sp, sp - qs);
+    STATS_ATOMIC_ADD(ctx, cyc_sp, sp - qs_end);
 
     /* flush transmit buffer */
     tx_flush(ctx);
@@ -228,33 +244,65 @@ void dataplane_dump_stats(void)
     if (ctx == NULL)
       continue;
 
-    TAS_LOG(INFO, MAIN, "DP [%u]> (POLL, EMPTY, TOTAL)\n", i);
-    TAS_LOG(INFO, MAIN, "qm=(%"PRIu64",%"PRIu64",%"PRIu64")  \n",
-            STATS_ATOMIC_FETCH(ctx, qm_poll),
-            STATS_ATOMIC_FETCH(ctx, qm_empty),
-            STATS_ATOMIC_FETCH(ctx, qm_total));
-    TAS_LOG(INFO, MAIN, "rx=(%"PRIu64",%"PRIu64",%"PRIu64")  \n",
-            STATS_ATOMIC_FETCH(ctx, rx_poll),
-            STATS_ATOMIC_FETCH(ctx, rx_empty),
-            STATS_ATOMIC_FETCH(ctx, rx_total));
-    TAS_LOG(INFO, MAIN, "qs=(%"PRIu64",%"PRIu64",%"PRIu64")  \n",
-            STATS_ATOMIC_FETCH(ctx, qs_poll),
-            STATS_ATOMIC_FETCH(ctx, qs_empty),
-            STATS_ATOMIC_FETCH(ctx, qs_total));
-    TAS_LOG(INFO, MAIN, "sp=(%"PRIu64",%"PRIu64",%"PRIu64")  \n",
-            STATS_ATOMIC_FETCH(ctx, sp_poll),
-            STATS_ATOMIC_FETCH(ctx, sp_empty),
-            STATS_ATOMIC_FETCH(ctx, sp_total));
-    TAS_LOG(INFO, MAIN, "tx=(%"PRIu64",%"PRIu64",%"PRIu64")  \n",
-            STATS_ATOMIC_FETCH(ctx, tx_poll),
-            STATS_ATOMIC_FETCH(ctx, tx_empty),
-            STATS_ATOMIC_FETCH(ctx, tx_total));
-    TAS_LOG(INFO, MAIN, "cyc=(%"PRIu64",%"PRIu64",%"PRIu64",%"PRIu64",%"PRIu64") \n",
-            STATS_ATOMIC_FETCH(ctx, cyc_qm),
-            STATS_ATOMIC_FETCH(ctx, cyc_rx),
-            STATS_ATOMIC_FETCH(ctx, cyc_qs),
-            STATS_ATOMIC_FETCH(ctx, cyc_sp),
-            STATS_ATOMIC_FETCH(ctx, cyc_tx));
+    uint64_t qm_total = STATS_ATOMIC_FETCH(ctx, qm_total);
+    uint64_t rx_total = STATS_ATOMIC_FETCH(ctx, rx_total);
+    uint64_t qs_total = STATS_ATOMIC_FETCH(ctx, qs_total);
+    uint64_t sp_total = STATS_ATOMIC_FETCH(ctx, sp_total);
+    uint64_t tx_total = STATS_ATOMIC_FETCH(ctx, tx_total);
+
+    uint64_t cyc_qm = STATS_ATOMIC_FETCH(ctx, cyc_qm);
+    uint64_t cyc_qm_useful = STATS_ATOMIC_FETCH(ctx, cyc_qm_useful);
+    uint64_t cyc_rx = STATS_ATOMIC_FETCH(ctx, cyc_rx);
+    uint64_t cyc_qs = STATS_ATOMIC_FETCH(ctx, cyc_qs);
+    uint64_t cyc_qs_useful = STATS_ATOMIC_FETCH(ctx, cyc_qs_useful);
+    uint64_t cyc_sp = STATS_ATOMIC_FETCH(ctx, cyc_sp);
+    uint64_t cyc_tx = STATS_ATOMIC_FETCH(ctx, cyc_tx);
+
+    uint64_t qm_poll = STATS_ATOMIC_FETCH(ctx, qm_poll);
+    uint64_t rx_poll = STATS_ATOMIC_FETCH(ctx, rx_poll);
+    uint64_t qs_poll = STATS_ATOMIC_FETCH(ctx, qs_poll);
+    uint64_t sp_poll = STATS_ATOMIC_FETCH(ctx, sp_poll);
+    uint64_t tx_poll = STATS_ATOMIC_FETCH(ctx, tx_poll);
+
+    uint64_t qm_empty  = STATS_ATOMIC_FETCH(ctx, qm_empty);
+    uint64_t rx_empty  = STATS_ATOMIC_FETCH(ctx, rx_empty);
+    uint64_t qs_empty  = STATS_ATOMIC_FETCH(ctx, qs_empty);
+    uint64_t sp_empty  = STATS_ATOMIC_FETCH(ctx, sp_empty);
+    uint64_t tx_empty  = STATS_ATOMIC_FETCH(ctx, tx_empty);
+
+    uint64_t act_timewheel_cnt = STATS_FETCH(&(ctx->qman), act_timewheel_cnt);
+    uint64_t queue_new_ts_wrap_cnt = STATS_FETCH(&(ctx->qman), queue_new_ts_wrap_cnt);
+    uint64_t timewheel_delta_high = STATS_FETCH(&(ctx->qman), timewheel_delta_high);
+    uint64_t cyc_queue_activate = STATS_FETCH(&(ctx->qman), cyc_queue_activate);
+    uint64_t cyc_qman_poll = STATS_FETCH(&(ctx->qman), cyc_qman_poll);
+
+    TAS_LOG(INFO, MAIN, "DP [%u]> (POLL, EMPTY, TOTAL, CYC/POLL, CYC/TOTAL, EMPTY/POLL)\n", i);
+    TAS_LOG(INFO, MAIN, "qm       =(%"PRIu64",%"PRIu64",%"PRIu64", %lF, %lF, %lF)\n",
+            qm_poll, qm_empty,
+            qm_total, (double) cyc_qm/qm_poll, (double) cyc_qm/qm_total, (double) qm_empty/qm_poll);
+    TAS_LOG(INFO, MAIN, "rx       =(%"PRIu64",%"PRIu64",%"PRIu64", %lF, %lF, %lF)\n",
+            rx_poll, rx_empty,
+            rx_total, (double) cyc_rx/rx_poll, (double) cyc_rx/rx_total, (double) rx_empty/rx_poll);
+    TAS_LOG(INFO, MAIN, "qs       =(%"PRIu64",%"PRIu64",%"PRIu64", %lF, %lF, %lF)\n",
+            qs_poll, qs_empty,
+            qs_total, (double) cyc_qs/qs_poll, (double) cyc_qs/qs_total, (double) qs_empty/qs_poll);
+    TAS_LOG(INFO, MAIN, "sp       =(%"PRIu64",%"PRIu64",%"PRIu64", %lF, %lF, %lF)\n",
+            sp_poll, sp_empty,
+            sp_total, (double) cyc_sp/sp_poll, (double) cyc_sp/sp_total, (double) sp_empty/sp_poll);
+    TAS_LOG(INFO, MAIN, "tx       =(%"PRIu64",%"PRIu64",%"PRIu64", %lF, %lF, %lF)\n",
+            tx_poll, tx_empty,
+            tx_total, (double) cyc_tx/tx_poll, (double) cyc_tx/tx_total, (double) tx_empty/tx_poll);
+    TAS_LOG(INFO, MAIN, "cyc       =(\n\t\t\t\t\t\tcyc_qm = %"PRIu64",\n\t\t\t\t\t\tcyc_qm_useful = %"PRIu64",\n\t\t\t\t\t\tcyc_rx = %"PRIu64",\n\t\t\t\t\t\tcyc_qs = %"PRIu64",\n\t\t\t\t\t\tcyc_qs_useful = %"PRIu64",\n\t\t\t\t\t\tcyc_sp = %"PRIu64",\n\t\t\t\t\t\tcyc_tx = %"PRIu64"\n)\n",
+            cyc_qm, cyc_qm_useful, cyc_rx, cyc_qs, cyc_qs_useful, cyc_sp, cyc_tx);
+    TAS_LOG(INFO, MAIN, "act_timewheel_cnt=%"PRIu64", queue_new_ts_wrap_cnt=%"PRIu64", timewheel_delta_high=%"PRIu64", cyc_queue_activate=%"PRIu64", cyc_qman_poll=%"PRIu64"\n", act_timewheel_cnt, queue_new_ts_wrap_cnt, timewheel_delta_high, cyc_queue_activate, cyc_qman_poll);
+
+    uint64_t cyc_total = STATS_ATOMIC_FETCH(ctx, cyc_qm) + STATS_ATOMIC_FETCH(ctx, cyc_rx) + STATS_ATOMIC_FETCH(ctx, cyc_qs) + STATS_ATOMIC_FETCH(ctx, cyc_sp) + STATS_ATOMIC_FETCH(ctx, cyc_tx) + 1;
+    TAS_LOG(INFO, MAIN, "ratio=(%lf, %lf, %lf, %lf, %lf) \n",
+            (double) STATS_ATOMIC_FETCH(ctx, cyc_qm)/cyc_total,
+            (double) STATS_ATOMIC_FETCH(ctx, cyc_rx)/cyc_total,
+            (double) STATS_ATOMIC_FETCH(ctx, cyc_qs)/cyc_total,
+            (double) STATS_ATOMIC_FETCH(ctx, cyc_sp)/cyc_total,
+            (double) STATS_ATOMIC_FETCH(ctx, cyc_tx)/cyc_total);
 
 #ifdef QUEUE_STATS
     TAS_LOG(INFO, MAIN, "slow -> fast (%"PRIu64",%"PRIu64") avg_queuing_delay=%lF\n", 
@@ -457,6 +505,8 @@ static unsigned poll_qman(struct dataplane_context *ctx, uint32_t ts)
     return 0;
   }
 
+  //TAS_LOG(ERR, MAIN, "poll_qman: Send %u bytes for flow_id=%u\n", q_bytes[0], q_ids[0]);
+
   STATS_ADD(ctx, qm_total, ret);
 
   for (i = 0; i < ret; i++) {
@@ -482,6 +532,8 @@ static unsigned poll_qman(struct dataplane_context *ctx, uint32_t ts)
     //STATS_TS(end_fast_flows_qman);
     //STATS_ATOMIC_ADD(ctx, cyc_fast_flows_qman, );
 
+    //if (use != 0)
+    //  fprintf(stderr, "Didn't send anything!!!\n");
     if (use == 0)
      off++;
   }