[ltt-dev] [PATCH 00/10] call_rcu: futex wakeup and miscellaneous improvements

[ltt-dev] [PATCH 00/10] call_rcu: futex wakeup and miscellaneous improvements

[Paolo Bonzini]

Hi,

this series redoes the futex implementation so that it does not busily
wait for the list to be filled.

I'm reusing the flags field as the futex variable, so I'm at the same
time converting it from mutex-protected to atomic (patch 5) and adding
more atomic operations to the uatomic library (patches 3, 4, 6).

Patches 1 and 2 are unrelated bugfixes.  Patch 7 is a simple
cleanup in the call_rcu areas that I was touching anyway.

Patch 8 is the actual futex implementation, with patches 9 and 10 further
optimizing the readers to avoid system calls when a call_rcu is already
scheduled; I separated them to simplify bisection in case things go wrong.

Paolo Bonzini (10):
  urcu-qsbr: fix typo
  rcutorture: make goflag volatile
  uatomic: fix typo in x86 compat implementation
  uatomic: add uatomic_or
  call_rcu: drop mutex
  uatomic: add uatomic_and
  call_rcu: remove write-only qlen variable
  call_rcu: redo futex implementation
  call_rcu: factor polling from RT and non-RT cases
  call_rcu: avoid useless futex wakeups

 compat_arch_x86.c       |   50 ++++++++++++
 tests/rcutorture.h      |    3 +-
 tests/test_uatomic.c    |    8 ++-
 urcu-call-rcu-impl.h    |   84 +++++++++------------
 urcu-call-rcu.h         |    1 +
 urcu-qsbr.h             |    2 +-
 urcu/uatomic_arch_x86.h |  130 +++++++++++++++++++++++++++++++-
 urcu/uatomic_generic.h  |  194 +++++++++++++++++++++++++++++++++++++++++++++++
 8 files changed, 418 insertions(+), 54 deletions(-)

-- 
1.7.4.4

[ltt-dev] [PATCH 01/10] urcu-qsbr: fix typo

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-qsbr.h |    2 +-
 1 files changed, 1 insertions(+), 1 deletions(-)

diff --git a/urcu-qsbr.h b/urcu-qsbr.h
index a691c52..75ea14c 100644
--- a/urcu-qsbr.h
+++ b/urcu-qsbr.h
@@ -95,7 +95,7 @@ static inline void rcu_read_lock(void)
 {
 }
 
-static inline void rcu_read_lock(void)
+static inline void rcu_read_unlock(void)
 {
 }
 
-- 
1.7.4.4

[ltt-dev] [PATCH 02/10] rcutorture: make goflag volatile

[Paolo Bonzini]

Even cmm_barrier may not be enough as a compiler barrier in the presence
of static variables, and cmm_mb would not be either!  The compiler assumes
that calling a function does not clobber static variables if it can prove
that they do not escape (which also implies the correct phase of the moon).

Rather than sprinkling the code with compiler barriers, I am taking the
easy way out and declaring the variable volatile.  Without this patch,
rcutorture_qsbr never finishes for me (GCC 4.5.0).

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 tests/rcutorture.h |    3 ++-
 1 files changed, 2 insertions(+), 1 deletions(-)

diff --git a/tests/rcutorture.h b/tests/rcutorture.h
index c5253d9..5ba3d2b 100644
--- a/tests/rcutorture.h
+++ b/tests/rcutorture.h
@@ -79,7 +79,8 @@ char argsbuf[64];
 #define GOFLAG_RUN  1
 #define GOFLAG_STOP 2
 
-int goflag __attribute__((__aligned__(CAA_CACHE_LINE_SIZE))) = GOFLAG_INIT;
+volatile int goflag __attribute__((__aligned__(CAA_CACHE_LINE_SIZE)))
+        = GOFLAG_INIT;
 
 #define RCU_READ_RUN 1000
 
-- 
1.7.4.4

[ltt-dev] [PATCH 03/10] uatomic: fix typo in x86 compat implementation

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu/uatomic_arch_x86.h |    4 ++--
 1 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
index aed513b..30418c8 100644
--- a/urcu/uatomic_arch_x86.h
+++ b/urcu/uatomic_arch_x86.h
@@ -428,8 +428,8 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
 						(unsigned long)(_new), 	       \
 						sizeof(*(addr))))
 
-extern unsigned long _compat_uatomic_xchg(void *addr,
-					  unsigned long _new, int len);
+extern unsigned long _compat_uatomic_add_return(void *addr,
+					        unsigned long _new, int len);
 #define compat_uatomic_add_return(addr, v)				       \
 	((__typeof__(*(addr))) _compat_uatomic_add_return((addr),	       \
 						(unsigned long)(v), 	       \
-- 
1.7.4.4

[ltt-dev] [PATCH 04/10] uatomic: add uatomic_or

[Paolo Bonzini]

For now, only the version returning void is used.  Also, I am not
providing PPC versions because I would be unable to test it right
now.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 compat_arch_x86.c       |   25 ++++++++++++
 tests/test_uatomic.c    |    6 ++-
 urcu/uatomic_arch_x86.h |   63 ++++++++++++++++++++++++++++++
 urcu/uatomic_generic.h  |   97 +++++++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 189 insertions(+), 2 deletions(-)

diff --git a/compat_arch_x86.c b/compat_arch_x86.c
index 5342c7b..33bf13d 100644
--- a/compat_arch_x86.c
+++ b/compat_arch_x86.c
@@ -201,6 +201,31 @@ unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
 	return retval;
 }
 
+void _compat_uatomic_or(void *addr, unsigned long v, int len)
+{
+	sigset_t mask;
+
+	mutex_lock_signal_save(&compat_mutex, &mask);
+	switch (len) {
+	case 1:
+		*(unsigned char *)addr |= (unsigned char)v;
+		break;
+	case 2:
+		*(unsigned short *)addr |= (unsigned short)v;
+		break;
+	case 4:
+		*(unsigned int *)addr |= (unsigned int)v;
+		break;
+	default:
+		/*
+		 * generate an illegal instruction. Cannot catch this with
+		 * linker tricks when optimizations are disabled.
+		 */
+		__asm__ __volatile__("ud2");
+	}
+	mutex_lock_signal_restore(&compat_mutex, &mask);
+}
+
 unsigned long _compat_uatomic_add_return(void *addr, unsigned long v, int len)
 {
 	sigset_t mask;
diff --git a/tests/test_uatomic.c b/tests/test_uatomic.c
index 5682655..37f95a6 100644
--- a/tests/test_uatomic.c
+++ b/tests/test_uatomic.c
@@ -33,8 +33,10 @@ do {						\
 	assert(uatomic_read(ptr) == 23);		\
 	uatomic_dec(ptr);			\
 	assert(uatomic_read(ptr) == 22);		\
-	v = uatomic_add_return(ptr, 100);	\
-	assert(v == 122);			\
+	v = uatomic_add_return(ptr, 74);	\
+	assert(v == 96);			\
+	assert(uatomic_read(ptr) == 96);	\
+	uatomic_or(ptr, 58);			\
 	assert(uatomic_read(ptr) == 122);	\
 	v = uatomic_sub_return(ptr, 1);		\
 	assert(v == 121);			\
diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
index 30418c8..c3b5333 100644
--- a/urcu/uatomic_arch_x86.h
+++ b/urcu/uatomic_arch_x86.h
@@ -231,6 +231,60 @@ unsigned long __uatomic_add_return(void *addr, unsigned long val,
 						  (unsigned long)(v),	\
 						  sizeof(*(addr))))
 
+/* uatomic_or */
+
+static inline __attribute__((always_inline))
+void __uatomic_or(void *addr, unsigned long val, int len)
+{
+	switch (len) {
+	case 1:
+	{
+		__asm__ __volatile__(
+		"lock; orb %1, %0"
+			: "=m"(*__hp(addr))
+			: "iq" ((unsigned char)val)
+			: "memory");
+		return;
+	}
+	case 2:
+	{
+		__asm__ __volatile__(
+		"lock; orw %1, %0"
+			: "=m"(*__hp(addr))
+			: "ir" ((unsigned short)val)
+			: "memory");
+		return;
+	}
+	case 4:
+	{
+		__asm__ __volatile__(
+		"lock; orl %1, %0"
+			: "=m"(*__hp(addr))
+			: "ir" ((unsigned int)val)
+			: "memory");
+		return;
+	}
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+	{
+		__asm__ __volatile__(
+		"lock; orq %1, %0"
+			: "=m"(*__hp(addr))
+			: "er" ((unsigned long)val)
+			: "memory");
+		return;
+	}
+#endif
+	}
+	/* generate an illegal instruction. Cannot catch this with linker tricks
+	 * when optimizations are disabled. */
+	__asm__ __volatile__("ud2");
+	return;
+}
+
+#define _uatomic_or(addr, v)						   \
+	(__uatomic_or((addr), (unsigned long)(v), sizeof(*(addr))))
+
 /* uatomic_add */
 
 static inline __attribute__((always_inline))
@@ -428,6 +482,13 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
 						(unsigned long)(_new), 	       \
 						sizeof(*(addr))))
 
+extern unsigned long _compat_uatomic_or(void *addr,
+				        unsigned long _new, int len);
+#define compat_uatomic_or(addr, v)				       \
+	((__typeof__(*(addr))) _compat_uatomic_or((addr),	       \
+						  (unsigned long)(v),  \
+						  sizeof(*(addr))))
+
 extern unsigned long _compat_uatomic_add_return(void *addr,
 					        unsigned long _new, int len);
 #define compat_uatomic_add_return(addr, v)				       \
@@ -454,6 +515,8 @@ extern unsigned long _compat_uatomic_add_return(void *addr,
 		UATOMIC_COMPAT(cmpxchg(addr, old, _new))
 #define uatomic_xchg(addr, v)			\
 		UATOMIC_COMPAT(xchg(addr, v))
+#define uatomic_or(addr, v)		\
+		UATOMIC_COMPAT(or(addr, v))
 #define uatomic_add_return(addr, v)		\
 		UATOMIC_COMPAT(add_return(addr, v))
 
diff --git a/urcu/uatomic_generic.h b/urcu/uatomic_generic.h
index 347e73f..556846f 100644
--- a/urcu/uatomic_generic.h
+++ b/urcu/uatomic_generic.h
@@ -87,6 +87,39 @@ unsigned long _uatomic_cmpxchg(void *addr, unsigned long old,
 						sizeof(*(addr))))
 
 
