ethdev: fix QinQ offload

[dpdk.git] / lib / librte_rcu / rte_rcu_qsbr.h

diff --git a/lib/librte_rcu/rte_rcu_qsbr.h b/lib/librte_rcu/rte_rcu_qsbr.h
index 53a5a71..0b55859 100644 (file)
--- a/lib/librte_rcu/rte_rcu_qsbr.h
+++ b/lib/librte_rcu/rte_rcu_qsbr.h
@@ -24,6 +24,7 @@
 extern "C" {
 #endif
 
+#include <stdbool.h>
 #include <stdio.h>
 #include <stdint.h>
 #include <inttypes.h>
@@ -82,6 +83,7 @@ struct rte_rcu_qsbr_cnt {
 
 #define __RTE_QSBR_CNT_THR_OFFLINE 0
 #define __RTE_QSBR_CNT_INIT 1
+#define __RTE_QSBR_CNT_MAX ((uint64_t)~0)
 
 /* RTE Quiescent State variable structure.
  * This structure has two elements that vary in size based on the
@@ -92,6 +94,10 @@ struct rte_rcu_qsbr_cnt {
 struct rte_rcu_qsbr {
        uint64_t token __rte_cache_aligned;
        /**< Counter to allow for multiple concurrent quiescent state queries */
+       uint64_t acked_token;
+       /**< Least token acked by all the threads in the last call to
+        *   rte_rcu_qsbr_check API.
+        */
 
        uint32_t num_elems __rte_cache_aligned;
        /**< Number of elements in the thread ID array */
@@ -211,7 +217,7 @@ rte_rcu_qsbr_thread_unregister(struct rte_rcu_qsbr *v, unsigned int thread_id);
  * call this API before calling rte_rcu_qsbr_quiescent. This can be called
  * during initialization or as part of the packet processing loop.
  *
- * The reader thread must call rte_rcu_thread_offline API, before
+ * The reader thread must call rte_rcu_qsbr_thread_offline API, before
  * calling any functions that block, to ensure that rte_rcu_qsbr_check
  * API does not wait indefinitely for the reader thread to update its QS.
  *
@@ -277,7 +283,7 @@ rte_rcu_qsbr_thread_online(struct rte_rcu_qsbr *v, unsigned int thread_id)
  * This can be called during initialization or as part of the packet
  * processing loop.
  *
- * The reader thread must call rte_rcu_thread_offline API, before
+ * The reader thread must call rte_rcu_qsbr_thread_offline API, before
  * calling any functions that block, to ensure that rte_rcu_qsbr_check
  * API does not wait indefinitely for the reader thread to update its QS.
  *
@@ -450,12 +456,14 @@ rte_rcu_qsbr_quiescent(struct rte_rcu_qsbr *v, unsigned int thread_id)
         */
        t = __atomic_load_n(&v->token, __ATOMIC_ACQUIRE);
 
-       /* Inform the writer that updates are visible to this reader.
+       /* Check if there are updates available from the writer.
+        * Inform the writer that updates are visible to this reader.
         * Prior loads of the shared data structure should not move
         * beyond this store. Hence use store-release.
         */
-       __atomic_store_n(&v->qsbr_cnt[thread_id].cnt,
-                        t, __ATOMIC_RELEASE);
+       if (t != __atomic_load_n(&v->qsbr_cnt[thread_id].cnt, __ATOMIC_RELAXED))
+               __atomic_store_n(&v->qsbr_cnt[thread_id].cnt,
+                                        t, __ATOMIC_RELEASE);
 
        __RTE_RCU_DP_LOG(DEBUG, "%s: update: token = %"PRIu64", Thread ID = %d",
                __func__, t, thread_id);
@@ -471,6 +479,7 @@ __rte_rcu_qsbr_check_selective(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
        uint64_t bmap;
        uint64_t c;
        uint64_t *reg_thread_id;
+       uint64_t acked_token = __RTE_QSBR_CNT_MAX;
 
        for (i = 0, reg_thread_id = __RTE_QSBR_THRID_ARRAY_ELM(v, 0);
                i < v->num_elems;
@@ -492,6 +501,7 @@ __rte_rcu_qsbr_check_selective(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
                        __RTE_RCU_DP_LOG(DEBUG,
                                "%s: status: token = %"PRIu64", wait = %d, Thread QS cnt = %"PRIu64", Thread ID = %d",
                                __func__, t, wait, c, id+j);
+
                        /* Counter is not checked for wrap-around condition
                         * as it is a 64b counter.
                         */
@@ -511,10 +521,25 @@ __rte_rcu_qsbr_check_selective(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
                                continue;
                        }
 
+                       /* This thread is in quiescent state. Use the counter
+                        * to find the least acknowledged token among all the
+                        * readers.
+                        */
+                       if (c != __RTE_QSBR_CNT_THR_OFFLINE && acked_token > c)
+                               acked_token = c;
+
                        bmap &= ~(1UL << j);
                }
        }
 
+       /* All readers are checked, update least acknowledged token.
+        * There might be multiple writers trying to update this. There is
+        * no need to update this very accurately using compare-and-swap.
+        */
+       if (acked_token != __RTE_QSBR_CNT_MAX)
+               __atomic_store_n(&v->acked_token, acked_token,
+                       __ATOMIC_RELAXED);
+
        return 1;
 }
 
@@ -527,6 +552,7 @@ __rte_rcu_qsbr_check_all(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
        uint32_t i;
        struct rte_rcu_qsbr_cnt *cnt;
        uint64_t c;
+       uint64_t acked_token = __RTE_QSBR_CNT_MAX;
 
        for (i = 0, cnt = v->qsbr_cnt; i < v->max_threads; i++, cnt++) {
                __RTE_RCU_DP_LOG(DEBUG,
@@ -537,6 +563,7 @@ __rte_rcu_qsbr_check_all(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
                        __RTE_RCU_DP_LOG(DEBUG,
                                "%s: status: token = %"PRIu64", wait = %d, Thread QS cnt = %"PRIu64", Thread ID = %d",
                                __func__, t, wait, c, i);
+
                        /* Counter is not checked for wrap-around condition
                         * as it is a 64b counter.
                         */
@@ -549,8 +576,22 @@ __rte_rcu_qsbr_check_all(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
 
                        rte_pause();
                }
+
+               /* This thread is in quiescent state. Use the counter to find
+                * the least acknowledged token among all the readers.
+                */
+               if (likely(c != __RTE_QSBR_CNT_THR_OFFLINE && acked_token > c))
+                       acked_token = c;
        }
 
+       /* All readers are checked, update least acknowledged token.
+        * There might be multiple writers trying to update this. There is
+        * no need to update this very accurately using compare-and-swap.
+        */
+       if (acked_token != __RTE_QSBR_CNT_MAX)
+               __atomic_store_n(&v->acked_token, acked_token,
+                       __ATOMIC_RELAXED);
+
        return 1;
 }
 
@@ -594,6 +635,10 @@ rte_rcu_qsbr_check(struct rte_rcu_qsbr *v, uint64_t t, bool wait)
 {
        RTE_ASSERT(v != NULL);
 
+       /* Check if all the readers have already acknowledged this token */
+       if (likely(t <= v->acked_token))
+               return 1;
+
        if (likely(v->num_threads == v->max_threads))
                return __rte_rcu_qsbr_check_all(v, t, wait);
        else