RCU lock implementation

Introduce an RCU lock implementation as an alternative locking mechanism
to openssl.  The api is documented in the ossl_rcu.pod
file

Read side implementaiton is comparable to that of RWLOCKS:
ossl_rcu_read_lock(lock);
<
critical section in which data can be accessed via
ossl_derefrence
>
ossl_rcu_read_unlock(lock);

Write side implementation is:
ossl_rcu_write_lock(lock);
<
critical section in which data can be updated via
ossl_assign_pointer
and stale data can optionally be scheduled for removal
via ossl_rcu_call
>
ossl_rcu_write_unlock(lock);
...
ossl_synchronize_rcu(lock);

ossl_rcu_call fixup

Reviewed-by: Hugo Landau <hlandau@openssl.org>
Reviewed-by: Matt Caswell <matt@openssl.org>
(Merged from https://github.com/openssl/openssl/pull/22729)

1 files changed, 358 insertions, 2 deletions

diff --git a/crypto/threads_win.c b/crypto/threads_win.c
index 4cdc62339d..fd51c735f8 100644
--- a/crypto/threads_win.c
+++ b/crypto/threads_win.c
@@ -13,6 +13,7 @@
 #  define USE_RWLOCK
 # endif
 #endif
+#include <assert.h>
 
 /*
  * VC++ 2008 or earlier x86 compilers do not have an inline implementation
@@ -27,6 +28,11 @@
 #endif
 
 #include <openssl/crypto.h>
+#include <crypto/cryptlib.h>
+#include "internal/common.h"
+#include "internal/thread_arch.h"
+#include "internal/rcu.h"
+#include "rcu_internal.h"
 
 #if defined(OPENSSL_THREADS) && !defined(CRYPTO_TDEBUG) && defined(OPENSSL_SYS_WINDOWS)
 
@@ -37,20 +43,370 @@ typedef struct {
 } CRYPTO_win_rwlock;
 # endif
 
+static CRYPTO_THREAD_LOCAL rcu_thr_key;
+
+# define READER_SHIFT 0
+# define ID_SHIFT 32 
+# define READER_SIZE 32 
+# define ID_SIZE 32 
+
+# define READER_MASK     (((LONG64)1 << READER_SIZE)-1)
+# define ID_MASK         (((LONG64)1 << ID_SIZE)-1)
+# define READER_COUNT(x) (((LONG64)(x) >> READER_SHIFT) & READER_MASK)
+# define ID_VAL(x)       (((LONG64)(x) >> ID_SHIFT) & ID_MASK)
+# define VAL_READER      ((LONG64)1 << READER_SHIFT)
+# define VAL_ID(x)       ((LONG64)x << ID_SHIFT)
+
+/*
+ * This defines a quescent point (qp)
+ * This is the barrier beyond which a writer
+ * must wait before freeing data that was
+ * atomically updated
+ */
+struct rcu_qp {
+    volatile LONG64 users;
+};
+
+struct thread_qp {
+    struct rcu_qp *qp;
+    unsigned int depth;
+    CRYPTO_RCU_LOCK *lock;
+};
+
+#define MAX_QPS 10
+/*
+ * This is the per thread tracking data
+ * that is assigned to each thread participating
+ * in an rcu qp
+ *
+ * qp points to the qp that it last acquired
+ *
+ */
+struct rcu_thr_data {
+    struct thread_qp thread_qps[MAX_QPS];
+};
+
+/*
+ * This is the internal version of a CRYPTO_RCU_LOCK
+ * it is cast from CRYPTO_RCU_LOCK
+ */
+struct rcu_lock_st {
+    struct rcu_cb_item *cb_items;
+    uint32_t id_ctr;
+    struct rcu_qp *qp_group;
+    size_t group_count;
+    uint32_t next_to_retire;
+    volatile long int reader_idx;
+    uint32_t current_alloc_idx;
+    uint32_t writers_alloced;
+    CRYPTO_MUTEX *write_lock;
+    CRYPTO_MUTEX *alloc_lock;
+    CRYPTO_CONDVAR *alloc_signal;
+    CRYPTO_MUTEX *prior_lock;
+    CRYPTO_CONDVAR *prior_signal;
+};
+
+/*
+ * Called on thread exit to free the pthread key
+ * associated with this thread, if any
+ */
+static void free_rcu_thr_data(void *ptr)
+{
+    struct rcu_thr_data *data =
+                        (struct rcu_thr_data *)CRYPTO_THREAD_get_local(&rcu_thr_key);
+
+    OPENSSL_free(data);
+    CRYPTO_THREAD_set_local(&rcu_thr_key, NULL);
+}
+
+
+static void ossl_rcu_init(void)
+{
+    CRYPTO_THREAD_init_local(&rcu_thr_key, NULL);
+    ossl_init_thread_start(NULL, NULL, free_rcu_thr_data);
+}
+
+static struct rcu_qp *allocate_new_qp_group(struct rcu_lock_st *lock,
+                                            int count)
+{
+    struct rcu_qp *new =
+        OPENSSL_zalloc(sizeof(*new) * count);
+
+    lock->group_count = count;
+    return new;
+}
+
+static CRYPTO_ONCE rcu_init_once = CRYPTO_ONCE_STATIC_INIT;
+
+CRYPTO_RCU_LOCK *ossl_rcu_lock_new(int num_writers)
+{
+    struct rcu_lock_st *new;
+
+    if (!CRYPTO_THREAD_run_once(&rcu_init_once, ossl_rcu_init))
+        return NULL;
+
+    if (num_writers < 1)
+        num_writers = 1;
+
+    new = OPENSSL_zalloc(sizeof(*new));
+
+    if (new == NULL)
+        return NULL;
+
+    new->write_lock = ossl_crypto_mutex_new();
+    new->alloc_signal = ossl_crypto_condvar_new();
+    new->prior_signal = ossl_crypto_condvar_new();
+    new->alloc_lock = ossl_crypto_mutex_new();
+    new->prior_lock = ossl_crypto_mutex_new();
+    new->write_lock = ossl_crypto_mutex_new();
+    new->qp_group = allocate_new_qp_group(new, num_writers + 1);
+    if (new->qp_group == NULL
+        || new->alloc_signal == NULL
+        || new->prior_signal == NULL
+        || new->write_lock == NULL
+        || new->alloc_lock == NULL
+        || new->prior_lock == NULL) {
+        OPENSSL_free(new->qp_group);
+        ossl_crypto_condvar_free(&new->alloc_signal);
+        ossl_crypto_condvar_free(&new->prior_signal);
+        ossl_crypto_mutex_free(&new->alloc_lock);
+        ossl_crypto_mutex_free(&new->prior_lock);
+        ossl_crypto_mutex_free(&new->write_lock);
+        OPENSSL_free(new);
+        new = NULL;
+    }
+    return new;
+
+}
+
+void ossl_rcu_lock_free(CRYPTO_RCU_LOCK *lock)
+{
+    OPENSSL_free(lock->qp_group);
+    ossl_crypto_condvar_free(&lock->alloc_signal);
+    ossl_crypto_condvar_free(&lock->prior_signal);
+    ossl_crypto_mutex_free(&lock->alloc_lock);
+    ossl_crypto_mutex_free(&lock->prior_lock);
+    ossl_crypto_mutex_free(&lock->write_lock);
+    OPENSSL_free(lock);
+}
+
+static inline struct rcu_qp *get_hold_current_qp(CRYPTO_RCU_LOCK *lock)
+{
+    uint32_t qp_idx;
+
+    /* get the current qp index */
+    for (;;) {
+        qp_idx = InterlockedOr(&lock->reader_idx, 0);
+        InterlockedAdd64(&lock->qp_group[qp_idx].users, VAL_READER);
+        if (qp_idx == InterlockedOr(&lock->reader_idx, 0))
+            break;
+        InterlockedAdd64(&lock->qp_group[qp_idx].users, -VAL_READER);
+    }
+
+    return &lock->qp_group[qp_idx];
+}
+
+void ossl_rcu_read_lock(CRYPTO_RCU_LOCK *lock)
+{
+    struct rcu_thr_data *data;
+    int i;
+    int available_qp = -1;
+
+    /*
+     * we're going to access current_qp here so ask the
+     * processor to fetch it
+     */
+    data = CRYPTO_THREAD_get_local(&rcu_thr_key);
+
+    if (data == NULL) {
+        data = OPENSSL_zalloc(sizeof(*data));
+        OPENSSL_assert(data != NULL);
+        CRYPTO_THREAD_set_local(&rcu_thr_key, data);
+    }
+
+    for (i = 0; i < MAX_QPS; i++) {
+        if (data->thread_qps[i].qp == NULL && available_qp == -1)
+            available_qp = i;
+        /* If we have a hold on this lock already, we're good */
+        if (data->thread_qps[i].lock == lock)
+            return;
+    }
+
+    /*
+     * if we get here, then we don't have a hold on this lock yet
+     */
+    assert(available_qp != -1);
+
+    data->thread_qps[available_qp].qp = get_hold_current_qp(lock);
+    data->thread_qps[available_qp].depth = 1;
+    data->thread_qps[available_qp].lock = lock;
+}
+
+void ossl_rcu_write_lock(CRYPTO_RCU_LOCK *lock)
+{
+    ossl_crypto_mutex_lock(lock->write_lock);
+}
+
+void ossl_rcu_write_unlock(CRYPTO_RCU_LOCK *lock)
+{
+    ossl_crypto_mutex_unlock(lock->write_lock);
+}
+
+void ossl_rcu_read_unlock(CRYPTO_RCU_LOCK *lock)
+{
+    struct rcu_thr_data *data = CRYPTO_THREAD_get_local(&rcu_thr_key);
+    int i;
+    LONG64 ret;
+
+    assert(data != NULL);
+
+    for (i = 0; i < MAX_QPS; i++) {
+        if (data->thread_qps[i].lock == lock) {
+            data->thread_qps[i].depth--;
+            if (data->thread_qps[i].depth == 0) {
+                ret = InterlockedAdd64(&data->thread_qps[i].qp->users, -VAL_READER);
+                OPENSSL_assert(ret >= 0);
+                data->thread_qps[i].qp = NULL;
+                data->thread_qps[i].lock = NULL;
+            }
+            return;
+        }
+    }
+}
+
+static struct rcu_qp *update_qp(CRYPTO_RCU_LOCK *lock)
+{
+    uint64_t new_id;
+    uint32_t current_idx;
+    uint32_t tmp;
+
+    ossl_crypto_mutex_lock(lock->alloc_lock);
+    /*
+     * we need at least one qp to be available with one
+     * left over, so that readers can start working on
+     * one that isn't yet being waited on
+     */
+    while (lock->group_count - lock->writers_alloced < 2)
+        ossl_crypto_condvar_wait(lock->alloc_signal, lock->alloc_lock);
+
+    current_idx = lock->current_alloc_idx;
+    /* Allocate the qp */
+    lock->writers_alloced++;
+
+    /* increment the allocation index */
+    lock->current_alloc_idx =
+        (lock->current_alloc_idx + 1) % lock->group_count;
+
+    /* get and insert a new id */
+    new_id = lock->id_ctr;
+    lock->id_ctr++;
+
+    new_id = VAL_ID(new_id);
+    InterlockedAnd64(&lock->qp_group[current_idx].users, ID_MASK);
+    InterlockedAdd64(&lock->qp_group[current_idx].users, new_id);
+
+    /* update the reader index to be the prior qp */
+    tmp = lock->current_alloc_idx;
+    InterlockedExchange(&lock->reader_idx, tmp);
+
+    /* wake up any waiters */
+    ossl_crypto_condvar_broadcast(lock->alloc_signal);
+    ossl_crypto_mutex_unlock(lock->alloc_lock);
+    return &lock->qp_group[current_idx];
+}
+
+static void retire_qp(CRYPTO_RCU_LOCK *lock,
+                      struct rcu_qp *qp)
+{
+    ossl_crypto_mutex_lock(lock->alloc_lock);
+    lock->writers_alloced--;
+    ossl_crypto_condvar_broadcast(lock->alloc_signal);
+    ossl_crypto_mutex_unlock(lock->alloc_lock);
+}
+
+
+void ossl_synchronize_rcu(CRYPTO_RCU_LOCK *lock)
+{
+    struct rcu_qp *qp;
+    uint64_t count;
+    struct rcu_cb_item *cb_items, *tmpcb;
+
+    /* before we do anything else, lets grab the cb list */
+    cb_items = InterlockedExchangePointer((void * volatile *)&lock->cb_items, NULL);
+
+    qp = update_qp(lock);
+
+    /* wait for the reader count to reach zero */
+    do {
+        count = InterlockedOr64(&qp->users, 0);
+    } while (READER_COUNT(count) != 0);
+
+    /* retire in order */
+    ossl_crypto_mutex_lock(lock->prior_lock);
+    while (lock->next_to_retire != ID_VAL(count))
+        ossl_crypto_condvar_wait(lock->prior_signal, lock->prior_lock);
+
+    lock->next_to_retire++;
+    ossl_crypto_condvar_broadcast(lock->prior_signal);
+    ossl_crypto_mutex_unlock(lock->prior_lock);
+
+    retire_qp(lock, qp);
+
+    /* handle any callbacks that we have */
+    while (cb_items != NULL) {
+        tmpcb = cb_items;
+        cb_items = cb_items->next;
+        tmpcb->fn(tmpcb->data);
+        OPENSSL_free(tmpcb);
+    }
+
+    /* and we're done */
+    return;
+
+}
+
+int ossl_rcu_call(CRYPTO_RCU_LOCK *lock, rcu_cb_fn cb, void *data)
+{
+    struct rcu_cb_item *new;
+    struct rcu_cb_item *prev;
+
+    new = OPENSSL_zalloc(sizeof(struct rcu_cb_item));
+    if (new == NULL)
+        return 0;
+    prev = new;
+    new->data = data;
+    new->fn = cb;
+
+    InterlockedExchangePointer((void * volatile *)&lock->cb_items, prev);
+    new->next = prev;
+    return 1;
+}
+
+void *ossl_rcu_uptr_deref(void **p)
+{
+    return (void *)*p;
+}
+
+void ossl_rcu_assign_uptr(void **p, void **v)
+{
+    InterlockedExchangePointer((void * volatile *)p, (void *)*v);
+}
+
+
 CRYPTO_RWLOCK *CRYPTO_THREAD_lock_new(void)
 {
     CRYPTO_RWLOCK *lock;
 # ifdef USE_RWLOCK
     CRYPTO_win_rwlock *rwlock;
 
-    if ((lock = CRYPTO_zalloc(sizeof(CRYPTO_win_rwlock), NULL, 0)) == NULL)
+    if ((lock = OPENSSL_zalloc(sizeof(CRYPTO_win_rwlock))) == NULL)
         /* Don't set error, to avoid recursion blowup. */
         return NULL;
     rwlock = lock;
     InitializeSRWLock(&rwlock->lock);
 # else
 
-    if ((lock = CRYPTO_zalloc(sizeof(CRITICAL_SECTION), NULL, 0)) == NULL)
+    if ((lock = OPENSSL_zalloc(sizeof(CRITICAL_SECTION))) == NULL)
         /* Don't set error, to avoid recursion blowup. */
         return NULL;