+/* uatomic_or */
+
+#ifndef uatomic_or
+static inline __attribute__((always_inline))
+void _uatomic_or(void *addr, unsigned long val,
+		 int len)
+{
+	switch (len) {
+#ifdef UATOMIC_HAS_ATOMIC_BYTE
+	case 1:
+		__sync_or_and_fetch_1(addr, val);
+#endif
+#ifdef UATOMIC_HAS_ATOMIC_SHORT
+	case 2:
+		__sync_or_and_fetch_2(addr, val);
+#endif
+	case 4:
+		__sync_or_and_fetch_4(addr, val);
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+		__sync_or_and_fetch_8(addr, val);
+#endif
+	}
+	_uatomic_link_error();
+	return 0;
+}
+
+#define uatomic_or(addr, v)			\
+	(_uatomic_or((addr),			\
+		     (unsigned long)(v),	\
+		     sizeof(*(addr))))
+#endif
+
 /* uatomic_add_return */
 
 #ifndef uatomic_add_return
@@ -186,6 +219,70 @@ unsigned long _uatomic_exchange(void *addr, unsigned long val, int len)
 
 #else /* #ifndef uatomic_cmpxchg */
 
+#ifndef uatomic_or
+/* uatomic_or */
+
+static inline __attribute__((always_inline))
+void _uatomic_or(void *addr, unsigned long val, int len)
+{
+	switch (len) {
+#ifdef UATOMIC_HAS_ATOMIC_BYTE
+	case 1:
+	{
+		unsigned char old, oldt;
+
+		oldt = uatomic_read((unsigned char *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old | val, 1);
+		} while (oldt != old);
+	}
+#endif
+#ifdef UATOMIC_HAS_ATOMIC_SHORT
+	case 2:
+	{
+		unsigned short old, oldt;
+
+		oldt = uatomic_read((unsigned short *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old | val, 2);
+		} while (oldt != old);
+	}
+#endif
+	case 4:
+	{
+		unsigned int old, oldt;
+
+		oldt = uatomic_read((unsigned int *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old | val, 4);
+		} while (oldt != old);
+	}
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+	{
+		unsigned long old, oldt;
+
+		oldt = uatomic_read((unsigned long *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old | val, 8);
+		} while (oldt != old);
+	}
+#endif
+	}
+	_uatomic_link_error();
+	return 0;
+}
+
+#define uatomic_or(addr, v)		\
+	(uatomic_or((addr),		\
+		    (unsigned long)(v),	\
+		    sizeof(*(addr))))
+#endif /* #ifndef uatomic_or */
+
 #ifndef uatomic_add_return
 /* uatomic_add_return */
 
-- 
1.7.4.4

[ltt-dev] [PATCH 05/10] call_rcu: drop mutex

[Paolo Bonzini]

The mutex is being used only to protect OR accesses to the flags.
Just use atomic operations for that.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |   23 +++++++----------------
 1 files changed, 7 insertions(+), 16 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 69edd49..7e6acdd 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -46,7 +46,6 @@
 struct call_rcu_data {
 	struct cds_wfq_queue cbs;
 	unsigned long flags;
-	pthread_mutex_t mtx;
 	int futex;
 	unsigned long qlen;
 	pthread_t tid;
@@ -235,9 +234,9 @@ static void *call_rcu_thread(void *arg)
 			} while (cbs != NULL);
 			uatomic_sub(&crdp->qlen, cbcount);
 		}
-		if (crdp->flags & URCU_CALL_RCU_STOP)
+		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 			break;
-		if (crdp->flags & URCU_CALL_RCU_RT)
+		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT)
 			poll(NULL, 0, 10);
 		else {
 			if (&crdp->cbs.head == _CMM_LOAD_SHARED(crdp->cbs.tail))
@@ -245,9 +244,7 @@ static void *call_rcu_thread(void *arg)
 			poll(NULL, 0, 10);
 		}
 	}
-	call_rcu_lock(&crdp->mtx);
-	crdp->flags |= URCU_CALL_RCU_STOPPED;
-	call_rcu_unlock(&crdp->mtx);
+	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
 	return NULL;
 }
 
@@ -271,10 +268,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 	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);
@@ -557,12 +550,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 	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);
+	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
+		uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
 		wake_call_rcu_thread(crdp);
-		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
+		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
 			poll(NULL, 0, 1);
 	}
 	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
@@ -646,7 +637,7 @@ void call_rcu_after_fork_child(void)
 		if (crdp == default_call_rcu_data)
 			crdp = cds_list_entry(crdp->list.prev,
 					      struct call_rcu_data, list);
-		crdp->flags = URCU_CALL_RCU_STOPPED;
+		uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
 		call_rcu_data_free(crdp);
 	}
 }
-- 
1.7.4.4

[ltt-dev] [PATCH 06/10] uatomic: add uatomic_and

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 compat_arch_x86.c       |   25 ++++++++++++
 tests/test_uatomic.c    |    2 +
 urcu/uatomic_arch_x86.h |   63 ++++++++++++++++++++++++++++++
 urcu/uatomic_generic.h  |   97 +++++++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 187 insertions(+), 0 deletions(-)

diff --git a/compat_arch_x86.c b/compat_arch_x86.c
index 33bf13d..692417e 100644
--- a/compat_arch_x86.c
+++ b/compat_arch_x86.c
@@ -226,6 +226,31 @@ void _compat_uatomic_or(void *addr, unsigned long v, int len)
 	mutex_lock_signal_restore(&compat_mutex, &mask);
 }
 
+void _compat_uatomic_and(void *addr, unsigned long v, int len)
+{
+	sigset_t mask;
+
+	mutex_lock_signal_save(&compat_mutex, &mask);
+	switch (len) {
+	case 1:
+		*(unsigned char *)addr &= (unsigned char)v;
+		break;
+	case 2:
+		*(unsigned short *)addr &= (unsigned short)v;
+		break;
+	case 4:
+		*(unsigned int *)addr &= (unsigned int)v;
+		break;
+	default:
+		/*
+		 * generate an illegal instruction. Cannot catch this with
+		 * linker tricks when optimizations are disabled.
+		 */
+		__asm__ __volatile__("ud2");
+	}
+	mutex_lock_signal_restore(&compat_mutex, &mask);
+}
+
 unsigned long _compat_uatomic_add_return(void *addr, unsigned long v, int len)
 {
 	sigset_t mask;
diff --git a/tests/test_uatomic.c b/tests/test_uatomic.c
index 37f95a6..2c8c232 100644
--- a/tests/test_uatomic.c
+++ b/tests/test_uatomic.c
@@ -41,6 +41,8 @@ do {						\
 	v = uatomic_sub_return(ptr, 1);		\
 	assert(v == 121);			\
 	assert(uatomic_read(ptr) == 121);	\
+	uatomic_and(ptr, 129);			\
+	assert(uatomic_read(ptr) == 1);		\
 } while (0)
 
 int main(int argc, char **argv)
diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
index c3b5333..c861208 100644
--- a/urcu/uatomic_arch_x86.h
+++ b/urcu/uatomic_arch_x86.h
@@ -231,6 +231,60 @@ unsigned long __uatomic_add_return(void *addr, unsigned long val,
 						  (unsigned long)(v),	\
 						  sizeof(*(addr))))
 
+/* uatomic_and */
+
+static inline __attribute__((always_inline))
+void __uatomic_and(void *addr, unsigned long val, int len)
+{
+	switch (len) {
+	case 1:
+	{
+		__asm__ __volatile__(
+		"lock; andb %1, %0"
+			: "=m"(*__hp(addr))
+			: "iq" ((unsigned char)val)
+			: "memory");
+		return;
+	}
+	case 2:
+	{
+		__asm__ __volatile__(
+		"lock; andw %1, %0"
+			: "=m"(*__hp(addr))
+			: "ir" ((unsigned short)val)
+			: "memory");
+		return;
+	}
+	case 4:
+	{
+		__asm__ __volatile__(
+		"lock; andl %1, %0"
+			: "=m"(*__hp(addr))
+			: "ir" ((unsigned int)val)
+			: "memory");
+		return;
+	}
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+	{
+		__asm__ __volatile__(
+		"lock; andq %1, %0"
+			: "=m"(*__hp(addr))
+			: "er" ((unsigned long)val)
+			: "memory");
+		return;
+	}
+#endif
+	}
+	/* generate an illegal instruction. Cannot catch this with linker tricks
+	 * when optimizations are disabled. */
+	__asm__ __volatile__("ud2");
+	return;
+}
+
+#define _uatomic_and(addr, v)						   \
+	(__uatomic_and((addr), (unsigned long)(v), sizeof(*(addr))))
+
 /* uatomic_or */
 
 static inline __attribute__((always_inline))
@@ -482,6 +536,13 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
 						(unsigned long)(_new), 	       \
 						sizeof(*(addr))))
 
+extern unsigned long _compat_uatomic_and(void *addr,
+				         unsigned long _new, int len);
+#define compat_uatomic_and(addr, v)				       \
+	((__typeof__(*(addr))) _compat_uatomic_and((addr),	       \
+						   (unsigned long)(v), \
+						   sizeof(*(addr))))
+
 extern unsigned long _compat_uatomic_or(void *addr,
 				        unsigned long _new, int len);
 #define compat_uatomic_or(addr, v)				       \
@@ -515,6 +576,8 @@ extern unsigned long _compat_uatomic_add_return(void *addr,
 		UATOMIC_COMPAT(cmpxchg(addr, old, _new))
 #define uatomic_xchg(addr, v)			\
 		UATOMIC_COMPAT(xchg(addr, v))
+#define uatomic_and(addr, v)		\
+		UATOMIC_COMPAT(and(addr, v))
 #define uatomic_or(addr, v)		\
 		UATOMIC_COMPAT(or(addr, v))
 #define uatomic_add_return(addr, v)		\
diff --git a/urcu/uatomic_generic.h b/urcu/uatomic_generic.h
index 556846f..cef58f3 100644
--- a/urcu/uatomic_generic.h
+++ b/urcu/uatomic_generic.h
@@ -87,6 +87,39 @@ unsigned long _uatomic_cmpxchg(void *addr, unsigned long old,
 						sizeof(*(addr))))
 
 
+/* uatomic_and */
+
+#ifndef uatomic_and
+static inline __attribute__((always_inline))
+void _uatomic_and(void *addr, unsigned long val,
+		  int len)
+{
+	switch (len) {
+#ifdef UATOMIC_HAS_ATOMIC_BYTE
+	case 1:
+		__sync_and_and_fetch_1(addr, val);
+#endif
+#ifdef UATOMIC_HAS_ATOMIC_SHORT
+	case 2:
+		__sync_and_and_fetch_2(addr, val);
+#endif
+	case 4:
+		__sync_and_and_fetch_4(addr, val);
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+		__sync_and_and_fetch_8(addr, val);
+#endif
+	}
+	_uatomic_link_error();
+	return 0;
+}
+
+#define uatomic_and(addr, v)			\
+	(_uatomic_and((addr),			\
+		      (unsigned long)(v),	\
+		      sizeof(*(addr))))
+#endif
+
 /* uatomic_or */
 
 #ifndef uatomic_or
