[urcu.git] / urcu-call-rcu-impl.h

/*
 * urcu-call-rcu.c
 *
 * Userspace RCU library - batch memory reclamation with kernel API
 *
 * Copyright (c) 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

#define _GNU_SOURCE
#include <stdio.h>
#include <pthread.h>
#include <signal.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <poll.h>
#include <sys/time.h>
#include <syscall.h>
#include <unistd.h>
#include <sched.h>

#include "config.h"
#include "urcu/wfqueue.h"
#include "urcu-call-rcu.h"
#include "urcu-pointer.h"
#include "urcu/list.h"
#include "urcu/urcu-futex.h"

/* Data structure that identifies a call_rcu thread. */

struct call_rcu_data {
	struct cds_wfq_queue cbs;
	unsigned long flags;
	pthread_mutex_t mtx;
	int futex;
	unsigned long qlen;
	pthread_t tid;
	int cpu_affinity;
	struct cds_list_head list;
} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));

/*
 * List of all call_rcu_data structures to keep valgrind happy.
 * Protected by call_rcu_mutex.
 */

CDS_LIST_HEAD(call_rcu_data_list);

/* Link a thread using call_rcu() to its call_rcu thread. */

static __thread struct call_rcu_data *thread_call_rcu_data;

/* Guard call_rcu thread creation. */

static pthread_mutex_t call_rcu_mutex = PTHREAD_MUTEX_INITIALIZER;

/* If a given thread does not have its own call_rcu thread, this is default. */

static struct call_rcu_data *default_call_rcu_data;

/*
 * If the sched_getcpu() and sysconf(_SC_NPROCESSORS_CONF) calls are
 * available, then we can have call_rcu threads assigned to individual
 * CPUs rather than only to specific threads.
 */

#if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF)

/*
 * Pointer to array of pointers to per-CPU call_rcu_data structures
 * and # CPUs.
 */

static struct call_rcu_data **per_cpu_call_rcu_data;
static long maxcpus;

static void call_rcu_wait(struct call_rcu_data *crdp)
{
	/* Read call_rcu list before read futex */
	cmm_smp_mb();
	if (uatomic_read(&crdp->futex) == -1)
		futex_async(&crdp->futex, FUTEX_WAIT, -1,
		      NULL, NULL, 0);
}

static void call_rcu_wake_up(struct call_rcu_data *crdp)
{
	/* Write to call_rcu list before reading/writing futex */
	cmm_smp_mb();
	if (unlikely(uatomic_read(&crdp->futex) == -1)) {
		uatomic_set(&crdp->futex, 0);
		futex_async(&crdp->futex, FUTEX_WAKE, 1,
		      NULL, NULL, 0);
	}
}

/* Allocate the array if it has not already been allocated. */

static void alloc_cpu_call_rcu_data(void)
{
	struct call_rcu_data **p;
	static int warned = 0;

	if (maxcpus != 0)
		return;
	maxcpus = sysconf(_SC_NPROCESSORS_CONF);
	if (maxcpus <= 0) {
		return;
	}
	p = malloc(maxcpus * sizeof(*per_cpu_call_rcu_data));
	if (p != NULL) {
		memset(p, '\0', maxcpus * sizeof(*per_cpu_call_rcu_data));
		per_cpu_call_rcu_data = p;
	} else {
		if (!warned) {
			fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
		}
		warned = 1;
	}
}

#else /* #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */

static const struct call_rcu_data **per_cpu_call_rcu_data = NULL;
static const long maxcpus = -1;

static void alloc_cpu_call_rcu_data(void)
{
}

static int sched_getcpu(void)
{
	return -1;
}

#endif /* #else #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */

/* Acquire the specified pthread mutex. */

static void call_rcu_lock(pthread_mutex_t *pmp)
{
	if (pthread_mutex_lock(pmp) != 0) {
		perror("pthread_mutex_lock");
		exit(-1);
	}
}

/* Release the specified pthread mutex. */

static void call_rcu_unlock(pthread_mutex_t *pmp)
{
	if (pthread_mutex_unlock(pmp) != 0) {
		perror("pthread_mutex_unlock");
		exit(-1);
	}
}

#if HAVE_SCHED_SETAFFINITY
static
int set_thread_cpu_affinity(struct call_rcu_data *crdp)
{
	cpu_set_t mask;

	if (crdp->cpu_affinity < 0)
		return 0;

	CPU_ZERO(&mask);
	CPU_SET(crdp->cpu_affinity, &mask);
#if SCHED_SETAFFINITY_ARGS == 2
	return sched_setaffinity(0, &mask);
#else
	return sched_setaffinity(0, sizeof(mask), &mask);
#endif
}
#else
static
int set_thread_cpu_affinity(struct call_rcu_data *crdp)
{
	return 0;
}
#endif

/* This is the code run by each call_rcu thread. */

static void *call_rcu_thread(void *arg)
{
	unsigned long cbcount;
	struct cds_wfq_node *cbs;
	struct cds_wfq_node **cbs_tail;
	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
	struct rcu_head *rhp;

	if (set_thread_cpu_affinity(crdp) != 0) {
		perror("pthread_setaffinity_np");
		exit(-1);
	}

	thread_call_rcu_data = crdp;
	for (;;) {
		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
			uatomic_dec(&crdp->futex);
			/* Decrement futex before reading call_rcu list */
			cmm_smp_mb();
		}
		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
				poll(NULL, 0, 1);
			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
			cbs_tail = (struct cds_wfq_node **)
				uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
			synchronize_rcu();
			cbcount = 0;
			do {
				while (cbs->next == NULL &&
				       &cbs->next != cbs_tail)
				       	poll(NULL, 0, 1);
				if (cbs == &crdp->cbs.dummy) {
					cbs = cbs->next;
					continue;
				}
				rhp = (struct rcu_head *)cbs;
				cbs = cbs->next;
				rhp->func(rhp);
				cbcount++;
			} while (cbs != NULL);
			uatomic_sub(&crdp->qlen, cbcount);
		}
		if (crdp->flags & URCU_CALL_RCU_STOP) {
			if (!(crdp->flags & URCU_CALL_RCU_RT)) {
				/*
				 * Read call_rcu list before write futex.
				 */
				cmm_smp_mb();
				uatomic_set(&crdp->futex, 0);
			}
			break;
		}
		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
			if (&crdp->cbs.head == _CMM_LOAD_SHARED(crdp->cbs.tail))
				call_rcu_wait(crdp);
		}
		poll(NULL, 0, 10);
	}
	call_rcu_lock(&crdp->mtx);
	crdp->flags |= URCU_CALL_RCU_STOPPED;
	call_rcu_unlock(&crdp->mtx);
	return NULL;
}

