pvector: Use acquire-release semantics for size.

Read/write concurrency of pvector library is implemented by a temp vector
and RCU protection. Considering performance reason, insertion does not
follow this scheme.
In insertion function, a thread fence ensures size increment is done
after new entry is stored. But there is no barrier in the iteration
fuction(pvector_cursor_init). Entry point access may be reordered before
loading vector size, so the invalid entry point may be loaded when vector
iteration.
This patch fixes it by acquire-release pair. It can guarantee new size is
observed by reader after new entry stored by writer. And this is
implemented by one-way barrier instead of two-way memory fence.

Fixes: fe7cfa5c3f19 ("lib/pvector: Non-intrusive RCU priority vector.")
Reviewed-by: Gavin Hu <Gavin.Hu@arm.com>
Reviewed-by: Lijian Zhang <Lijian.Zhang@arm.com>
Signed-off-by: Yanqin Wei <Yanqin.Wei@arm.com>
Signed-off-by: Ilya Maximets <i.maximets@ovn.org>

2 files changed, 20 insertions, 11 deletions

diff --git a/lib/pvector.c b/lib/pvector.c
index aaeee9214..cc527fdc4 100644
--- a/lib/pvector.c
+++ b/lib/pvector.c
@@ -33,7 +33,7 @@ pvector_impl_alloc(size_t size)
     struct pvector_impl *impl;
 
     impl = xmalloc(sizeof *impl + size * sizeof impl->vector[0]);
-    impl->size = 0;
+    atomic_init(&impl->size, 0);
     impl->allocated = size;
 
     return impl;
@@ -117,18 +117,22 @@ pvector_insert(struct pvector *pvec, void *ptr, int priority)
 {
     struct pvector_impl *temp = pvec->temp;
     struct pvector_impl *old = pvector_impl_get(pvec);
+    size_t size;
 
     ovs_assert(ptr != NULL);
 
+    /* There is no possible concurrent writer. Insertions must be protected
+     * by mutex or be always excuted from the same thread. */
+    atomic_read_relaxed(&old->size, &size);
+
     /* Check if can add to the end without reallocation. */
-    if (!temp && old->allocated > old->size &&
-        (!old->size || priority <= old->vector[old->size - 1].priority)) {
-        old->vector[old->size].ptr = ptr;
-        old->vector[old->size].priority = priority;
+    if (!temp && old->allocated > size &&
+        (!size || priority <= old->vector[size - 1].priority)) {
+        old->vector[size].ptr = ptr;
+        old->vector[size].priority = priority;
         /* Size increment must not be visible to the readers before the new
          * entry is stored. */
-        atomic_thread_fence(memory_order_release);
-        ++old->size;
+        atomic_store_explicit(&old->size, size + 1, memory_order_release);
     } else {
         if (!temp) {
             temp = pvector_impl_dup(old);
diff --git a/lib/pvector.h b/lib/pvector.h
index b175b213d..3d491ee25 100644
--- a/lib/pvector.h
+++ b/lib/pvector.h
@@ -62,8 +62,8 @@ struct pvector_entry {
 };
 
 struct pvector_impl {
-    size_t size;       /* Number of entries in the vector. */
-    size_t allocated;  /* Number of allocated entries. */
+    atomic_size_t size;   /* Number of entries in the vector. */
+    size_t allocated;     /* Number of allocated entries. */
     struct pvector_entry vector[];
 };
 
@@ -174,12 +174,17 @@ pvector_cursor_init(const struct pvector *pvec,
 {
     const struct pvector_impl *impl;
     struct pvector_cursor cursor;
+    size_t size;
 
     impl = ovsrcu_get(struct pvector_impl *, &pvec->impl);
 
-    ovs_prefetch_range(impl->vector, impl->size * sizeof impl->vector[0]);
+    /* Use memory_order_acquire to ensure entry access can not be
+     * reordered to happen before size read. */
+    atomic_read_explicit(&CONST_CAST(struct pvector_impl *, impl)->size,
+                         &size, memory_order_acquire);
+    ovs_prefetch_range(impl->vector, size * sizeof impl->vector[0]);
 
-    cursor.size = impl->size;
+    cursor.size = size;
     cursor.vector = impl->vector;
     cursor.entry_idx = -1;