@@ -219,6 +252,70 @@ unsigned long _uatomic_exchange(void *addr, unsigned long val, int len)
 
 #else /* #ifndef uatomic_cmpxchg */
 
+#ifndef uatomic_and
+/* uatomic_and */
+
+static inline __attribute__((always_inline))
+void _uatomic_and(void *addr, unsigned long val, int len)
+{
+	switch (len) {
+#ifdef UATOMIC_HAS_ATOMIC_BYTE
+	case 1:
+	{
+		unsigned char old, oldt;
+
+		oldt = uatomic_read((unsigned char *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old & val, 1);
+		} while (oldt != old);
+	}
+#endif
+#ifdef UATOMIC_HAS_ATOMIC_SHORT
+	case 2:
+	{
+		unsigned short old, oldt;
+
+		oldt = uatomic_read((unsigned short *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old & val, 2);
+		} while (oldt != old);
+	}
+#endif
+	case 4:
+	{
+		unsigned int old, oldt;
+
+		oldt = uatomic_read((unsigned int *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old & val, 4);
+		} while (oldt != old);
+	}
+#if (CAA_BITS_PER_LONG == 64)
+	case 8:
+	{
+		unsigned long old, oldt;
+
+		oldt = uatomic_read((unsigned long *)addr);
+		do {
+			old = oldt;
+			oldt = _uatomic_cmpxchg(addr, old, old & val, 8);
+		} while (oldt != old);
+	}
+#endif
+	}
+	_uatomic_link_error();
+	return 0;
+}
+
+#define uatomic_and(addr, v)		\
+	(uatomic_and((addr),		\
+		    (unsigned long)(v),	\
+		    sizeof(*(addr))))
+#endif /* #ifndef uatomic_and */
+
 #ifndef uatomic_or
 /* uatomic_or */
 
-- 
1.7.4.4

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Paolo Bonzini]

The qlen variable is write-only, we can remove it easily.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |    9 ---------
 1 files changed, 0 insertions(+), 9 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 7e6acdd..ca597d0 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -47,7 +47,6 @@ struct call_rcu_data {
 	struct cds_wfq_queue cbs;
 	unsigned long flags;
 	int futex;
-	unsigned long qlen;
 	pthread_t tid;
 	int cpu_affinity;
 	struct cds_list_head list;
@@ -198,7 +197,6 @@ int set_thread_cpu_affinity(struct call_rcu_data *crdp)
 
 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;
@@ -218,7 +216,6 @@ static void *call_rcu_thread(void *arg)
 			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)
@@ -230,9 +227,7 @@ static void *call_rcu_thread(void *arg)
 				rhp = (struct rcu_head *)cbs;
 				cbs = cbs->next;
 				rhp->func(rhp);
-				cbcount++;
 			} while (cbs != NULL);
-			uatomic_sub(&crdp->qlen, cbcount);
 		}
 		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 			break;
@@ -267,7 +262,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 	}
 	memset(crdp, '\0', sizeof(*crdp));
 	cds_wfq_init(&crdp->cbs);
-	crdp->qlen = 0;
 	crdp->futex = 0;
 	crdp->flags = flags;
 	cds_list_add(&crdp->list, &call_rcu_data_list);
@@ -519,7 +513,6 @@ void call_rcu(struct rcu_head *head,
 	head->func = func;
 	crdp = get_call_rcu_data();
 	cds_wfq_enqueue(&crdp->cbs, &head->next);
-	uatomic_inc(&crdp->qlen);
 	wake_call_rcu_thread(crdp);
 }
 
@@ -565,8 +558,6 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 		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);
 	}
-- 
1.7.4.4

[ltt-dev] [PATCH 08/10] call_rcu: redo futex implementation

[Paolo Bonzini]

The current futex implementation is actually busy waiting on the
list, because futex is never set to -1.  This one instead piggybacks
on the flags field, so that the futex can also be used for the STOP flag.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |   43 +++++++++++++++++++++++++------------------
 urcu-call-rcu.h      |    1 +
 2 files changed, 26 insertions(+), 18 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index ca597d0..9824515 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -45,8 +45,7 @@
 
 struct call_rcu_data {
 	struct cds_wfq_queue cbs;
-	unsigned long flags;
-	int futex;
+	int flags;
 	pthread_t tid;
 	int cpu_affinity;
 	struct cds_list_head list;
@@ -89,22 +88,18 @@ 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);
+	int flags;
+	for (;;) {
+		flags = uatomic_read(&crdp->flags);
+		if (flags & (URCU_CALL_RCU_BUSY | URCU_CALL_RCU_STOP))
+			break;
+		futex_async(&crdp->flags, FUTEX_WAIT, flags, NULL, 0, 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);
-	}
+	futex_async(&crdp->flags, FUTEX_WAKE, 1, NULL, NULL, 0);
 }
 
 /* Allocate the array if it has not already been allocated. */
@@ -209,7 +204,16 @@ static void *call_rcu_thread(void *arg)
 
 	thread_call_rcu_data = crdp;
 	for (;;) {
-		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
+		for (;;) {
+			if (&crdp->cbs.head
+			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
+				uatomic_and(&crdp->flags, ~URCU_CALL_RCU_BUSY);
+				/* Write flags before rechecking the list.  */
+				cmm_smp_mb();
+				if (&crdp->cbs.head
+				    == _CMM_LOAD_SHARED(crdp->cbs.tail))
+					break;
+			}
 			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
 				poll(NULL, 0, 1);
 			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
@@ -229,13 +233,14 @@ static void *call_rcu_thread(void *arg)
 				rhp->func(rhp);
 			} while (cbs != NULL);
 		}
+		/* Read call_rcu list before reading the flags.  */
+		cmm_smp_mb();
 		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 			break;
 		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT)
 			poll(NULL, 0, 10);
 		else {
-			if (&crdp->cbs.head == _CMM_LOAD_SHARED(crdp->cbs.tail))
-				call_rcu_wait(crdp);
+			call_rcu_wait(crdp);
 			poll(NULL, 0, 10);
 		}
 	}
@@ -262,7 +267,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 	}
 	memset(crdp, '\0', sizeof(*crdp));
 	cds_wfq_init(&crdp->cbs);
-	crdp->futex = 0;
 	crdp->flags = flags;
 	cds_list_add(&crdp->list, &call_rcu_data_list);
 	crdp->cpu_affinity = cpu_affinity;
@@ -513,6 +517,9 @@ void call_rcu(struct rcu_head *head,
 	head->func = func;
 	crdp = get_call_rcu_data();
 	cds_wfq_enqueue(&crdp->cbs, &head->next);
+	/* Write list before writing the flags.  */
+	cmm_smp_mb();
+	uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
 	wake_call_rcu_thread(crdp);
 }
 
diff --git a/urcu-call-rcu.h b/urcu-call-rcu.h
index e76a844..bfd56ec 100644
--- a/urcu-call-rcu.h
+++ b/urcu-call-rcu.h
@@ -48,6 +48,7 @@ struct call_rcu_data;
 #define URCU_CALL_RCU_RUNNING	0x2
 #define URCU_CALL_RCU_STOP	0x4
 #define URCU_CALL_RCU_STOPPED	0x8
+#define URCU_CALL_RCU_BUSY	0x10
 
 /*
  * The rcu_head data structure is placed in the structure to be freed
-- 
1.7.4.4

[ltt-dev] [PATCH 09/10] call_rcu: factor polling from RT and non-RT cases

[Paolo Bonzini]

This has the additional advantage of introducing a small delay before
consecutive checks of the list.  In case call_rcu is very busy, this and
the following patch practically eliminate the difference between the RT
and non-RT version.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |    7 ++-----
 1 files changed, 2 insertions(+), 5 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 9824515..d8570e3 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -205,6 +205,7 @@ static void *call_rcu_thread(void *arg)
 	thread_call_rcu_data = crdp;
 	for (;;) {
 		for (;;) {
+			poll(NULL, 0, 10);
 			if (&crdp->cbs.head
 			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
 				uatomic_and(&crdp->flags, ~URCU_CALL_RCU_BUSY);
@@ -237,12 +238,8 @@ static void *call_rcu_thread(void *arg)
 		cmm_smp_mb();
 		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 			break;
-		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT)
-			poll(NULL, 0, 10);
-		else {
+		if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT) == 0)
 			call_rcu_wait(crdp);
-			poll(NULL, 0, 10);
-		}
 	}
 	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
 	return NULL;
-- 
1.7.4.4

[ltt-dev] [PATCH 10/10] call_rcu: avoid useless futex wakeups

[Paolo Bonzini]

If the call_rcu thread is waiting for entries to accumulate, it is
not necessary to wake it up.  Use the BUSY flag to detect this.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |    8 +++++---
 1 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index d8570e3..165956e 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -514,10 +514,12 @@ void call_rcu(struct rcu_head *head,
 	head->func = func;
 	crdp = get_call_rcu_data();
 	cds_wfq_enqueue(&crdp->cbs, &head->next);
-	/* Write list before writing the flags.  */
+	/* Write list before checking/writing the flags.  */
 	cmm_smp_mb();
-	uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
-	wake_call_rcu_thread(crdp);
+	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_BUSY) == 0) {
+		uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
+		wake_call_rcu_thread(crdp);
+	}
 }
 
 /*
-- 
1.7.4.4

[ltt-dev] [PATCH] call_rcu: keep BUSY flag set as long as possible

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
        Not a correctness issue, and it will just save a syscall in
        rare cases.  Can be committed separately, or squashed in patch 8
        (that would be my favorite option), or left out altogether.

 urcu-call-rcu-impl.h |    3 +++
 1 files changed, 3 insertions(+), 0 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 165956e..c45928a 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -214,6 +214,9 @@ static void *call_rcu_thread(void *arg)
 				if (&crdp->cbs.head
 				    == _CMM_LOAD_SHARED(crdp->cbs.tail))
 					break;
+                                /* False alarm... another bunch is ready.  */
+				cmm_smp_mb();
+				uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
 			}
 			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
 				poll(NULL, 0, 1);
-- 
1.7.4.4

[ltt-dev] [PATCH 01/10] urcu-qsbr: fix typo

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>

Good catch, patch merged, thanks!

Mathieu

> ---
>  urcu-qsbr.h |    2 +-
>  1 files changed, 1 insertions(+), 1 deletions(-)
> 
> diff --git a/urcu-qsbr.h b/urcu-qsbr.h
> index a691c52..75ea14c 100644
> --- a/urcu-qsbr.h
> +++ b/urcu-qsbr.h
> @@ -95,7 +95,7 @@ static inline void rcu_read_lock(void)
>  {
>  }
>  
> -static inline void rcu_read_lock(void)
> +static inline void rcu_read_unlock(void)
>  {
>  }
>  
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 02/10] rcutorture: make goflag volatile

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> Even cmm_barrier may not be enough as a compiler barrier in the presence
> of static variables, and cmm_mb would not be either!  The compiler assumes
> that calling a function does not clobber static variables if it can prove
> that they do not escape (which also implies the correct phase of the moon).
> 
> Rather than sprinkling the code with compiler barriers, I am taking the
> easy way out and declaring the variable volatile.  Without this patch,
> rcutorture_qsbr never finishes for me (GCC 4.5.0).