/*
 * Create both a call_rcu thread and the corresponding call_rcu_data
 * structure, linking the structure in as specified.  Caller must hold
 * call_rcu_mutex.
 */

static void call_rcu_data_init(struct call_rcu_data **crdpp,
			       unsigned long flags,
			       int cpu_affinity)
{
	struct call_rcu_data *crdp;

	crdp = malloc(sizeof(*crdp));
	if (crdp == NULL) {
		fprintf(stderr, "Out of memory.\n");
		exit(-1);
	}
	memset(crdp, '\0', sizeof(*crdp));
	cds_wfq_init(&crdp->cbs);
	crdp->qlen = 0;
	if (pthread_mutex_init(&crdp->mtx, NULL) != 0) {
		perror("pthread_mutex_init");
		exit(-1);
	}
	crdp->futex = 0;
	crdp->flags = flags;
	cds_list_add(&crdp->list, &call_rcu_data_list);
	crdp->cpu_affinity = cpu_affinity;
	cmm_smp_mb();  /* Structure initialized before pointer is planted. */
	*crdpp = crdp;
	if (pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp) != 0) {
		perror("pthread_create");
		exit(-1);
	}
}

/*
 * Return a pointer to the call_rcu_data structure for the specified
 * CPU, returning NULL if there is none.  We cannot automatically
 * created it because the platform we are running on might not define
 * sched_getcpu().
 */

struct call_rcu_data *get_cpu_call_rcu_data(int cpu)
{
	static int warned = 0;

	if (per_cpu_call_rcu_data == NULL)
		return NULL;
	if (!warned && maxcpus > 0 && (cpu < 0 || maxcpus <= cpu)) {
		fprintf(stderr, "[error] liburcu: get CPU # out of range\n");
		warned = 1;
	}
	if (cpu < 0 || maxcpus <= cpu)
		return NULL;
	return per_cpu_call_rcu_data[cpu];
}

/*
 * Return the tid corresponding to the call_rcu thread whose
 * call_rcu_data structure is specified.
 */

pthread_t get_call_rcu_thread(struct call_rcu_data *crdp)
{
	return crdp->tid;
}

/*
 * Create a call_rcu_data structure (with thread) and return a pointer.
 */

static struct call_rcu_data *__create_call_rcu_data(unsigned long flags,
						    int cpu_affinity)
{
	struct call_rcu_data *crdp;

	call_rcu_data_init(&crdp, flags, cpu_affinity);
	return crdp;
}

struct call_rcu_data *create_call_rcu_data(unsigned long flags,
					   int cpu_affinity)
{
	struct call_rcu_data *crdp;

	call_rcu_lock(&call_rcu_mutex);
	crdp = __create_call_rcu_data(flags, cpu_affinity);
	call_rcu_unlock(&call_rcu_mutex);
	return crdp;
}

/*
 * Set the specified CPU to use the specified call_rcu_data structure.
 *
 * Use NULL to remove a CPU's call_rcu_data structure, but it is
 * the caller's responsibility to dispose of the removed structure.
 * Use get_cpu_call_rcu_data() to obtain a pointer to the old structure
 * (prior to NULLing it out, of course).
 */

int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp)
{
	int warned = 0;

	call_rcu_lock(&call_rcu_mutex);
	if (cpu < 0 || maxcpus <= cpu) {
		if (!warned) {
			fprintf(stderr, "[error] liburcu: set CPU # out of range\n");
			warned = 1;
		}
		call_rcu_unlock(&call_rcu_mutex);
		errno = EINVAL;
		return -EINVAL;
	}
	alloc_cpu_call_rcu_data();
	call_rcu_unlock(&call_rcu_mutex);
	if (per_cpu_call_rcu_data == NULL) {
		errno = ENOMEM;
		return -ENOMEM;
	}
	per_cpu_call_rcu_data[cpu] = crdp;
	return 0;
}

/*
 * Return a pointer to the default call_rcu_data structure, creating
 * one if need be.  Because we never free call_rcu_data structures,
 * we don't need to be in an RCU read-side critical section.
 */

struct call_rcu_data *get_default_call_rcu_data(void)
{
	if (default_call_rcu_data != NULL)
		return rcu_dereference(default_call_rcu_data);
	call_rcu_lock(&call_rcu_mutex);
	if (default_call_rcu_data != NULL) {
		call_rcu_unlock(&call_rcu_mutex);
		return default_call_rcu_data;
	}
	call_rcu_data_init(&default_call_rcu_data, 0, -1);
	call_rcu_unlock(&call_rcu_mutex);
	return default_call_rcu_data;
}

/*
 * Return the call_rcu_data structure that applies to the currently
 * running thread.  Any call_rcu_data structure assigned specifically
 * to this thread has first priority, followed by any call_rcu_data
 * structure assigned to the CPU on which the thread is running,
 * followed by the default call_rcu_data structure.  If there is not
 * yet a default call_rcu_data structure, one will be created.
 */
