Changeset 1da7397 for libcfa

Timestamp:

Mar 27, 2021, 6:04:14 PM (5 years ago)

Author:

Peter A. Buhr <pabuhr@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

fec3e9a

Parents:

2644610 (diff), f8a7fed (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

fix conflict

Location:

libcfa/src

Files:

• concurrency/clib/cfathread.cfa (modified) (4 diffs)
• concurrency/invoke.h (modified) (1 diff)
• concurrency/kernel.hfa (modified) (1 diff)
• concurrency/kernel/startup.cfa (modified) (1 diff)
• concurrency/kernel_private.hfa (modified) (1 diff)
• concurrency/locks.hfa (modified) (3 diffs)
• concurrency/ready_queue.cfa (modified) (7 diffs)
• containers/queueLockFree.hfa (modified) (7 diffs)

libcfa/src/concurrency/clib/cfathread.cfa

@@ -58 +58 @@
         this.themain = themain;
         this.arg = arg;
-        ((thread&)this){"C-thread", cl};
+        (this.self){"C-thread", cl};
         __thrd_start(this, main);
 }
@@ -102 +102 @@
         this.init = init;
         this.arg = arg;
-        ((thread&)this){"Processir Init"};
+        (this.self){"Processir Init"};
 
         // Don't use __thrd_start! just prep the context manually
@@ -312 +312 @@
 
         ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
-                return cfa_write(fildes, buf, nbyte, CFA_IO_LAZY);
+                // Use send rather then write for socket since it's faster
+                return cfa_send(fildes, buf, nbyte, 0, CFA_IO_LAZY);
         }
 
@@ -335 +336 @@
 
         ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
-                return cfa_read(fildes, buf, nbyte, CFA_IO_LAZY);
-        }
-
-}
+                // Use recv rather then read for socket since it's faster
+                return cfa_recv(fildes, buf, nbyte, 0, CFA_IO_LAZY);
+        }
+
+}

libcfa/src/concurrency/invoke.h