merged, thanks!

Mathieu

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  tests/rcutorture.h |    3 ++-
>  1 files changed, 2 insertions(+), 1 deletions(-)
> 
> diff --git a/tests/rcutorture.h b/tests/rcutorture.h
> index c5253d9..5ba3d2b 100644
> --- a/tests/rcutorture.h
> +++ b/tests/rcutorture.h
> @@ -79,7 +79,8 @@ char argsbuf[64];
>  #define GOFLAG_RUN  1
>  #define GOFLAG_STOP 2
>  
> -int goflag __attribute__((__aligned__(CAA_CACHE_LINE_SIZE))) = GOFLAG_INIT;
> +volatile int goflag __attribute__((__aligned__(CAA_CACHE_LINE_SIZE)))
> +        = GOFLAG_INIT;
>  
>  #define RCU_READ_RUN 1000
>  
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 03/10] uatomic: fix typo in x86 compat implementation

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> 

Merged, thanks!

Mathieu

> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu/uatomic_arch_x86.h |    4 ++--
>  1 files changed, 2 insertions(+), 2 deletions(-)
> 
> diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
> index aed513b..30418c8 100644
> --- a/urcu/uatomic_arch_x86.h
> +++ b/urcu/uatomic_arch_x86.h
> @@ -428,8 +428,8 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
>  						(unsigned long)(_new), 	       \
>  						sizeof(*(addr))))
>  
> -extern unsigned long _compat_uatomic_xchg(void *addr,
> -					  unsigned long _new, int len);
> +extern unsigned long _compat_uatomic_add_return(void *addr,
> +					        unsigned long _new, int len);
>  #define compat_uatomic_add_return(addr, v)				       \
>  	((__typeof__(*(addr))) _compat_uatomic_add_return((addr),	       \
>  						(unsigned long)(v), 	       \
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> The qlen variable is write-only, we can remove it easily.

I defer to Paul for this one: he likes keeping qlen there so the
debugger can show the list state, which is useful for debugging. I'll
let you two figure out how to come to an agreement. ;)

Mathieu

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |    9 ---------
>  1 files changed, 0 insertions(+), 9 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 7e6acdd..ca597d0 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -47,7 +47,6 @@ struct call_rcu_data {
>  	struct cds_wfq_queue cbs;
>  	unsigned long flags;
>  	int futex;
> -	unsigned long qlen;
>  	pthread_t tid;
>  	int cpu_affinity;
>  	struct cds_list_head list;
> @@ -198,7 +197,6 @@ int set_thread_cpu_affinity(struct call_rcu_data *crdp)
>  
>  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;
> @@ -218,7 +216,6 @@ static void *call_rcu_thread(void *arg)
>  			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)
> @@ -230,9 +227,7 @@ static void *call_rcu_thread(void *arg)
>  				rhp = (struct rcu_head *)cbs;
>  				cbs = cbs->next;
>  				rhp->func(rhp);
> -				cbcount++;
>  			} while (cbs != NULL);
> -			uatomic_sub(&crdp->qlen, cbcount);
>  		}
>  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
>  			break;
> @@ -267,7 +262,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
>  	}
>  	memset(crdp, '\0', sizeof(*crdp));
>  	cds_wfq_init(&crdp->cbs);
> -	crdp->qlen = 0;
>  	crdp->futex = 0;
>  	crdp->flags = flags;
>  	cds_list_add(&crdp->list, &call_rcu_data_list);
> @@ -519,7 +513,6 @@ void call_rcu(struct rcu_head *head,
>  	head->func = func;
>  	crdp = get_call_rcu_data();
>  	cds_wfq_enqueue(&crdp->cbs, &head->next);
> -	uatomic_inc(&crdp->qlen);
>  	wake_call_rcu_thread(crdp);
>  }
>  
> @@ -565,8 +558,6 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
>  		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);
>  	}
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Paul E. McKenney]

On Wed, Jun 08, 2011 at 06:15:30PM -0400, Mathieu Desnoyers wrote:
> * Paolo Bonzini (pbonzini at redhat.com) wrote:
> > The qlen variable is write-only, we can remove it easily.
> 
> I defer to Paul for this one: he likes keeping qlen there so the
> debugger can show the list state, which is useful for debugging. I'll
> let you two figure out how to come to an agreement. ;)

Indeed, I have often used gdb to read ->qlen.  ;-)

							Thanx, Paul

> Mathieu
> 
> > 
> > Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> > ---
> >  urcu-call-rcu-impl.h |    9 ---------
> >  1 files changed, 0 insertions(+), 9 deletions(-)
> > 
> > diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> > index 7e6acdd..ca597d0 100644
> > --- a/urcu-call-rcu-impl.h
> > +++ b/urcu-call-rcu-impl.h
> > @@ -47,7 +47,6 @@ struct call_rcu_data {
> >  	struct cds_wfq_queue cbs;
> >  	unsigned long flags;
> >  	int futex;
> > -	unsigned long qlen;
> >  	pthread_t tid;
> >  	int cpu_affinity;
> >  	struct cds_list_head list;
> > @@ -198,7 +197,6 @@ int set_thread_cpu_affinity(struct call_rcu_data *crdp)
> >  
> >  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;
> > @@ -218,7 +216,6 @@ static void *call_rcu_thread(void *arg)
> >  			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)
> > @@ -230,9 +227,7 @@ static void *call_rcu_thread(void *arg)
> >  				rhp = (struct rcu_head *)cbs;
> >  				cbs = cbs->next;
> >  				rhp->func(rhp);
> > -				cbcount++;
> >  			} while (cbs != NULL);
> > -			uatomic_sub(&crdp->qlen, cbcount);
> >  		}
> >  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
> >  			break;
> > @@ -267,7 +262,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
> >  	}
> >  	memset(crdp, '\0', sizeof(*crdp));
> >  	cds_wfq_init(&crdp->cbs);
> > -	crdp->qlen = 0;
> >  	crdp->futex = 0;
> >  	crdp->flags = flags;
> >  	cds_list_add(&crdp->list, &call_rcu_data_list);
> > @@ -519,7 +513,6 @@ void call_rcu(struct rcu_head *head,
> >  	head->func = func;
> >  	crdp = get_call_rcu_data();
> >  	cds_wfq_enqueue(&crdp->cbs, &head->next);
> > -	uatomic_inc(&crdp->qlen);
> >  	wake_call_rcu_thread(crdp);
> >  }
> >  
> > @@ -565,8 +558,6 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
> >  		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);
> >  	}
> > -- 
> > 1.7.4.4
> > 
> > 
> > 
> > _______________________________________________
> > ltt-dev mailing list
> > ltt-dev at lists.casi.polymtl.ca
> > http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> > 
> 
> -- 
> Mathieu Desnoyers
> Operating System Efficiency R&D Consultant
> EfficiOS Inc.
> http://www.efficios.com

[ltt-dev] [PATCH 08/10] call_rcu: redo futex implementation

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> The current futex implementation is actually busy waiting on the
> list, because futex is never set to -1.  This one instead piggybacks
> on the flags field, so that the futex can also be used for the STOP flag.

Please see if my commit fixes the problem you refer to here. Some
comments below,

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |   43 +++++++++++++++++++++++++------------------
>  urcu-call-rcu.h      |    1 +
>  2 files changed, 26 insertions(+), 18 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index ca597d0..9824515 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -45,8 +45,7 @@
>  
>  struct call_rcu_data {
>  	struct cds_wfq_queue cbs;
> -	unsigned long flags;
> -	int futex;
> +	int flags;
>  	pthread_t tid;
>  	int cpu_affinity;
>  	struct cds_list_head list;
> @@ -89,22 +88,18 @@ 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);
> +	int flags;
> +	for (;;) {
> +		flags = uatomic_read(&crdp->flags);
> +		if (flags & (URCU_CALL_RCU_BUSY | URCU_CALL_RCU_STOP))
> +			break;
> +		futex_async(&crdp->flags, FUTEX_WAIT, flags, NULL, 0, 0);

Hrm, not sure this works. We need to order call_rcu list access wrt
"futex" (or here flag) value accesses. This seems to be missing here.
(uatomic_read does not imply any memory barrier).

I'm tempted to just leave the mutex in place for now until we see a
clearly measurable performance gain by going with the "or/and" flag
approach compared to the simpler (in terms of memory ordering) mutex
implementation.

As far as the or/and implementations you provide in separate patches are
concerned, I'm tempted to pull them for completeness sake (they might
end up being useful). Before I do that though, I'm just curious about
the memory ordering guarantees they should provide: no-ordering seems
like a good thing for or/and: it would be similar to add/sub.
Thoughts ?

Thanks,

Mathieu

