--- /dev/null
+DIRS = barrier chase-lev-deque dekker-fences linuxrwlocks mcs-lock mpmc-queue ms-queue spsc-queue
+
+all :
+ set -e; set -u; for d in $(DIRS); do (cd $$d; $(MAKE) ); done
+
+clean :
+ set -e; set -u; for d in $(DIRS); do (cd $$d; $(MAKE) clean ); done
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = barrier
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(TESTNAME).cc $(TESTNAME).h
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include <stdio.h>
+
+#include "cds_threads.h"
+
+#include "barrier.h"
+
+#include "librace.h"
+
+spinning_barrier *barr;
+int var = 0;
+
+void threadA(void *arg)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ store_32(&var, 1);
+ barr->wait();
+}
+
+void threadB(void *arg)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ barr->wait();
+ printf("var = %d\n", load_32(&var));
+}
+
+#define NUMREADERS 3
+int user_main(int argc, char **argv)
+{
+ std::thread A, B[NUMREADERS];
+ int i;
+
+ barr = new spinning_barrier(NUMREADERS + 1);
+
+ A = std::thread(threadA, (void *)NULL);
+ for (i = 0; i < NUMREADERS; i++)
+ B[i] = std::thread(threadB, (void *)NULL);
+
+ for (i = 0; i < NUMREADERS; i++)
+ B[i].join();
+ A.join();
+
+ return 0;
+}
--- /dev/null
+#include "cds_atomic.h"
+
+class spinning_barrier {
+ public:
+ spinning_barrier (unsigned int n) : n_ (n) {
+ nwait_ = 0;
+ step_ = 0;
+ }
+
+ bool wait() {
+ unsigned int step = step_.load ();
+
+ if (nwait_.fetch_add (1) == n_ - 1) {
+ /* OK, last thread to come. */
+ nwait_.store (0); // XXX: maybe can use relaxed ordering here ??
+ step_.fetch_add (1, std::memory_order_relaxed);
+ return true;
+ } else {
+ /* Run in circles and scream like a little girl. */
+ while (step_.load () == step)
+ thrd_yield();
+ return false;
+ }
+ }
+
+ protected:
+ /* Number of synchronized threads. */
+ const unsigned int n_;
+
+ /* Number of threads currently spinning. */
+ std::atomic<unsigned int> nwait_;
+
+ /* Number of barrier syncronizations completed so far,
+ * * it's OK to wrap. */
+ std::atomic<unsigned int> step_;
+};
--- /dev/null
+# A few common Makefile items
+
+CC=../../clang
+CXX=../../clang++
+
+CXXFLAGS=-std=c++0x -pthread -Wall $(SANITIZE) -g -I../include
+
+UNAME = $(shell uname)
+
+
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = chase-lev-deque
+
+HEADERS = deque.h
+OBJECTS = main.o deque.o
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(HEADERS) $(OBJECTS)
+ $(CXX) -o $@ $(OBJECTS) $(CXXFLAGS) $(LDFLAGS)
+
+%.o: %.c
+ $(CXX) -c -o $@ $< $(CXXFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include "cds_atomic.h"
+#include <inttypes.h>
+#include "deque.h"
+#include <stdlib.h>
+#include <stdio.h>
+
+Deque * create() {
+ Deque * q = (Deque *) calloc(1, sizeof(Deque));
+ Array * a = (Array *) calloc(1, sizeof(Array)+2*sizeof(atomic_int));
+ atomic_store_explicit(&q->array, (uintptr_t)a, memory_order_relaxed);
+ atomic_store_explicit(&q->top, 0, memory_order_relaxed);
+ atomic_store_explicit(&q->bottom, 0, memory_order_relaxed);
+ atomic_store_explicit(&a->size, 2, memory_order_relaxed);
+ return q;
+}
+
+int take(Deque *q) {
+ size_t b = atomic_load_explicit(&q->bottom, memory_order_relaxed) - 1;
+ Array *a = (Array *) atomic_load_explicit(&q->array, memory_order_relaxed);
+ atomic_store_explicit(&q->bottom, b, memory_order_relaxed);
+ atomic_thread_fence(memory_order_seq_cst);
+ size_t t = atomic_load_explicit(&q->top, memory_order_relaxed);
+ int x;
+ if (t <= b) {
+ /* Non-empty queue. */
+ x = atomic_load_explicit(&a->buffer[b % atomic_load_explicit(&a->size,memory_order_relaxed)], memory_order_relaxed);
+ if (t == b) {
+ /* Single last element in queue. */
+ if (!atomic_compare_exchange_strong_explicit(&q->top, &t, t + 1, memory_order_seq_cst, memory_order_relaxed))
+ /* Failed race. */
+ x = EMPTY;
+ atomic_store_explicit(&q->bottom, b + 1, memory_order_relaxed);
+ }
+ } else { /* Empty queue. */
+ x = EMPTY;
+ atomic_store_explicit(&q->bottom, b + 1, memory_order_relaxed);
+ }
+ return x;
+}
+
+void resize(Deque *q) {
+ Array *a = (Array *) atomic_load_explicit(&q->array, memory_order_relaxed);
+ size_t size=atomic_load_explicit(&a->size, memory_order_relaxed);
+ size_t new_size=size << 1;
+ Array *new_a = (Array *) calloc(1, new_size * sizeof(atomic_int) + sizeof(Array));
+ size_t top=atomic_load_explicit(&q->top, memory_order_relaxed);
+ size_t bottom=atomic_load_explicit(&q->bottom, memory_order_relaxed);
+ atomic_store_explicit(&new_a->size, new_size, memory_order_relaxed);
+ size_t i;
+ for(i=top; i < bottom; i++) {
+ atomic_store_explicit(&new_a->buffer[i % new_size], atomic_load_explicit(&a->buffer[i % size], memory_order_relaxed), memory_order_relaxed);
+ }
+ atomic_store_explicit(&q->array, (uintptr_t)new_a, memory_order_relaxed);
+ printf("resize\n");
+}
+
+void push(Deque *q, int x) {
+ size_t b = atomic_load_explicit(&q->bottom, memory_order_relaxed);
+ size_t t = atomic_load_explicit(&q->top, memory_order_acquire);
+ Array *a = (Array *) atomic_load_explicit(&q->array, memory_order_relaxed);
+ if (b - t > atomic_load_explicit(&a->size, memory_order_relaxed) - 1) /* Full queue. */ {
+ resize(q);
+ //Bug in paper...should have next line...
+ a = (Array *) atomic_load_explicit(&q->array, memory_order_relaxed);
+ }
+ atomic_store_explicit(&a->buffer[b % atomic_load_explicit(&a->size, memory_order_relaxed)], x, memory_order_relaxed);
+ atomic_thread_fence(memory_order_release);
+ atomic_store_explicit(&q->bottom, b + 1, memory_order_relaxed);
+}
+
+int steal(Deque *q) {
+ size_t t = atomic_load_explicit(&q->top, memory_order_acquire);
+ atomic_thread_fence(memory_order_seq_cst);
+ size_t b = atomic_load_explicit(&q->bottom, memory_order_acquire);
+ int x = EMPTY;
+ if (t < b) {
+ /* Non-empty queue. */
+ Array *a = (Array *) atomic_load_explicit(&q->array, memory_order_relaxed);
+ x = atomic_load_explicit(&a->buffer[t % atomic_load_explicit(&a->size, memory_order_relaxed)], memory_order_relaxed);
+ if (!atomic_compare_exchange_strong_explicit(&q->top, &t, t + 1, memory_order_seq_cst, memory_order_relaxed))
+ /* Failed race. */
+ return ABORT;
+ }
+ return x;
+}
--- /dev/null
+#ifndef DEQUE_H
+#define DEQUE_H
+
+#include "cds_atomic.h"
+
+typedef struct {
+ atomic_size_t size;
+ atomic_int buffer[];
+} Array;
+
+typedef struct {
+ atomic_size_t top, bottom;
+ atomic_uintptr_t array; /* Atomic(Array *) */
+} Deque;
+
+Deque * create();
+int take(Deque *q);
+int steal(Deque *q);
+void resize(Deque *q);
+void push(Deque *q, int x);
+
+#define EMPTY 0xffffffff
+#define ABORT 0xfffffffe
+
+#endif
--- /dev/null
+#include <stdlib.h>
+#include <assert.h>
+#include <stdio.h>
+#include "cds_threads.h"
+//#include <atomic>
+
+#include "model-assert.h"
+
+#include "deque.h"
+
+#define ITERATION 1
+
+Deque *q;
+int a;
+int b;
+int c;
+
+static void task(void * param) {
+ for (int i = 0; i < ITERATION; i++)
+ a=steal(q);
+}
+
+int user_main(int argc, char **argv)
+{
+ q=create();
+ std::thread t(task, (void *)0);
+
+ for (int i = 0; i < ITERATION; i++) {
+ push(q, 1);
+ push(q, 2);
+ push(q, 4);
+ b=take(q);
+ c=take(q);
+ }
+
+ t.join();
+
+ bool correct=true;
+ if (a!=1 && a!=2 && a!=4 && a!= EMPTY)
+ correct=false;
+ if (b!=1 && b!=2 && b!=4 && b!= EMPTY)
+ correct=false;
+ if (c!=1 && c!=2 && c!=4 && a!= EMPTY)
+ correct=false;
+ if (a!=EMPTY && b!=EMPTY && c!=EMPTY && (a+b+c)!=7)
+ correct=false;
+ if (!correct)
+ printf("a=%d b=%d c=%d\n",a,b,c);
+ MODEL_ASSERT(correct);
+
+ return 0;
+}
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = dekker-fences
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(TESTNAME).cc
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+/*
+ * Dekker's critical section algorithm, implemented with fences.
+ *
+ * URL:
+ * http://www.justsoftwaresolutions.co.uk/threading/
+ */
+
+
+#include "cds_threads.h"
+//#include "cds_atomic.h"
+#include <atomic>
+
+#include "librace.h"
+
+std::atomic<bool> flag0, flag1;
+std::atomic<int> turn;
+
+uint32_t var = 0;
+
+void p0()
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+ flag0.store(true,std::memory_order_relaxed);
+ //std::atomic_thread_fence(std::memory_order_seq_cst);
+
+ while (flag1.load(std::memory_order_relaxed))
+ {
+ if (turn.load(std::memory_order_relaxed) != 0)
+ {
+ flag0.store(false,std::memory_order_relaxed);
+ while (turn.load(std::memory_order_relaxed) != 0)
+ {
+ thrd_yield();
+ }
+ flag0.store(true,std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_seq_cst);
+ } else
+ thrd_yield();
+ }
+ std::atomic_thread_fence(std::memory_order_acquire);
+
+ // critical section
+ store_32(&var, 1);
+
+ turn.store(1,std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_release);
+ flag0.store(false,std::memory_order_relaxed);
+}
+
+void p1()
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+ flag1.store(true,std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_seq_cst);
+
+ while (flag0.load(std::memory_order_relaxed))
+ {
+ if (turn.load(std::memory_order_relaxed) != 1)
+ {
+ flag1.store(false,std::memory_order_relaxed);
+ while (turn.load(std::memory_order_relaxed) != 1)
+ {
+ thrd_yield();
+ }
+ flag1.store(true,std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_seq_cst);
+ } else
+ thrd_yield();
+ }
+ std::atomic_thread_fence(std::memory_order_acquire);
+
+ // critical section
+ store_32(&var, 2);
+
+ turn.store(0,std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_release);
+ flag1.store(false,std::memory_order_relaxed);
+}
+
+int user_main(int argc, char **argv)
+{
+ flag0 = false;
+ flag1 = false;
+ turn = 0;
+
+ std::thread a(p0);
+ std::thread b(p1);
+
+ a.join();
+ b.join();
+
+ return 0;
+}
--- /dev/null
+#ifndef __STDATOMIC_H__
+#define __STDATOMIC_H__
+
+#ifdef __cplusplus
+
+#include <atomic>
+
+using std::atomic_flag;
+using std::atomic_bool;
+//using std::atomic_address;
+using std::atomic_char;
+using std::atomic_schar;
+using std::atomic_uchar;
+using std::atomic_short;
+using std::atomic_ushort;
+using std::atomic_int;
+using std::atomic_uint;
+using std::atomic_long;
+using std::atomic_ulong;
+using std::atomic_llong;
+using std::atomic_ullong;
+using std::atomic_wchar_t;
+
+using std::atomic_size_t;
+using std::atomic_uintptr_t;
+
+using std::atomic;
+using std::memory_order;
+using std::memory_order_relaxed;
+using std::memory_order_acquire;
+using std::memory_order_release;
+using std::memory_order_acq_rel;
+using std::memory_order_seq_cst;
+
+using std::atomic_thread_fence;
+using std::atomic_signal_fence;
+
+using std::atomic_init;
+
+#define atomic_init(A, V) *A=V
+
+#define atomic_load_explicit(A, MO) (*A).load(MO)
+#define atomic_store_explicit(A, V, MO) (*A).store(V, MO)
+
+#define atomic_fetch_add_explicit(A, V, MO) (*A).fetch_add(V, MO);
+#define atomic_fetch_sub_explicit(A, V, MO) (*A).fetch_sub(V, MO);
+#define atomic_compare_exchange_strong_explicit(A, E, V, MO, FMO) \
+ (*A).compare_exchange_strong(*E, V, MO, FMO)
+
+#endif
+
+#endif // __STDATOMIC_H__
--- /dev/null
+// Header to handle CDSChecker bullshit.
+// Handles most things, not just threads:
+// - Catches C11 threads and forwards to C++11 threads.
+// - Catches store_n and load_n and turns them into normal stores/loads.
+// - Replaces user_main with just main.
+
+#ifndef __THREADS_H__
+#define __THREADS_H__
+
+#ifdef __cplusplus
+#include <thread>
+
+#include <map>
+#include <string>
+#include <cassert>
+
+#define thrd_t std::thread
+typedef int (*thrd_start_t)(void *);
+
+static std::map<std::string, std::thread> thrs;
+
+#define thrd_create(T, F, A) new_thread(#T, F, A)
+static void new_thread(const char *t, thrd_start_t f, void *args) {
+ assert(thrs.emplace(std::string(++t), std::thread(*f, args)).second);
+}
+
+#define thrd_join(T) end_thread(#T)
+static void end_thread(const char *t) {
+ thrs[std::string(t)].join();
+}
+
+#define thrd_yield() std::this_thread::yield()
+
+#define user_main main
+
+#endif
+
+#endif // __THREADS_H__
--- /dev/null
+#ifndef __LIBRACE_H__
+#define __LIBRACE_H__
+
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+ void store_8(void *addr, uint8_t val) { *((uint8_t *)addr) = val; }
+ void store_16(void *addr, uint16_t val) { *((uint16_t *)addr) = val; }
+ void store_32(void *addr, uint32_t val) { *((uint32_t *)addr) = val; }
+ void store_64(void *addr, uint64_t val) { *((uint64_t *)addr) = val; }
+
+ void store_8_(void *addr);
+ void store_16_(void *addr);
+ void store_32_(void *addr);
+ void store_64_(void *addr);
+
+ uint8_t load_8(const void *addr) { return *((uint8_t *)addr); }
+ uint16_t load_16(const void *addr) { return *((uint16_t *)addr); }
+ uint32_t load_32(const void *addr) { return *((uint32_t *)addr); }
+ uint64_t load_64(const void *addr) { return *((uint64_t *)addr); }
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // __LIBRACE_H__
--- /dev/null
+#ifndef __MODEL_ASSERT_H__
+#define __MODEL_ASSERT_H__
+
+#if __cplusplus
+extern "C" {
+#else
+#include <stdbool.h>
+#endif
+
+#include <cassert>
+
+static void model_assert(bool expr, const char *file, int line) {
+ assert(expr && file && line);
+}
+#define MODEL_ASSERT(expr) model_assert((expr), __FILE__, __LINE__)
+
+#if __cplusplus
+}
+#endif
+
+#endif /* __MODEL_ASSERT_H__ */
--- /dev/null
+#ifndef __UNRELACY_H__
+#define __UNRELACY_H__
+
+#include <atomic>
+#include <stdlib.h>
+#include <stdio.h>
+#include <mutex>
+#include <condition_variable>
+
+#include <model-assert.h>
+#include <librace.h>
+
+#define $
+
+#define ASSERT(expr) MODEL_ASSERT(expr)
+#define RL_ASSERT(expr) MODEL_ASSERT(expr)
+
+#define RL_NEW new
+#define RL_DELETE(expr) delete expr
+
+#define mo_seqcst memory_order_relaxed
+#define mo_release memory_order_release
+#define mo_acquire memory_order_acquire
+#define mo_acq_rel memory_order_acq_rel
+#define mo_relaxed memory_order_relaxed
+
+namespace rl {
+
+ /* This 'useless' struct is declared just so we can use partial template
+ * specialization in our store and load functions. */
+ template <typename T, size_t n>
+ struct useless {
+ static void store(void *addr, T val);
+ static T load(const void *addr);
+ };
+
+ template <typename T>
+ struct useless<T, 1> {
+ static void store(void *addr, T val) { store_8(addr, (uint8_t)val); }
+ static T load(const void *addr) { return (T)load_8(addr); }
+ };
+
+ template <typename T>
+ struct useless<T, 2> {
+ static void store(void *addr, T val) { store_16(addr, (uint16_t)val); }
+ static T load(const void *addr) { return (T)load_16(addr); }
+ };
+
+ template <typename T>
+ struct useless<T, 4> {
+ static void store(void *addr, T val) { store_32(addr, (uint32_t)val); }
+ static T load(const void *addr) { return (T)load_32(addr); }
+ };
+
+ template <typename T>
+ struct useless<T, 8> {
+ static void store(void *addr, T val) { store_64(addr, (uint64_t)val); }
+ static T load(const void *addr) { return (T)load_64(addr); }
+ };
+
+ template <typename T>
+ struct var {
+ var() { useless<T, sizeof(T)>::store(&value, 0); }
+ var(T v) { useless<T, sizeof(T)>::store(&value, v); }
+ var(var const& r) {
+ value = r.value;
+ }
+ ~var() { }
+
+ void operator = (T v) { useless<T, sizeof(T)>::store(&value, v); }
+ T operator () () { return useless<T, sizeof(T)>::load(&value); }
+ void operator += (T v) {
+ useless<T, sizeof(T)>::store(&value,
+ useless<T, sizeof(T)>::load(&value) + v);
+ }
+ bool operator == (const struct var<T> v) const { return useless<T, sizeof(T)>::load(&value) == useless<T, sizeof(T)>::load(&v.value); }
+
+ T value;
+ };
+
+ class backoff_t
+ {
+ public:
+ typedef int debug_info_param;
+ void yield(debug_info_param info) { }
+ void yield() { }
+ };
+
+
+ typedef backoff_t backoff;
+ typedef backoff_t linear_backoff;
+ typedef backoff_t exp_backoff;
+
+}
+
+#endif /* __UNRELACY_H__ */
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = linuxrwlocks
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(TESTNAME).cc
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include <stdio.h>
+#include "cds_threads.h"
+#include "cds_atomic.h"
+
+#include "librace.h"
+
+#define RW_LOCK_BIAS 0x00100000
+#define WRITE_LOCK_CMP RW_LOCK_BIAS
+
+/** Example implementation of linux rw lock along with 2 thread test
+ * driver... */
+
+typedef union {
+ atomic_int lock;
+} rwlock_t;
+
+static inline int read_can_lock(rwlock_t *lock)
+{
+ return atomic_load_explicit(&lock->lock, memory_order_relaxed) > 0;
+}
+
+static inline int write_can_lock(rwlock_t *lock)
+{
+ return atomic_load_explicit(&lock->lock, memory_order_relaxed) == RW_LOCK_BIAS;
+}
+
+static inline void read_lock(rwlock_t *rw)
+{
+ int priorvalue = atomic_fetch_sub_explicit(&rw->lock, 1, memory_order_acquire);
+ while (priorvalue <= 0) {
+ atomic_fetch_add_explicit(&rw->lock, 1, memory_order_relaxed);
+ while (atomic_load_explicit(&rw->lock, memory_order_relaxed) <= 0) {
+ thrd_yield();
+ }
+ priorvalue = atomic_fetch_sub_explicit(&rw->lock, 1, memory_order_relaxed);
+ }
+}
+
+static inline void write_lock(rwlock_t *rw)
+{
+ int priorvalue = atomic_fetch_sub_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_acquire);
+ while (priorvalue != RW_LOCK_BIAS) {
+ atomic_fetch_add_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_relaxed);
+ while (atomic_load_explicit(&rw->lock, memory_order_relaxed) != RW_LOCK_BIAS) {
+ thrd_yield();
+ }
+ priorvalue = atomic_fetch_sub_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_relaxed);
+ }
+}
+
+static inline int read_trylock(rwlock_t *rw)
+{
+ int priorvalue = atomic_fetch_sub_explicit(&rw->lock, 1, memory_order_acquire);
+ if (priorvalue > 0)
+ return 1;
+
+ atomic_fetch_add_explicit(&rw->lock, 1, memory_order_relaxed);
+ return 0;
+}
+
+static inline int write_trylock(rwlock_t *rw)
+{
+ int priorvalue = atomic_fetch_sub_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_acquire);
+ if (priorvalue == RW_LOCK_BIAS)
+ return 1;
+
+ atomic_fetch_add_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_relaxed);
+ return 0;
+}
+
+static inline void read_unlock(rwlock_t *rw)
+{
+ atomic_fetch_add_explicit(&rw->lock, 1, memory_order_release);
+}
+
+static inline void write_unlock(rwlock_t *rw)
+{
+ atomic_fetch_add_explicit(&rw->lock, RW_LOCK_BIAS, memory_order_release);
+}
+
+rwlock_t mylock;
+int shareddata;
+
+void a()
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ int i, rs;
+ for(i = 0; i < 2; i++) {
+ if ((i % 2) == 0) {
+ read_lock(&mylock);
+ rs = load_32(&shareddata);
+ read_unlock(&mylock);
+ } else {
+ write_lock(&mylock);
+ std::this_thread::yield();
+ store_32(&shareddata,(unsigned int)i);
+ write_unlock(&mylock);
+ }
+ }
+}
+
+int user_main(int argc, char **argv)
+{
+ //thrd_t t1, t2;
+ atomic_init(&mylock.lock, RW_LOCK_BIAS);
+
+ std::thread t1(a);
+ std::thread t2(a);
+ //thrd_create(&t1, (thrd_start_t)&a, NULL);
+ //thrd_create(&t2, (thrd_start_t)&a, NULL);
+
+ t1.join();
+ t2.join();
+ //thrd_join(t1);
+ //thrd_join(t2);
+
+ return 0;
+}
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = mcs-lock
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(TESTNAME).cc $(TESTNAME).h
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include <stdio.h>
+#include "cds_threads.h"
+
+#include "mcs-lock.h"
+
+/* For data race instrumentation */
+#include "librace.h"
+
+struct mcs_mutex *mutex;
+static uint32_t shared;
+
+void threadA(void *arg)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ mcs_mutex::guard g(mutex);
+ printf("store: %d\n", 17);
+ store_32(&shared, 17);
+ mutex->unlock(&g);
+ mutex->lock(&g);
+ printf("load: %u\n", load_32(&shared));
+}
+
+void threadB(void *arg)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ mcs_mutex::guard g(mutex);
+ printf("load: %u\n", load_32(&shared));
+ mutex->unlock(&g);
+ mutex->lock(&g);
+ printf("store: %d\n", 17);
+ store_32(&shared, 17);
+}
+
+int user_main(int argc, char **argv)
+{
+ mutex = new mcs_mutex();
+
+ std::thread A(threadA, (void *)NULL);
+ std::thread B(threadB, (void *)NULL);
+ A.join();
+ B.join();
+ return 0;
+}
--- /dev/null
+// mcs on stack
+
+#include <atomic>
+#include <unrelacy.h>
+
+struct mcs_node {
+ std::atomic<mcs_node *> next;
+ std::atomic<int> gate;
+
+ mcs_node() {
+ next.store(0);
+ gate.store(0);
+ }
+};
+
+struct mcs_mutex {
+public:
+ // tail is null when lock is not held
+ std::atomic<mcs_node *> m_tail;
+
+ mcs_mutex() {
+ m_tail.store( NULL );
+ }
+ ~mcs_mutex() {
+ ASSERT( m_tail.load() == NULL );
+ }
+
+ class guard {
+ public:
+ mcs_mutex * m_t;
+ mcs_node m_node; // node held on the stack
+
+ guard(mcs_mutex * t) : m_t(t) { t->lock(this); }
+ ~guard() { m_t->unlock(this); }
+ };
+
+ void lock(guard * I) {
+ mcs_node * me = &(I->m_node);
+
+ // set up my node :
+ // not published yet so relaxed :
+ me->next.store(NULL, std::mo_relaxed );
+ me->gate.store(1, std::mo_relaxed );
+
+ // publish my node as the new tail :
+ mcs_node * pred = m_tail.exchange(me, std::mo_acq_rel);
+ if ( pred != NULL ) {
+ // (*1) race here
+ // unlock of pred can see me in the tail before I fill next
+
+ // publish me to previous lock-holder :
+ pred->next.store(me, std::mo_release );
+
+ // (*2) pred not touched any more
+
+ // now this is the spin -
+ // wait on predecessor setting my flag -
+ rl::linear_backoff bo;
+ while ( me->gate.load(std::mo_relaxed) ) {
+ thrd_yield();
+ }
+ }
+ }
+
+ void unlock(guard * I) {
+ mcs_node * me = &(I->m_node);
+
+ mcs_node * next = me->next.load(std::mo_acquire);
+ if ( next == NULL )
+ {
+ mcs_node * tail_was_me = me;
+ if ( m_tail.compare_exchange_strong( tail_was_me,NULL,std::mo_acq_rel) ) {
+ // got null in tail, mutex is unlocked
+ return;
+ }
+
+ // (*1) catch the race :
+ rl::linear_backoff bo;
+ for(;;) {
+ next = me->next.load(std::mo_relaxed);
+ if ( next != NULL )
+ break;
+ thrd_yield();
+ }
+ }
+
+ // (*2) - store to next must be done,
+ // so no locker can be viewing my node any more
+
+ // let next guy in :
+ next->gate.store( 0, std::mo_release );
+ }
+};
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = mpmc-queue
+#TESTS = mpmc-queue mpmc-1r2w mpmc-2r1w mpmc-queue-rdwr
+#TESTS += mpmc-queue-noinit mpmc-1r2w-noinit mpmc-2r1w-noinit mpmc-rdwr-noinit
+TESTS = mpmc-queue
+
+all: $(TESTS)
+
+#mpmc-queue: CPPFLAGS += -DCONFIG_MPMC_READERS=2 -DCONFIG_MPMC_WRITERS=2
+mpmc-queue: CXXFLAGS += -DCONFIG_MPMC_READERS=0 -DCONFIG_MPMC_WRITERS=0 -DCONFIG_MPMC_RDWR=2
+mpmc-queue-rdwr: CPPFLAGS += -DCONFIG_MPMC_READERS=0 -DCONFIG_MPMC_WRITERS=0 -DCONFIG_MPMC_RDWR=2
+mpmc-1r2w: CPPFLAGS += -DCONFIG_MPMC_READERS=1 -DCONFIG_MPMC_WRITERS=2
+mpmc-2r1w: CPPFLAGS += -DCONFIG_MPMC_READERS=2 -DCONFIG_MPMC_WRITERS=1
+mpmc-queue-noinit: CPPFLAGS += -DCONFIG_MPMC_READERS=2 -DCONFIG_MPMC_WRITERS=2 -DCONFIG_MPMC_NO_INITIAL_ELEMENT
+mpmc-1r2w-noinit: CPPFLAGS += -DCONFIG_MPMC_READERS=1 -DCONFIG_MPMC_WRITERS=2 -DCONFIG_MPMC_NO_INITIAL_ELEMENT
+mpmc-2r1w-noinit: CPPFLAGS += -DCONFIG_MPMC_READERS=2 -DCONFIG_MPMC_WRITERS=1 -DCONFIG_MPMC_NO_INITIAL_ELEMENT
+mpmc-rdwr-noinit: CPPFLAGS += -DCONFIG_MPMC_READERS=0 -DCONFIG_MPMC_WRITERS=0 -DCONFIG_MPMC_RDWR=2 -DCONFIG_MPMC_NO_INITIAL_ELEMENT
+
+$(TESTS): $(TESTNAME).cc $(TESTNAME).h
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+clean:
+ rm -f $(TESTS) *.o
--- /dev/null
+#include <inttypes.h>
+#include "cds_threads.h"
+#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
+
+#include <librace.h>
+
+#include "mpmc-queue.h"
+
+void threadA(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ int32_t *bin = queue->write_prepare();
+ store_32(bin, 1);
+ queue->write_publish();
+}
+
+void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ int32_t *bin;
+ while ((bin = queue->read_fetch()) != NULL) {
+ printf("Read: %d\n", load_32(bin));
+ queue->read_consume();
+ }
+}
+
+void threadC(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+// std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ int32_t *bin = queue->write_prepare();
+ store_32(bin, 1);
+ queue->write_publish();
+
+ while ((bin = queue->read_fetch()) != NULL) {
+ printf("Read: %d\n", load_32(bin));
+ queue->read_consume();
+ }
+}
+
+#define MAXREADERS 3
+#define MAXWRITERS 3
+#define MAXRDWR 3
+
+#ifdef CONFIG_MPMC_READERS
+#define DEFAULT_READERS (CONFIG_MPMC_READERS)
+#else
+#define DEFAULT_READERS 2
+#endif
+
+#ifdef CONFIG_MPMC_WRITERS
+#define DEFAULT_WRITERS (CONFIG_MPMC_WRITERS)
+#else
+#define DEFAULT_WRITERS 2
+#endif
+
+#ifdef CONFIG_MPMC_RDWR
+#define DEFAULT_RDWR (CONFIG_MPMC_RDWR)
+#else
+#define DEFAULT_RDWR 0
+#endif
+
+int readers = DEFAULT_READERS, writers = DEFAULT_WRITERS, rdwr = DEFAULT_RDWR;
+
+void print_usage()
+{
+ printf("Error: use the following options\n"
+ " -r <num> Choose number of reader threads\n"
+ " -w <num> Choose number of writer threads\n");
+ exit(EXIT_FAILURE);
+}
+
+void process_params(int argc, char **argv)
+{
+ const char *shortopts = "hr:w:";
+ int opt;
+ bool error = false;
+
+ while (!error && (opt = getopt(argc, argv, shortopts)) != -1) {
+ switch (opt) {
+ case 'h':
+ print_usage();
+ break;
+ case 'r':
+ readers = atoi(optarg);
+ break;
+ case 'w':
+ writers = atoi(optarg);
+ break;
+ default: /* '?' */
+ error = true;
+ break;
+ }
+ }
+
+ if (writers < 1 || writers > MAXWRITERS)
+ error = true;
+ if (readers < 1 || readers > MAXREADERS)
+ error = true;
+
+ if (error)
+ print_usage();
+}
+
+int user_main(int argc, char **argv)
+{
+ struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
+ std::thread A[MAXWRITERS], B[MAXREADERS], C[MAXRDWR];
+
+ /* Note: optarg() / optind is broken in model-checker - workaround is
+ * to just copy&paste this test a few times */
+ //process_params(argc, argv);
+ printf("%d reader(s), %d writer(s)\n", readers, writers);
+
+#ifndef CONFIG_MPMC_NO_INITIAL_ELEMENT
+ printf("Adding initial element\n");
+ int32_t *bin = queue.write_prepare();
+ store_32(bin, 17);
+ queue.write_publish();
+#endif
+
+ printf("Start threads\n");
+
+ for (int i = 0; i < writers; i++)
+ A[i] = std::thread(threadA, &queue);
+ for (int i = 0; i < readers; i++)
+ B[i] = std::thread(threadB, &queue);
+
+ for (int i = 0; i < rdwr; i++)
+ C[i] = std::thread(threadC, &queue);
+
+ for (int i = 0; i < writers; i++)
+ A[i].join();
+ for (int i = 0; i < readers; i++)
+ B[i].join();
+ for (int i = 0; i < rdwr; i++)
+ C[i].join();
+
+ printf("Threads complete\n");
+
+ return 0;
+}
--- /dev/null
+#include "cds_atomic.h"
+#include <unrelacy.h>
+
+template <typename t_element, size_t t_size>
+struct mpmc_boundq_1_alt
+{
+private:
+
+ // elements should generally be cache-line-size padded :
+ t_element m_array[t_size];
+
+ // rdwr counts the reads & writes that have started
+ atomic<unsigned int> m_rdwr;
+ // "read" and "written" count the number completed
+ atomic<unsigned int> m_read;
+ atomic<unsigned int> m_written;
+
+public:
+
+ mpmc_boundq_1_alt()
+ {
+ m_rdwr = 0;
+ m_read = 0;
+ m_written = 0;
+ }
+
+ //-----------------------------------------------------
+
+ t_element * read_fetch() {
+ unsigned int rdwr = m_rdwr.load(mo_acquire);
+ unsigned int rd,wr;
+ for(;;) {
+ rd = (rdwr>>16) & 0xFFFF;
+ wr = rdwr & 0xFFFF;
+
+ if ( wr == rd ) // empty
+ return NULL;
+
+ if ( m_rdwr.compare_exchange_weak(rdwr,rdwr+(1<<16),mo_acquire) )
+ break;
+ else
+ thrd_yield();
+ }
+
+ // (*1)
+ rl::backoff bo;
+ while ( (m_written.load(mo_acquire) & 0xFFFF) != wr ) {
+ thrd_yield();
+ }
+
+ t_element * p = & ( m_array[ rd % t_size ] );
+
+ return p;
+ }
+
+ void read_consume() {
+ m_read.fetch_add(1,mo_release);
+ }
+
+ //-----------------------------------------------------
+
+ t_element * write_prepare() {
+ unsigned int rdwr = m_rdwr.load(mo_acquire);
+ unsigned int rd,wr;
+ for(;;) {
+ rd = (rdwr>>16) & 0xFFFF;
+ wr = rdwr & 0xFFFF;
+
+ if ( wr == ((rd + t_size)&0xFFFF) ) // full
+ return NULL;
+
+ if ( m_rdwr.compare_exchange_weak(rdwr,(rd<<16) | ((wr+1)&0xFFFF),mo_acq_rel) )
+ break;
+ else
+ thrd_yield();
+ }
+
+ // (*1)
+ rl::backoff bo;
+ while ( (m_read.load(mo_acquire) & 0xFFFF) != rd ) {
+ thrd_yield();
+ }
+
+ t_element * p = & ( m_array[ wr % t_size ] );
+
+ return p;
+ }
+
+ void write_publish()
+ {
+ m_written.fetch_add(1,mo_relaxed);
+ }
+
+ //-----------------------------------------------------
+
+
+};
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = ms-queue
+
+HEADERS = my_queue.h
+OBJECTS = main.o my_queue.o
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(HEADERS) $(OBJECTS)
+ $(CXX) -o $@ $(OBJECTS) $(CXXFLAGS) $(LDFLAGS)
+
+%.o: %.c
+ $(CXX) -c -o $@ $< $(CXXFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include <stdlib.h>
+#include <stdio.h>
+#include "cds_threads.h"
+
+#include "my_queue.h"
+#include "model-assert.h"
+
+static int procs = 4;
+static queue_t *queue;
+static thrd_t *threads;
+static unsigned int *input;
+static unsigned int *output;
+static int num_threads;
+
+int get_thread_num()
+{
+ //thrd_t curr = thrd_current();
+ int i;
+ printf("thread id: %d\n", std::this_thread::get_id());
+ for (i = 0; i < num_threads; i++)
+ if (std::this_thread::get_id() == threads[i].get_id())
+ return i;
+ MODEL_ASSERT(0);
+ return -1;
+}
+
+bool succ1, succ2;
+
+static void main_task(void *param)
+{
+ unsigned int val;
+ int pid = *((int *)param);
+ if (!pid) {
+ input[0] = 17;
+ enqueue(queue, input[0]);
+ succ1 = dequeue(queue, &output[0]);
+ //printf("Dequeue: %d\n", output[0]);
+ } else {
+ input[1] = 37;
+ enqueue(queue, input[1]);
+ succ2 = dequeue(queue, &output[1]);
+ }
+}
+
+int user_main(int argc, char **argv)
+{
+ int i;
+ int *param;
+ unsigned int in_sum = 0, out_sum = 0;
+
+ queue = (queue_t *)calloc(1, sizeof(*queue));
+ MODEL_ASSERT(queue);
+
+ num_threads = procs;
+ threads = (std::thread *)malloc(num_threads * sizeof(std::thread));
+ param = (int *)malloc(num_threads * sizeof(*param));
+ input = (unsigned *)calloc(num_threads, sizeof(*input));
+ output = (unsigned *)calloc(num_threads, sizeof(*output));
+
+ init_queue(queue, num_threads);
+ for (i = 0; i < num_threads; i++) {
+ param[i] = i;
+ //threads[i] = std::thread(main_task, ¶m[i]);
+ new (&threads[i])std::thread(main_task, ¶m[i]);
+ }
+ for (i = 0; i < num_threads; i++)
+ threads[i].join();
+
+ for (i = 0; i < num_threads; i++) {
+ in_sum += input[i];
+ out_sum += output[i];
+ }
+ for (i = 0; i < num_threads; i++)
+ printf("input[%d] = %u\n", i, input[i]);
+ for (i = 0; i < num_threads; i++)
+ printf("output[%d] = %u\n", i, output[i]);
+ if (succ1 && succ2)
+ MODEL_ASSERT(in_sum == out_sum);
+ else
+ MODEL_ASSERT (false);
+
+ free(param);
+ free(threads);
+ free(queue);
+
+ return 0;
+}
--- /dev/null
+#include "cds_threads.h"
+#include <stdlib.h>
+#include "librace.h"
+#include "model-assert.h"
+
+#include "my_queue.h"
+
+#define relaxed memory_order_relaxed
+#define release memory_order_release
+#define acquire memory_order_acquire
+
+#define MAX_FREELIST 4 /* Each thread can own up to MAX_FREELIST free nodes */
+#define INITIAL_FREE 2 /* Each thread starts with INITIAL_FREE free nodes */
+
+#define POISON_IDX 0x666
+
+static unsigned int (*free_lists)[MAX_FREELIST];
+
+/* Search this thread's free list for a "new" node */
+static unsigned int new_node()
+{
+ int i;
+ int t = get_thread_num();
+ for (i = 0; i < MAX_FREELIST; i++) {
+ unsigned int node = load_32(&free_lists[t][i]);
+ if (node) {
+ store_32(&free_lists[t][i], 0);
+ return node;
+ }
+ }
+ /* free_list is empty? */
+ MODEL_ASSERT(0);
+ return 0;
+}
+
+/* Place this node index back on this thread's free list */
+static void reclaim(unsigned int node)
+{
+ int i;
+ int t = get_thread_num();
+
+ /* Don't reclaim NULL node */
+ MODEL_ASSERT(node);
+
+ for (i = 0; i < MAX_FREELIST; i++) {
+ /* Should never race with our own thread here */
+ unsigned int idx = load_32(&free_lists[t][i]);
+
+ /* Found empty spot in free list */
+ if (idx == 0) {
+ store_32(&free_lists[t][i], node);
+ return;
+ }
+ }
+ /* free list is full? */
+ MODEL_ASSERT(0);
+}
+
+void init_queue(queue_t *q, int num_threads)
+{
+ int i, j;
+
+ /* Initialize each thread's free list with INITIAL_FREE pointers */
+ /* The actual nodes are initialized with poison indexes */
+ free_lists = (unsigned (*)[MAX_FREELIST])malloc(num_threads * sizeof(*free_lists));
+ for (i = 0; i < num_threads; i++) {
+ for (j = 0; j < INITIAL_FREE; j++) {
+ free_lists[i][j] = 2 + i * MAX_FREELIST + j;
+ atomic_init(&q->nodes[free_lists[i][j]].next, MAKE_POINTER(POISON_IDX, 0));
+ }
+ }
+
+ /* initialize queue */
+ atomic_init(&q->head, MAKE_POINTER(1, 0));
+ atomic_init(&q->tail, MAKE_POINTER(1, 0));
+ atomic_init(&q->nodes[1].next, MAKE_POINTER(0, 0));
+}
+
+void enqueue(queue_t *q, unsigned int val)
+{
+ int success = 0;
+ unsigned int node;
+ pointer tail;
+ pointer next;
+ pointer tmp;
+
+ node = new_node();
+ store_32(&q->nodes[node].value, val);
+ tmp = atomic_load_explicit(&q->nodes[node].next, relaxed);
+ set_ptr(&tmp, 0); // NULL
+ atomic_store_explicit(&q->nodes[node].next, tmp, relaxed);
+
+ while (!success) {
+ tail = atomic_load_explicit(&q->tail, acquire);
+ next = atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire);
+ if (tail == atomic_load_explicit(&q->tail, relaxed)) {
+
+ /* Check for uninitialized 'next' */
+ MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // == NULL
+ pointer value = MAKE_POINTER(node, get_count(next) + 1);
+ success = atomic_compare_exchange_strong_explicit(&q->nodes[get_ptr(tail)].next,
+ &next, value, release, release);
+ }
+ if (!success) {
+ unsigned int ptr = get_ptr(atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire));
+ pointer value = MAKE_POINTER(ptr,
+ get_count(tail) + 1);
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail, value,
+ release, release);
+ thrd_yield();
+ }
+ }
+ }
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(node, get_count(tail) + 1),
+ release, release);
+}
+
+bool dequeue(queue_t *q, unsigned int *retVal)
+{
+ int success = 0;
+ pointer head;
+ pointer tail;
+ pointer next;
+
+ while (!success) {
+ head = atomic_load_explicit(&q->head, acquire);
+ tail = atomic_load_explicit(&q->tail, relaxed);
+ next = atomic_load_explicit(&q->nodes[get_ptr(head)].next, acquire);
+ if (atomic_load_explicit(&q->head, relaxed) == head) {
+ if (get_ptr(head) == get_ptr(tail)) {
+
+ /* Check for uninitialized 'next' */
+ MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // NULL
+ return false; // NULL
+ }
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(get_ptr(next), get_count(tail) + 1),
+ release, release);
+ thrd_yield();
+ } else {
+ *retVal = load_32(&q->nodes[get_ptr(next)].value);
+ success = atomic_compare_exchange_strong_explicit(&q->head,
+ &head,
+ MAKE_POINTER(get_ptr(next), get_count(head) + 1),
+ release, release);
+ if (!success)
+ thrd_yield();
+ }
+ }
+ }
+ reclaim(get_ptr(head));
+ return true;
+}
--- /dev/null
+#include "cds_atomic.h"
+
+#define MAX_NODES 0xf
+
+typedef unsigned long long pointer;
+typedef atomic_ullong pointer_t;
+
+#define MAKE_POINTER(ptr, count) ((((pointer)count) << 32) | ptr)
+#define PTR_MASK 0xffffffffLL
+#define COUNT_MASK (0xffffffffLL << 32)
+
+static inline void set_count(pointer *p, unsigned int val) { *p = (*p & ~COUNT_MASK) | ((pointer)val << 32); }
+static inline void set_ptr(pointer *p, unsigned int val) { *p = (*p & ~PTR_MASK) | val; }
+static inline unsigned int get_count(pointer p) { return (p & COUNT_MASK) >> 32; }
+static inline unsigned int get_ptr(pointer p) { return p & PTR_MASK; }
+
+typedef struct node {
+ unsigned int value;
+ pointer_t next;
+} node_t;
+
+typedef struct {
+ pointer_t head;
+ pointer_t tail;
+ node_t nodes[MAX_NODES + 1];
+} queue_t;
+
+void init_queue(queue_t *q, int num_threads);
+void enqueue(queue_t *q, unsigned int val);
+bool dequeue(queue_t *q, unsigned int *retVal);
+int get_thread_num();
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = ms-queue
+
+HEADERS = my_queue.h
+OBJECTS = main.o my_queue.o
+
+CXXFLAGS += -I /scratch/fuzzer/random-fuzzer/include
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(HEADERS) $(OBJECTS)
+ $(CC) -o $@ $(OBJECTS) $(CFLAGS) $(LDFLAGS)
+
+%.o: %.c
+ $(CC) -c -o $@ $(CFLAGS)
+
+clean:
+ rm -f $(TESTNAME) *.o
--- /dev/null
+#include <stdlib.h>
+#include <stdio.h>
+#include <threads.h>
+//#include "cds_threads.h"
+
+#include "my_queue.h"
+#include "model-assert.h"
+
+#define user_main main
+
+static int procs = 4;
+static queue_t *queue;
+static thrd_t *threads;
+static unsigned int *input;
+static unsigned int *output;
+static int num_threads;
+
+int get_thread_num()
+{
+ thrd_t curr = thrd_current();
+ int i;
+ for (i = 0; i < num_threads; i++)
+ if (curr.priv == threads[i].priv)
+ return i;
+ MODEL_ASSERT(0);
+ return -1;
+}
+
+bool succ1, succ2;
+
+static void main_task(void *param)
+{
+ int pid = *((int *)param);
+ if (!pid) {
+ input[0] = 17;
+ enqueue(queue, input[0]);
+ succ1 = dequeue(queue, &output[0]);
+ //printf("Dequeue: %d\n", output[0]);
+ } else {
+ input[1] = 37;
+ enqueue(queue, input[1]);
+ succ2 = dequeue(queue, &output[1]);
+ }
+}
+
+int user_main(int argc, char **argv)
+{
+ int i;
+ int *param;
+ unsigned int in_sum = 0, out_sum = 0;
+
+ queue = (queue_t *)calloc(1, sizeof(*queue));
+ MODEL_ASSERT(queue);
+
+ num_threads = procs;
+ threads = (thrd_t *)malloc(num_threads * sizeof(thrd_t));
+ param = (int *)malloc(num_threads * sizeof(*param));
+ input = (unsigned *)calloc(num_threads, sizeof(*input));
+ output = (unsigned *)calloc(num_threads, sizeof(*output));
+
+ init_queue(queue, num_threads);
+ for (i = 0; i < num_threads; i++) {
+ param[i] = i;
+ thrd_create(&threads[i], main_task, ¶m[i]);
+ }
+ for (i = 0; i < num_threads; i++)
+ thrd_join(threads[i]);
+
+ for (i = 0; i < num_threads; i++) {
+ in_sum += input[i];
+ out_sum += output[i];
+ }
+ for (i = 0; i < num_threads; i++)
+ printf("input[%d] = %u\n", i, input[i]);
+ for (i = 0; i < num_threads; i++)
+ printf("output[%d] = %u\n", i, output[i]);
+/*
+ if (succ1 && succ2)
+ MODEL_ASSERT(in_sum == out_sum);
+ else
+ MODEL_ASSERT(false);
+*/
+ free(param);
+ free(threads);
+ free(queue);
+
+ return 0;
+}
--- /dev/null
+#include <threads.h>
+#include <stdlib.h>
+#include "librace.h"
+#include "model-assert.h"
+
+#include "queue.h"
+
+#define relaxed memory_order_relaxed
+#define release memory_order_release
+#define acquire memory_order_acquire
+
+#define MAX_FREELIST 4 /* Each thread can own up to MAX_FREELIST free nodes */
+#define INITIAL_FREE 2 /* Each thread starts with INITIAL_FREE free nodes */
+
+#define POISON_IDX 0x666
+
+static unsigned int (*free_lists)[MAX_FREELIST];
+
+/* Search this thread's free list for a "new" node */
+static unsigned int new_node()
+{
+ int i;
+ int t = get_thread_num();
+ for (i = 0; i < MAX_FREELIST; i++) {
+ //unsigned int node = load_32(&free_lists[t][i]);
+ unsigned int node = free_lists[t][i];
+ if (node) {
+ //store_32(&free_lists[t][i], 0);
+ free_lists[t][i] = 0;
+ return node;
+ }
+ }
+ /* free_list is empty? */
+ MODEL_ASSERT(0);
+ return 0;
+}
+
+/* Simulate the fact that when a node got recycled, it will get assigned to the
+ * same queue or for other usage */
+void simulateRecycledNodeUpdate(queue_t *q, unsigned int node) {
+ atomic_store_explicit(&q->nodes[node].next, -1, memory_order_release);
+}
+
+
+/* Place this node index back on this thread's free list */
+static void reclaim(unsigned int node)
+{
+ int i;
+ int t = get_thread_num();
+
+ /* Don't reclaim NULL node */
+ //MODEL_ASSERT(node);
+
+ for (i = 0; i < MAX_FREELIST; i++) {
+ /* Should never race with our own thread here */
+ //unsigned int idx = load_32(&free_lists[t][i]);
+ unsigned int idx = free_lists[t][i];
+
+ /* Found empty spot in free list */
+ if (idx == 0) {
+ //store_32(&free_lists[t][i], node);
+ free_lists[t][i] = node;
+ return;
+ }
+ }
+ /* free list is full? */
+ //MODEL_ASSERT(0);
+}
+
+void init_queue(queue_t *q, int num_threads)
+{
+ int i, j;
+
+ /* Initialize each thread's free list with INITIAL_FREE pointers */
+ /* The actual nodes are initialized with poison indexes */
+ free_lists = ( unsigned int (*)[MAX_FREELIST] ) malloc(num_threads * sizeof(*free_lists));
+ for (i = 0; i < num_threads; i++) {
+ for (j = 0; j < INITIAL_FREE; j++) {
+ free_lists[i][j] = 2 + i * MAX_FREELIST + j;
+ atomic_init(&q->nodes[free_lists[i][j]].next, MAKE_POINTER(POISON_IDX, 0));
+ }
+ }
+
+ /* initialize queue */
+ atomic_init(&q->head, MAKE_POINTER(1, 0));
+ atomic_init(&q->tail, MAKE_POINTER(1, 0));
+ atomic_init(&q->nodes[1].next, MAKE_POINTER(0, 0));
+}
+
+/** @DeclareState: IntList *q;
+@Initial: q = new IntList;
+@Print:
+ model_print("\tSTATE(q): ");
+ printContainer(q);
+ model_print("\n"); */
+
+/** @Transition: STATE(q)->push_back(val);
+@Print: model_print("\tENQ #%d: val=%d\n", ID, val); */
+void enqueue(queue_t *q, unsigned int val, int n)
+{
+ int success = 0;
+ unsigned int node;
+ pointer tail;
+ pointer next;
+ pointer tmp;
+
+ node = new_node();
+ //store_32(&q->nodes[node].value, val);
+ q->nodes[node].value = val;
+ tmp = atomic_load_explicit(&q->nodes[node].next, relaxed);
+ set_ptr(&tmp, 0); // NULL
+ // XXX-known-bug-#1: This is a found bug in AutoMO, and testcase4 can reveal
+ // this known bug.
+ // To reproduce, weaken the parameter "memory_order_release" to
+ // "memory_order_relaxed", run "make" to recompile, and then run:
+ // "./run.sh ./ms-queue/testcase4 -m2 -y -u3 -tSPEC"
+ /********** Detected KNOWN BUG (testcase4) **********/
+ atomic_store_explicit(&q->nodes[node].next, tmp, release);
+
+ while (!success) {
+ // XXX-injection-#1: To reproduce, weaken the parameter
+ // "memory_order_acquire" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase2 -m2 -y -u3 -tSPEC"
+ /********** Detected UL (testcase2) **********/
+ tail = atomic_load_explicit(&q->tail, acquire);
+ // XXX-injection-#2: To reproduce, weaken the parameter
+ // "memory_order_acquire" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase4 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase4) **********/
+ next = atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire);
+ if (tail == atomic_load_explicit(&q->tail, relaxed)) {
+
+ /* Check for uninitialized 'next' */
+ //MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // == NULL
+ pointer value = MAKE_POINTER(node, get_count(next) + 1);
+ // XXX-injection-#3: To reproduce, weaken the parameter
+ // "memory_order_release" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase1 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase1) **********/
+ success = atomic_compare_exchange_strong_explicit(&q->nodes[get_ptr(tail)].next,
+ &next, value, release, release);
+ /** @OPClearDefine: success */
+ }
+ if (!success) {
+ // XXX-injection-#4: To reproduce, weaken the parameter
+ // "memory_order_acquire" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase2 -m2 -y -u3 -tSPEC"
+ /********** Detected UL (testcase2) **********/
+ unsigned int ptr = get_ptr(atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire));
+ pointer value = MAKE_POINTER(ptr,
+ get_count(tail) + 1);
+ // XXX-injection-#5: To reproduce, weaken the parameter
+ // "memory_order_release" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase2 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase2) **********/
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail, value,
+ release, release);
+ thrd_yield();
+ }
+ }
+ }
+
+ // XXX-injection-#6: To reproduce, weaken the parameter
+ // "memory_order_release" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase1 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase1) **********/
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(node, get_count(tail) + 1),
+ release, release);
+}
+
+/** @Transition: S_RET = STATE(q)->empty() ? 0 : STATE(q)->front();
+if (S_RET && C_RET) STATE(q)->pop_front();
+@JustifyingPostcondition: if (!C_RET)
+ return S_RET == C_RET;
+@PostCondition: return C_RET ? *retVal == S_RET : true;
+@Print: model_print("\tDEQ #%d: C_RET=%d && *retVal=%d && S_RET=%d\n", ID,
+ C_RET, *retVal, S_RET);
+*/
+int dequeue(queue_t *q, unsigned int *retVal, unsigned int *reclaimNode)
+{
+ int success = 0;
+ pointer head;
+ pointer tail;
+ pointer next;
+
+ while (!success) {
+ // XXX-injection-#7: To reproduce, weaken the parameter
+ // "memory_order_acquire" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase3 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase3) **********/
+ head = atomic_load_explicit(&q->head, acquire);
+ // To reproduce, weaken the parameter "memory_order_acquire" to
+ // "memory_order_relaxed", run "make" to recompile, and then run:
+ // "./run.sh ./ms-queue/testcase4 -m2 -y -u3 -tSPEC"
+ // XXX-known-bug-#2: This is another known bug, and testcase2 can reveal
+ // this known bug
+ /********** Detected KNOWN BUG (testcase2) **********/
+ tail = atomic_load_explicit(&q->tail, acquire);
+
+ // XXX-injection-#8: To reproduce, weaken the parameter
+ // "memory_order_acquire" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase1 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase1) **********/
+ next = atomic_load_explicit(&q->nodes[get_ptr(head)].next, acquire);
+ /** @OPClearDefine: true */
+ if (atomic_load_explicit(&q->head, relaxed) == head) {
+ if (get_ptr(head) == get_ptr(tail)) {
+
+ /* Check for uninitialized 'next' */
+ MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // NULL
+ return false; // NULL
+ }
+
+ // XXX-injection-#9: To reproduce, weaken the parameter
+ // "memory_order_release" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase2 -m2 -y -u3 -tSPEC"
+ /********** Detected UL (testcase2) **********/
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(get_ptr(next), get_count(tail) + 1),
+ release, release);
+ thrd_yield();
+ } else {
+ //*retVal = load_32(&q->nodes[get_ptr(next)].value);
+ *retVal = q->nodes[get_ptr(next)].value;
+
+ // XXX-injection-#10: To reproduce, weaken the parameter
+ // "memory_order_release" to "memory_order_relaxed", run "make" to
+ // recompile, and then run:
+ // "./run.sh ./ms-queue/testcase3 -m2 -y -u3 -tSPEC"
+ /********** Detected Correctness (testcase3) **********/
+ success = atomic_compare_exchange_strong_explicit(&q->head,
+ &head,
+ MAKE_POINTER(get_ptr(next), get_count(head) + 1),
+ release, release);
+ if (!success)
+ thrd_yield();
+ }
+ }
+ }
+ *reclaimNode = get_ptr(head);
+ reclaim(get_ptr(head));
+ return true;
+}
--- /dev/null
+#include <threads.h>
+#include <stdlib.h>
+#include "librace.h"
+#include "model-assert.h"
+
+#include "my_queue.h"
+
+#define relaxed memory_order_relaxed
+#define release memory_order_release
+#define acquire memory_order_acquire
+
+#define MAX_FREELIST 4 /* Each thread can own up to MAX_FREELIST free nodes */
+#define INITIAL_FREE 2 /* Each thread starts with INITIAL_FREE free nodes */
+
+#define POISON_IDX 0x666
+
+static unsigned int (*free_lists)[MAX_FREELIST];
+
+/* Search this thread's free list for a "new" node */
+static unsigned int new_node()
+{
+ int i;
+ int t = get_thread_num();
+ for (i = 0; i < MAX_FREELIST; i++) {
+ unsigned int node = load_32(&free_lists[t][i]);
+ if (node) {
+ store_32(&free_lists[t][i], 0);
+ return node;
+ }
+ }
+ /* free_list is empty? */
+ MODEL_ASSERT(0);
+ return 0;
+}
+
+/* Place this node index back on this thread's free list */
+static void reclaim(unsigned int node)
+{
+ int i;
+ int t = get_thread_num();
+
+ /* Don't reclaim NULL node */
+ MODEL_ASSERT(node);
+
+ for (i = 0; i < MAX_FREELIST; i++) {
+ /* Should never race with our own thread here */
+ unsigned int idx = load_32(&free_lists[t][i]);
+
+ /* Found empty spot in free list */
+ if (idx == 0) {
+ store_32(&free_lists[t][i], node);
+ return;
+ }
+ }
+ /* free list is full? */
+ MODEL_ASSERT(0);
+}
+
+void init_queue(queue_t *q, int num_threads)
+{
+ int i, j;
+
+ /* Initialize each thread's free list with INITIAL_FREE pointers */
+ /* The actual nodes are initialized with poison indexes */
+ free_lists = (unsigned (*)[MAX_FREELIST])malloc(num_threads * sizeof(*free_lists));
+ for (i = 0; i < num_threads; i++) {
+ for (j = 0; j < INITIAL_FREE; j++) {
+ free_lists[i][j] = 2 + i * MAX_FREELIST + j;
+ atomic_init(&q->nodes[free_lists[i][j]].next, MAKE_POINTER(POISON_IDX, 0));
+ }
+ }
+
+ /* initialize queue */
+ atomic_init(&q->head, MAKE_POINTER(1, 0));
+ atomic_init(&q->tail, MAKE_POINTER(1, 0));
+ atomic_init(&q->nodes[1].next, MAKE_POINTER(0, 0));
+}
+
+void enqueue(queue_t *q, unsigned int val)
+{
+ int success = 0;
+ unsigned int node;
+ pointer tail;
+ pointer next;
+ pointer tmp;
+
+ node = new_node();
+ store_32(&q->nodes[node].value, val);
+ tmp = atomic_load_explicit(&q->nodes[node].next, relaxed);
+ set_ptr(&tmp, 0); // NULL
+
+ // Bug 1
+ atomic_store_explicit(&q->nodes[node].next, tmp, relaxed);
+ //atomic_store_explicit(&q->nodes[node].next, tmp, release);
+
+ while (!success) {
+ tail = atomic_load_explicit(&q->tail, acquire);
+ next = atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire);
+ if (tail == atomic_load_explicit(&q->tail, relaxed)) {
+ /* Check for uninitialized 'next' */
+ MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // == NULL
+ pointer value = MAKE_POINTER(node, get_count(next) + 1);
+ success = atomic_compare_exchange_strong_explicit(&q->nodes[get_ptr(tail)].next,
+ &next, value, release, release);
+ }
+ if (!success) {
+ unsigned int ptr = get_ptr(atomic_load_explicit(&q->nodes[get_ptr(tail)].next, acquire));
+ pointer value = MAKE_POINTER(ptr,
+ get_count(tail) + 1);
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail, value,
+ release, release);
+ thrd_yield();
+ }
+ }
+ }
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(node, get_count(tail) + 1),
+ release, release);
+}
+
+bool dequeue(queue_t *q, unsigned int *retVal)
+{
+ int success = 0;
+ pointer head;
+ pointer tail;
+ pointer next;
+
+ while (!success) {
+ head = atomic_load_explicit(&q->head, acquire);
+
+ // Bug 2
+ tail = atomic_load_explicit(&q->tail, relaxed);
+ //tail = atomic_load_explicit(&q->tail, acquire);
+
+ next = atomic_load_explicit(&q->nodes[get_ptr(head)].next, acquire);
+ if (atomic_load_explicit(&q->head, relaxed) == head) {
+ if (get_ptr(head) == get_ptr(tail)) {
+
+ /* Check for uninitialized 'next' */
+ MODEL_ASSERT(get_ptr(next) != POISON_IDX);
+
+ if (get_ptr(next) == 0) { // NULL
+ return false; // NULL
+ }
+ atomic_compare_exchange_strong_explicit(&q->tail,
+ &tail,
+ MAKE_POINTER(get_ptr(next), get_count(tail) + 1),
+ release, release);
+ thrd_yield();
+ } else {
+ *retVal = load_32(&q->nodes[get_ptr(next)].value);
+ success = atomic_compare_exchange_strong_explicit(&q->head,
+ &head,
+ MAKE_POINTER(get_ptr(next), get_count(head) + 1),
+ release, release);
+ if (!success)
+ thrd_yield();
+ }
+ }
+ }
+ reclaim(get_ptr(head));
+ return true;
+}
--- /dev/null
+#include <stdatomic.h>
+#include <stdbool.h>
+
+#define MAX_NODES 0xf
+
+typedef unsigned long long pointer;
+typedef atomic_ullong pointer_t;
+
+#define MAKE_POINTER(ptr, count) ((((pointer)count) << 32) | ptr)
+#define PTR_MASK 0xffffffffLL
+#define COUNT_MASK (0xffffffffLL << 32)
+
+static inline void set_count(pointer *p, unsigned int val) { *p = (*p & ~COUNT_MASK) | ((pointer)val << 32); }
+static inline void set_ptr(pointer *p, unsigned int val) { *p = (*p & ~PTR_MASK) | val; }
+static inline unsigned int get_count(pointer p) { return (p & COUNT_MASK) >> 32; }
+static inline unsigned int get_ptr(pointer p) { return p & PTR_MASK; }
+
+typedef struct node {
+ unsigned int value;
+ pointer_t next;
+} node_t;
+
+typedef struct {
+ pointer_t head;
+ pointer_t tail;
+ node_t nodes[MAX_NODES + 1];
+} queue_t;
+
+void init_queue(queue_t *q, int num_threads);
+void enqueue(queue_t *q, unsigned int val);
+bool dequeue(queue_t *q, unsigned int *retVal);
+int get_thread_num();
--- /dev/null
+include ../benchmarks.mk
+
+TESTNAME = spsc-queue
+RELACYNAME = spsc-relacy
+
+all: $(TESTNAME)
+
+$(TESTNAME): $(TESTNAME).cc queue.h eventcount.h
+ $(CXX) -o $@ $< $(CXXFLAGS) $(LDFLAGS)
+
+relacy: $(RELACYNAME)
+
+$(RELACYNAME): spsc-relacy.cc queue-relacy.h eventcount-relacy.h
+ifdef RELACYPATH
+ $(CXX) -o $(RELACYNAME) spsc-relacy.cc -I$(RELACYPATH) -Wno-deprecated
+else
+ @echo "Please define RELACYPATH"
+ @echo " e.g., make RELACYPATH=/path-to-relacy"
+ @exit 1
+endif
+
+clean:
+ rm -f $(TESTNAME) $(RELACYNAME) *.o
--- /dev/null
+class eventcount
+{
+public:
+ eventcount() : waiters(0)
+ {
+ count($) = 0;
+ }
+
+ void signal_relaxed()
+ {
+ unsigned cmp = count.load(std::memory_order_relaxed);
+ signal_impl(cmp);
+ }
+
+ void signal()
+ {
+ unsigned cmp = count.fetch_add(0, std::memory_order_seq_cst);
+ signal_impl(cmp);
+ }
+
+ unsigned get()
+ {
+ unsigned cmp = count.fetch_or(0x80000000,
+std::memory_order_seq_cst);
+ return cmp & 0x7FFFFFFF;
+ }
+
+ void wait(unsigned cmp)
+ {
+ unsigned ec = count.load(std::memory_order_seq_cst);
+ if (cmp == (ec & 0x7FFFFFFF))
+ {
+ guard.lock($);
+ ec = count.load(std::memory_order_seq_cst);
+ if (cmp == (ec & 0x7FFFFFFF))
+ {
+ waiters($) += 1;
+ cv.wait(guard, $);
+ }
+ guard.unlock($);
+ }
+ }
+
+private:
+ std::atomic<unsigned> count;
+ rl::var<unsigned> waiters;
+ std::mutex guard;
+ std::condition_variable cv;
+
+ void signal_impl(unsigned cmp)
+ {
+ if (cmp & 0x80000000)
+ {
+ guard.lock($);
+ while (false == count.compare_exchange_weak(cmp,
+ (cmp + 1) & 0x7FFFFFFF, std::memory_order_relaxed));
+ unsigned w = waiters($);
+ waiters($) = 0;
+ guard.unlock($);
+ if (w)
+ cv.notify_all($);
+ }
+ }
+};
--- /dev/null
+#include <unrelacy.h>
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
+
+class eventcount
+{
+public:
+ eventcount() : waiters(0)
+ {
+ count = 0;
+ }
+
+ void signal_relaxed()
+ {
+ unsigned cmp = count.load(std::memory_order_relaxed);
+ signal_impl(cmp);
+ }
+
+ void signal()
+ {
+ unsigned cmp = count.fetch_add(0, std::memory_order_seq_cst);
+ signal_impl(cmp);
+ }
+
+ unsigned get()
+ {
+ unsigned cmp = count.fetch_or(0x80000000,
+std::memory_order_seq_cst);
+ return cmp & 0x7FFFFFFF;
+ }
+
+ void wait(unsigned cmp)
+ {
+ unsigned ec = count.load(std::memory_order_seq_cst);
+ if (cmp == (ec & 0x7FFFFFFF))
+ {
+ //guard.lock($);
+ std::unique_lock<std::mutex> lck(guard);
+ ec = count.load(std::memory_order_seq_cst);
+ if (cmp == (ec & 0x7FFFFFFF))
+ {
+ waiters += 1;
+ //cv.wait(guard);
+ cv.wait(lck);
+ }
+ guard.unlock($);
+ //lck.unlock();
+ }
+ }
+
+private:
+ std::atomic<unsigned> count;
+ rl::var<unsigned> waiters;
+ std::mutex guard;
+ std::condition_variable cv;
+
+ void signal_impl(unsigned cmp)
+ {
+ if (cmp & 0x80000000)
+ {
+ guard.lock($);
+ while (false == count.compare_exchange_weak(cmp,
+ (cmp + 1) & 0x7FFFFFFF, std::memory_order_relaxed));
+ unsigned w = waiters($);
+ waiters = 0;
+ guard.unlock($);
+ if (w)
+ cv.notify_all($);
+ }
+ }
+};
--- /dev/null
+#include "eventcount-relacy.h"
+
+template<typename T>
+class spsc_queue
+{
+public:
+ spsc_queue()
+ {
+ node* n = new node ();
+ head($) = n;
+ tail($) = n;
+ }
+
+ ~spsc_queue()
+ {
+ RL_ASSERT(head($) == tail($));
+ delete ((node*)head($));
+ }
+
+ void enqueue(T data)
+ {
+ node* n = new node (data);
+ head($)->next.store(n, std::memory_order_release);
+ head($) = n;
+ ec.signal_relaxed();
+ }
+
+ T dequeue()
+ {
+ T data = try_dequeue();
+ while (0 == data)
+ {
+ int cmp = ec.get();
+ data = try_dequeue();
+ if (data)
+ break;
+ ec.wait(cmp);
+ data = try_dequeue();
+ if (data)
+ break;
+ }
+ return data;
+ }
+
+private:
+ struct node
+ {
+ std::atomic<node*> next;
+ rl::var<T> data;
+
+ node(T data = T())
+ : data(data)
+ {
+ next($) = 0;
+ }
+ };
+
+ rl::var<node*> head;
+ rl::var<node*> tail;
+
+ eventcount ec;
+
+ T try_dequeue()
+ {
+ node* t = tail($);
+ node* n = t->next.load(std::memory_order_acquire);
+ if (0 == n)
+ return 0;
+ T data = n->data($);
+ delete (t);
+ tail($) = n;
+ return data;
+ }
+};
--- /dev/null
+#include <unrelacy.h>
+#include <atomic>
+
+#include "eventcount.h"
+
+template<typename T>
+class spsc_queue
+{
+public:
+ spsc_queue()
+ {
+ node* n = new node ();
+ head = n;
+ tail = n;
+ }
+
+ ~spsc_queue()
+ {
+ RL_ASSERT(head == tail);
+ delete ((node*)head($));
+ }
+
+ void enqueue(T data)
+ {
+ node* n = new node (data);
+ head($)->next.store(n, std::memory_order_release);
+ head = n;
+ ec.signal_relaxed();
+ }
+
+ T dequeue()
+ {
+ T data = try_dequeue();
+ while (0 == data)
+ {
+ int cmp = ec.get();
+ data = try_dequeue();
+ if (data)
+ break;
+ ec.wait(cmp);
+ data = try_dequeue();
+ if (data)
+ break;
+ }
+ return data;
+ }
+
+private:
+ struct node
+ {
+ std::atomic<node*> next;
+ rl::var<T> data;
+
+ node(T data = T())
+ : data(data)
+ {
+ next = 0;
+ }
+ };
+
+ rl::var<node*> head;
+ rl::var<node*> tail;
+
+ eventcount ec;
+
+ T try_dequeue()
+ {
+ node* t = tail($);
+ node* n = t->next.load(std::memory_order_acquire);
+ if (0 == n)
+ return 0;
+ T data = n->data($);
+ delete (t);
+ tail = n;
+ return data;
+ }
+};
--- /dev/null
+#include "cds_threads.h"
+
+#include "queue.h"
+
+spsc_queue<int> *q;
+
+ void thread(unsigned thread_index)
+ {
+ if (0 == thread_index)
+ {
+ q->enqueue(11);
+ }
+ else
+ {
+ int d = q->dequeue();
+ RL_ASSERT(11 == d);
+ }
+ }
+
+int user_main(int argc, char **argv)
+{
+ thrd_t A, B;
+
+ q = new spsc_queue<int>();
+
+ thrd_create(&A, (thrd_start_t)&thread, (void *)0);
+ thrd_create(&B, (thrd_start_t)&thread, (void *)1);
+ thrd_join(A);
+ thrd_join(B);
+
+ delete q;
+
+ return 0;
+}
--- /dev/null
+#include <relacy/relacy_std.hpp>
+
+#include "queue-relacy.h"
+
+struct spsc_queue_test : rl::test_suite<spsc_queue_test, 2>
+{
+ spsc_queue<int> q;
+
+ void thread(unsigned thread_index)
+ {
+ if (0 == thread_index)
+ {
+ q.enqueue(11);
+ }
+ else
+ {
+ int d = q.dequeue();
+ RL_ASSERT(11 == d);
+ }
+ }
+};
+
+
+int main()
+{
+ rl::simulate<spsc_queue_test>();
+}
--- /dev/null
+#!/bin/bash
+
+EXE=$1
+TOTAL_RUN=5 #00
+CDSLIB="/scratch/fuzzer/random-fuzzer"
+export LD_LIBRARY_PATH=${CDSLIB}
+export C11TESTER='-x1'
+
+#ERROR_FILE="data-structure.log"
+TASKSET=""
+
+COUNT_DATA_RACE=0
+COUNT_TIME=0
+
+for i in `seq 1 1 $TOTAL_RUN` ; do
+# time ${TASKSET} $EXE &> $ERROR_FILE
+# OUTPUT=$(< $ERROR_FILE)
+
+ OUTPUT="$(/usr/bin/time -f "time: %U %S" $EXE -x1 2>&1)"
+ RACE="$(echo "$OUTPUT" | grep "race")"
+ if [ -n "$RACE" ] ; then
+ ((++COUNT_DATA_RACE))
+ fi
+
+ TIME="$(echo "$OUTPUT" | grep -o "time: .\... .\...")"
+ TIME_USER_S="$(echo "$TIME" | cut -d' ' -f2 | cut -d'.' -f1)"
+ TIME_USER_CS="$(echo "$TIME" | cut -d' ' -f2 | cut -d'.' -f2)"
+ TIME_SYSTEM_S="$(echo "$TIME" | cut -d' ' -f3 | cut -d'.' -f1)"
+ TIME_SYSTEM_CS="$(echo "$TIME" | cut -d' ' -f3 | cut -d'.' -f2)"
+
+ TIME_EXE=$((10#$TIME_USER_S * 1000 + 10#$TIME_USER_CS * 10 + 10#$TIME_SYSTEM_S * 1000 + 10#$TIME_SYSTEM_CS * 10))
+ COUNT_TIME=$((COUNT_TIME + TIME_EXE))
+done
+
+#rm $ERROR_FILE
+
+AVG_DATA_RACE=$(echo "${COUNT_DATA_RACE} * 100 / ${TOTAL_RUN}" | bc -l | xargs printf "%.1f")
+AVG_TIME_INT=$(echo "${COUNT_TIME} / ${TOTAL_RUN} + 0.5" | bc -l | xargs printf "%.0f")
+
+# -3 / log(1 - p) < n
+#NO_99=$(echo "-3 / (l(1 - (${AVG_DATA_RACE} / 100)) / l(10)) + 0.5" | bc -l | xargs printf "%.0f")
+#TIME_99=$(echo "${NO_99} * ${AVG_TIME_INT}" | bc -l)
+
+echo "Runs: $TOTAL_RUN | Data races: $COUNT_DATA_RACE | Total time: ${COUNT_TIME}ms"
+echo "Time: ${AVG_TIME_INT}ms | Race rate: ${AVG_DATA_RACE}%"
+#echo "Time: ${AVG_TIME_INT}ms | Race rate: ${AVG_DATA_RACE}% | No. 99.9%: ${NO_99} | Time 99.9%: ${TIME_99}ms"
--- /dev/null
+#!/bin/bash
+set -e
+set -u
+
+# Paul: skip `spsc-queue` as it deadlocks.
+
+for t in barrier chase-lev-deque dekker-fences linuxrwlocks mcs-lock mpmc-queue ms-queue; do
+ cd $t
+ echo -n "$t "
+ ../test.sh ./$t
+ cd ..
+done
+
projects / c11concurrency-benchmarks.git / commitdiff