struct call_rcu_data *get_call_rcu_data(void)
{
	int curcpu;
	static int warned = 0;

	if (thread_call_rcu_data != NULL)
		return thread_call_rcu_data;
	if (maxcpus <= 0)
		return get_default_call_rcu_data();
	curcpu = sched_getcpu();
	if (!warned && (curcpu < 0 || maxcpus <= curcpu)) {
		fprintf(stderr, "[error] liburcu: gcrd CPU # out of range\n");
		warned = 1;
	}
	if (curcpu >= 0 && maxcpus > curcpu &&
	    per_cpu_call_rcu_data != NULL &&
	    per_cpu_call_rcu_data[curcpu] != NULL)
	    	return per_cpu_call_rcu_data[curcpu];
	return get_default_call_rcu_data();
}

/*
 * Return a pointer to this task's call_rcu_data if there is one.
 */

struct call_rcu_data *get_thread_call_rcu_data(void)
{
	return thread_call_rcu_data;
}

/*
 * Set this task's call_rcu_data structure as specified, regardless
 * of whether or not this task already had one.  (This allows switching
 * to and from real-time call_rcu threads, for example.)
 *
 * Use NULL to remove a thread's call_rcu_data structure, but it is
 * the caller's responsibility to dispose of the removed structure.
 * Use get_thread_call_rcu_data() to obtain a pointer to the old structure
 * (prior to NULLing it out, of course).
 */

void set_thread_call_rcu_data(struct call_rcu_data *crdp)
{
	thread_call_rcu_data = crdp;
}

/*
 * Create a separate call_rcu thread for each CPU.  This does not
 * replace a pre-existing call_rcu thread -- use the set_cpu_call_rcu_data()
 * function if you want that behavior.
 */

int create_all_cpu_call_rcu_data(unsigned long flags)
{
	int i;
	struct call_rcu_data *crdp;
	int ret;

	call_rcu_lock(&call_rcu_mutex);
	alloc_cpu_call_rcu_data();
	call_rcu_unlock(&call_rcu_mutex);
	if (maxcpus <= 0) {
		errno = EINVAL;
		return -EINVAL;
	}
	if (per_cpu_call_rcu_data == NULL) {
		errno = ENOMEM;
		return -ENOMEM;
	}
	for (i = 0; i < maxcpus; i++) {
		call_rcu_lock(&call_rcu_mutex);
		if (get_cpu_call_rcu_data(i)) {
			call_rcu_unlock(&call_rcu_mutex);
			continue;
		}
		crdp = __create_call_rcu_data(flags, i);
		if (crdp == NULL) {
			call_rcu_unlock(&call_rcu_mutex);
			errno = ENOMEM;
			return -ENOMEM;
		}
		call_rcu_unlock(&call_rcu_mutex);
		if ((ret = set_cpu_call_rcu_data(i, crdp)) != 0) {
			/* FIXME: Leaks crdp for now. */
			return ret; /* Can happen on race. */
		}
	}
	return 0;
}

/*
 * Wake up the call_rcu thread corresponding to the specified
 * call_rcu_data structure.
 */
static void wake_call_rcu_thread(struct call_rcu_data *crdp)
{
	if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT))
		call_rcu_wake_up(crdp);
}

/*
 * Schedule a function to be invoked after a following grace period.
 * This is the only function that must be called -- the others are
 * only present to allow applications to tune their use of RCU for
 * maximum performance.
 *
 * Note that unless a call_rcu thread has not already been created,
 * the first invocation of call_rcu() will create one.  So, if you
 * need the first invocation of call_rcu() to be fast, make sure
 * to create a call_rcu thread first.  One way to accomplish this is
 * "get_call_rcu_data();", and another is create_all_cpu_call_rcu_data().
 */

void call_rcu(struct rcu_head *head,
	      void (*func)(struct rcu_head *head))
{
	struct call_rcu_data *crdp;

	cds_wfq_node_init(&head->next);
	head->func = func;
	crdp = get_call_rcu_data();
	cds_wfq_enqueue(&crdp->cbs, &head->next);
	uatomic_inc(&crdp->qlen);
	wake_call_rcu_thread(crdp);
}

/*
 * Free up the specified call_rcu_data structure, terminating the
 * associated call_rcu thread.  The caller must have previously
 * removed the call_rcu_data structure from per-thread or per-CPU
 * usage.  For example, set_cpu_call_rcu_data(cpu, NULL) for per-CPU
 * call_rcu_data structures or set_thread_call_rcu_data(NULL) for
 * per-thread call_rcu_data structures.
 *
 * We silently refuse to free up the default call_rcu_data structure
 * because that is where we put any leftover callbacks.  Note that
 * the possibility of self-spawning callbacks makes it impossible
 * to execute all the callbacks in finite time without putting any
 * newly spawned callbacks somewhere else.  The "somewhere else" of
 * last resort is the default call_rcu_data structure.
 *
 * We also silently refuse to free NULL pointers.  This simplifies
 * the calling code.
 */
void call_rcu_data_free(struct call_rcu_data *crdp)
{
	struct cds_wfq_node *cbs;
	struct cds_wfq_node **cbs_tail;
	struct cds_wfq_node **cbs_endprev;

	if (crdp == NULL || crdp == default_call_rcu_data) {
		return;
	}
	if ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0) {
		call_rcu_lock(&crdp->mtx);
		crdp->flags |= URCU_CALL_RCU_STOP;
		call_rcu_unlock(&crdp->mtx);
		wake_call_rcu_thread(crdp);
		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
			poll(NULL, 0, 1);
	}
	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
		while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
			poll(NULL, 0, 1);
		_CMM_STORE_SHARED(crdp->cbs.head, NULL);
		cbs_tail = (struct cds_wfq_node **)
			uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
		cbs_endprev = (struct cds_wfq_node **)
			uatomic_xchg(&default_call_rcu_data, cbs_tail);
		*cbs_endprev = cbs;
		uatomic_add(&default_call_rcu_data->qlen,
			    uatomic_read(&crdp->qlen));
		cds_list_del(&crdp->list);
		free(crdp);
	}
}

