Fix: protect shared port access with proper sync

Signed-off-by: Kasiewicz, Marek <marek.kasiewicz@intel.com>

Author: Sakoram

lib/src/datapath/mt_queue.c

@@ -45,7 +45,20 @@ static uint16_t rx_csq_burst(struct mt_rxq_entry* entry, struct rte_mbuf** rx_pk
 
 static uint16_t rx_dpdk_burst(struct mt_rxq_entry* entry, struct rte_mbuf** rx_pkts,
                               const uint16_t nb_pkts) {
-  return mt_dpdk_rx_burst(entry->rxq, rx_pkts, nb_pkts);
+  enum mtl_port port_id = entry->rxq->port;
+  struct mt_interface* inf = mt_if(entry->parent, port_id);
+  int ret;
+
+  /* Trylock as we should not block in tasklates */
+  ret = mt_pthread_rwlock_tryrdlock(&inf->rl_rwlock);
+  if (ret) {
+    dbg("%s(%d), try lock fail %d\n", __func__, port_id, ret);
+    return 0;
+  }
+  ret = mt_dpdk_rx_burst(entry->rxq, rx_pkts, nb_pkts);
+  mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+
+  return ret;
 }
 
 struct mt_rxq_entry* mt_rxq_get(struct mtl_main_impl* impl, enum mtl_port port,
@@ -163,7 +176,20 @@ static uint16_t tx_tsq_burst(struct mt_txq_entry* entry, struct rte_mbuf** tx_pk
 
 static uint16_t tx_dpdk_burst(struct mt_txq_entry* entry, struct rte_mbuf** tx_pkts,
                               uint16_t nb_pkts) {
-  return mt_dpdk_tx_burst(entry->txq, tx_pkts, nb_pkts);
+  enum mtl_port port_id = entry->txq->port;
+  struct mt_interface* inf = mt_if(entry->parent, port_id);
+  uint16_t ret;
+
+  /* Trylock as we should not block in tasklates */
+  ret = mt_pthread_rwlock_tryrdlock(&inf->rl_rwlock);
+  if (ret) {
+    dbg("%s(%d), try lock fail %d\n", __func__, port_id, ret);
+    return 0;
+  }
+  ret = mt_dpdk_tx_burst(entry->txq, tx_pkts, nb_pkts);
+  mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+
+  return ret;
 }
 
 struct mt_txq_entry* mt_txq_get(struct mtl_main_impl* impl, enum mtl_port port,

lib/src/dev/mt_dev.c

@@ -707,8 +707,11 @@ static int dev_tx_queue_set_rl_rate(struct mt_interface* inf, uint16_t queue,
   /* not changed */
   if (bps == tx_queue->bps) return 0;
 
-  rte_atomic32_set(&inf->resetting, true);
-  mt_pthread_mutex_lock(&inf->vf_cmd_mutex);
+  ret = mt_pthread_rwlock_wrlock(&inf->rl_rwlock);
+  if (ret) {
+    err("%s(%d), failed to acquire write lock, ret %d\n", __func__, port, ret);
+    return ret;
+  }
 
   ret = rte_eth_dev_stop(port_id);
   if (ret) {
@@ -800,8 +803,9 @@ static int dev_tx_queue_set_rl_rate(struct mt_interface* inf, uint16_t queue,
   tx_queue->bps = bps;
 
 exit:
-  mt_pthread_mutex_unlock(&inf->vf_cmd_mutex);
-  rte_atomic32_set(&inf->resetting, false);
+  ret = mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+  if (ret) err("%s(%d), failed to release write lock, ret %d\n", __func__, port, ret);
+
   return ret;
 }
 
@@ -2115,7 +2119,7 @@ int mt_dev_if_uinit(struct mtl_main_impl* impl) {
 
     mt_pthread_mutex_destroy(&inf->tx_queues_mutex);
     mt_pthread_mutex_destroy(&inf->rx_queues_mutex);
-    mt_pthread_mutex_destroy(&inf->vf_cmd_mutex);
+    mt_pthread_rwlock_destroy(&inf->rl_rwlock);
 
     dev_close_port(inf);
   }
@@ -2181,7 +2185,7 @@ int mt_dev_if_init(struct mtl_main_impl* impl) {
     inf->tx_pacing_way = p->pacing;
     mt_pthread_mutex_init(&inf->tx_queues_mutex, NULL);
     mt_pthread_mutex_init(&inf->rx_queues_mutex, NULL);
-    mt_pthread_mutex_init(&inf->vf_cmd_mutex, NULL);
+    mt_pthread_rwlock_pref_wr_init(&inf->rl_rwlock);
     rte_spinlock_init(&inf->stats_lock);
 
     if (mt_user_ptp_tsc_source(impl)) {

lib/src/mt_cni.c

@@ -344,7 +344,6 @@ static int cni_traffic(struct mtl_main_impl* impl) {
   for (int i = 0; i < num_ports; i++) {
     cni = cni_get_entry(impl, i);
     if (!cni->rxq) continue;
-    if (rte_atomic32_read(&impl->inf[i].resetting)) continue;
 
     struct mt_rx_pcap* pcap = &cni->pcap;
     /* if any pcap progress */

lib/src/mt_flow.c

@@ -27,6 +27,7 @@ static struct rte_flow* rte_rx_flow_create_raw(struct mt_interface* inf, uint16_
   struct rte_flow_item_raw spec = {0};
   struct rte_flow_item_raw mask = {0};
   struct rte_flow_action_queue to_queue = {0};
+  int ret;
 
   uint16_t port_id = inf->port_id;
   char pkt_buf[] =
@@ -58,9 +59,17 @@ static struct rte_flow* rte_rx_flow_create_raw(struct mt_interface* inf, uint16_
   action[0].conf = &to_queue;
   action[1].type = RTE_FLOW_ACTION_TYPE_END;
 
-  mt_pthread_mutex_lock(&inf->vf_cmd_mutex);
+  ret = mt_pthread_rwlock_wrlock(&inf->rl_rwlock);
+  if (ret) {
+    err("%s(%d), failed to acquire write lock, ret %d\n", __func__, port_id, ret);
+    return NULL;
+  }
   r_flow = rte_flow_create(port_id, &attr, pattern, action, &error);
-  mt_pthread_mutex_unlock(&inf->vf_cmd_mutex);
+  ret = mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+  if (ret) {
+    err("%s(%d), failed to release write lock, ret %d\n", __func__, port_id, ret);
+    return NULL;
+  }
   if (!r_flow) {
     err("%s(%d), rte_flow_create fail for queue %d, %s\n", __func__, port_id, q,
         mt_string_safe(error.message));
@@ -175,9 +184,17 @@ static struct rte_flow* rte_rx_flow_create(struct mt_interface* inf, uint16_t q,
     return NULL;
   }
 
-  mt_pthread_mutex_lock(&inf->vf_cmd_mutex);
+  ret = mt_pthread_rwlock_wrlock(&inf->rl_rwlock);
+  if (ret) {
+    err("%s(%d), failed to acquire write lock, ret %d\n", __func__, port_id, ret);
+    return NULL;
+  }
   r_flow = rte_flow_create(port_id, &attr, pattern, action, &error);
-  mt_pthread_mutex_unlock(&inf->vf_cmd_mutex);
+  ret = mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+  if (ret) {
+    err("%s(%d), failed to release write lock, ret %d\n", __func__, port_id, ret);
+    return NULL;
+  }
 
   /* WA specific for e810 for PF interfaces */
   if (!has_ip_flow && !r_flow) {
@@ -192,9 +209,17 @@ static struct rte_flow* rte_rx_flow_create(struct mt_interface* inf, uint16_t q,
       return NULL;
     }
 
-    mt_pthread_mutex_lock(&inf->vf_cmd_mutex);
+    ret = mt_pthread_rwlock_wrlock(&inf->rl_rwlock);
+    if (ret) {
+      err("%s(%d), failed to acquire write lock, ret %d\n", __func__, port_id, ret);
+      return NULL;
+    }
     r_flow = rte_flow_create(port_id, &attr, pattern, action, &error);
-    mt_pthread_mutex_unlock(&inf->vf_cmd_mutex);
+    ret = mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+    if (ret) {
+      err("%s(%d), failed to release write lock, ret %d\n", __func__, port_id, ret);
+      return NULL;
+    }
   }
 
   if (!r_flow) {
@@ -264,6 +289,7 @@ static int rx_flow_free(struct mt_interface* inf, struct mt_rx_flow_rsp* rsp) {
   enum mtl_port port = inf->port;
   struct rte_flow_error error;
   int ret;
+  int rwlock_ret;
   int max_retry = 5;
   int retry = 0;
 
@@ -273,9 +299,18 @@ static int rx_flow_free(struct mt_interface* inf, struct mt_rx_flow_rsp* rsp) {
     rsp->flow_id = -1;
   }
   if (rsp->flow) {
-    mt_pthread_mutex_lock(&inf->vf_cmd_mutex);
+    rwlock_ret = mt_pthread_rwlock_wrlock(&inf->rl_rwlock);
+    if (rwlock_ret) {
+      err("%s(%d), failed to acquire write lock, ret %d\n", __func__, port, rwlock_ret);
+      return rwlock_ret;
+    }
     ret = rte_flow_destroy(inf->port_id, rsp->flow, &error);
-    mt_pthread_mutex_unlock(&inf->vf_cmd_mutex);
+    rwlock_ret = mt_pthread_rwlock_unlock(&inf->rl_rwlock);
+    if (rwlock_ret) {
+      err("%s(%d), failed to release write lock, ret %d\n", __func__, port, rwlock_ret);
+      return rwlock_ret;
+    }
+
     if (ret < 0) {
       err("%s(%d), flow destroy fail, queue %d, retry %d\n", __func__, port,
           rsp->queue_id, retry);

lib/src/mt_main.h

@@ -704,8 +704,6 @@ struct mt_interface {
   struct rte_ether_addr* mcast_mac_lists; /* pool of multicast mac addrs */
   uint32_t mcast_nb;                      /* number of address */
   uint32_t status;                        /* MT_IF_STAT_* */
-  /* The port is temporarily off, e.g. during rte_tm_hierarchy_commit */
-  rte_atomic32_t resetting;
 
   /* default tx mbuf_pool */
   struct rte_mempool* tx_mbuf_pool;
@@ -715,11 +713,9 @@ struct mt_interface {
   uint16_t nb_rx_desc;
 
   struct rte_mbuf* pad;
-  /*
-   * protect rl and fdir for vf.
-   * _atomic_set_cmd(): There is incomplete cmd 112
-   */
-  pthread_mutex_t vf_cmd_mutex;
+
+  /* protect port during reset when changing RL speed */
+  pthread_rwlock_t rl_rwlock;
 
   /* tx queue resources */
   uint16_t nb_tx_q;

lib/src/mt_platform.h

@@ -93,6 +93,55 @@ static inline int mt_pthread_mutex_destroy(pthread_mutex_t* mutex) {
   return pthread_mutex_destroy(mutex);
 }
 
+static inline int mt_pthread_rwlock_init(pthread_rwlock_t* rwlock,
+                                         pthread_rwlockattr_t* attr) {
+  return pthread_rwlock_init(rwlock, attr);
+}
+
+static inline int mt_pthread_rwlock_pref_wr_init(pthread_rwlock_t* rwlock) {
+  pthread_rwlockattr_t rwlock_attr;
+  int ret;
+
+  ret = pthread_rwlockattr_init(&rwlock_attr);
+  if (ret) return ret;
+
+  ret = pthread_rwlockattr_setkind_np(&rwlock_attr,
+                                      PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
+  if (ret) {
+    pthread_rwlockattr_destroy(&rwlock_attr);
+    return ret;
+  }
+
+  ret = pthread_rwlock_init(rwlock, &rwlock_attr);
+  pthread_rwlockattr_destroy(&rwlock_attr);
+
+  return ret;
+}
+
+static inline int mt_pthread_rwlock_rdlock(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_rdlock(rwlock);
+}
+
+static inline int mt_pthread_rwlock_tryrdlock(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_tryrdlock(rwlock);
+}
+
+static inline int mt_pthread_rwlock_wrlock(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_wrlock(rwlock);
+}
+
+static inline int mt_pthread_rwlock_trywrlock(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_trywrlock(rwlock);
+}
+
+static inline int mt_pthread_rwlock_unlock(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_unlock(rwlock);
+}
+
+static inline int mt_pthread_rwlock_destroy(pthread_rwlock_t* rwlock) {
+  return pthread_rwlock_destroy(rwlock);
+}
+
 static inline int mt_pthread_cond_init(pthread_cond_t* cond,
                                        pthread_condattr_t* cond_attr) {
   return pthread_cond_init(cond, cond_attr);

lib/src/st2110/st_tx_video_session.c

@@ -430,15 +430,15 @@ static int tv_train_pacing(struct mtl_main_impl* impl, struct st_tx_video_sessio
   /* If the measured speed is lower than expected. Set higher bps and retrain to add
    * padding */
   if (measured_bps < rl_bps) {
-    info("%s(%d), measured bps %"PRIu64" is lower then set bps %"PRIu64"\n",
-         __func__, idx, (uint64_t)measured_bps, rl_bps);
+    info("%s(%d), measured bps %" PRIu64 " is lower than set bps %" PRIu64 "\n", __func__,
+         idx, (uint64_t)measured_bps, rl_bps);
     if (!mt_pacing_train_bps_result_search(impl, port, rl_bps, &bps_to_set)) {
       err("%s(%d), measured speed is too low on already trained bps\n", __func__, idx);
       return -EINVAL;
     }
 
     bps_to_set = (rl_bps * rl_bps) / measured_bps;
-    info("%s(%d), increase bps to %"PRIu64"\n", __func__, idx, bps_to_set);
+    info("%s(%d), increase bps to %" PRIu64 "\n", __func__, idx, bps_to_set);
     mt_pacing_train_bps_result_add(impl, port, rl_bps, bps_to_set);
     mt_txq_set_tx_bps(queue, bps_to_set);
     ret = tv_train_pacing(impl, s, s_port);