Re: [RFC PATCH 1/1] hpref: Hazard Pointers with Reference Counter

[Jonas Oberhauser <jonas.oberhauser@huaweicloud.com>]

Hi Mathieu,

interesting idea. Conceptually it looks good.

There's another approach of using hazard pointer to optimize shared 
reference counting (to make it lock-free in common cases).

https://github.com/cmuparlay/concurrent_deferred_rc

It doesn't go as far as what you're doing, but they also use the hazard 
pointer to protect the refcount (like you do in the promote function).

I haven't read your code in detail but it seems to me you have an ABA 
bug: as I explained elsewhere, you could read the same pointer after ABA 
but you don't synchronize with the newer store that gave you node2, 
leaving you to speculatively read stale values through *ctx->hp.
(I am assuming here that ctx->hp is essentially an out parameter used to 
let the caller know which node got protected).

Have fun,
   jonas

> +/*
> + * hpref_hp_get: Obtain a reference to a stable object, protected either
> + *               by hazard pointer (fast-path) or using reference
> + *               counter as fall-back.
> + */
> +static inline
> +bool hpref_hp_get(struct hpref_node **node_p, struct hpref_ctx *ctx)
> +{
> +	int cpu = rseq_current_cpu_raw();
> +	struct hpref_percpu_slots *cpu_slots = rseq_percpu_ptr(hpref_percpu_slots, cpu);
> +	struct hpref_slot *slot = &cpu_slots->slots[cpu_slots->current_slot];
> +	bool use_refcount = false;
> +	struct hpref_node *node, *node2;
> +	unsigned int next_slot;
> +
> +retry:
> +	node = uatomic_load(node_p, CMM_RELAXED);
> +	if (!node)
> +		return false;
> +	/* Use rseq to try setting current slot hp. Store B. */
> +	if (rseq_load_cbne_store__ptr(RSEQ_MO_RELAXED, RSEQ_PERCPU_CPU_ID,
> +				(intptr_t *) &slot->node, (intptr_t) NULL,
> +				(intptr_t) node, cpu)) {
> +		slot = &cpu_slots->slots[HPREF_EMERGENCY_SLOT];
> +		use_refcount = true;
> +		/*
> +		 * This may busy-wait for another reader using the
> +		 * emergency slot to transition to refcount.
> +		 */
> +		caa_cpu_relax();
> +		goto retry;
> +	}
> +	/* Memory ordering: Store B before Load A. */
> +	cmm_smp_mb();
> +	node2 = uatomic_load(node_p, CMM_RELAXED);	/* Load A */
> +	if (node != node2) {
> +		uatomic_store(&slot->node, NULL, CMM_RELAXED);
> +		if (!node2)
> +			return false;
> +		goto retry;
> +	}
> +	ctx->type = HPREF_TYPE_HP;
> +	ctx->hp = node;
                    ^---- here

> +	ctx->slot = slot;
> +	if (use_refcount) {
> +		hpref_promote_hp_to_ref(ctx);
> +		return true;
> +	}
> +	/*
> +	 * Increment current slot (racy increment is OK because it is
> +	 * just a position hint). Skip the emergency slot.
> +	 */
> +	next_slot = uatomic_load(&cpu_slots->current_slot, CMM_RELAXED) + 1;
> +	if (next_slot >= HPREF_EMERGENCY_SLOT)
> +		next_slot = 0;
> +	uatomic_store(&cpu_slots->current_slot, next_slot, CMM_RELAXED);
> +	return true;
> +}
> +
> +static inline
> +void hpref_put(struct hpref_ctx *ctx)
> +{
> +	if (ctx->type == HPREF_TYPE_REF) {
> +		urcu_ref_put(&ctx->hp->refcount, hpref_release);
> +	} else {
> +		/* Release HP. */
> +		uatomic_store(&ctx->slot->node, NULL, CMM_RELEASE);
> +	}
> +	ctx->hp = NULL;
> +}
> +
> +/*
> + * hpref_set_pointer: Store pointer @node to @ptr, with RCU publication
> + *                    guarantees.
> + */
> +static inline
> +void hpref_set_pointer(struct hpref_node **ptr, struct hpref_node *node)
> +{
> +	if (__builtin_constant_p(node) && node == NULL)
> +		uatomic_store(ptr, NULL, CMM_RELAXED);
> +	else
> +		uatomic_store(ptr, node, CMM_RELEASE);
> +}
> +
> +/*
> + * Return the content of the hpref context hazard pointer field.
> + */
> +static inline
> +struct hpref_node *hpref_ctx_pointer(struct hpref_ctx *ctx)
> +{
> +	return ctx->hp;
> +}
> +
> +#ifdef __cplusplus
> +}
> +#endif
> +
> +#endif /* _URCU_HPREF_H */
> diff --git a/src/Makefile.am b/src/Makefile.am
> index b555c818..7312c9f7 100644
> --- a/src/Makefile.am
> +++ b/src/Makefile.am
> @@ -19,7 +19,8 @@ RCULFHASH = rculfhash.c rculfhash-mm-order.c rculfhash-mm-chunk.c \
>   lib_LTLIBRARIES = liburcu-common.la \
>   		liburcu.la liburcu-qsbr.la \
>   		liburcu-mb.la liburcu-bp.la \
> -		liburcu-memb.la liburcu-cds.la
> +		liburcu-memb.la liburcu-cds.la \
> +		liburcu-hpref.la
>   
>   #
>   # liburcu-common contains wait-free queues (needed by call_rcu) as well
> @@ -50,6 +51,9 @@ liburcu_cds_la_SOURCES = rculfqueue.c rculfstack.c lfstack.c \
>   	workqueue.c workqueue.h $(RCULFHASH) $(COMPAT)
>   liburcu_cds_la_LIBADD = liburcu-common.la
>   
> +liburcu_hpref_la_SOURCES = hpref.c
> +liburcu_hpref_la_LIBADD = -lrseq
> +
>   pkgconfigdir = $(libdir)/pkgconfig
>   pkgconfig_DATA = liburcu-cds.pc liburcu.pc liburcu-bp.pc liburcu-qsbr.pc \
>   	liburcu-mb.pc liburcu-memb.pc
> diff --git a/src/hpref.c b/src/hpref.c
> new file mode 100644
> index 00000000..f63530f5
> --- /dev/null
> +++ b/src/hpref.c
> @@ -0,0 +1,78 @@
> +// SPDX-FileCopyrightText: 2024 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
> +//
> +// SPDX-License-Identifier: LGPL-2.1-or-later
> +
> +/*
> + * HPREF: Hazard pointers with reference counters
> + */
> +
> +#define _LGPL_SOURCE
> +#include <urcu/hpref.h>
> +#include <rseq/mempool.h>	/* Per-CPU memory */
> +
> +static struct rseq_mempool *mempool;
> +struct hpref_percpu_slots *hpref_percpu_slots;
> +
> +void hpref_release(struct urcu_ref *ref)
> +{
> +	struct hpref_node *node = caa_container_of(ref, struct hpref_node, refcount);
> +
> +	node->release(node);
> +}
> +
> +/*
> + * hpref_synchronize: Wait for any reader possessing a hazard pointer to
> + *                    @node to clear its hazard pointer slot.
> + */
> +void hpref_synchronize(struct hpref_node *node)
> +{
> +	int nr_cpus = rseq_get_max_nr_cpus(), cpu;
> +
> +	/* Memory ordering: Store A before Load B. */
> +	cmm_smp_mb();
> +	/* Scan all CPUs slots. */
> +	for (cpu = 0; cpu < nr_cpus; cpu++) {
> +		struct hpref_percpu_slots *cpu_slots = rseq_percpu_ptr(hpref_percpu_slots, cpu);
> +		struct hpref_slot *slot;
> +		unsigned int i;
> +
> +		for (i = 0; i < HPREF_NR_PERCPU_SLOTS; i++) {
> +			slot = &cpu_slots->slots[i];
> +			/* Busy-wait if node is found. */
> +			while (uatomic_load(&slot->node, CMM_ACQUIRE) == node)	/* Load B */
> +				caa_cpu_relax();
> +		}
> +	}
> +}
> +
> +/*
> + * hpref_synchronize_put: Wait for any reader possessing a hazard
> + *                        pointer to clear its slot and put reference
> + *                        count.
> + */
> +void hpref_synchronize_put(struct hpref_node *node)
> +{
> +	if (!node)
> +		return;
> +	hpref_synchronize(node);
> +	urcu_ref_put(&node->refcount, hpref_release);
> +}
> +
> +static __attribute__((constructor))
> +void hpref_init(void)
> +{
> +	mempool = rseq_mempool_create("hpref", sizeof(struct hpref_percpu_slots), NULL);
> +	if (!mempool)
> +		abort();
> +	hpref_percpu_slots = rseq_mempool_percpu_zmalloc(mempool);
> +	if (!hpref_percpu_slots)
> +		abort();
> +}
> +
> +static __attribute__((destructor))
> +void hpref_exit(void)
> +{
> +	rseq_mempool_percpu_free(hpref_percpu_slots);
> +	if (rseq_mempool_destroy(mempool))
> +		abort();
> +}