/*
 * Clean up all the per-CPU call_rcu threads.
 */
void free_all_cpu_call_rcu_data(void)
{
	int cpu;
	struct call_rcu_data *crdp;

	if (maxcpus <= 0)
		return;
	for (cpu = 0; cpu < maxcpus; cpu++) {
		crdp = get_cpu_call_rcu_data(cpu);
		if (crdp == NULL)
			continue;
		set_cpu_call_rcu_data(cpu, NULL);
		call_rcu_data_free(crdp);
	}
}

/*
 * Acquire the call_rcu_mutex in order to ensure that the child sees
 * all of the call_rcu() data structures in a consistent state.
 * Suitable for pthread_atfork() and friends.
 */
void call_rcu_before_fork(void)
{
	call_rcu_lock(&call_rcu_mutex);
}

/*
 * Clean up call_rcu data structures in the parent of a successful fork()
 * that is not followed by exec() in the child.  Suitable for
 * pthread_atfork() and friends.
 */
void call_rcu_after_fork_parent(void)
{
	call_rcu_unlock(&call_rcu_mutex);
}

/*
 * Clean up call_rcu data structures in the child of a successful fork()
 * that is not followed by exec().  Suitable for pthread_atfork() and
 * friends.
 */
void call_rcu_after_fork_child(void)
{
	struct call_rcu_data *crdp;

	/* Release the mutex. */
	call_rcu_unlock(&call_rcu_mutex);

	/*
	 * Allocate a new default call_rcu_data structure in order
	 * to get a working call_rcu thread to go with it.
	 */
	default_call_rcu_data = NULL;
	(void)get_default_call_rcu_data();

	/* Dispose of all of the rest of the call_rcu_data structures. */
	while (call_rcu_data_list.next != call_rcu_data_list.prev) {
		crdp = cds_list_entry(call_rcu_data_list.prev,
				      struct call_rcu_data, list);
		if (crdp == default_call_rcu_data)
			crdp = cds_list_entry(crdp->list.prev,
					      struct call_rcu_data, list);
		crdp->flags = URCU_CALL_RCU_STOPPED;
		call_rcu_data_free(crdp);
	}
}
Commit	Line	Data
b57aee66 PM	1	/*
	2	* urcu-call-rcu.c
	3	*
	4	* Userspace RCU library - batch memory reclamation with kernel API
	5	*
	6	* Copyright (c) 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
	7	*
	8	* This library is free software; you can redistribute it and/or
	9	* modify it under the terms of the GNU Lesser General Public
	10	* License as published by the Free Software Foundation; either
	11	* version 2.1 of the License, or (at your option) any later version.
	12	*
	13	* This library is distributed in the hope that it will be useful,
	14	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	15	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	16	* Lesser General Public License for more details.
	17	*
	18	* You should have received a copy of the GNU Lesser General Public
	19	* License along with this library; if not, write to the Free Software
	20	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
	21	*/
	22
c1d2c60b	23	#define _GNU_SOURCE
b57aee66 PM	24	#include <stdio.h>
	25	#include <pthread.h>
	26	#include <signal.h>
	27	#include <assert.h>
	28	#include <stdlib.h>
	29	#include <string.h>
	30	#include <errno.h>
	31	#include <poll.h>
	32	#include <sys/time.h>
	33	#include <syscall.h>
	34	#include <unistd.h>
c1d2c60b	35	#include <sched.h>
b57aee66 PM	36
	37	#include "config.h"
	38	#include "urcu/wfqueue.h"
	39	#include "urcu-call-rcu.h"
	40	#include "urcu-pointer.h"
3c24913f	41	#include "urcu/list.h"
263e3cf9	42	#include "urcu/urcu-futex.h"
b57aee66 PM	43
	44	/* Data structure that identifies a call_rcu thread. */
	45
	46	struct call_rcu_data {
	47	struct cds_wfq_queue cbs;
	48	unsigned long flags;
	49	pthread_mutex_t mtx;
263e3cf9	50	int futex;
b57aee66 PM	51	unsigned long qlen;
b57aee66 PM	52	pthread_t tid;
c1d2c60b	53	int cpu_affinity;
3c24913f	54	struct cds_list_head list;
b57aee66 PM	55	} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
b57aee66 PM	56
3c24913f PM	57	/*
	58	* List of all call_rcu_data structures to keep valgrind happy.
	59	* Protected by call_rcu_mutex.
	60	*/
	61
	62	CDS_LIST_HEAD(call_rcu_data_list);
	63
b57aee66 PM	64	/* Link a thread using call_rcu() to its call_rcu thread. */
	65
	66	static __thread struct call_rcu_data *thread_call_rcu_data;
	67
	68	/* Guard call_rcu thread creation. */
	69
	70	static pthread_mutex_t call_rcu_mutex = PTHREAD_MUTEX_INITIALIZER;
	71
	72	/* If a given thread does not have its own call_rcu thread, this is default. */
	73
	74	static struct call_rcu_data *default_call_rcu_data;
	75
b57aee66 PM	76	/*
	77	* If the sched_getcpu() and sysconf(_SC_NPROCESSORS_CONF) calls are
	78	* available, then we can have call_rcu threads assigned to individual
	79	* CPUs rather than only to specific threads.
	80	*/
	81
	82	#if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF)
	83
	84	/*
	85	* Pointer to array of pointers to per-CPU call_rcu_data structures
	86	* and # CPUs.
	87	*/
	88
	89	static struct call_rcu_data **per_cpu_call_rcu_data;
	90	static long maxcpus;
	91