> +	}
>  }
>  
>  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);
> -	}
> +	futex_async(&crdp->flags, FUTEX_WAKE, 1, NULL, NULL, 0);
>  }
>  
>  /* Allocate the array if it has not already been allocated. */
> @@ -209,7 +204,16 @@ static void *call_rcu_thread(void *arg)
>  
>  	thread_call_rcu_data = crdp;
>  	for (;;) {
> -		if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> +		for (;;) {
> +			if (&crdp->cbs.head
> +			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> +				uatomic_and(&crdp->flags, ~URCU_CALL_RCU_BUSY);
> +				/* Write flags before rechecking the list.  */
> +				cmm_smp_mb();
> +				if (&crdp->cbs.head
> +				    == _CMM_LOAD_SHARED(crdp->cbs.tail))
> +					break;
> +			}
>  			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
>  				poll(NULL, 0, 1);
>  			_CMM_STORE_SHARED(crdp->cbs.head, NULL);
> @@ -229,13 +233,14 @@ static void *call_rcu_thread(void *arg)
>  				rhp->func(rhp);
>  			} while (cbs != NULL);
>  		}
> +		/* Read call_rcu list before reading the flags.  */
> +		cmm_smp_mb();
>  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
>  			break;
>  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT)
>  			poll(NULL, 0, 10);
>  		else {
> -			if (&crdp->cbs.head == _CMM_LOAD_SHARED(crdp->cbs.tail))
> -				call_rcu_wait(crdp);
> +			call_rcu_wait(crdp);
>  			poll(NULL, 0, 10);
>  		}
>  	}
> @@ -262,7 +267,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
>  	}
>  	memset(crdp, '\0', sizeof(*crdp));
>  	cds_wfq_init(&crdp->cbs);
> -	crdp->futex = 0;
>  	crdp->flags = flags;
>  	cds_list_add(&crdp->list, &call_rcu_data_list);
>  	crdp->cpu_affinity = cpu_affinity;
> @@ -513,6 +517,9 @@ void call_rcu(struct rcu_head *head,
>  	head->func = func;
>  	crdp = get_call_rcu_data();
>  	cds_wfq_enqueue(&crdp->cbs, &head->next);
> +	/* Write list before writing the flags.  */
> +	cmm_smp_mb();
> +	uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
>  	wake_call_rcu_thread(crdp);
>  }
>  
> diff --git a/urcu-call-rcu.h b/urcu-call-rcu.h
> index e76a844..bfd56ec 100644
> --- a/urcu-call-rcu.h
> +++ b/urcu-call-rcu.h
> @@ -48,6 +48,7 @@ struct call_rcu_data;
>  #define URCU_CALL_RCU_RUNNING	0x2
>  #define URCU_CALL_RCU_STOP	0x4
>  #define URCU_CALL_RCU_STOPPED	0x8
> +#define URCU_CALL_RCU_BUSY	0x10
>  
>  /*
>   * The rcu_head data structure is placed in the structure to be freed
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Mathieu Desnoyers]

* Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> On Wed, Jun 08, 2011 at 06:15:30PM -0400, Mathieu Desnoyers wrote:
> > * Paolo Bonzini (pbonzini at redhat.com) wrote:
> > > The qlen variable is write-only, we can remove it easily.
> > 
> > I defer to Paul for this one: he likes keeping qlen there so the
> > debugger can show the list state, which is useful for debugging. I'll
> > let you two figure out how to come to an agreement. ;)
> 
> Indeed, I have often used gdb to read ->qlen.  ;-)

So maybe adding a comment besides qlen stating the reason why it's there
(e.g. for debugging purposes) might be appropriate ?

Thanks,

Mathieu

> 
> 							Thanx, Paul
> 
> > Mathieu
> > 
> > > 
> > > Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> > > ---
> > >  urcu-call-rcu-impl.h |    9 ---------
> > >  1 files changed, 0 insertions(+), 9 deletions(-)
> > > 
> > > diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> > > index 7e6acdd..ca597d0 100644
> > > --- a/urcu-call-rcu-impl.h
> > > +++ b/urcu-call-rcu-impl.h
> > > @@ -47,7 +47,6 @@ struct call_rcu_data {
> > >  	struct cds_wfq_queue cbs;
> > >  	unsigned long flags;
> > >  	int futex;
> > > -	unsigned long qlen;
> > >  	pthread_t tid;
> > >  	int cpu_affinity;
> > >  	struct cds_list_head list;
> > > @@ -198,7 +197,6 @@ int set_thread_cpu_affinity(struct call_rcu_data *crdp)
> > >  
> > >  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;
> > > @@ -218,7 +216,6 @@ static void *call_rcu_thread(void *arg)
> > >  			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)
> > > @@ -230,9 +227,7 @@ static void *call_rcu_thread(void *arg)
> > >  				rhp = (struct rcu_head *)cbs;
> > >  				cbs = cbs->next;
> > >  				rhp->func(rhp);
> > > -				cbcount++;
> > >  			} while (cbs != NULL);
> > > -			uatomic_sub(&crdp->qlen, cbcount);
> > >  		}
> > >  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
> > >  			break;
> > > @@ -267,7 +262,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
> > >  	}
> > >  	memset(crdp, '\0', sizeof(*crdp));
> > >  	cds_wfq_init(&crdp->cbs);
> > > -	crdp->qlen = 0;
> > >  	crdp->futex = 0;
> > >  	crdp->flags = flags;
> > >  	cds_list_add(&crdp->list, &call_rcu_data_list);
> > > @@ -519,7 +513,6 @@ void call_rcu(struct rcu_head *head,
> > >  	head->func = func;
> > >  	crdp = get_call_rcu_data();
> > >  	cds_wfq_enqueue(&crdp->cbs, &head->next);
> > > -	uatomic_inc(&crdp->qlen);
> > >  	wake_call_rcu_thread(crdp);
> > >  }
> > >  
> > > @@ -565,8 +558,6 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
> > >  		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);
> > >  	}
> > > -- 
> > > 1.7.4.4
> > > 
> > > 
> > > 
> > > _______________________________________________
> > > ltt-dev mailing list
> > > ltt-dev at lists.casi.polymtl.ca
> > > http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> > > 
> > 
> > -- 
> > Mathieu Desnoyers
> > Operating System Efficiency R&D Consultant
> > EfficiOS Inc.
> > http://www.efficios.com
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Paul E. McKenney]

On Wed, Jun 08, 2011 at 06:46:20PM -0400, Mathieu Desnoyers wrote:
> * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> > On Wed, Jun 08, 2011 at 06:15:30PM -0400, Mathieu Desnoyers wrote:
> > > * Paolo Bonzini (pbonzini at redhat.com) wrote:
> > > > The qlen variable is write-only, we can remove it easily.
> > > 
> > > I defer to Paul for this one: he likes keeping qlen there so the
> > > debugger can show the list state, which is useful for debugging. I'll
> > > let you two figure out how to come to an agreement. ;)
> > 
> > Indeed, I have often used gdb to read ->qlen.  ;-)
> 
> So maybe adding a comment besides qlen stating the reason why it's there
> (e.g. for debugging purposes) might be appropriate ?

Good point!  How about the following?

							Thanx, Paul

------------------------------------------------------------------------

Add comment to flag purpose of the ->qlen field

It is write-only, but is needed for debugging purposes.

Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index cfe1cce..9beb58c 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -48,7 +48,7 @@ struct call_rcu_data {
 	unsigned long flags;
 	pthread_mutex_t mtx;
 	int futex;
-	unsigned long qlen;
+	unsigned long qlen; /* maintained for debugging. */
 	pthread_t tid;
 	int cpu_affinity;
 	struct cds_list_head list;

[ltt-dev] [PATCH 07/10] call_rcu: remove write-only qlen variable

[Mathieu Desnoyers]

* Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> On Wed, Jun 08, 2011 at 06:46:20PM -0400, Mathieu Desnoyers wrote:
> > * Paul E. McKenney (paulmck at linux.vnet.ibm.com) wrote:
> > > On Wed, Jun 08, 2011 at 06:15:30PM -0400, Mathieu Desnoyers wrote:
> > > > * Paolo Bonzini (pbonzini at redhat.com) wrote:
> > > > > The qlen variable is write-only, we can remove it easily.
> > > > 
> > > > I defer to Paul for this one: he likes keeping qlen there so the
> > > > debugger can show the list state, which is useful for debugging. I'll
> > > > let you two figure out how to come to an agreement. ;)
> > > 
> > > Indeed, I have often used gdb to read ->qlen.  ;-)
> > 
> > So maybe adding a comment besides qlen stating the reason why it's there
> > (e.g. for debugging purposes) might be appropriate ?
> 
> Good point!  How about the following?

Merged, thanks!

Mathieu

> 
> 							Thanx, Paul
> 
> ------------------------------------------------------------------------
> 
> Add comment to flag purpose of the ->qlen field
> 
> It is write-only, but is needed for debugging purposes.
> 
> Signed-off-by: Paul E. McKenney <paulmck at linux.vnet.ibm.com>
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index cfe1cce..9beb58c 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -48,7 +48,7 @@ struct call_rcu_data {
>  	unsigned long flags;
>  	pthread_mutex_t mtx;
>  	int futex;
> -	unsigned long qlen;
> +	unsigned long qlen; /* maintained for debugging. */
>  	pthread_t tid;
>  	int cpu_affinity;
>  	struct cds_list_head list;
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 08/10] call_rcu: redo futex implementation

[Paolo Bonzini]

On 06/09/2011 12:42 AM, Mathieu Desnoyers wrote:
> I'm tempted to just leave the mutex in place for now until we see a
> clearly measurable performance gain by going with the "or/and" flag
> approach compared to the simpler (in terms of memory ordering) mutex
> implementation.

I don't think the memory ordering guarantees are much different:
URCU_CALL_RCU_STOP is always written before a futex system call, and
URCU_CALL_RCU_STOPPED is always written as the last thing before the
thread exits. In both cases there is no need for a memory barrier. So,
the version with atomic accesses seems simpler to me, and I generally
like patches that remove more lines than they add.  Also, the mutex
in code that strives to be lockless (in the RT case) seemed out of
place...

One thing is that the RT/non-RT flag is not modifiable. So perhaps it is
better to move it to a separate variable and limit the uatomic_read noise.
See the attached patch.

> As far as the or/and implementations you provide in separate patches are
> concerned, I'm tempted to pull them for completeness sake (they might
> end up being useful). Before I do that though, I'm just curious about
> the memory ordering guarantees they should provide: no-ordering seems
> like a good thing for or/and: it would be similar to add/sub.

Yes, I agree.

Paolo

------------------ 8< --------------------------

From 9eb4f2821f212705a72ea38568b2a6157665b203 Mon Sep 17 00:00:00 2001
From: Paolo Bonzini <pbonzini@redhat.com>
Date: Wed, 8 Jun 2011 09:33:51 +0200
Subject: [PATCH] call_rcu: drop mutex

The mutex is being used only to protect OR accesses to the flags.
Just use atomic operations for that.

Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
---
 urcu-call-rcu-impl.h |   28 ++++++++++------------------
 1 files changed, 10 insertions(+), 18 deletions(-)

diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
index 9beb58c..82fec80 100644
--- a/urcu-call-rcu-impl.h
+++ b/urcu-call-rcu-impl.h
@@ -46,7 +46,6 @@
 struct call_rcu_data {
 	struct cds_wfq_queue cbs;
 	unsigned long flags;
-	pthread_mutex_t mtx;
 	int futex;
 	unsigned long qlen; /* maintained for debugging. */
 	pthread_t tid;
@@ -204,6 +203,7 @@ static void *call_rcu_thread(void *arg)
 	struct cds_wfq_node **cbs_tail;
 	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
 	struct rcu_head *rhp;
+	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
 
 	if (set_thread_cpu_affinity(crdp) != 0) {
 		perror("pthread_setaffinity_np");
@@ -212,7 +212,7 @@ static void *call_rcu_thread(void *arg)
 
 	thread_call_rcu_data = crdp;
 	for (;;) {
-		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
+		if (!rt) {
 			uatomic_dec(&crdp->futex);
 			/* Decrement futex before reading call_rcu list */
 			cmm_smp_mb();
@@ -240,8 +240,8 @@ static void *call_rcu_thread(void *arg)
 			} while (cbs != NULL);
 			uatomic_sub(&crdp->qlen, cbcount);
 		}
-		if (crdp->flags & URCU_CALL_RCU_STOP) {
-			if (!(crdp->flags & URCU_CALL_RCU_RT)) {
+		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP) {
+			if (!rt) {
 				/*
 				 * Read call_rcu list before write futex.
 				 */
@@ -250,15 +250,13 @@ static void *call_rcu_thread(void *arg)
 			}
 			break;
 		}
-		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
+		if (!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);
+	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
 	return NULL;
 }
 
@@ -282,10 +280,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
 	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);
@@ -568,12 +562,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
 	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);
