Changeset 28f3a19 for src/libcfa/concurrency

Timestamp:

Jun 27, 2018, 3:28:41 PM (8 years ago)

Author:

Aaron Moss <a3moss@…>

Branches:

new-env, with_gc

Children:

b21c77a

Parents:

0182bfa (diff), 63238a4 (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' into with_gc

Location:

src/libcfa/concurrency

Files:

• invoke.h (modified) (3 diffs)
• kernel (modified) (5 diffs)
• kernel.c (modified) (14 diffs)
• kernel_private.h (modified) (3 diffs)
• monitor.c (modified) (1 diff)
• mutex (added)
• mutex.c (added)
• preemption.c (modified) (9 diffs)

src/libcfa/concurrency/invoke.h

@@ -10 +10 @@
 // Created On       : Tue Jan 17 12:27:26 2016
 // Last Modified By : Peter A. Buhr
-// Last Modified On : Fri Mar 30 22:33:59 2018
-// Update Count     : 30
+// Last Modified On : Sat May 19 08:23:21 2018
+// Update Count     : 31
 //
 
@@ -18 +18 @@
 #include "bits/locks.h"
 
-#define TL_GET( member ) kernelTLS.member
-#define TL_SET( member, value ) kernelTLS.member = value;
-
 #ifdef __cforall
 extern "C" {
@@ -28 +25 @@
 #ifndef _INVOKE_H_
 #define _INVOKE_H_
+
+#ifdef __ARM_ARCH
+        // function prototypes are only really used by these macros on ARM
+        void disable_global_interrupts();
+        void enable_global_interrupts();
+
+        #define TL_GET( member ) ( { __typeof__( kernelTLS.member ) target; \
+                disable_global_interrupts(); \
+                target = kernelTLS.member; \
+                enable_global_interrupts(); \
+                target; } )
+        #define TL_SET( member, value ) disable_global_interrupts(); \
+                kernelTLS.member = value; \
+                enable_global_interrupts();
+#else
+        #define TL_GET( member ) kernelTLS.member
+        #define TL_SET( member, value ) kernelTLS.member = value;
+#endif
 
         #ifdef __cforall

src/libcfa/concurrency/kernel

@@ -23 +23 @@
 extern "C" {
 #include <pthread.h>
+#include <semaphore.h>
 }
 
@@ -43 +44 @@
 extern struct cluster * mainCluster;
 
