ovs-rcu: Add ovsrcu_barrier.

ovsrcu_barrier will block the current thread until all the postponed
rcu job has been finished. it's like a OVS version of the Linux
kernel rcu_barrier().

Signed-off-by: Peng He <hepeng.0320@bytedance.com>
Co-authored-by: Eelco Chaudron <echaudro@redhat.com>
Signed-off-by: Eelco Chaudron <echaudro@redhat.com>
Reviewed-by: David Marchand <david.marchand@redhat.com>
Acked-by: Eelco Chaudron <echaudro@redhat.com>
Acked-by: Aaron Conole <aconole@redhat.com>
Signed-off-by: Ilya Maximets <i.maximets@ovn.org>

lib/ovs-rcu.c

@@ -444,3 +444,40 @@ ovsrcu_init_module(void)
         ovsthread_once_done(&once);
     }
 }
+
+static void
+ovsrcu_barrier_func(void *seq_)
+{
+    struct seq *seq = (struct seq *) seq_;
+    seq_change(seq);
+}
+
+/* Similar to the kernel rcu_barrier, ovsrcu_barrier waits for all outstanding
+ * RCU callbacks to complete. However, unlike the kernel rcu_barrier, which
+ * might return immediately if there are no outstanding RCU callbacks,
+ * this API will at least wait for a grace period.
+ *
+ * Another issue the caller might need to know is that the barrier is just
+ * for "one-shot", i.e. if inside some RCU callbacks, another RCU callback is
+ * registered, this API only guarantees the first round of RCU callbacks have
+ * been executed after it returns.
+ */
+void
+ovsrcu_barrier(void)
+{
+    struct seq *seq = seq_create();
+    /* First let all threads flush their cbsets. */
+    ovsrcu_synchronize();
+
+    /* Then register a new cbset, ensure this cbset
+     * is at the tail of the global list. */
+    uint64_t seqno = seq_read(seq);
+    ovsrcu_postpone__(ovsrcu_barrier_func, (void *) seq);
+
+    do {
+        seq_wait(seq, seqno);
+        poll_block();
+    } while (seqno == seq_read(seq));
+
+    seq_destroy(seq);
+}

lib/ovs-rcu.h

@@ -155,6 +155,19 @@
  *         port_delete(id);
  *     }
  *
+ * Use ovsrcu_barrier() to wait for all the outstanding RCU callbacks to
+ * finish. This is useful when you have to destroy some resources however
+ * these resources are referenced in the outstanding RCU callbacks.
+ *
+ *     void rcu_cb(void *A) {
+ *         do_something(A);
+ *     }
+ *
+ *     void destroy_A() {
+ *         ovsrcu_postpone(rcu_cb, A); // will use A later
+ *         ovsrcu_barrier(); // wait for rcu_cb done
+ *         do_destroy_A(); // free A
+ *     }
  */
 
 #include "compiler.h"
@@ -310,4 +323,6 @@ void ovsrcu_synchronize(void);
 
 void ovsrcu_exit(void);
 
+void ovsrcu_barrier(void);
+
 #endif /* ovs-rcu.h */

tests/library.at

@@ -252,7 +252,7 @@ AT_CHECK([ovstest test-barrier], [0], [])
 AT_CLEANUP
 
 AT_SETUP([rcu])
-AT_CHECK([ovstest test-rcu-quiesce], [0], [])
+AT_CHECK([ovstest test-rcu], [0], [])
 AT_CLEANUP
 
 AT_SETUP([stopwatch module])

tests/test-rcu.c

@@ -35,7 +35,7 @@ quiescer_main(void *aux OVS_UNUSED)
 }
 
 static void
-test_rcu_quiesce(int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
+test_rcu_quiesce(void)
 {
     pthread_t quiescer;
 
@@ -48,4 +48,29 @@ test_rcu_quiesce(int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
     xpthread_join(quiescer, NULL);
 }
 
-OVSTEST_REGISTER("test-rcu-quiesce", test_rcu_quiesce);
+static void
+add_count(void *_count)
+{
+    unsigned *count = (unsigned *)_count;
+    (*count) ++;
+}
+
+static void
+test_rcu_barrier(void)
+{
+    unsigned count = 0;
+    for (int i = 0; i < 10; i ++) {
+        ovsrcu_postpone(add_count, &count);
+    }
+
+    ovsrcu_barrier();
+    ovs_assert(count == 10);
+}
+
+static void
+test_rcu(int argc OVS_UNUSED, char *argv[] OVS_UNUSED) {
+    test_rcu_quiesce();
+    test_rcu_barrier();
+}
+
+OVSTEST_REGISTER("test-rcu", test_rcu);