Changeset 93d2219 for libcfa/src/concurrency

Timestamp:

Oct 28, 2022, 3:12:16 PM (3 years ago)

Author:

JiadaL <j82liang@…>

Branches:

ADT, ast-experimental, master

Children:

fa2e183

Parents:

e874605 (diff), 22a0e87 (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:

• clib/cfathread.cfa (modified) (1 diff)
• invoke.h (modified) (1 diff)
• io.cfa (modified) (1 diff)
• io/setup.cfa (modified) (1 diff)
• kernel.cfa (modified) (2 diffs)
• kernel.hfa (modified) (3 diffs)
• kernel/private.hfa (modified) (3 diffs)
• kernel/startup.cfa (modified) (12 diffs)
• locks.hfa (modified) (2 diffs)
• preemption.cfa (modified) (9 diffs)
• pthread.cfa (added)
• thread.cfa (modified) (2 diffs)
• thread.hfa (modified) (1 diff)

libcfa/src/concurrency/clib/cfathread.cfa

@@ -172 +172 @@
 
                 pthread_attr_t attr;
-                if (int ret = pthread_attr_init(&attr); 0 != ret) {
+                if (int ret = __cfaabi_pthread_attr_init(&attr); 0 != ret) {
                         abort | "failed to create master epoll thread attr: " | ret | strerror(ret);
                 }
 
-                if (int ret = pthread_create(&master_poller, &attr, master_epoll, 0p); 0 != ret) {
+                if (int ret = __cfaabi_pthread_create(&master_poller, &attr, master_epoll, 0p); 0 != ret) {
                         abort | "failed to create master epoll thread: " | ret | strerror(ret);
                 }

libcfa/src/concurrency/invoke.h

@@ -214 +214 @@
 
                 #if defined( __CFA_WITH_VERIFY__ )
+                        struct processor * volatile executing;
                         void * canary;
                 #endif

libcfa/src/concurrency/io.cfa

@@ -610 +610 @@
                 if( we ) {
                         sigval_t value = { PREEMPT_IO };
-                        pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
+                        __cfaabi_pthread_sigqueue(ctx->proc->kernel_thread, SIGUSR1, value);
                 }
 

libcfa/src/concurrency/io/setup.cfa

@@ -344 +344 @@
         //      iopoll.run = false;
         //      sigval val = { 1 };
-        //      pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
+        //      __cfaabi_pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
 
         //      // Make sure all this is done

libcfa/src/concurrency/kernel.cfa

@@ -321 +321 @@
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too small.\n", thrd_dst ); // add escape condition if we are setting up the processor
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->curr_cor == proc_cor || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too large.\n", thrd_dst ); // add escape condition if we are setting up the processor
+                /* paranoid */ verify( __atomic_exchange_n( &thrd_dst->executing, this, __ATOMIC_SEQ_CST) == 0p );
                 /* paranoid */ verify( 0x0D15EA5E0D15EA5Ep == thrd_dst->canary );
 
@@ -332 +333 @@
 
                 /* paranoid */ verify( 0x0D15EA5E0D15EA5Ep == thrd_dst->canary );
+                /* paranoid */ verify( __atomic_exchange_n( &thrd_dst->executing, 0p, __ATOMIC_SEQ_CST) == this );
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too large.\n", thrd_dst );
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->corctx_flag, "ERROR : Destination thread$ %p has been corrupted.\n StackPointer too small.\n", thrd_dst );
+                /* paranoid */ verify( thrd_dst->state != Halted );
                 /* paranoid */ verify( thrd_dst->context.SP );
-                /* paranoid */ verify( thrd_dst->curr_cluster == this->cltr );
                 /* paranoid */ verify( kernelTLS().this_thread == thrd_dst );
                 /* paranoid */ verify( ! __preemption_enabled() );

libcfa/src/concurrency/kernel.hfa

@@ -160 +160 @@
 // P9_EMBEDDED( processor, dlink(processor) )
 static inline tytagref( dlink(processor), dlink(processor) ) ?`inner( processor & this ) {
-    dlink(processor) & b = this.link;
-    tytagref( dlink(processor), dlink(processor) ) result = { b };
-    return result;
+        dlink(processor) & b = this.link;
+        tytagref( dlink(processor), dlink(processor) ) result = { b };
+        return result;
 }
 
@@ -269 +269 @@
                 io_context_params params;
         } io;
+
+        struct {
+                struct processor ** procs;
+                unsigned cnt;
+        } managed;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -298 +303 @@
 static inline struct cluster   * active_cluster  () { return publicTLS_get( this_processor )->cltr; }
 
+// set the number of internal processors
+// these processors are in addition to any explicitly declared processors
+unsigned set_concurrency( cluster & this, unsigned new_count );
+
 #if !defined(__CFA_NO_STATISTICS__)
         void print_stats_now( cluster & this, int flags );

libcfa/src/concurrency/kernel/private.hfa

@@ -20 +20 @@
 #endif
 
+#include <signal.h>
+
 #include "kernel.hfa"
 #include "thread.hfa"
@@ -48 +50 @@
         #endif
 #endif
-
 // #define READYQ_USE_LINEAR_AVG
 #define READYQ_USE_LOGDBL_AVG
@@ -62 +63 @@
 #error must pick a scheme for averaging
 #endif
+
+extern "C" {
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_create(pthread_t *_thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_join(pthread_t _thread, void **retval);
+        __attribute__((visibility("protected"))) pthread_t __cfaabi_pthread_self(void);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_init(pthread_attr_t *attr);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_destroy(pthread_attr_t *attr);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_setstack( pthread_attr_t *attr, void *stackaddr, size_t stacksize );
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_getstacksize( const pthread_attr_t *attr, size_t *stacksize );
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_sigqueue(pthread_t _thread, int sig, const union sigval value);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_sigmask( int how, const sigset_t *set, sigset_t *oset);
+}
 
 //-----------------------------------------------------------------------------

libcfa/src/concurrency/kernel/startup.cfa

@@ -16 +16 @@
 #define __cforall_thread__
 #define _GNU_SOURCE
+
+// #define __CFA_DEBUG_PRINT_RUNTIME_CORE__
 
 // C Includes
@@ -222 +224 @@
                 ( this.runner ){};
                 init( this, "Main Processor", *mainCluster, 0p );
-                kernel_thread = pthread_self();
+                kernel_thread = __cfaabi_pthread_self();
 
                 runner{ &this };
@@ -283 +285 @@
 }
 
+extern "C"{
+        void pthread_delete_kernel_threads_();
+}
+
+
 static void __kernel_shutdown(void) {
         if(!cfa_main_returned) return;
+
+        //delete kernel threads for pthread_concurrency
+        pthread_delete_kernel_threads_();
+
         /* paranoid */ verify( __preemption_enabled() );
         disable_interrupts();
@@ -327 +338 @@
 
                 /* paranoid */ verify( this.do_terminate == true );
-                __cfaabi_dbg_print_safe("Kernel : destroyed main processor context %p\n", &runner);
+                __cfadbg_print_safe(runtime_core, "Kernel : destroyed main processor context %p\n", &runner);
         }
 
@@ -388 +399 @@
         (proc->runner){ proc, &info };
 
-        __cfaabi_dbg_print_safe("Coroutine : created stack %p\n", get_coroutine(proc->runner)->stack.storage);
+        __cfadbg_print_safe(runtime_core, "Coroutine : created stack %p\n", get_coroutine(proc->runner)->stack.storage);
 
         //Set global state
@@ -520 +531 @@
         random_state = __global_random_mask ? __global_random_prime : __global_random_prime ^ rdtscl();
         #if defined( __CFA_WITH_VERIFY__ )
+                executing = 0p;
                 canary = 0x0D15EA5E0D15EA5Ep;
         #endif
@@ -652 +664 @@
         io.params = io_params;
 
+        managed.procs = 0p;
+        managed.cnt = 0;
+
         doregister(this);
 
@@ -667 +682 @@
 
 void ^?{}(cluster & this) libcfa_public {
+        set_concurrency( this, 0 );
+
         destroy(this.io.arbiter);
 
@@ -777 +794 @@
         pthread_attr_t attr;
 
-        check( pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
+        check( __cfaabi_pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
 
         size_t stacksize = max( PTHREAD_STACK_MIN, DEFAULT_STACK_SIZE );
@@ -804 +821 @@
         #endif
 
-        check( pthread_attr_setstack( &attr, stack, stacksize ), "pthread_attr_setstack" );
-        check( pthread_create( pthread, &attr, start, arg ), "pthread_create" );
+        check( __cfaabi_pthread_attr_setstack( &attr, stack, stacksize ), "pthread_attr_setstack" );
+        check( __cfaabi_pthread_create( pthread, &attr, start, arg ), "pthread_create" );
         return stack;
 }
 
 void __destroy_pthread( pthread_t pthread, void * stack, void ** retval ) {
-        int err = pthread_join( pthread, retval );
+        int err = __cfaabi_pthread_join( pthread, retval );
         if( err != 0 ) abort("KERNEL ERROR: joining pthread %p caused error %s\n", (void*)pthread, strerror(err));
 
@@ -816 +833 @@
                 pthread_attr_t attr;
 
-                check( pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
+                check( __cfaabi_pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
 
                 size_t stacksize;
                 // default stack size, normally defined by shell limit
-                check( pthread_attr_getstacksize( &attr, &stacksize ), "pthread_attr_getstacksize" );
+                check( __cfaabi_pthread_attr_getstacksize( &attr, &stacksize ), "pthread_attr_getstacksize" );
                 assert( stacksize >= PTHREAD_STACK_MIN );
                 stacksize += __page_size;
@@ -838 +855 @@
 }
 
+unsigned set_concurrency( cluster & this, unsigned new ) libcfa_public {
+        unsigned old = this.managed.cnt;
+
+        __cfadbg_print_safe(runtime_core, "Kernel : resizing cluster from %u to %u\n", old, (unsigned)new);
+
+        // Delete all the old unneeded procs
+        if(old > new) for(i; (unsigned)new ~ old) {
+                __cfadbg_print_safe(runtime_core, "Kernel : destroying %u\n", i);
+                delete( this.managed.procs[i] );
+        }
+
+        // Allocate new array (uses realloc and memcpies the data)
+        this.managed.procs = alloc( new, this.managed.procs`realloc );
+        this.managed.cnt = new;
+
+        // Create the desired new procs
+        if(old < new) for(i; old ~ new) {
+                __cfadbg_print_safe(runtime_core, "Kernel : constructing %u\n", i);
+                (*(this.managed.procs[i] = alloc())){ this };
+        }
+
+        // return the old count
+        return old;
+}
+
 #if defined(__CFA_WITH_VERIFY__)
 static bool verify_fwd_bck_rng(void) {

libcfa/src/concurrency/locks.hfa

@@ -21 +21 @@
 
 #include "bits/weakso_locks.hfa"
-#include "containers/queueLockFree.hfa"
+#include "containers/lockfree.hfa"
 #include "containers/list.hfa"
 
@@ -423 +423 @@
 }
 
