ufd: fix memleak detected by mt_asan_check (#906)

Fix below fail:

MTL: 2024-06-15 00:45:30, mt_asan_check, leak of 1624 byte(s) at
0x7fc17ff3b900
MTL: 2024-06-15 00:45:30, mt_asan_check, backtrace info: MTL: 2024-06-15
00:45:30, mt_asan_check,
/lib/x86_64-linux-gnu/libasan.so.5(+0x6cd40) [0x7fc9823fdd40] MTL:
2024-06-15 00:45:30, mt_asan_check,
/usr/local/lib/x86_64-linux-gnu/libmtl.so(mt_rte_zmalloc_socket+0x18a)
[0x7fc981d07c69]
MTL: 2024-06-15 00:45:30, mt_asan_check,
/usr/local/lib/x86_64-linux-gnu/libmtl.so(+0x6ecbfe) [0x7fc982200bfe]
MTL: 2024-06-15 00:45:30, mt_asan_check,
/usr/local/lib/x86_64-linux-gnu/libmtl.so(+0x6ed322) [0x7fc982201322]
MTL: 2024-06-15 00:45:30, mt_asan_check,
/usr/local/lib/x86_64-linux-gnu/libmtl.so(mufd_socket_port+0x132)
[0x7fc982201b26]
MTL: 2024-06-15 00:45:30, mt_asan_check,
/usr/local/lib/x86_64-linux-gnu/libmtl.so(mufd_socket+0x55)
[0x7fc982202d20]
MTL: 2024-06-15 00:45:30, mt_asan_check,
./build/app/UfdServerSample(+0x81f0) [0x559cc2cc11f0] MTL: 2024-06-15
00:45:30, mt_asan_check,
/lib/x86_64-linux-gnu/libc.so.6(__libc_start_main+0xf3) [0x7fc981923083]
MTL: 2024-06-15 00:45:30, mt_asan_check,
./build/app/UfdServerSample(+0x588e) [0x559cc2cbe88e]

Signed-off-by: Frank Du <frank.du@intel.com>
(cherry picked from commit b087f1f)

lib/src/mt_cni.c

@@ -393,7 +393,7 @@ static void* cni_traffic_thread(void* arg) {
   info("%s, start\n", __func__);
   while (rte_atomic32_read(&cni->stop_thread) == 0) {
     cni_traffic(impl);
-    mt_sleep_ms(1);
+    mt_sleep_ms(cni->thread_sleep_ms);
   }
   info("%s, stop\n", __func__);
 
@@ -563,6 +563,7 @@ int mt_cni_init(struct mtl_main_impl* impl) {
 
   cni_impl->lcore_tasklet = (p->flags & MTL_FLAG_CNI_THREAD) ? false : true;
   rte_atomic32_set(&cni_impl->stop_thread, 0);
+  cni_impl->thread_sleep_ms = 1;
 
   for (int i = 0; i < num_ports; i++) {
     struct mt_cni_entry* cni = cni_get_entry(impl, i);
@@ -719,6 +720,9 @@ struct mt_csq_entry* mt_csq_get(struct mtl_main_impl* impl, enum mtl_port port,
   cni->csq_idx++;
   csq_unlock(cni);
 
+  /* csq enabled, disable the sleep */
+  mt_get_cni(impl)->thread_sleep_ms = 0;
+
   uint8_t* ip = flow->dip_addr;
   info("%s(%d), ip %u.%u.%u.%u port %u on %d\n", __func__, port, ip[0], ip[1], ip[2],
        ip[3], flow->dst_port, idx);

lib/src/mt_main.h

@@ -389,6 +389,7 @@ struct mt_cni_impl {
   rte_atomic32_t stop_thread;
   bool lcore_tasklet;
   struct mt_sch_tasklet_impl* tasklet;
+  int thread_sleep_ms;
 
   struct mt_cni_entry entries[MTL_PORT_MAX];
 

lib/src/udp/udp_rxq.c

@@ -231,6 +231,7 @@ static struct mur_queue* urq_get(struct mudp_rxq_mgr* mgr,
   if (q) {
     if (!q->reuse_port || !create->reuse_port) {
       err("%s(%d,%u), already used\n", __func__, port, dst_port);
+      q = NULL;
       goto out_unlock_fail;
     }
 
@@ -257,6 +258,7 @@ static struct mur_queue* urq_get(struct mudp_rxq_mgr* mgr,
   q->rx_burst_pkts = 128;
   MT_TAILQ_INIT(&q->client_head);
   mt_pthread_mutex_init(&q->mutex, NULL);
+  rte_atomic32_inc(&q->refcnt);
 
   /* create flow */
   struct mt_rxq_flow flow;
@@ -265,28 +267,37 @@ static struct mur_queue* urq_get(struct mudp_rxq_mgr* mgr,
   flow.dst_port = dst_port;
   q->rxq = mt_rxq_get(impl, port, &flow);
   if (!q->rxq) {
-    err("%s(%d,%u), get rxq fail\n", __func__, port, dst_port);
-    urq_put(q);
-    goto out_unlock_fail;
+    /* wa for e810 pf mode since it doesn't support MT_RXQ_FLOW_F_NO_IP */
+    warn("%s(%d,%u), get rxq fail with no ip flow, try cni queue\n", __func__, port,
+         dst_port);
+    flow.flags |= MT_RXQ_FLOW_F_FORCE_CNI;
+    q->rxq = mt_rxq_get(impl, port, &flow);
+    if (!q->rxq) {
+      err("%s(%d,%u), get rxq fail with CNI also\n", __func__, port, dst_port);
+      goto out_unlock_fail;
+    }
+    /* start mtl sch with CNI tasklet mode */
+    if (!mt_started(impl)) {
+      mtl_start(impl);
+    }
   }
   q->rxq_id = mt_rxq_queue_id(q->rxq);
 
   ret = urq_mgr_add(mgr, q);
   if (ret < 0) {
     err("%s(%d,%u), urq mgr add fail %d\n", __func__, port, dst_port, ret);
-    urq_put(q);
     goto out_unlock_fail;
   }
 
   *idx = 0;
-  rte_atomic32_inc(&q->refcnt);
   urq_mgr_unlock(mgr);
 
   info("%s(%d,%u), new q %p\n", __func__, port, dst_port, q);
   return q;
 
 out_unlock_fail:
   urq_mgr_unlock(mgr);
+  if (q) urq_put(q);
   return NULL;
 }
 

lib/src/udp/ufd_main.c

@@ -59,6 +59,8 @@ static int ufd_free_slot(struct ufd_mt_ctx* ctx, struct ufd_slot* slot) {
 }
 
 static int ufd_free_mt_ctx(struct ufd_mt_ctx* ctx) {
+  struct mtl_main_impl* mt = ctx->mt;
+
   if (ctx->slots) {
     for (int i = 0; i < ufd_max_slot(ctx); i++) {
       /* check if any not free slot */
@@ -69,15 +71,18 @@ static int ufd_free_mt_ctx(struct ufd_mt_ctx* ctx) {
     mt_rte_free(ctx->slots);
     ctx->slots = NULL;
   }
-  if (ctx->mt) {
-    mtl_uninit(ctx->mt);
-    ctx->mt = NULL;
-  }
   mt_pthread_mutex_destroy(&ctx->slots_lock);
   if (ctx->alloc_with_rte)
     mt_rte_free(ctx);
   else
     mt_free(ctx);
+
+  /* always mtl_uninit at the last */
+  if (mt) {
+    mtl_uninit(mt);
+    mt = NULL;
+  }
+
   return 0;
 }