+ uint64_t qp_idx;
+
+ /* get the current qp index */
+ for (;;) {
+ /*
+ * Notes on use of __ATOMIC_ACQUIRE
+ * We need to ensure the following:
+ * 1) That subsequent operations aren't optimized by hoisting them above
+ * this operation. Specifically, we don't want the below re-load of
+ * qp_idx to get optimized away
+ * 2) We want to ensure that any updating of reader_idx on the write side
+ * of the lock is flushed from a local cpu cache so that we see any
+ * updates prior to the load. This is a non-issue on cache coherent
+ * systems like x86, but is relevant on other arches
+ * Note: This applies to the reload below as well
+ */
+ qp_idx = ATOMIC_LOAD_N(uint64_t, &lock->reader_idx, __ATOMIC_ACQUIRE);
+
+ /*
+ * Notes of use of __ATOMIC_RELEASE
+ * This counter is only read by the write side of the lock, and so we
+ * specify __ATOMIC_RELEASE here to ensure that the write side of the
+ * lock see this during the spin loop read of users, as it waits for the
+ * reader count to approach zero
+ */
+ ATOMIC_ADD_FETCH(&lock->qp_group[qp_idx].users, VAL_READER,
+ __ATOMIC_RELEASE);
+
+ /* if the idx hasn't changed, we're good, else try again */
+ if (qp_idx == ATOMIC_LOAD_N(uint64_t, &lock->reader_idx, __ATOMIC_ACQUIRE))
+ break;
+
+ /*
+ * Notes on use of __ATOMIC_RELEASE
+ * As with the add above, we want to ensure that this decrement is
+ * seen by the write side of the lock as soon as it happens to prevent
+ * undue spinning waiting for write side completion
+ */
+ ATOMIC_SUB_FETCH(&lock->qp_group[qp_idx].users, VAL_READER,
+ __ATOMIC_RELEASE);
+ }
+
+ return &lock->qp_group[qp_idx];
+}
+
+static void ossl_rcu_free_local_data(void *arg)
+{
+ OSSL_LIB_CTX *ctx = arg;
+ CRYPTO_THREAD_LOCAL *lkey = ossl_lib_ctx_get_rcukey(ctx);
+ struct rcu_thr_data *data = CRYPTO_THREAD_get_local(lkey);
+ OPENSSL_free(data);
+}
+
+void ossl_rcu_read_lock(CRYPTO_RCU_LOCK *lock)
+{
+ struct rcu_thr_data *data;
+ int i, available_qp = -1;
+ CRYPTO_THREAD_LOCAL *lkey = ossl_lib_ctx_get_rcukey(lock->ctx);
+
+ /*
+ * we're going to access current_qp here so ask the
+ * processor to fetch it
+ */
+ data = CRYPTO_THREAD_get_local(lkey);
+
+ if (data == NULL) {
+ data = OPENSSL_zalloc(sizeof(*data));
+ OPENSSL_assert(data != NULL);
+ CRYPTO_THREAD_set_local(lkey, data);
+ ossl_init_thread_start(NULL, lock->ctx, ossl_rcu_free_local_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) {
+ data->thread_qps[i].depth++;
+ 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_read_unlock(CRYPTO_RCU_LOCK *lock)
+{
+ int i;
+ CRYPTO_THREAD_LOCAL *lkey = ossl_lib_ctx_get_rcukey(lock->ctx);
+ struct rcu_thr_data *data = CRYPTO_THREAD_get_local(lkey);
+ uint64_t ret;
+
+ assert(data != NULL);
+
+ for (i = 0; i < MAX_QPS; i++) {
+ if (data->thread_qps[i].lock == lock) {
+ /*
+ * As with read side acquisition, we use __ATOMIC_RELEASE here
+ * to ensure that the decrement is published immediately
+ * to any write side waiters
+ */
+ data->thread_qps[i].depth--;
+ if (data->thread_qps[i].depth == 0) {
+ ret = ATOMIC_SUB_FETCH(&data->thread_qps[i].qp->users, VAL_READER,
+ __ATOMIC_RELEASE);
+ OPENSSL_assert(ret != UINT64_MAX);
+ data->thread_qps[i].qp = NULL;
+ data->thread_qps[i].lock = NULL;
+ }
+ return;
+ }
+ }
+ /*
+ * If we get here, we're trying to unlock a lock that we never acquired -
+ * that's fatal.
+ */
+ assert(0);
+}
+
+/*
+ * Write side allocation routine to get the current qp
+ * and replace it with a new one
+ */
+static struct rcu_qp *update_qp(CRYPTO_RCU_LOCK *lock)
+{
+ uint64_t new_id;
+ uint64_t current_idx;
+
+ pthread_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)
+ /* we have to wait for one to be free */
+ pthread_cond_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);
+ /*
+ * Even though we are under a write side lock here
+ * We need to use atomic instructions to ensure that the results
+ * of this update are published to the read side prior to updating the
+ * reader idx below
+ */
+ ATOMIC_AND_FETCH(&lock->qp_group[current_idx].users, ID_MASK,
+ __ATOMIC_RELEASE);
+ ATOMIC_OR_FETCH(&lock->qp_group[current_idx].users, new_id,
+ __ATOMIC_RELEASE);
+
+ /*
+ * Update the reader index to be the prior qp.
+ * Note the use of __ATOMIC_RELEASE here is based on the corresponding use
+ * of __ATOMIC_ACQUIRE in get_hold_current_qp, as we want any publication
+ * of this value to be seen on the read side immediately after it happens
+ */
+ ATOMIC_STORE_N(uint64_t, &lock->reader_idx, lock->current_alloc_idx,
+ __ATOMIC_RELEASE);
+
+ /* wake up any waiters */
+ pthread_cond_signal(&lock->alloc_signal);
+ pthread_mutex_unlock(&lock->alloc_lock);
+ return &lock->qp_group[current_idx];
+}
+
+static void retire_qp(CRYPTO_RCU_LOCK *lock, struct rcu_qp *qp)
+{
+ pthread_mutex_lock(&lock->alloc_lock);
+ lock->writers_alloced--;
+ pthread_cond_signal(&lock->alloc_signal);
+ pthread_mutex_unlock(&lock->alloc_lock);
+}
+
+static struct rcu_qp *allocate_new_qp_group(CRYPTO_RCU_LOCK *lock,
+ int count)
+{
+ struct rcu_qp *new =
+ OPENSSL_zalloc(sizeof(*new) * count);
+
+ lock->group_count = count;
+ return new;
+}
+
+void ossl_rcu_write_lock(CRYPTO_RCU_LOCK *lock)
+{
+ pthread_mutex_lock(&lock->write_lock);
+ TSAN_FAKE_UNLOCK(&lock->write_lock);
+}
+
+void ossl_rcu_write_unlock(CRYPTO_RCU_LOCK *lock)
+{
+ TSAN_FAKE_LOCK(&lock->write_lock);
+ pthread_mutex_unlock(&lock->write_lock);
+}
+
+void ossl_synchronize_rcu(CRYPTO_RCU_LOCK *lock)
+{
+ struct rcu_qp *qp;
+ uint64_t count;
+ struct rcu_cb_item *cb_items, *tmpcb;
+
+ pthread_mutex_lock(&lock->write_lock);
+ cb_items = lock->cb_items;
+ lock->cb_items = NULL;
+ pthread_mutex_unlock(&lock->write_lock);
+
+ qp = update_qp(lock);
+
+ /*
+ * wait for the reader count to reach zero
+ * Note the use of __ATOMIC_ACQUIRE here to ensure that any
+ * prior __ATOMIC_RELEASE write operation in get_hold_current_qp
+ * is visible prior to our read
+ */
+ do {
+ count = ATOMIC_LOAD_N(uint64_t, &qp->users, __ATOMIC_ACQUIRE);
+ } while (READER_COUNT(count) != 0);
+
+ /* retire in order */
+ pthread_mutex_lock(&lock->prior_lock);
+ while (lock->next_to_retire != ID_VAL(count))
+ pthread_cond_wait(&lock->prior_signal, &lock->prior_lock);
+ lock->next_to_retire++;
+ pthread_cond_broadcast(&lock->prior_signal);
+ pthread_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);
+ }
+}
+
+int ossl_rcu_call(CRYPTO_RCU_LOCK *lock, rcu_cb_fn cb, void *data)
+{
+ struct rcu_cb_item *new =
+ OPENSSL_zalloc(sizeof(*new));
+
+ if (new == NULL)
+ return 0;
+
+ new->data = data;
+ new->fn = cb;
+ /*
+ * Use __ATOMIC_ACQ_REL here to indicate that any prior writes to this
+ * list are visible to us prior to reading, and publish the new value
+ * immediately
+ */
+ new->next = ATOMIC_EXCHANGE_N(prcu_cb_item, &lock->cb_items, new,
+ __ATOMIC_ACQ_REL);
+
+ return 1;
+}
+
+void *ossl_rcu_uptr_deref(void **p)
+{
+ return ATOMIC_LOAD_N(pvoid, p, __ATOMIC_ACQUIRE);
+}
+
+void ossl_rcu_assign_uptr(void **p, void **v)
+{
+ ATOMIC_STORE(pvoid, p, v, __ATOMIC_RELEASE);
+}
+
+CRYPTO_RCU_LOCK *ossl_rcu_lock_new(int num_writers, OSSL_LIB_CTX *ctx)
+{
+ struct rcu_lock_st *new;
+
+ if (num_writers < 1)
+ num_writers = 1;
+
+ ctx = ossl_lib_ctx_get_concrete(ctx);
+ if (ctx == NULL)
+ return 0;
+
+ new = OPENSSL_zalloc(sizeof(*new));
+ if (new == NULL)
+ return NULL;
+
+ new->ctx = ctx;
+ pthread_mutex_init(&new->write_lock, NULL);
+ pthread_mutex_init(&new->prior_lock, NULL);
+ pthread_mutex_init(&new->alloc_lock, NULL);
+ pthread_cond_init(&new->prior_signal, NULL);
+ pthread_cond_init(&new->alloc_signal, NULL);
+ new->qp_group = allocate_new_qp_group(new, num_writers + 1);
+ if (new->qp_group == NULL) {
+ OPENSSL_free(new);
+ new = NULL;
+ }
+ return new;
+}
+
+void ossl_rcu_lock_free(CRYPTO_RCU_LOCK *lock)
+{
+ struct rcu_lock_st *rlock = (struct rcu_lock_st *)lock;
+