-enum FinishOpCode { No_Action, Release, Schedule, Release_Schedule, Release_Multi, Release_Multi_Schedule };
+enum FinishOpCode { No_Action, Release, Schedule, Release_Schedule, Release_Multi, Release_Multi_Schedule, Callback };
+
+typedef void (*__finish_callback_fptr_t)(void);
 
 //TODO use union, many of these fields are mutually exclusive (i.e. MULTI vs NOMULTI)
 struct FinishAction {
         FinishOpCode action_code;
+        /*
+        // Union of possible actions
+        union {
+                // Option 1 : locks and threads
+                struct {
+                        // 1 thread or N thread
+                        union {
+                                thread_desc * thrd;
+                                struct {
+                                        thread_desc ** thrds;
+                                        unsigned short thrd_count;
+                                };
+                        };
+                        // 1 lock or N lock
+                        union {
+                                __spinlock_t * lock;
+                                struct {
+                                        __spinlock_t ** locks;
+                                        unsigned short lock_count;
+                                };
+                        };
+                };
+                // Option 2 : action pointer
+                __finish_callback_fptr_t callback;
+        };
+        /*/
         thread_desc * thrd;
+        thread_desc ** thrds;
+        unsigned short thrd_count;
         __spinlock_t * lock;
         __spinlock_t ** locks;
         unsigned short lock_count;
-        thread_desc ** thrds;
-        unsigned short thrd_count;
+        __finish_callback_fptr_t callback;
+        //*/
 };
 static inline void ?{}(FinishAction & this) {
@@ -82 +113 @@
         pthread_t kernel_thread;
 
+        // RunThread data
+        // Action to do after a thread is ran
+        struct FinishAction finish;
+
+        // Preemption data
+        // Node which is added in the discrete event simulaiton
+        struct alarm_node_t * preemption_alarm;
+
+        // If true, a preemption was triggered in an unsafe region, the processor must preempt as soon as possible
+        bool pending_preemption;
+
+        // Idle lock
+        __bin_sem_t idleLock;
+
         // Termination
         // Set to true to notify the processor should terminate
@@ -89 +134 @@
         semaphore terminated;
 
-        // RunThread data
-        // Action to do after a thread is ran
-        struct FinishAction finish;
-
-        // Preemption data
-        // Node which is added in the discrete event simulaiton
-        struct alarm_node_t * preemption_alarm;
-
-        // If true, a preemption was triggered in an unsafe region, the processor must preempt as soon as possible
-        bool pending_preemption;
-
-        // Idle lock
-
         // Link lists fields
-        struct {
+        struct __dbg_node_proc {
                 struct processor * next;
                 struct processor * prev;
@@ -150 +182 @@
 
         // Link lists fields
-        struct {
+        struct __dbg_node_cltr {
                 cluster * next;
                 cluster * prev;

src/libcfa/concurrency/kernel.c

@@ -16 +16 @@
 //C Includes
 #include <stddef.h>
+#include <errno.h>
+#include <string.h>
 extern "C" {
 #include <stdio.h>
@@ -49 +51 @@
 thread_desc * mainThread;
 
-struct { __dllist_t(cluster    ) list; __spinlock_t lock; } global_clusters;
+extern "C" {
+struct { __dllist_t(cluster) list; __spinlock_t lock; } __cfa_dbg_global_clusters;
+}
 
 //-----------------------------------------------------------------------------
@@ -143 +147 @@
         runner.proc = &this;
 
+        idleLock{};
+
         start( &this );
 }
 
 void ^?{}(processor & this) with( this ){
-        if( ! do_terminate ) {
+        if( ! __atomic_load_n(&do_terminate, __ATOMIC_ACQUIRE) ) {
                 __cfaabi_dbg_print_safe("Kernel : core %p signaling termination\n", &this);
-                terminate(&this);
-                verify(this.do_terminate);
-                verify( kernelTLS.this_processor != &this);
+
+                __atomic_store_n(&do_terminate, true, __ATOMIC_RELAXED);
+                wake( &this );
+
                 P( terminated );
                 verify( kernelTLS.this_processor != &this);
-                pthread_join( kernel_thread, NULL );
-        }
+        }
+
+        pthread_join( kernel_thread, NULL );
 }
 
@@ -194 +202 @@
 
                 thread_desc * readyThread = NULL;
-                for( unsigned int spin_count = 0; ! this->do_terminate; spin_count++ )
+                for( unsigned int spin_count = 0; ! __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST); spin_count++ )
                 {
                         readyThread = nextThread( this->cltr );
@@ -213 +221 @@
                         else
                         {
-                                spin(this, &spin_count);
+                                // spin(this, &spin_count);
+                                halt(this);
                         }
                 }
@@ -257 +266 @@
 // its final actions must be executed from the kernel
 void finishRunning(processor * this) with( this->finish ) {
-        if( action_code == Release ) {
-                verify( ! kernelTLS.preemption_state.enabled );
+        verify( ! kernelTLS.preemption_state.enabled );
+        choose( action_code ) {
+        case No_Action:
+                break;
+        case Release:
                 unlock( *lock );
-        }
-        else if( action_code == Schedule ) {
+        case Schedule:
                 ScheduleThread( thrd );
-        }
-        else if( action_code == Release_Schedule ) {
-                verify( ! kernelTLS.preemption_state.enabled );
+        case Release_Schedule:
                 unlock( *lock );
                 ScheduleThread( thrd );
-        }
-        else if( action_code == Release_Multi ) {
-                verify( ! kernelTLS.preemption_state.enabled );
+        case Release_Multi:
                 for(int i = 0; i < lock_count; i++) {
                         unlock( *locks[i] );
                 }
-        }
-        else if( action_code == Release_Multi_Schedule ) {
+        case Release_Multi_Schedule:
                 for(int i = 0; i < lock_count; i++) {
                         unlock( *locks[i] );
@@ -282 +288 @@
                         ScheduleThread( thrds[i] );
                 }
-        }
-        else {
-                assert(action_code == No_Action);
-        }
-}
-
-// Handles spinning logic
-// TODO : find some strategy to put cores to sleep after some time
-void spin(processor * this, unsigned int * spin_count) {
-        (*spin_count)++;
+        case Callback:
+                callback();
+        default:
+                abort("KERNEL ERROR: Unexpected action to run after thread");
+        }
 }
 
@@ -396 +397 @@
         with( *thrd->curr_cluster ) {
                 lock  ( ready_queue_lock __cfaabi_dbg_ctx2 );
+                bool was_empty = !(ready_queue != 0);
                 append( ready_queue, thrd );
                 unlock( ready_queue_lock );
+
+                if(was_empty) {
+                        lock      (proc_list_lock __cfaabi_dbg_ctx2);
+                        if(idles) {
+                                wake_fast(idles.head);
+                        }
+                        unlock    (proc_list_lock);
+                }
+                else if( struct processor * idle = idles.head ) {
+                        wake_fast(idle);
+                }
+
         }
 
@@ -497 +511 @@
 }
 
+void BlockInternal(__finish_callback_fptr_t callback) {
+        disable_interrupts();
+        with( *kernelTLS.this_processor ) {
+                finish.action_code = Callback;
+                finish.callback    = callback;
+        }
+
+        verify( ! kernelTLS.preemption_state.enabled );
+        returnToKernel();
+        verify( ! kernelTLS.preemption_state.enabled );
+
+        enable_interrupts( __cfaabi_dbg_ctx );
+}
+
 // KERNEL ONLY
 void LeaveThread(__spinlock_t * lock, thread_desc * thrd) {
@@ -518 +546 @@
         __cfaabi_dbg_print_safe("Kernel : Starting\n");
 
-        global_clusters.list{ __get };
-        global_clusters.lock{};
+        __cfa_dbg_global_clusters.list{ __get };
+        __cfa_dbg_global_clusters.lock{};
 
         // Initialize the main cluster
@@ -600 +628 @@
         // When its coroutine terminates, it return control to the mainThread
         // which is currently here
-        mainProcessor->do_terminate = true;
+        __atomic_store_n(&mainProcessor->do_terminate, true, __ATOMIC_RELEASE);
         returnToKernel();
+        mainThread->self_cor.state = Halted;
 
         // THE SYSTEM IS NOW COMPLETELY STOPPED
@@ -617 +646 @@
         ^(mainThread){};
 
-        ^(global_clusters.list){};
-        ^(global_clusters.lock){};
+        ^(__cfa_dbg_global_clusters.list){};
+        ^(__cfa_dbg_global_clusters.lock){};
 
         __cfaabi_dbg_print_safe("Kernel : Shutdown complete\n");
@@ -627 +656 @@
 //=============================================================================================
 
-// void halt(processor * this) with( this ) {
-//      pthread_mutex_lock( &idle.lock );
-
-
-
-//      // SKULLDUGGERY: Even if spurious wake-up is a thing
-//      // spuriously waking up a kernel thread is not a big deal
-//      // if it is very rare.
-//      pthread_cond_wait( &idle.cond, &idle.lock);
-//      pthread_mutex_unlock( &idle.lock );
-// }
-
-// void wake(processor * this) with( this ) {
-//      pthread_mutex_lock  (&idle.lock);
-//      pthread_cond_signal (&idle.cond);
-//      pthread_mutex_unlock(&idle.lock);
-// }
+void halt(processor * this) with( *this ) {
+        // verify( ! __atomic_load_n(&do_terminate, __ATOMIC_SEQ_CST) );
+
+        with( *cltr ) {
+                lock      (proc_list_lock __cfaabi_dbg_ctx2);
+                remove    (procs, *this);
+                push_front(idles, *this);
+                unlock    (proc_list_lock);
+        }
+
+        __cfaabi_dbg_print_safe("Kernel : Processor %p ready to sleep\n", this);
+
+        wait( idleLock );
+
+        __cfaabi_dbg_print_safe("Kernel : Processor %p woke up and ready to run\n", this);
+
+        with( *cltr ) {
+                lock      (proc_list_lock __cfaabi_dbg_ctx2);
+                remove    (idles, *this);
+                push_front(procs, *this);
+                unlock    (proc_list_lock);
+        }
+}
 
 //=============================================================================================
@@ -758 +793 @@
 // Global Queues
 void doregister( cluster     & cltr ) {
-        lock      ( global_clusters.lock __cfaabi_dbg_ctx2);
-        push_front( global_clusters.list, cltr );
-        unlock    ( global_clusters.lock );
+        lock      ( __cfa_dbg_global_clusters.lock __cfaabi_dbg_ctx2);
+        push_front( __cfa_dbg_global_clusters.list, cltr );
+        unlock    ( __cfa_dbg_global_clusters.lock );
 }
 
 void unregister( cluster     & cltr ) {
-        lock  ( global_clusters.lock __cfaabi_dbg_ctx2);
-        remove( global_clusters.list, cltr );
-        unlock( global_clusters.lock );
+        lock  ( __cfa_dbg_global_clusters.lock __cfaabi_dbg_ctx2);
+        remove( __cfa_dbg_global_clusters.list, cltr );
+        unlock( __cfa_dbg_global_clusters.lock );
 }
 

src/libcfa/concurrency/kernel_private.h

@@ -48 +48 @@
 void BlockInternal(__spinlock_t * locks [], unsigned short count);
 void BlockInternal(__spinlock_t * locks [], unsigned short count, thread_desc * thrds [], unsigned short thrd_count);
+void BlockInternal(__finish_callback_fptr_t callback);
 void LeaveThread(__spinlock_t * lock, thread_desc * thrd);
 
@@ -56 +57 @@
 void runThread(processor * this, thread_desc * dst);
 void finishRunning(processor * this);
-void terminate(processor * this);
-void spin(processor * this, unsigned int * spin_count);
+void halt(processor * this);
+
+static inline void wake_fast(processor * this) {
+        __cfaabi_dbg_print_safe("Kernel : Waking up processor %p\n", this);
+        post( this->idleLock );
+}
+
+static inline void wake(processor * this) {
+        disable_interrupts();
+        wake_fast(this);
+        enable_interrupts( __cfaabi_dbg_ctx );
+}
 
 struct event_kernel_t {
@@ -65 +76 @@
 
 extern event_kernel_t * event_kernel;
-
-//extern thread_local coroutine_desc * volatile this_coroutine;
-//extern thread_local thread_desc *    volatile this_thread;
-//extern thread_local processor *      volatile this_processor;
-
-// extern volatile thread_local bool preemption_in_progress;
-// extern volatile thread_local bool preemption_enabled;
-// extern volatile thread_local unsigned short disable_preempt_count;
 
 struct __cfa_kernel_preemption_state_t {

src/libcfa/concurrency/monitor.c

@@ -297 +297 @@
         this.count = count;
 
-        // Sort monitors based on address -> TODO use a sort specialized for small numbers
+        // Sort monitors based on address
         __libcfa_small_sort(this.m, count);
 

src/libcfa/concurrency/preemption.c

@@ -10 +10 @@
 // Created On       : Mon Jun 5 14:20:42 2017
 // Last Modified By : Peter A. Buhr
-// Last Modified On : Mon Apr  9 13:52:39 2018
-// Update Count     : 36
+// Last Modified On : Tue Jun  5 17:35:49 2018
+// Update Count     : 37
 //
 
@@ -116 +116 @@
         // If there are still alarms pending, reset the timer
         if( alarms->head ) {
-                __cfaabi_dbg_print_buffer_decl( " KERNEL: @%lu(%lu) resetting alarm to %lu.\n", currtime.tv, __kernel_get_time().tv, (alarms->head->alarm - currtime).tv);
+                __cfaabi_dbg_print_buffer_decl( " KERNEL: @%ju(%ju) resetting alarm to %ju.\n", currtime.tv, __kernel_get_time().tv, (alarms->head->alarm - currtime).tv);
                 Duration delta = alarms->head->alarm - currtime;
                 Duration caped = max(delta, 50`us);
@@ -161 +161 @@
         void disable_interrupts() {
                 with( kernelTLS.preemption_state ) {
+                        #if GCC_VERSION > 50000
                         static_assert(__atomic_always_lock_free(sizeof(enabled), &enabled), "Must be lock-free");
+                        #endif
 
                         // Set enabled flag to false
@@ -190 +192 @@
                         // Check if we need to prempt the thread because an interrupt was missed
                         if( prev == 1 ) {
+                                #if GCC_VERSION > 50000
                                 static_assert(__atomic_always_lock_free(sizeof(enabled), &enabled), "Must be lock-free");
+                                #endif
 
                                 // Set enabled flag to true
@@ -217 +221 @@
                 verifyf( prev != 0u, "Incremented from %u\n", prev );                     // If this triggers someone is enabled already enabled interrupts
                 if( prev == 1 ) {
+                        #if GCC_VERSION > 50000
                         static_assert(__atomic_always_lock_free(sizeof(kernelTLS.preemption_state.enabled), &kernelTLS.preemption_state.enabled), "Must be lock-free");
+                        #endif
                         // Set enabled flag to true
                         // should be atomic to avoid preemption in the middle of the operation.
@@ -254 +260 @@
 static void preempt( processor * this ) {
         sigval_t value = { PREEMPT_NORMAL };
-        pthread_sigqueue( this->kernel_thread, SIGUSR1, value );
-}
-
-// kill wrapper : signal a processor
-void terminate(processor * this) {
-        this->do_terminate = true;
-        sigval_t value = { PREEMPT_TERMINATE };
         pthread_sigqueue( this->kernel_thread, SIGUSR1, value );
 }
@@ -362 +361 @@
         choose(sfp->si_value.sival_int) {
                 case PREEMPT_NORMAL   : ;// Normal case, nothing to do here
-                case PREEMPT_TERMINATE: verify( kernelTLS.this_processor->do_terminate);
+                case PREEMPT_TERMINATE: verify( __atomic_load_n( &kernelTLS.this_processor->do_terminate, __ATOMIC_SEQ_CST ) );
                 default:
                         abort( "internal error, signal value is %d", sfp->si_value.sival_int );
@@ -376 +375 @@
 
         // Clear sighandler mask before context switching.
+        #if GCC_VERSION > 50000
         static_assert( sizeof( sigset_t ) == sizeof( cxt->uc_sigmask ), "Expected cxt->uc_sigmask to be of sigset_t" );
+        #endif
         if ( pthread_sigmask( SIG_SETMASK, (sigset_t *)&(cxt->uc_sigmask), NULL ) == -1 ) {
                 abort( "internal error, sigprocmask" );
@@ -479 +480 @@
 }
 
+#ifdef __CFA_WITH_VERIFY__
+bool __cfaabi_dbg_in_kernel() {
+        return !kernelTLS.preemption_state.enabled;
+}
+#endif
+
 // Local Variables: //
 // mode: c //