-static inline size_t on_wait(simple_owner_lock & this) with(this) { 
+static inline size_t on_wait(simple_owner_lock & this) with(this) {
         lock( lock __cfaabi_dbg_ctx2 );
         /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );

libcfa/src/concurrency/preemption.cfa

@@ -352 +352 @@
         sigset_t oldset;
         int ret;
-        ret = pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
+        ret = __cfaabi_pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
         if(ret != 0) { abort("ERROR sigprocmask returned %d", ret); }
 
@@ -385 +385 @@
         sigaddset( &mask, sig );
 
-        if ( pthread_sigmask( SIG_UNBLOCK, &mask, 0p ) == -1 ) {
+        if ( __cfaabi_pthread_sigmask( SIG_UNBLOCK, &mask, 0p ) == -1 ) {
             abort( "internal error, pthread_sigmask" );
         }
@@ -396 +396 @@
         sigaddset( &mask, sig );
 
-        if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
+        if ( __cfaabi_pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
                 abort( "internal error, pthread_sigmask" );
         }
@@ -404 +404 @@
 static void preempt( processor * this ) {
         sigval_t value = { PREEMPT_NORMAL };
-        pthread_sigqueue( this->kernel_thread, SIGUSR1, value );
+        __cfaabi_pthread_sigqueue( this->kernel_thread, SIGUSR1, value );
 }
 
@@ -415 +415 @@
         sigset_t oldset;
         int ret;
-        ret = pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
+        ret = __cfaabi_pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
         if(ret != 0) { abort("ERROR sigprocmask returned %d", ret); }
 
@@ -434 +434 @@
         sigset_t oldset;
         int ret;
-        ret = pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
+        ret = __cfaabi_pthread_sigmask(0, ( const sigset_t * ) 0p, &oldset);  // workaround trac#208: cast should be unnecessary
         if(ret != 0) { abort("ERROR sigprocmask returned %d", ret); }
 
@@ -505 +505 @@
         sigval val;
         val.sival_int = 0;
-        pthread_sigqueue( alarm_thread, SIGALRM, val );
+        __cfaabi_pthread_sigqueue( alarm_thread, SIGALRM, val );
 
         // Wait for the preemption thread to finish
@@ -579 +579 @@
         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), 0p ) == -1 ) {
+        if ( __cfaabi_pthread_sigmask( SIG_SETMASK, (sigset_t *)&(cxt->uc_sigmask), 0p ) == -1 ) {
                 abort( "internal error, sigprocmask" );
         }
@@ -607 +607 @@
         sigset_t mask;
         sigfillset(&mask);
-        if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
+        if ( __cfaabi_pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
             abort( "internal error, pthread_sigmask" );
         }

libcfa/src/concurrency/thread.cfa

@@ -50 +50 @@
         random_state = __global_random_mask ? __global_random_prime : __global_random_prime ^ rdtscl();
         #if defined( __CFA_WITH_VERIFY__ )
+                executing = 0p;
                 canary = 0x0D15EA5E0D15EA5Ep;
         #endif
@@ -177 +178 @@
 
 //-----------------------------------------------------------------------------
+bool migrate( thread$ * thrd, struct cluster & cl ) {
+
+        monitor$ * tmon = get_monitor(thrd);
+        monitor$ * __monitors[] = { tmon };
+        monitor_guard_t __guard = { __monitors, 1 };
+
+
+        {
+                // if nothing needs to be done, return false
+                if( thrd->curr_cluster == &cl ) return false;
+
+                // are we migrating ourself?
+                const bool local = thrd == active_thread();
+
+                /* paranoid */ verify( !local || &cl != active_cluster() );
+                /* paranoid */ verify( !local || thrd->curr_cluster == active_cluster() );
+                /* paranoid */ verify( !local || thrd->curr_cluster == active_processor()->cltr );
+                /* paranoid */ verify( local || tmon->signal_stack.top->owner->waiting_thread == thrd );
+                /* paranoid */ verify( local || tmon->signal_stack.top );
+
+                // make sure we aren't interrupted while doing this
+                // not as important if we aren't local
+                disable_interrupts();
+
+                // actually move the thread
+                unregister( thrd->curr_cluster, *thrd );
+                thrd->curr_cluster = &cl;
+                doregister( thrd->curr_cluster, *thrd );
+
+                // restore interrupts
+                enable_interrupts();
+
+                // if this is the local thread, we are still running on the old cluster
+                if(local) yield();
+
+                /* paranoid */ verify( !local || &cl == active_cluster() );
+                /* paranoid */ verify( !local || thrd->curr_cluster == active_cluster() );
+                /* paranoid */ verify( !local || thrd->curr_cluster == active_processor()->cltr );
+                /* paranoid */ verify(  local || tmon->signal_stack.top );
+                /* paranoid */ verify(  local || tmon->signal_stack.top->owner->waiting_thread == thrd );
+
+                return true;
+        }
+}
+
+//-----------------------------------------------------------------------------
 #define GENERATOR LCG
 

libcfa/src/concurrency/thread.hfa

@@ -132 +132 @@
 
 //----------
+// misc
+bool migrate( thread$ * thrd, struct cluster & cl );
+
+forall( T & | is_thread(T) )
+static inline bool migrate( T & mutex thrd, struct cluster & cl ) { return migrate( &(thread&)thrd, cl ); }
+
+
+//----------
 // prng
 static inline {