@@ -225 +225 @@
 
                 static inline $thread * volatile & ?`next ( $thread * this )  __attribute__((const)) {
-                        return this->seqable.back;
+                        return this->seqable.next;
                 }
 

libcfa/src/concurrency/kernel.hfa

@@ -69 +69 @@
         // Cluster from which to get threads
         struct cluster * cltr;
+
+        // Id within the cluster
+        unsigned cltr_id;
 
         // Set to true to notify the processor should terminate

libcfa/src/concurrency/kernel/startup.cfa

@@ -486 +486 @@
 
                 // Adjust the ready queue size
-                ready_queue_grow( cltr, target );
+                this.cltr_id = ready_queue_grow( cltr, target );
 
         // Unlock the RWlock

libcfa/src/concurrency/kernel_private.hfa

@@ -278 +278 @@
 //-----------------------------------------------------------------------
 // Increase the width of the ready queue (number of lanes) by 4
-void ready_queue_grow  (struct cluster * cltr, int target);
+unsigned ready_queue_grow  (struct cluster * cltr, int target);
 
 //-----------------------------------------------------------------------

libcfa/src/concurrency/locks.hfa

@@ -20 +20 @@
 
 #include "bits/weakso_locks.hfa"
+#include "containers/queueLockFree.hfa"
+
+#include "thread.hfa"
 
 #include "time_t.hfa"
 #include "time.hfa"
+
+//-----------------------------------------------------------------------------
+// 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) __attribute__((artificial));
+static inline bool P(Semaphore0nary & this, $thread * thrd) {
+        /* paranoid */ verify(!(thrd->seqable.next));
+        /* paranoid */ verify(!(thrd`next));
+
+        push(this.queue, thrd);
+        return true;
+}
+
+static inline bool P(Semaphore0nary & this) __attribute__((artificial));
+static inline bool P(Semaphore0nary & this) {
+    $thread * thrd = active_thread();
+    P(this, thrd);
+    park();
+    return true;
+}
+
+static inline $thread * V(Semaphore0nary & this, const bool doUnpark = true) __attribute__((artificial));
+static inline $thread * V(Semaphore0nary & this, const 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;
+                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 () {
+                        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 P(ThreadBenaphore & this, $thread * t) { return /* P(this.ben) ? false : */ P(this.sem, t ); }
+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, const bool doUnpark = true) {
+        // if (V(this.ben)) return 0p;
+        return V(this.sem, doUnpark);
+}
+
+//-----------------------------------------------------------------------------
+// Semaphore
+struct semaphore {
+        __spinlock_t lock;
+        int count;
+        __queue_t($thread) waiting;
+};
+
+void  ?{}(semaphore & this, int count = 1);
+void ^?{}(semaphore & this);
+bool   P (semaphore & this);
+bool   V (semaphore & this);
+bool   V (semaphore & this, unsigned count);
 
 //----------
@@ -54 +169 @@
 static inline size_t get_recursion_count( owner_lock & this ) { return get_recursion_count( (blocking_lock &)this ); }
 
+struct fast_lock {
+        $thread * volatile owner;
+        ThreadBenaphore sem;
+};
+
+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, $thread * thrd) {
+        /* paranoid */verify(thrd != this.owner);
+
+        for (;;) {
+                if ($try_lock(this, thrd)) return;
+                P(this.sem, thrd);
+        }
+}
+
+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 void try_lock ( fast_lock & this ) {
+        $thread * thrd = active_thread();
+        /* paranoid */ verify(thrd != this.owner);
+        return $try_lock(this, thrd);
+}
+
+static inline void unlock( fast_lock & this ) {
+        $thread * thrd = active_thread();
+        /* paranoid */ verify(thrd == this.owner);
+        $thread * next = V(this.sem, false); // implicit fence
+        // open 'owner' only after fence
+        this.owner = 0p;
+
+        // Unpark the next person (can be 0p, unpark handles it)
+        unpark(next);
+}
+
+static inline void on_wait( fast_lock & this ) {
+        unlock(this);
+        #warning this is broken
+}
+
+static inline void on_notify( fast_lock & this, struct $thread * t ) {
+        $lock(this, t);
+        #warning this is broken
+}
+
+static inline void   set_recursion_count( fast_lock & this, size_t recursion ) {}
+static inline size_t get_recursion_count( fast_lock & this ) { return 0; }
+
+struct mcs_node {
+        mcs_node * volatile next;
+        single_sem sem;
+};
+
+static inline void ?{}(mcs_node & this) { this.next = 0p; }
+
+static inline mcs_node * volatile & ?`next ( mcs_node * node ) {
+        return node->next;
+}
+
+struct mcs_lock {
+        mcs_queue(mcs_node) queue;
+};
+
+static inline void lock(mcs_lock & l, mcs_node & n) {
+        if(push(l.queue, &n))
+                wait(n.sem);
+}
+
+static inline void unlock(mcs_lock & l, mcs_node & n) {
+        mcs_node * next = advance(l.queue, &n);
+        if(next) post(next->sem);
+}
+
 //-----------------------------------------------------------------------------
 // is_blocking_lock
@@ -121 +320 @@
         bool wait( condition_variable(L) & this, L & l, uintptr_t info, Time time );
 }
-
-//-----------------------------------------------------------------------------
-// Semaphore
-struct semaphore {
-        __spinlock_t lock;
-        int count;
-        __queue_t($thread) waiting;
-};
-
-void  ?{}(semaphore & this, int count = 1);
-void ^?{}(semaphore & this);
-bool   P (semaphore & this);
-bool   V (semaphore & this);
-bool   V (semaphore & this, unsigned count);

libcfa/src/concurrency/ready_queue.cfa

@@ -39 +39 @@
 #endif
 
-#define BIAS 16
+#define BIAS 4
 
 // returns the maximum number of processors the RWLock support
@@ -252 +252 @@
                 preferred =
                         //*
-                        kernelTLS().this_processor ? kernelTLS().this_processor->id * 4 : -1;
+                        kernelTLS().this_processor ? kernelTLS().this_processor->cltr_id : -1;
                         /*/
                         thrd->link.preferred * 4;
@@ -330 +330 @@
         #if defined(BIAS)
                 // Don't bother trying locally too much
-                preferred = kernelTLS().this_processor->id * 4;
+                preferred = kernelTLS().this_processor->cltr_id;
         #endif
 
@@ -352 +352 @@
 
                 #if !defined(__CFA_NO_STATISTICS__)
-                        if(locali) {
-                                __tls_stats()->ready.pick.pop.local++;
-                        }
-                        if(localj) {
+                        if(locali && localj) {
                                 __tls_stats()->ready.pick.pop.local++;
                         }
@@ -528 +525 @@
 
 // Grow the ready queue
-void ready_queue_grow  (struct cluster * cltr, int target) {
+unsigned ready_queue_grow(struct cluster * cltr, int target) {
+        unsigned preferred;
+        size_t ncount;
+
         /* paranoid */ verify( ready_mutate_islocked() );
         __cfadbg_print_safe(ready_queue, "Kernel : Growing ready queue\n");
@@ -543 +543 @@
                 // Find new count
                 // Make sure we always have atleast 1 list
-                size_t ncount = target >= 2 ? target * 4: 1;
+                if(target >= 2) {
+                        ncount = target * 4;
+                        preferred = ncount - 4;
+                } else {
+                        ncount = 1;
+                        preferred = 0;
+                }
 
                 // Allocate new array (uses realloc and memcpies the data)
@@ -578 +584 @@
 
         /* paranoid */ verify( ready_mutate_islocked() );
+        return preferred;
 }
 

libcfa/src/containers/queueLockFree.hfa

@@ -20 +20 @@
                 // Adds an element to the list
                 // Multi-Thread Safe, Lock-Free
-                T * push(mcs_queue(T) & this, T & elem) {
-                        /* paranoid */ verify(!(&elem)`next);
+                T * push(mcs_queue(T) & this, T * elem) __attribute__((artificial));
+                T * push(mcs_queue(T) & this, T * elem) {
+                        /* paranoid */ verify(!(elem`next));
                         // Race to add to the tail
-                        T * prev = __atomic_exchange_n(&this.tail, &elem, __ATOMIC_SEQ_CST);
+                        T * prev = __atomic_exchange_n(&this.tail, elem, __ATOMIC_SEQ_CST);
                         // If we aren't the first, we need to tell the person before us
                         // No need to
-                        if (prev) prev`next = &elem;
+                        if (prev) prev`next = elem;
                         return prev;
                 }
@@ -33 +34 @@
                 // Passing an element that is not the head is undefined behavior
                 // NOT Multi-Thread Safe, concurrent pushes are safe
-                T * advance(mcs_queue(T) & this, T & elem) {
-                        T * expected = &elem;
+                T * advance(mcs_queue(T) & this, T * elem) __attribute__((artificial));
+                T * advance(mcs_queue(T) & this, T * elem) {
+                        T * expected = elem;
                         // Check if this is already the last item
                         if (__atomic_compare_exchange_n(&this.tail, &expected, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) return 0p;
 
-                        // If not wait for next item to show-up
-                        // added by push
-                        while (!(&elem)`next) Pause();
-                        return (&elem)`next;
+                        // If not wait for next item to show-up, filled by push
+                        while (!(elem`next)) Pause();
+
+                        // we need to return if the next link was empty
+                        T * ret = elem`next;
+
+                        // invalidate link to reset to initial state
+                        elem`next = 0p;
+                        return ret;
                 }
         }
