Changeset fc12f05 for libcfa/src/concurrency

Timestamp:

Nov 13, 2023, 3:43:43 AM (2 years ago)

Author:

JiadaL <j82liang@…>

Branches:

master

Children:

25f2798

Parents:

0030b508 (diff), 2174191 (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:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

Location:

libcfa/src/concurrency

Files:

• channel.hfa (modified) (2 diffs)
• cofor.cfa (modified) (3 diffs)
• cofor.hfa (modified) (3 diffs)
• kernel.hfa (modified) (1 diff)
• locks.hfa (modified) (1 diff)

libcfa/src/concurrency/channel.hfa

@@ -130 +130 @@
 static inline void __cons_handoff( channel(T) & chan, T & elem ) with(chan) {
     memcpy( cons`first.extra, (void *)&elem, sizeof(T) ); // do waiting consumer work
-    __atomic_thread_fence( __ATOMIC_SEQ_CST );
     wake_one( cons );
 }
@@ -137 +136 @@
 static inline void __prods_handoff( channel(T) & chan, T & retval ) with(chan) {
     memcpy( (void *)&retval, prods`first.extra, sizeof(T) );
-    __atomic_thread_fence( __ATOMIC_SEQ_CST );
     wake_one( prods );
 }

libcfa/src/concurrency/cofor.cfa

@@ -4 +4 @@
 // cofor ( uC++ COFOR )
 
-thread cofor_runner {
+thread cofor_task {
         ssize_t low, high;
         __cofor_body_t loop_body;
 };
 
-static void ?{}( cofor_runner & this, ssize_t low, ssize_t high, __cofor_body_t loop_body ) {
+static void ?{}( cofor_task & this, ssize_t low, ssize_t high, __cofor_body_t loop_body ) {
         this.low = low;
         this.high = high;
@@ -15 +15 @@
 }
 
-void main( cofor_runner & this ) with( this ) {
+void main( cofor_task & this ) with( this ) {
         for ( ssize_t i = low; i < high; i++ )
                 loop_body(i);
@@ -29 +29 @@
         ssize_t i = 0;
         ssize_t stride_iter = low;
-        cofor_runner * runners[ threads ];
+        cofor_task * runners[ threads ];
         for ( i; threads ) {
                 runners[i] = alloc();

libcfa/src/concurrency/cofor.hfa

@@ -1 +1 @@
 #include <thread.hfa>
+#include <locks.hfa>
+#include <list.hfa>
 
 //////////////////////////////////////////////////////////////////////////////////////////
@@ -14 +16 @@
                 __Cofor__( low, high, __CFA_loopLambda__ ); \
         }
+
+struct runner_node {
+    void * value;
+    inline dlink(runner_node);
+};
+P9_EMBEDDED( runner_node, dlink(runner_node) )
+
+thread cofor_runner {
+        go_mutex mutex_lock;              // MX lock
+    dlist( runner_node ) items;
+    void (*func)(void *);
+    volatile bool done;
+};
+
+void ?{}( cofor_runner & this ) { this.done = false; }
+
+void main( cofor_runner & this ) with(this) {
+    while ( !done || !items`isEmpty ) {
+                lock( mutex_lock );
+        runner_node * node = &try_pop_front( items );
+                unlock( mutex_lock );
+        func( node->value );
+        free( node->value );
+        free( node );
+    }
+}
+
+void start_runners( cofor_runner * thds, unsigned nprocs, void (*func)(void *) ) {
+        for ( i; nprocs ) {
+                thds[i].func = func;
+        }
+}
+
+void end_runners( cofor_runner * thds, unsigned nprocs ) {
+        for ( i; nprocs ) {
+                thds[i].done = true;
+        }
+}
+
+void send_work( cofor_runner * thds, unsigned nprocs, unsigned & curr_proc, void * value ) {
+        runner_node * node = malloc();
+        (*node){};
+        node->value = value;
+        lock( thds[curr_proc].mutex_lock );
+        insert_last( thds[curr_proc].items, *node );
+        unlock( thds[curr_proc].mutex_lock );
+        curr_proc = ( curr_proc + 1 ) % nprocs;
+}
 
 //////////////////////////////////////////////////////////////////////////////////////////
@@ -42 +92 @@
     delete( this.runner );
 }
-

libcfa/src/concurrency/kernel.hfa

@@ -303 +303 @@
 // gets the number of constructed processors on the cluster
 static inline unsigned get_proc_count( cluster & this ) { return this.procs.constructed; }
+static inline unsigned get_proc_count() { return publicTLS_get( this_processor )->cltr->procs.constructed; }
 
 // set the number of internal processors

libcfa/src/concurrency/locks.hfa

@@ -182 +182 @@
 static inline void lock( mcs_spin_lock & l, mcs_spin_node & n ) {
     n.locked = true;
+
+        #if defined(__ARM_ARCH)
+        __asm__ __volatile__ ( "DMB ISH" ::: );
+        #endif
+
         mcs_spin_node * prev = __atomic_exchange_n(&l.queue.tail, &n, __ATOMIC_SEQ_CST);
         if( prev == 0p ) return;
         prev->next = &n;
+        
+        #if defined(__ARM_ARCH)
+        __asm__ __volatile__ ( "DMB ISH" ::: );
+        #endif
+
         while( __atomic_load_n(&n.locked, __ATOMIC_RELAXED) ) Pause();
+
+        #if defined(__ARM_ARCH)
+        __asm__ __volatile__ ( "DMB ISH" ::: );
+        #endif
 }
 
 static inline void unlock(mcs_spin_lock & l, mcs_spin_node & n) {
+        #if defined(__ARM_ARCH)
+        __asm__ __volatile__ ( "DMB ISH" ::: );
+        #endif
+
         mcs_spin_node * n_ptr = &n;
         if (__atomic_compare_exchange_n(&l.queue.tail, &n_ptr, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) return;
         while (__atomic_load_n(&n.next, __ATOMIC_RELAXED) == 0p) Pause();
+
+        #if defined(__ARM_ARCH)
+        __asm__ __volatile__ ( "DMB ISH" ::: );
+        #endif
+
         n.next->locked = false;
 }