+	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
+		uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
 		wake_call_rcu_thread(crdp);
-		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
+		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
 			poll(NULL, 0, 1);
 	}
 	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
@@ -657,7 +649,7 @@ void call_rcu_after_fork_child(void)
 		if (crdp == default_call_rcu_data)
 			crdp = cds_list_entry(crdp->list.prev,
 					      struct call_rcu_data, list);
-		crdp->flags = URCU_CALL_RCU_STOPPED;
+		uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
 		call_rcu_data_free(crdp);
 	}
 }
-- 
1.7.4.4

[ltt-dev] [PATCH 08/10] call_rcu: redo futex implementation

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> On 06/09/2011 12:42 AM, Mathieu Desnoyers wrote:
> > I'm tempted to just leave the mutex in place for now until we see a
> > clearly measurable performance gain by going with the "or/and" flag
> > approach compared to the simpler (in terms of memory ordering) mutex
> > implementation.
> 
> I don't think the memory ordering guarantees are much different:
> URCU_CALL_RCU_STOP is always written before a futex system call, and
> URCU_CALL_RCU_STOPPED is always written as the last thing before the
> thread exits. In both cases there is no need for a memory barrier.
> So, the version with atomic accesses seems simpler to me, and I
> generally like patches that remove more lines than they add.

Good point. Also, I notice that your patch fix the flag accesses that
are performed outside of mutex protection (e.g. busy loop waiting for
"stopped"), which busy loop on a non-volatile variable. So yes, your
patch is a clear improvement.

> Also, the mutex
> in code that strives to be lockless (in the RT case) seemed out of
> place...

The "RT" guarantee only apply to the thread calling call_rcu() AFAIK.
The call_rcu worker thread is waiting on a poll() for a certain delay in
RT mode, which makes it entirely non-RT. So the mutex there does not
change the RT/non-RT guarantees provided here.

> 
> One thing is that the RT/non-RT flag is not modifiable. So perhaps it is
> better to move it to a separate variable and limit the uatomic_read noise.
> See the attached patch.

Good point. Will merge.

> 
> > As far as the or/and implementations you provide in separate patches are
> > concerned, I'm tempted to pull them for completeness sake (they might
> > end up being useful). Before I do that though, I'm just curious about
> > the memory ordering guarantees they should provide: no-ordering seems
> > like a good thing for or/and: it would be similar to add/sub.
> 
> Yes, I agree.

Thanks!

Mathieu

> 
> Paolo
> 
> ------------------ 8< --------------------------
> 
> From 9eb4f2821f212705a72ea38568b2a6157665b203 Mon Sep 17 00:00:00 2001
> From: Paolo Bonzini <pbonzini@redhat.com>
> Date: Wed, 8 Jun 2011 09:33:51 +0200
> Subject: [PATCH] call_rcu: drop mutex
> 
> The mutex is being used only to protect OR accesses to the flags.
> Just use atomic operations for that.
> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |   28 ++++++++++------------------
>  1 files changed, 10 insertions(+), 18 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 9beb58c..82fec80 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -46,7 +46,6 @@
>  struct call_rcu_data {
>  	struct cds_wfq_queue cbs;
>  	unsigned long flags;
> -	pthread_mutex_t mtx;
>  	int futex;
>  	unsigned long qlen; /* maintained for debugging. */
>  	pthread_t tid;
> @@ -204,6 +203,7 @@ static void *call_rcu_thread(void *arg)
>  	struct cds_wfq_node **cbs_tail;
>  	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
>  	struct rcu_head *rhp;
> +	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
>  
>  	if (set_thread_cpu_affinity(crdp) != 0) {
>  		perror("pthread_setaffinity_np");
> @@ -212,7 +212,7 @@ static void *call_rcu_thread(void *arg)
>  
>  	thread_call_rcu_data = crdp;
>  	for (;;) {
> -		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (!rt) {
>  			uatomic_dec(&crdp->futex);
>  			/* Decrement futex before reading call_rcu list */
>  			cmm_smp_mb();
> @@ -240,8 +240,8 @@ static void *call_rcu_thread(void *arg)
>  			} while (cbs != NULL);
>  			uatomic_sub(&crdp->qlen, cbcount);
>  		}
> -		if (crdp->flags & URCU_CALL_RCU_STOP) {
> -			if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP) {
> +			if (!rt) {
>  				/*
>  				 * Read call_rcu list before write futex.
>  				 */
> @@ -250,15 +250,13 @@ static void *call_rcu_thread(void *arg)
>  			}
>  			break;
>  		}
> -		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (!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);
> +	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
>  	return NULL;
>  }
>  
> @@ -282,10 +280,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
>  	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);
> @@ -568,12 +562,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
>  	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);
> +	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
> +		uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
>  		wake_call_rcu_thread(crdp);
> -		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
> +		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
>  			poll(NULL, 0, 1);
>  	}
>  	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> @@ -657,7 +649,7 @@ void call_rcu_after_fork_child(void)
>  		if (crdp == default_call_rcu_data)
>  			crdp = cds_list_entry(crdp->list.prev,
>  					      struct call_rcu_data, list);
> -		crdp->flags = URCU_CALL_RCU_STOPPED;
> +		uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
>  		call_rcu_data_free(crdp);
>  	}
>  }
> -- 
> 1.7.4.4
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 04/10] uatomic: add uatomic_or

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> For now, only the version returning void is used.  Also, I am not
> providing PPC versions because I would be unable to test it right
> now.

merged, thanks!

Mathieu

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  compat_arch_x86.c       |   25 ++++++++++++
>  tests/test_uatomic.c    |    6 ++-
>  urcu/uatomic_arch_x86.h |   63 ++++++++++++++++++++++++++++++
>  urcu/uatomic_generic.h  |   97 +++++++++++++++++++++++++++++++++++++++++++++++
>  4 files changed, 189 insertions(+), 2 deletions(-)
> 
> diff --git a/compat_arch_x86.c b/compat_arch_x86.c
> index 5342c7b..33bf13d 100644
> --- a/compat_arch_x86.c
> +++ b/compat_arch_x86.c
> @@ -201,6 +201,31 @@ unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
>  	return retval;
>  }
>  
> +void _compat_uatomic_or(void *addr, unsigned long v, int len)
> +{
> +	sigset_t mask;
> +
> +	mutex_lock_signal_save(&compat_mutex, &mask);
> +	switch (len) {
> +	case 1:
> +		*(unsigned char *)addr |= (unsigned char)v;
> +		break;
> +	case 2:
> +		*(unsigned short *)addr |= (unsigned short)v;
> +		break;
> +	case 4:
> +		*(unsigned int *)addr |= (unsigned int)v;
> +		break;
> +	default:
> +		/*
> +		 * generate an illegal instruction. Cannot catch this with
> +		 * linker tricks when optimizations are disabled.
> +		 */
> +		__asm__ __volatile__("ud2");
> +	}
> +	mutex_lock_signal_restore(&compat_mutex, &mask);
> +}
> +
>  unsigned long _compat_uatomic_add_return(void *addr, unsigned long v, int len)
>  {
>  	sigset_t mask;
> diff --git a/tests/test_uatomic.c b/tests/test_uatomic.c
> index 5682655..37f95a6 100644
> --- a/tests/test_uatomic.c
> +++ b/tests/test_uatomic.c
> @@ -33,8 +33,10 @@ do {						\
>  	assert(uatomic_read(ptr) == 23);		\
>  	uatomic_dec(ptr);			\
>  	assert(uatomic_read(ptr) == 22);		\
> -	v = uatomic_add_return(ptr, 100);	\
> -	assert(v == 122);			\
> +	v = uatomic_add_return(ptr, 74);	\
> +	assert(v == 96);			\
> +	assert(uatomic_read(ptr) == 96);	\
> +	uatomic_or(ptr, 58);			\
>  	assert(uatomic_read(ptr) == 122);	\
>  	v = uatomic_sub_return(ptr, 1);		\
>  	assert(v == 121);			\
> diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
> index 30418c8..c3b5333 100644
> --- a/urcu/uatomic_arch_x86.h
> +++ b/urcu/uatomic_arch_x86.h
> @@ -231,6 +231,60 @@ unsigned long __uatomic_add_return(void *addr, unsigned long val,
>  						  (unsigned long)(v),	\
>  						  sizeof(*(addr))))
>  
> +/* uatomic_or */
> +
> +static inline __attribute__((always_inline))
> +void __uatomic_or(void *addr, unsigned long val, int len)
> +{
> +	switch (len) {
> +	case 1:
> +	{
> +		__asm__ __volatile__(
> +		"lock; orb %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "iq" ((unsigned char)val)
> +			: "memory");
> +		return;
> +	}
> +	case 2:
> +	{
> +		__asm__ __volatile__(
> +		"lock; orw %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "ir" ((unsigned short)val)
> +			: "memory");
> +		return;
> +	}
> +	case 4:
> +	{
> +		__asm__ __volatile__(
> +		"lock; orl %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "ir" ((unsigned int)val)
> +			: "memory");
> +		return;
> +	}
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +	{
> +		__asm__ __volatile__(
> +		"lock; orq %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "er" ((unsigned long)val)
> +			: "memory");
> +		return;
> +	}
> +#endif
> +	}
> +	/* generate an illegal instruction. Cannot catch this with linker tricks
> +	 * when optimizations are disabled. */
> +	__asm__ __volatile__("ud2");
> +	return;
> +}
> +
> +#define _uatomic_or(addr, v)						   \
> +	(__uatomic_or((addr), (unsigned long)(v), sizeof(*(addr))))
> +
>  /* uatomic_add */
>  
>  static inline __attribute__((always_inline))
> @@ -428,6 +482,13 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
>  						(unsigned long)(_new), 	       \
>  						sizeof(*(addr))))
>  
> +extern unsigned long _compat_uatomic_or(void *addr,
> +				        unsigned long _new, int len);
> +#define compat_uatomic_or(addr, v)				       \
> +	((__typeof__(*(addr))) _compat_uatomic_or((addr),	       \
> +						  (unsigned long)(v),  \
> +						  sizeof(*(addr))))
> +
>  extern unsigned long _compat_uatomic_add_return(void *addr,
>  					        unsigned long _new, int len);
>  #define compat_uatomic_add_return(addr, v)				       \
> @@ -454,6 +515,8 @@ extern unsigned long _compat_uatomic_add_return(void *addr,
>  		UATOMIC_COMPAT(cmpxchg(addr, old, _new))
>  #define uatomic_xchg(addr, v)			\
>  		UATOMIC_COMPAT(xchg(addr, v))
> +#define uatomic_or(addr, v)		\
> +		UATOMIC_COMPAT(or(addr, v))
>  #define uatomic_add_return(addr, v)		\
>  		UATOMIC_COMPAT(add_return(addr, v))
>  
> diff --git a/urcu/uatomic_generic.h b/urcu/uatomic_generic.h
> index 347e73f..556846f 100644
> --- a/urcu/uatomic_generic.h
> +++ b/urcu/uatomic_generic.h
> @@ -87,6 +87,39 @@ unsigned long _uatomic_cmpxchg(void *addr, unsigned long old,
>  						sizeof(*(addr))))
>  
>  
> +/* uatomic_or */
> +
> +#ifndef uatomic_or
> +static inline __attribute__((always_inline))
> +void _uatomic_or(void *addr, unsigned long val,
> +		 int len)
> +{
> +	switch (len) {
> +#ifdef UATOMIC_HAS_ATOMIC_BYTE
> +	case 1:
> +		__sync_or_and_fetch_1(addr, val);
> +#endif
> +#ifdef UATOMIC_HAS_ATOMIC_SHORT
> +	case 2:
> +		__sync_or_and_fetch_2(addr, val);
> +#endif
> +	case 4:
> +		__sync_or_and_fetch_4(addr, val);
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +		__sync_or_and_fetch_8(addr, val);
> +#endif
> +	}
> +	_uatomic_link_error();
> +	return 0;
> +}
> +
> +#define uatomic_or(addr, v)			\
> +	(_uatomic_or((addr),			\
> +		     (unsigned long)(v),	\
> +		     sizeof(*(addr))))
> +#endif
> +
>  /* uatomic_add_return */
>  
>  #ifndef uatomic_add_return
> @@ -186,6 +219,70 @@ unsigned long _uatomic_exchange(void *addr, unsigned long val, int len)
>  
>  #else /* #ifndef uatomic_cmpxchg */
>  
> +#ifndef uatomic_or
> +/* uatomic_or */
> +
> +static inline __attribute__((always_inline))
> +void _uatomic_or(void *addr, unsigned long val, int len)
> +{
> +	switch (len) {
> +#ifdef UATOMIC_HAS_ATOMIC_BYTE
> +	case 1:
> +	{
> +		unsigned char old, oldt;
> +
> +		oldt = uatomic_read((unsigned char *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old | val, 1);
> +		} while (oldt != old);
> +	}
> +#endif
> +#ifdef UATOMIC_HAS_ATOMIC_SHORT
> +	case 2:
> +	{
> +		unsigned short old, oldt;
> +
> +		oldt = uatomic_read((unsigned short *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old | val, 2);
> +		} while (oldt != old);
> +	}
> +#endif
> +	case 4:
> +	{
> +		unsigned int old, oldt;
> +
> +		oldt = uatomic_read((unsigned int *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old | val, 4);
> +		} while (oldt != old);
> +	}
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +	{
> +		unsigned long old, oldt;
> +
> +		oldt = uatomic_read((unsigned long *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old | val, 8);
> +		} while (oldt != old);
> +	}
> +#endif
> +	}
> +	_uatomic_link_error();
> +	return 0;
> +}
> +
> +#define uatomic_or(addr, v)		\
> +	(uatomic_or((addr),		\
> +		    (unsigned long)(v),	\
> +		    sizeof(*(addr))))
> +#endif /* #ifndef uatomic_or */
> +
>  #ifndef uatomic_add_return
>  /* uatomic_add_return */
>  
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 06/10] uatomic: add uatomic_and

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>