@@ -64 +71 @@
                 // Added a new element to the queue
                 // Multi-Thread Safe, Lock-Free
-                T * push(mpsc_queue(T) & this, T & elem) {
+                T * push(mpsc_queue(T) & this, T * elem) __attribute__((artificial));
+                T * push(mpsc_queue(T) & this, T * elem) {
                         T * prev = push((mcs_queue(T)&)this, elem);
-                        if (!prev) this.head = &elem;
+                        if (!prev) this.head = elem;
                         return prev;
                 }
@@ -74 +82 @@
                 // next is set to the new head of the queue
                 // NOT Multi-Thread Safe
+                T * pop(mpsc_queue(T) & this, T *& next) __attribute__((artificial));
                 T * pop(mpsc_queue(T) & this, T *& next) {
                         T * elem = this.head;
@@ -84 +93 @@
                                 // force memory sync
                                 __atomic_thread_fence(__ATOMIC_SEQ_CST);
+
+                                // invalidate link to reset to initial state
+                                elem`next = 0p;
                         }
                         // Otherwise, there might be a race where it only looks but someone is enqueuing
@@ -91 +103 @@
                                 // after that point, it could overwrite the write in push
                                 this.head = 0p;
-                                next = advance((mcs_queue(T)&)this, (*elem));
+                                next = advance((mcs_queue(T)&)this, elem);
 
                                 // Only write to the head if there is a next element
@@ -98 +110 @@
                                 if (next) this.head = next;
                         }
-
-                        // invalidate link
-                        elem`next = 0p;
 
                         // return removed element