263e3cf9 MD	92	static void call_rcu_wait(struct call_rcu_data *crdp)
	93	{
	94	/* Read call_rcu list before read futex */
	95	cmm_smp_mb();
	96	if (uatomic_read(&crdp->futex) == -1)
	97	futex_async(&crdp->futex, FUTEX_WAIT, -1,
	98	NULL, NULL, 0);
	99	}
	100
	101	static void call_rcu_wake_up(struct call_rcu_data *crdp)
	102	{
	103	/* Write to call_rcu list before reading/writing futex */
	104	cmm_smp_mb();
	105	if (unlikely(uatomic_read(&crdp->futex) == -1)) {
	106	uatomic_set(&crdp->futex, 0);
	107	futex_async(&crdp->futex, FUTEX_WAKE, 1,
	108	NULL, NULL, 0);
	109	}
	110	}
	111
b57aee66 PM	112	/* Allocate the array if it has not already been allocated. */
	113
	114	static void alloc_cpu_call_rcu_data(void)
	115	{
	116	struct call_rcu_data **p;
	117	static int warned = 0;
	118
	119	if (maxcpus != 0)
	120	return;
	121	maxcpus = sysconf(_SC_NPROCESSORS_CONF);
	122	if (maxcpus <= 0) {
	123	return;
	124	}
	125	p = malloc(maxcpus * sizeof(*per_cpu_call_rcu_data));
	126	if (p != NULL) {
	127	memset(p, '\0', maxcpus * sizeof(*per_cpu_call_rcu_data));
	128	per_cpu_call_rcu_data = p;
	129	} else {
	130	if (!warned) {
	131	fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
	132	}
	133	warned = 1;
	134	}
	135	}
	136
	137	#else /* #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
	138
	139	static const struct call_rcu_data **per_cpu_call_rcu_data = NULL;
	140	static const long maxcpus = -1;
	141
	142	static void alloc_cpu_call_rcu_data(void)
	143	{
	144	}
	145
	146	static int sched_getcpu(void)
	147	{
	148	return -1;
	149	}
	150
	151	#endif /* #else #if defined(HAVE_SCHED_GETCPU) && defined(HAVE_SYSCONF) */
	152
	153	/* Acquire the specified pthread mutex. */
	154
	155	static void call_rcu_lock(pthread_mutex_t *pmp)
	156	{
	157	if (pthread_mutex_lock(pmp) != 0) {
	158	perror("pthread_mutex_lock");
	159	exit(-1);
	160	}
	161	}
	162
	163	/* Release the specified pthread mutex. */
	164
	165	static void call_rcu_unlock(pthread_mutex_t *pmp)
	166	{
	167	if (pthread_mutex_unlock(pmp) != 0) {
	168	perror("pthread_mutex_unlock");
	169	exit(-1);
	170	}
	171	}
	172
c1d2c60b MD	173	#if HAVE_SCHED_SETAFFINITY
	174	static
	175	int set_thread_cpu_affinity(struct call_rcu_data *crdp)
	176	{
	177	cpu_set_t mask;
	178
	179	if (crdp->cpu_affinity < 0)
	180	return 0;
	181
	182	CPU_ZERO(&mask);
	183	CPU_SET(crdp->cpu_affinity, &mask);
	184	#if SCHED_SETAFFINITY_ARGS == 2
	185	return sched_setaffinity(0, &mask);
	186	#else
	187	return sched_setaffinity(0, sizeof(mask), &mask);
	188	#endif
	189	}
	190	#else
	191	static
	192	int set_thread_cpu_affinity(struct call_rcu_data *crdp)
	193	{
	194	return 0;
	195	}
	196	#endif
	197
b57aee66 PM	198	/* This is the code run by each call_rcu thread. */
	199
	200	static void call_rcu_thread(void arg)
	201	{
	202	unsigned long cbcount;
	203	struct cds_wfq_node *cbs;
	204	struct cds_wfq_node **cbs_tail;
	205	struct call_rcu_data crdp = (struct call_rcu_data )arg;
	206	struct rcu_head *rhp;
	207
c1d2c60b MD	208	if (set_thread_cpu_affinity(crdp) != 0) {
	209	perror("pthread_setaffinity_np");
	210	exit(-1);
	211	}
	212
b57aee66 PM	213	thread_call_rcu_data = crdp;
b57aee66 PM	214	for (;;) {
c768e45e MD	215	if (!(crdp->flags & URCU_CALL_RCU_RT)) {
	216	uatomic_dec(&crdp->futex);
	217	/* Decrement futex before reading call_rcu list */
	218	cmm_smp_mb();
	219	}
b57aee66 PM	220	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
	221	while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
	222	poll(NULL, 0, 1);
	223	_CMM_STORE_SHARED(crdp->cbs.head, NULL);
	224	cbs_tail = (struct cds_wfq_node **)
	225	uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
	226	synchronize_rcu();
	227	cbcount = 0;
	228	do {
	229	while (cbs->next == NULL &&
	230	&cbs->next != cbs_tail)
	231	poll(NULL, 0, 1);
	232	if (cbs == &crdp->cbs.dummy) {
	233	cbs = cbs->next;
	234	continue;
	235	}
	236	rhp = (struct rcu_head *)cbs;
	237	cbs = cbs->next;
	238	rhp->func(rhp);
	239	cbcount++;
	240	} while (cbs != NULL);
	241	uatomic_sub(&crdp->qlen, cbcount);
	242	}
c768e45e MD	243	if (crdp->flags & URCU_CALL_RCU_STOP) {
	244	if (!(crdp->flags & URCU_CALL_RCU_RT)) {
	245	/*
	246	* Read call_rcu list before write futex.
	247	*/
	248	cmm_smp_mb();
	249	uatomic_set(&crdp->futex, 0);
	250	}
7106ddf8	251	break;
c768e45e MD	252	}
c768e45e MD	253	if (!(crdp->flags & URCU_CALL_RCU_RT)) {
263e3cf9 MD	254	if (&crdp->cbs.head == _CMM_LOAD_SHARED(crdp->cbs.tail))
263e3cf9 MD	255	call_rcu_wait(crdp);
b57aee66	256	}
c768e45e	257	poll(NULL, 0, 10);
b57aee66	258	}
7106ddf8 PM	259	call_rcu_lock(&crdp->mtx);
	260	crdp->flags \|= URCU_CALL_RCU_STOPPED;
	261	call_rcu_unlock(&crdp->mtx);
	262	return NULL;