Merged, thanks!

Mathieu

> ---
>  compat_arch_x86.c       |   25 ++++++++++++
>  tests/test_uatomic.c    |    2 +
>  urcu/uatomic_arch_x86.h |   63 ++++++++++++++++++++++++++++++
>  urcu/uatomic_generic.h  |   97 +++++++++++++++++++++++++++++++++++++++++++++++
>  4 files changed, 187 insertions(+), 0 deletions(-)
> 
> diff --git a/compat_arch_x86.c b/compat_arch_x86.c
> index 33bf13d..692417e 100644
> --- a/compat_arch_x86.c
> +++ b/compat_arch_x86.c
> @@ -226,6 +226,31 @@ void _compat_uatomic_or(void *addr, unsigned long v, int len)
>  	mutex_lock_signal_restore(&compat_mutex, &mask);
>  }
>  
> +void _compat_uatomic_and(void *addr, unsigned long v, int len)
> +{
> +	sigset_t mask;
> +
> +	mutex_lock_signal_save(&compat_mutex, &mask);
> +	switch (len) {
> +	case 1:
> +		*(unsigned char *)addr &= (unsigned char)v;
> +		break;
> +	case 2:
> +		*(unsigned short *)addr &= (unsigned short)v;
> +		break;
> +	case 4:
> +		*(unsigned int *)addr &= (unsigned int)v;
> +		break;
> +	default:
> +		/*
> +		 * generate an illegal instruction. Cannot catch this with
> +		 * linker tricks when optimizations are disabled.
> +		 */
> +		__asm__ __volatile__("ud2");
> +	}
> +	mutex_lock_signal_restore(&compat_mutex, &mask);
> +}
> +
>  unsigned long _compat_uatomic_add_return(void *addr, unsigned long v, int len)
>  {
>  	sigset_t mask;
> diff --git a/tests/test_uatomic.c b/tests/test_uatomic.c
> index 37f95a6..2c8c232 100644
> --- a/tests/test_uatomic.c
> +++ b/tests/test_uatomic.c
> @@ -41,6 +41,8 @@ do {						\
>  	v = uatomic_sub_return(ptr, 1);		\
>  	assert(v == 121);			\
>  	assert(uatomic_read(ptr) == 121);	\
> +	uatomic_and(ptr, 129);			\
> +	assert(uatomic_read(ptr) == 1);		\
>  } while (0)
>  
>  int main(int argc, char **argv)
> diff --git a/urcu/uatomic_arch_x86.h b/urcu/uatomic_arch_x86.h
> index c3b5333..c861208 100644
> --- a/urcu/uatomic_arch_x86.h
> +++ b/urcu/uatomic_arch_x86.h
> @@ -231,6 +231,60 @@ unsigned long __uatomic_add_return(void *addr, unsigned long val,
>  						  (unsigned long)(v),	\
>  						  sizeof(*(addr))))
>  
> +/* uatomic_and */
> +
> +static inline __attribute__((always_inline))
> +void __uatomic_and(void *addr, unsigned long val, int len)
> +{
> +	switch (len) {
> +	case 1:
> +	{
> +		__asm__ __volatile__(
> +		"lock; andb %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "iq" ((unsigned char)val)
> +			: "memory");
> +		return;
> +	}
> +	case 2:
> +	{
> +		__asm__ __volatile__(
> +		"lock; andw %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "ir" ((unsigned short)val)
> +			: "memory");
> +		return;
> +	}
> +	case 4:
> +	{
> +		__asm__ __volatile__(
> +		"lock; andl %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "ir" ((unsigned int)val)
> +			: "memory");
> +		return;
> +	}
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +	{
> +		__asm__ __volatile__(
> +		"lock; andq %1, %0"
> +			: "=m"(*__hp(addr))
> +			: "er" ((unsigned long)val)
> +			: "memory");
> +		return;
> +	}
> +#endif
> +	}
> +	/* generate an illegal instruction. Cannot catch this with linker tricks
> +	 * when optimizations are disabled. */
> +	__asm__ __volatile__("ud2");
> +	return;
> +}
> +
> +#define _uatomic_and(addr, v)						   \
> +	(__uatomic_and((addr), (unsigned long)(v), sizeof(*(addr))))
> +
>  /* uatomic_or */
>  
>  static inline __attribute__((always_inline))
> @@ -482,6 +536,13 @@ extern unsigned long _compat_uatomic_cmpxchg(void *addr, unsigned long old,
>  						(unsigned long)(_new), 	       \
>  						sizeof(*(addr))))
>  
> +extern unsigned long _compat_uatomic_and(void *addr,
> +				         unsigned long _new, int len);
> +#define compat_uatomic_and(addr, v)				       \
> +	((__typeof__(*(addr))) _compat_uatomic_and((addr),	       \
> +						   (unsigned long)(v), \
> +						   sizeof(*(addr))))
> +
>  extern unsigned long _compat_uatomic_or(void *addr,
>  				        unsigned long _new, int len);
>  #define compat_uatomic_or(addr, v)				       \
> @@ -515,6 +576,8 @@ extern unsigned long _compat_uatomic_add_return(void *addr,
>  		UATOMIC_COMPAT(cmpxchg(addr, old, _new))
>  #define uatomic_xchg(addr, v)			\
>  		UATOMIC_COMPAT(xchg(addr, v))
> +#define uatomic_and(addr, v)		\
> +		UATOMIC_COMPAT(and(addr, v))
>  #define uatomic_or(addr, v)		\
>  		UATOMIC_COMPAT(or(addr, v))
>  #define uatomic_add_return(addr, v)		\
> diff --git a/urcu/uatomic_generic.h b/urcu/uatomic_generic.h
> index 556846f..cef58f3 100644
> --- a/urcu/uatomic_generic.h
> +++ b/urcu/uatomic_generic.h
> @@ -87,6 +87,39 @@ unsigned long _uatomic_cmpxchg(void *addr, unsigned long old,
>  						sizeof(*(addr))))
>  
>  
> +/* uatomic_and */
> +
> +#ifndef uatomic_and
> +static inline __attribute__((always_inline))
> +void _uatomic_and(void *addr, unsigned long val,
> +		  int len)
> +{
> +	switch (len) {
> +#ifdef UATOMIC_HAS_ATOMIC_BYTE
> +	case 1:
> +		__sync_and_and_fetch_1(addr, val);
> +#endif
> +#ifdef UATOMIC_HAS_ATOMIC_SHORT
> +	case 2:
> +		__sync_and_and_fetch_2(addr, val);
> +#endif
> +	case 4:
> +		__sync_and_and_fetch_4(addr, val);
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +		__sync_and_and_fetch_8(addr, val);
> +#endif
> +	}
> +	_uatomic_link_error();
> +	return 0;
> +}
> +
> +#define uatomic_and(addr, v)			\
> +	(_uatomic_and((addr),			\
> +		      (unsigned long)(v),	\
> +		      sizeof(*(addr))))
> +#endif
> +
>  /* uatomic_or */
>  
>  #ifndef uatomic_or
> @@ -219,6 +252,70 @@ unsigned long _uatomic_exchange(void *addr, unsigned long val, int len)
>  
>  #else /* #ifndef uatomic_cmpxchg */
>  
> +#ifndef uatomic_and
> +/* uatomic_and */
> +
> +static inline __attribute__((always_inline))
> +void _uatomic_and(void *addr, unsigned long val, int len)
> +{
> +	switch (len) {
> +#ifdef UATOMIC_HAS_ATOMIC_BYTE
> +	case 1:
> +	{
> +		unsigned char old, oldt;
> +
> +		oldt = uatomic_read((unsigned char *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old & val, 1);
> +		} while (oldt != old);
> +	}
> +#endif
> +#ifdef UATOMIC_HAS_ATOMIC_SHORT
> +	case 2:
> +	{
> +		unsigned short old, oldt;
> +
> +		oldt = uatomic_read((unsigned short *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old & val, 2);
> +		} while (oldt != old);
> +	}
> +#endif
> +	case 4:
> +	{
> +		unsigned int old, oldt;
> +
> +		oldt = uatomic_read((unsigned int *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old & val, 4);
> +		} while (oldt != old);
> +	}
> +#if (CAA_BITS_PER_LONG == 64)
> +	case 8:
> +	{
> +		unsigned long old, oldt;
> +
> +		oldt = uatomic_read((unsigned long *)addr);
> +		do {
> +			old = oldt;
> +			oldt = _uatomic_cmpxchg(addr, old, old & val, 8);
> +		} while (oldt != old);
> +	}
> +#endif
> +	}
> +	_uatomic_link_error();
> +	return 0;
> +}
> +
> +#define uatomic_and(addr, v)		\
> +	(uatomic_and((addr),		\
> +		    (unsigned long)(v),	\
> +		    sizeof(*(addr))))
> +#endif /* #ifndef uatomic_and */
> +
>  #ifndef uatomic_or
>  /* uatomic_or */
>  
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 08/10] call_rcu: redo futex implementation

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
 [...]
