Changes in libcfa/src/concurrency/locks.hfa [0fc447c:bbe3719]

File:

• libcfa/src/concurrency/locks.hfa (modified) (2 diffs)

libcfa/src/concurrency/locks.hfa

@@ -31 +31 @@
 
 //-----------------------------------------------------------------------------
+// Semaphores
+
+// '0-nary' semaphore
+// Similar to a counting semaphore except the value of one is never reached
+// as a consequence, a V() that would bring the value to 1 *spins* until
+// a P consumes it
+struct Semaphore0nary {
+        __spinlock_t lock; // needed to protect
+        mpsc_queue(thread$) queue;
+};
+
+static inline bool P(Semaphore0nary & this, thread$ * thrd) {
+        /* paranoid */ verify(!thrd`next);
+        /* paranoid */ verify(!(&(*thrd)`next));
+
+        push(this.queue, thrd);
+        return true;
+}
+
+static inline bool P(Semaphore0nary & this) {
+    thread$ * thrd = active_thread();
+    P(this, thrd);
+    park();
+    return true;
+}
+
+static inline thread$ * V(Semaphore0nary & this, bool doUnpark = true) {
+        thread$ * next;
+        lock(this.lock __cfaabi_dbg_ctx2);
+                for (;;) {
+                        next = pop(this.queue);
+                        if (next) break;
+                        Pause();
+                }
+        unlock(this.lock);
+
+        if (doUnpark) unpark(next);
+        return next;
+}
+
+// Wrapper used on top of any sempahore to avoid potential locking
+struct BinaryBenaphore {
+        volatile ssize_t counter;
+};
+
+static inline {
+        void ?{}(BinaryBenaphore & this) { this.counter = 0; }
+        void ?{}(BinaryBenaphore & this, zero_t) { this.counter = 0; }
+        void ?{}(BinaryBenaphore & this, one_t ) { this.counter = 1; }
+
+        // returns true if no blocking needed
+        bool P(BinaryBenaphore & this) {
+                return __atomic_fetch_sub(&this.counter, 1, __ATOMIC_SEQ_CST) > 0;
+        }
+
+        bool tryP(BinaryBenaphore & this) {
+                ssize_t c = this.counter;
+                /* paranoid */ verify( c > MIN );
+                return (c >= 1) && __atomic_compare_exchange_n(&this.counter, &c, c-1, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
+        }
+
+        // returns true if notify needed
+        bool V(BinaryBenaphore & this) {
+                ssize_t c = 0;
+                for () {
+                        /* paranoid */ verify( this.counter < MAX );
+                        if (__atomic_compare_exchange_n(&this.counter, &c, c+1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                if (c == 0) return true;
+                                /* paranoid */ verify(c < 0);
+                                return false;
+                        } else {
+                                if (c == 1) return true;
+                                /* paranoid */ verify(c < 1);
+                                Pause();
+                        }
+                }
+        }
+}
+
+// Binary Semaphore based on the BinaryBenaphore on top of the 0-nary Semaphore
+struct ThreadBenaphore {
+        BinaryBenaphore ben;
+        Semaphore0nary  sem;
+};
+
+static inline void ?{}(ThreadBenaphore & this) {}
+static inline void ?{}(ThreadBenaphore & this, zero_t) { (this.ben){ 0 }; }
+static inline void ?{}(ThreadBenaphore & this, one_t ) { (this.ben){ 1 }; }
+
+static inline bool P(ThreadBenaphore & this)              { return P(this.ben) ? false : P(this.sem); }
+static inline bool tryP(ThreadBenaphore & this)           { return tryP(this.ben); }
+static inline bool P(ThreadBenaphore & this, bool wait)   { return wait ? P(this) : tryP(this); }
+
+static inline thread$ * V(ThreadBenaphore & this, bool doUnpark = true) {
+        if (V(this.ben)) return 0p;
+        return V(this.sem, doUnpark);
+}
+
+//-----------------------------------------------------------------------------
 // Semaphore
 struct semaphore {
@@ -72 +171 @@
 static inline void   on_wakeup( owner_lock & this, size_t v ) { on_wakeup ( (blocking_lock &)this, v ); }
 static inline void   on_notify( owner_lock & this, struct thread$ * t ) { on_notify( (blocking_lock &)this, t ); }
+
+struct fast_lock {
+        thread$ * volatile owner;
+        ThreadBenaphore sem;
+};
+
+static inline void ?{}(fast_lock & this) { this.owner = 0p; }
+
+static inline bool $try_lock(fast_lock & this, thread$ * thrd) {
+    thread$ * exp = 0p;
+    return __atomic_compare_exchange_n(&this.owner, &exp, thrd, false, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED);
+}
+
+static inline void lock( fast_lock & this ) __attribute__((artificial));
+static inline void lock( fast_lock & this ) {
+        thread$ * thrd = active_thread();
+        /* paranoid */verify(thrd != this.owner);
+
+        for (;;) {
+                if ($try_lock(this, thrd)) return;
+                P(this.sem);
+        }
+}
+
+static inline bool try_lock( fast_lock & this ) __attribute__((artificial));
+static inline bool try_lock ( fast_lock & this ) {
+        thread$ * thrd = active_thread();
+        /* paranoid */ verify(thrd != this.owner);
+        return $try_lock(this, thrd);
+}
+
+static inline thread$ * unlock( fast_lock & this ) __attribute__((artificial));
+static inline thread$ * unlock( fast_lock & this ) {
+        /* paranoid */ verify(active_thread() == this.owner);
+
+        // open 'owner' before unlocking anyone
+        // so new and unlocked threads don't park incorrectly.
+        // This may require additional fencing on ARM.
+        this.owner = 0p;
+
+        return V(this.sem);
+}
+
+static inline size_t on_wait( fast_lock & this ) { unlock(this); return 0; }
+static inline void on_wakeup( fast_lock & this, size_t ) { lock(this); }
+static inline void on_notify( fast_lock &, struct thread$ * t ) { unpark(t); }
 
 struct mcs_node {