b57aee66 PM	263	}
	264
	265	/*
	266	* Create both a call_rcu thread and the corresponding call_rcu_data
3c24913f PM	267	* structure, linking the structure in as specified. Caller must hold
3c24913f PM	268	* call_rcu_mutex.
b57aee66 PM	269	*/
b57aee66 PM	270
3c24913f	271	static void call_rcu_data_init(struct call_rcu_data **crdpp,
c1d2c60b MD	272	unsigned long flags,
c1d2c60b MD	273	int cpu_affinity)
b57aee66 PM	274	{
	275	struct call_rcu_data *crdp;
	276
	277	crdp = malloc(sizeof(*crdp));
	278	if (crdp == NULL) {
	279	fprintf(stderr, "Out of memory.\n");
	280	exit(-1);
	281	}
	282	memset(crdp, '\0', sizeof(*crdp));
	283	cds_wfq_init(&crdp->cbs);
	284	crdp->qlen = 0;
	285	if (pthread_mutex_init(&crdp->mtx, NULL) != 0) {
	286	perror("pthread_mutex_init");
	287	exit(-1);
	288	}
263e3cf9 MD	289	crdp->futex = 0;
263e3cf9 MD	290	crdp->flags = flags;
3c24913f	291	cds_list_add(&crdp->list, &call_rcu_data_list);
c1d2c60b	292	crdp->cpu_affinity = cpu_affinity;
b57aee66 PM	293	cmm_smp_mb(); /* Structure initialized before pointer is planted. */
	294	*crdpp = crdp;
	295	if (pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp) != 0) {
	296	perror("pthread_create");
	297	exit(-1);
	298	}
	299	}
	300
	301	/*
	302	* Return a pointer to the call_rcu_data structure for the specified
	303	* CPU, returning NULL if there is none. We cannot automatically
	304	* created it because the platform we are running on might not define
	305	* sched_getcpu().
	306	*/
	307
	308	struct call_rcu_data *get_cpu_call_rcu_data(int cpu)
	309	{
	310	static int warned = 0;
	311
	312	if (per_cpu_call_rcu_data == NULL)
	313	return NULL;
	314	if (!warned && maxcpus > 0 && (cpu < 0 \|\| maxcpus <= cpu)) {
	315	fprintf(stderr, "[error] liburcu: get CPU # out of range\n");
	316	warned = 1;
	317	}
	318	if (cpu < 0 \|\| maxcpus <= cpu)
	319	return NULL;
	320	return per_cpu_call_rcu_data[cpu];
	321	}
	322
	323	/*
	324	* Return the tid corresponding to the call_rcu thread whose
	325	* call_rcu_data structure is specified.
	326	*/
	327
	328	pthread_t get_call_rcu_thread(struct call_rcu_data *crdp)
	329	{
	330	return crdp->tid;
	331	}
	332
	333	/*
	334	* Create a call_rcu_data structure (with thread) and return a pointer.
	335	*/
	336
c1d2c60b MD	337	static struct call_rcu_data *__create_call_rcu_data(unsigned long flags,
c1d2c60b MD	338	int cpu_affinity)
b57aee66 PM	339	{
	340	struct call_rcu_data *crdp;
	341
c1d2c60b	342	call_rcu_data_init(&crdp, flags, cpu_affinity);
b57aee66 PM	343	return crdp;
	344	}
	345
c1d2c60b MD	346	struct call_rcu_data *create_call_rcu_data(unsigned long flags,
c1d2c60b MD	347	int cpu_affinity)
3c24913f PM	348	{
	349	struct call_rcu_data *crdp;
	350
	351	call_rcu_lock(&call_rcu_mutex);
c1d2c60b	352	crdp = __create_call_rcu_data(flags, cpu_affinity);
3c24913f PM	353	call_rcu_unlock(&call_rcu_mutex);
	354	return crdp;
	355	}
	356