> From 9eb4f2821f212705a72ea38568b2a6157665b203 Mon Sep 17 00:00:00 2001
> From: Paolo Bonzini <pbonzini@redhat.com>
> Date: Wed, 8 Jun 2011 09:33:51 +0200
> Subject: [PATCH] call_rcu: drop mutex
> 
> The mutex is being used only to protect OR accesses to the flags.
> Just use atomic operations for that.

Merged, thanks!

Mathieu

>
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |   28 ++++++++++------------------
>  1 files changed, 10 insertions(+), 18 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 9beb58c..82fec80 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -46,7 +46,6 @@
>  struct call_rcu_data {
>  	struct cds_wfq_queue cbs;
>  	unsigned long flags;
> -	pthread_mutex_t mtx;
>  	int futex;
>  	unsigned long qlen; /* maintained for debugging. */
>  	pthread_t tid;
> @@ -204,6 +203,7 @@ static void *call_rcu_thread(void *arg)
>  	struct cds_wfq_node **cbs_tail;
>  	struct call_rcu_data *crdp = (struct call_rcu_data *)arg;
>  	struct rcu_head *rhp;
> +	int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
>  
>  	if (set_thread_cpu_affinity(crdp) != 0) {
>  		perror("pthread_setaffinity_np");
> @@ -212,7 +212,7 @@ static void *call_rcu_thread(void *arg)
>  
>  	thread_call_rcu_data = crdp;
>  	for (;;) {
> -		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (!rt) {
>  			uatomic_dec(&crdp->futex);
>  			/* Decrement futex before reading call_rcu list */
>  			cmm_smp_mb();
> @@ -240,8 +240,8 @@ static void *call_rcu_thread(void *arg)
>  			} while (cbs != NULL);
>  			uatomic_sub(&crdp->qlen, cbcount);
>  		}
> -		if (crdp->flags & URCU_CALL_RCU_STOP) {
> -			if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP) {
> +			if (!rt) {
>  				/*
>  				 * Read call_rcu list before write futex.
>  				 */
> @@ -250,15 +250,13 @@ static void *call_rcu_thread(void *arg)
>  			}
>  			break;
>  		}
> -		if (!(crdp->flags & URCU_CALL_RCU_RT)) {
> +		if (!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);
> +	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
>  	return NULL;
>  }
>  
> @@ -282,10 +280,6 @@ static void call_rcu_data_init(struct call_rcu_data **crdpp,
>  	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);
> @@ -568,12 +562,10 @@ void call_rcu_data_free(struct call_rcu_data *crdp)
>  	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);
> +	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
> +		uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
>  		wake_call_rcu_thread(crdp);
> -		while ((crdp->flags & URCU_CALL_RCU_STOPPED) == 0)
> +		while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
>  			poll(NULL, 0, 1);
>  	}
>  	if (&crdp->cbs.head != _CMM_LOAD_SHARED(crdp->cbs.tail)) {
> @@ -657,7 +649,7 @@ void call_rcu_after_fork_child(void)
>  		if (crdp == default_call_rcu_data)
>  			crdp = cds_list_entry(crdp->list.prev,
>  					      struct call_rcu_data, list);
> -		crdp->flags = URCU_CALL_RCU_STOPPED;
> +		uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
>  		call_rcu_data_free(crdp);
>  	}
>  }
> -- 
> 1.7.4.4
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 09/10] call_rcu: factor polling from RT and non-RT cases

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> This has the additional advantage of introducing a small delay before
> consecutive checks of the list.  In case call_rcu is very busy, this and
> the following patch practically eliminate the difference between the RT
> and non-RT version.

Not sure this patch is still relevant with current git head ? If yes,
can you rebase it ?

Thanks,

Mathieu

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |    7 ++-----
>  1 files changed, 2 insertions(+), 5 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 9824515..d8570e3 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -205,6 +205,7 @@ static void *call_rcu_thread(void *arg)
>  	thread_call_rcu_data = crdp;
>  	for (;;) {
>  		for (;;) {
> +			poll(NULL, 0, 10);
>  			if (&crdp->cbs.head
>  			    == _CMM_LOAD_SHARED(crdp->cbs.tail)) {
>  				uatomic_and(&crdp->flags, ~URCU_CALL_RCU_BUSY);
> @@ -237,12 +238,8 @@ static void *call_rcu_thread(void *arg)
>  		cmm_smp_mb();
>  		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
>  			break;
> -		if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT)
> -			poll(NULL, 0, 10);
> -		else {
> +		if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT) == 0)
>  			call_rcu_wait(crdp);
> -			poll(NULL, 0, 10);
> -		}
>  	}
>  	uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
>  	return NULL;
> -- 
> 1.7.4.4
> 
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH 10/10] call_rcu: avoid useless futex wakeups

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> If the call_rcu thread is waiting for entries to accumulate, it is
> not necessary to wake it up.  Use the BUSY flag to detect this.

The crdp->futex value being -1, I think this case is already being taken
care of.

Thanks,

Mathieu

> 
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>  urcu-call-rcu-impl.h |    8 +++++---
>  1 files changed, 5 insertions(+), 3 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index d8570e3..165956e 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -514,10 +514,12 @@ void call_rcu(struct rcu_head *head,
>  	head->func = func;
>  	crdp = get_call_rcu_data();
>  	cds_wfq_enqueue(&crdp->cbs, &head->next);
> -	/* Write list before writing the flags.  */
> +	/* Write list before checking/writing the flags.  */
>  	cmm_smp_mb();
> -	uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
> -	wake_call_rcu_thread(crdp);
> +	if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_BUSY) == 0) {
> +		uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);
> +		wake_call_rcu_thread(crdp);
> +	}
>  }
>  
>  /*
> -- 
> 1.7.4.4
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH] call_rcu: keep BUSY flag set as long as possible

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> Signed-off-by: Paolo Bonzini <pbonzini at redhat.com>
> ---
>         Not a correctness issue, and it will just save a syscall in
>         rare cases.  Can be committed separately, or squashed in patch 8
>         (that would be my favorite option), or left out altogether.
> 
>  urcu-call-rcu-impl.h |    3 +++
>  1 files changed, 3 insertions(+), 0 deletions(-)
> 
> diff --git a/urcu-call-rcu-impl.h b/urcu-call-rcu-impl.h
> index 165956e..c45928a 100644
> --- a/urcu-call-rcu-impl.h
> +++ b/urcu-call-rcu-impl.h
> @@ -214,6 +214,9 @@ static void *call_rcu_thread(void *arg)
>  				if (&crdp->cbs.head
>  				    == _CMM_LOAD_SHARED(crdp->cbs.tail))
>  					break;
> +                                /* False alarm... another bunch is ready.  */
> +				cmm_smp_mb();
> +				uatomic_or(&crdp->flags, URCU_CALL_RCU_BUSY);

Would this be equivalent to not setting:

  uatomic_set(&crdp->futex, 0);

before breaking the loop, but then not decrementing

  uatomic_dec(&crdp->futex);

at the next loop iteration ?

Mathieu


>  			}
>  			while ((cbs = _CMM_LOAD_SHARED(crdp->cbs.head)) == NULL)
>  				poll(NULL, 0, 1);
> -- 
> 1.7.4.4
> 
> 
> _______________________________________________
> ltt-dev mailing list
> ltt-dev at lists.casi.polymtl.ca
> http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
> 

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com

[ltt-dev] [PATCH] call_rcu: keep BUSY flag set as long as possible

[Paolo Bonzini]

On 06/09/2011 04:09 PM, Mathieu Desnoyers wrote:
> Would this be equivalent to not setting:
>
>    uatomic_set(&crdp->futex, 0);
>
> before breaking the loop, but then not decrementing
>
>    uatomic_dec(&crdp->futex);
>
> at the next loop iteration ?

Something like that, yes.

Paolo

[ltt-dev] [PATCH 09/10] call_rcu: factor polling from RT and non-RT cases

[Paolo Bonzini]

On 06/09/2011 04:06 PM, Mathieu Desnoyers wrote:
> Not sure this patch is still relevant with current git head ? If yes,
> can you rebase it ?

No, it's not, thanks for doing the same! :)

Paolo

[ltt-dev] [PATCH] call_rcu: keep BUSY flag set as long as possible

[Mathieu Desnoyers]

* Paolo Bonzini (pbonzini at redhat.com) wrote:
> On 06/09/2011 04:09 PM, Mathieu Desnoyers wrote:
>> Would this be equivalent to not setting:
>>
>>    uatomic_set(&crdp->futex, 0);
>>
>> before breaking the loop, but then not decrementing
>>
>>    uatomic_dec(&crdp->futex);
>>
>> at the next loop iteration ?
>
> Something like that, yes.

Actually, my implementation had a problem with non-empty lists: it would
lead to decrementing the futex value below -1. commit
bc94ca9bada25f7403e3e859caa241146ae8e338 fixes this.

Thanks,

Mathieu

>
> Paolo
>

-- 
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com