b57aee66 PM	357	/*
b57aee66 PM	358	* Set the specified CPU to use the specified call_rcu_data structure.
7106ddf8 PM	359	*
	360	* Use NULL to remove a CPU's call_rcu_data structure, but it is
	361	* the caller's responsibility to dispose of the removed structure.
	362	* Use get_cpu_call_rcu_data() to obtain a pointer to the old structure
	363	* (prior to NULLing it out, of course).
b57aee66 PM	364	*/
	365
	366	int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp)
	367	{
	368	int warned = 0;
	369
	370	call_rcu_lock(&call_rcu_mutex);
	371	if (cpu < 0 \|\| maxcpus <= cpu) {
	372	if (!warned) {
	373	fprintf(stderr, "[error] liburcu: set CPU # out of range\n");
	374	warned = 1;
	375	}
	376	call_rcu_unlock(&call_rcu_mutex);
	377	errno = EINVAL;
	378	return -EINVAL;
	379	}
	380	alloc_cpu_call_rcu_data();
	381	call_rcu_unlock(&call_rcu_mutex);
	382	if (per_cpu_call_rcu_data == NULL) {
	383	errno = ENOMEM;
	384	return -ENOMEM;
	385	}
	386	per_cpu_call_rcu_data[cpu] = crdp;
	387	return 0;
	388	}
	389
	390	/*
	391	* Return a pointer to the default call_rcu_data structure, creating
	392	* one if need be. Because we never free call_rcu_data structures,
	393	* we don't need to be in an RCU read-side critical section.
	394	*/
	395
	396	struct call_rcu_data *get_default_call_rcu_data(void)
	397	{
	398	if (default_call_rcu_data != NULL)
	399	return rcu_dereference(default_call_rcu_data);
	400	call_rcu_lock(&call_rcu_mutex);
	401	if (default_call_rcu_data != NULL) {
	402	call_rcu_unlock(&call_rcu_mutex);
	403	return default_call_rcu_data;
	404	}
c1d2c60b	405	call_rcu_data_init(&default_call_rcu_data, 0, -1);
b57aee66 PM	406	call_rcu_unlock(&call_rcu_mutex);
	407	return default_call_rcu_data;
	408	}
	409
	410	/*
	411	* Return the call_rcu_data structure that applies to the currently
	412	* running thread. Any call_rcu_data structure assigned specifically
	413	* to this thread has first priority, followed by any call_rcu_data
	414	* structure assigned to the CPU on which the thread is running,
	415	* followed by the default call_rcu_data structure. If there is not
	416	* yet a default call_rcu_data structure, one will be created.
	417	*/
	418	struct call_rcu_data *get_call_rcu_data(void)
	419	{
	420	int curcpu;
	421	static int warned = 0;
	422
	423	if (thread_call_rcu_data != NULL)
	424	return thread_call_rcu_data;
	425	if (maxcpus <= 0)
	426	return get_default_call_rcu_data();
	427	curcpu = sched_getcpu();
	428	if (!warned && (curcpu < 0 \|\| maxcpus <= curcpu)) {
	429	fprintf(stderr, "[error] liburcu: gcrd CPU # out of range\n");
	430	warned = 1;
	431	}
	432	if (curcpu >= 0 && maxcpus > curcpu &&
	433	per_cpu_call_rcu_data != NULL &&
	434	per_cpu_call_rcu_data[curcpu] != NULL)
	435	return per_cpu_call_rcu_data[curcpu];
	436	return get_default_call_rcu_data();
	437	}
	438
	439	/*
	440	* Return a pointer to this task's call_rcu_data if there is one.
	441	*/
	442
	443	struct call_rcu_data *get_thread_call_rcu_data(void)
	444	{
	445	return thread_call_rcu_data;
	446	}
	447
	448	/*
	449	* Set this task's call_rcu_data structure as specified, regardless
	450	* of whether or not this task already had one. (This allows switching
	451	* to and from real-time call_rcu threads, for example.)
7106ddf8 PM	452	*
	453	* Use NULL to remove a thread's call_rcu_data structure, but it is
	454	* the caller's responsibility to dispose of the removed structure.
	455	* Use get_thread_call_rcu_data() to obtain a pointer to the old structure
	456	* (prior to NULLing it out, of course).
b57aee66 PM	457	*/
	458
	459	void set_thread_call_rcu_data(struct call_rcu_data *crdp)
	460	{
	461	thread_call_rcu_data = crdp;
	462	}
	463
	464	/*
	465	* Create a separate call_rcu thread for each CPU. This does not
	466	* replace a pre-existing call_rcu thread -- use the set_cpu_call_rcu_data()
	467	* function if you want that behavior.
	468	*/
	469
	470	int create_all_cpu_call_rcu_data(unsigned long flags)
	471	{
	472	int i;
	473	struct call_rcu_data *crdp;
	474	int ret;
	475
	476	call_rcu_lock(&call_rcu_mutex);
	477	alloc_cpu_call_rcu_data();
	478	call_rcu_unlock(&call_rcu_mutex);
	479	if (maxcpus <= 0) {
	480	errno = EINVAL;
	481	return -EINVAL;
	482	}
	483	if (per_cpu_call_rcu_data == NULL) {
	484	errno = ENOMEM;
	485	return -ENOMEM;
	486	}
	487	for (i = 0; i < maxcpus; i++) {
	488	call_rcu_lock(&call_rcu_mutex);
	489	if (get_cpu_call_rcu_data(i)) {
	490	call_rcu_unlock(&call_rcu_mutex);
	491	continue;
	492	}
c1d2c60b	493	crdp = __create_call_rcu_data(flags, i);
b57aee66 PM	494	if (crdp == NULL) {
	495	call_rcu_unlock(&call_rcu_mutex);
	496	errno = ENOMEM;
	497	return -ENOMEM;
	498	}
	499	call_rcu_unlock(&call_rcu_mutex);
	500	if ((ret = set_cpu_call_rcu_data(i, crdp)) != 0) {
	501	/* FIXME: Leaks crdp for now. */
	502	return ret; /* Can happen on race. */
	503	}
	504	}
	505	return 0;
	506	}
	507
7106ddf8 PM	508	/*
	509	* Wake up the call_rcu thread corresponding to the specified
	510	* call_rcu_data structure.
	511	*/
	512	static void wake_call_rcu_thread(struct call_rcu_data *crdp)
	513	{
263e3cf9 MD	514	if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT))
263e3cf9 MD	515	call_rcu_wake_up(crdp);
7106ddf8 PM	516	}
7106ddf8 PM	517
b57aee66 PM	518	/*
	519	* Schedule a function to be invoked after a following grace period.
	520	* This is the only function that must be called -- the others are
	521	* only present to allow applications to tune their use of RCU for
	522	* maximum performance.
	523	*
	524	* Note that unless a call_rcu thread has not already been created,
	525	* the first invocation of call_rcu() will create one. So, if you
	526	* need the first invocation of call_rcu() to be fast, make sure
	527	* to create a call_rcu thread first. One way to accomplish this is
	528	* "get_call_rcu_data();", and another is create_all_cpu_call_rcu_data().
	529	*/
	530
	531	void call_rcu(struct rcu_head *head,
	532	void (func)(struct rcu_head head))
	533	{
	534	struct call_rcu_data *crdp;
	535
	536	cds_wfq_node_init(&head->next);
	537	head->func = func;
	538	crdp = get_call_rcu_data();
	539	cds_wfq_enqueue(&crdp->cbs, &head->next);
	540	uatomic_inc(&crdp->qlen);
7106ddf8 PM	541	wake_call_rcu_thread(crdp);
	542	}
	543
	544	/*
	545	* Free up the specified call_rcu_data structure, terminating the
	546	* associated call_rcu thread. The caller must have previously
	547	* removed the call_rcu_data structure from per-thread or per-CPU
	548	* usage. For example, set_cpu_call_rcu_data(cpu, NULL) for per-CPU
	549	* call_rcu_data structures or set_thread_call_rcu_data(NULL) for
	550	* per-thread call_rcu_data structures.
	551	*
	552	* We silently refuse to free up the default call_rcu_data structure
	553	* because that is where we put any leftover callbacks. Note that
	554	* the possibility of self-spawning callbacks makes it impossible
	555	* to execute all the callbacks in finite time without putting any
	556	* newly spawned callbacks somewhere else. The "somewhere else" of
	557	* last resort is the default call_rcu_data structure.
	558	*
	559	* We also silently refuse to free NULL pointers. This simplifies
	560	* the calling code.
	561	*/
	562	void call_rcu_data_free(struct call_rcu_data *crdp)
	563	{
	564	struct cds_wfq_node *cbs;
	565	struct cds_wfq_node **cbs_tail;
	566	struct cds_wfq_node **cbs_endprev;
	567
	568	if (crdp == NULL \|\| crdp == default_call_rcu_data) {
	569	return;
	570	}
	571	if ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0) {
b57aee66	572	call_rcu_lock(&crdp->mtx);
7106ddf8	573	crdp->flags \|= URCU_CALL_RCU_STOP;
b57aee66	574	call_rcu_unlock(&crdp->mtx);
7106ddf8 PM	575	wake_call_rcu_thread(crdp);
	576	while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
	577	poll(NULL, 0, 1);
	578	}
	579	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
	580	while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
	581	poll(NULL, 0, 1);
	582	_CMM_STORE_SHARED(crdp->cbs.head, NULL);
	583	cbs_tail = (struct cds_wfq_node **)
	584	uatomic_xchg(&crdp->cbs.tail, &crdp->cbs.head);
	585	cbs_endprev = (struct cds_wfq_node **)
	586	uatomic_xchg(&default_call_rcu_data, cbs_tail);
	587	*cbs_endprev = cbs;
	588	uatomic_add(&default_call_rcu_data->qlen,
	589	uatomic_read(&crdp->qlen));
	590	cds_list_del(&crdp->list);
	591	free(crdp);
	592	}
	593	}
	594
	595	/*
	596	* Clean up all the per-CPU call_rcu threads.
	597	*/
	598	void free_all_cpu_call_rcu_data(void)
	599	{
	600	int cpu;
	601	struct call_rcu_data *crdp;
	602
	603	if (maxcpus <= 0)
	604	return;
	605	for (cpu = 0; cpu < maxcpus; cpu++) {
	606	crdp = get_cpu_call_rcu_data(cpu);
	607	if (crdp == NULL)
	608	continue;
	609	set_cpu_call_rcu_data(cpu, NULL);
	610	call_rcu_data_free(crdp);
	611	}
	612	}
	613
81ad2e19 PM	614	/*
	615	* Acquire the call_rcu_mutex in order to ensure that the child sees
	616	* all of the call_rcu() data structures in a consistent state.
	617	* Suitable for pthread_atfork() and friends.
	618	*/
	619	void call_rcu_before_fork(void)
	620	{
	621	call_rcu_lock(&call_rcu_mutex);
	622	}
	623
	624	/*
	625	* Clean up call_rcu data structures in the parent of a successful fork()
	626	* that is not followed by exec() in the child. Suitable for
	627	* pthread_atfork() and friends.
	628	*/
	629	void call_rcu_after_fork_parent(void)
	630	{
	631	call_rcu_unlock(&call_rcu_mutex);
	632	}
	633
7106ddf8 PM	634	/*
7106ddf8 PM	635	* Clean up call_rcu data structures in the child of a successful fork()
81ad2e19 PM	636	* that is not followed by exec(). Suitable for pthread_atfork() and
81ad2e19 PM	637	* friends.
7106ddf8 PM	638	*/
	639	void call_rcu_after_fork_child(void)
	640	{
	641	struct call_rcu_data *crdp;
	642
81ad2e19 PM	643	/* Release the mutex. */
	644	call_rcu_unlock(&call_rcu_mutex);
	645
7106ddf8 PM	646	/*
	647	* Allocate a new default call_rcu_data structure in order
	648	* to get a working call_rcu thread to go with it.
	649	*/
	650	default_call_rcu_data = NULL;
	651	(void)get_default_call_rcu_data();
	652
	653	/* Dispose of all of the rest of the call_rcu_data structures. */
	654	while (call_rcu_data_list.next != call_rcu_data_list.prev) {
	655	crdp = cds_list_entry(call_rcu_data_list.prev,
	656	struct call_rcu_data, list);
	657	if (crdp == default_call_rcu_data)
	658	crdp = cds_list_entry(crdp->list.prev,
	659	struct call_rcu_data, list);
	660	crdp->flags = URCU_CALL_RCU_STOPPED;
	661	call_rcu_data_free(crdp);
b57aee66 PM	662	}
b57aee66 PM	663	}