Changeset 0c92c9f for src/libcfa/concurrency

Timestamp:

Jan 27, 2017, 4:28:05 PM (8 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, aaron-thesis, arm-eh, ast-experimental, cleanup-dtors, deferred_resn, demangler, enum, forall-pointer-decay, jacob/cs343-translation, jenkins-sandbox, master, new-ast, new-ast-unique-expr, new-env, no_list, persistent-indexer, pthread-emulation, qualifiedEnum, resolv-new, with_gc

Children:

8804701

Parents:

0157ca7

Message:

Cleaned-up threading code and added temporary test for threads (single core)

Location:

src/libcfa/concurrency

Files:

• coroutines (modified) (3 diffs)
• coroutines.c (modified) (2 diffs)
• invoke.c (modified) (5 diffs)
• kernel.c (modified) (7 diffs)
• threads (modified) (1 diff)
• threads.c (modified) (5 diffs)

src/libcfa/concurrency/coroutines

@@ -65 +65 @@
 
 // Private wrappers for context switch and stack creation
-extern void corCxtSw(coroutine * src, coroutine * dst);
+extern void CoroutineCtxSwitch(coroutine * src, coroutine * dst);
 extern void create_stack( coStack_t * this, unsigned int storageSize );
 
@@ -81 +81 @@
                 src->name, src, src->last->name, src->last );
 
-        corCxtSw( src, src->last );
+        CoroutineCtxSwitch( src, src->last );
 }
 
@@ -107 +107 @@
 
       // always done for performance testing
-        corCxtSw( src, dst );
+        CoroutineCtxSwitch( src, dst );
 }
 

src/libcfa/concurrency/coroutines.c

@@ -98 +98 @@
 }
 
-// We need to call suspend from invoke.c, so we expose this wrapper that
-// is not inline (We can't inline Cforall in C)
-void suspend_no_inline(void) {
-        suspend();
-}
-
-void corCxtSw(coroutine* src, coroutine* dst) {
+// Wrapper for co
+void CoroutineCtxSwitch(coroutine* src, coroutine* dst) {
         // THREAD_GETMEM( This )->disableInterrupts();
 
@@ -167 +162 @@
 }
 
+// We need to call suspend from invoke.c, so we expose this wrapper that
+// is not inline (We can't inline Cforall in C)
+extern "C" {
+        void __suspend_internal(void) {
+                suspend();
+        }
+}
+
 // Local Variables: //
 // mode: c //

src/libcfa/concurrency/invoke.c

@@ -28 +28 @@
 // Called from the kernel when starting a coroutine or task so must switch back to user mode.
 
-extern void __suspend_no_inline__F___1(void);
-extern void __signal_termination__F_P7sthread__1(struct thread*);
+extern void __suspend_internal(void);
+extern void __thread_signal_termination(struct thread*);
 
 void CtxInvokeCoroutine(
@@ -41 +41 @@
 
       if(cor->state == Primed) {
-            __suspend_no_inline__F___1();
+            __suspend_internal();
       }
 
@@ -52 +52 @@
 
       //Final suspend, should never return
-      __suspend_no_inline__F___1();
-      assertf(false, "Resumed dead coroutine");
+      __suspend_internal();
+      abortf("Resumed dead coroutine");
 }
 
@@ -61 +61 @@
       void *this
 ) {
-      // LIB_DEBUG_PRINTF("Invoke Thread : Received %p (main %p, get_t %p)\n", this, main, get_thread);
-
-      __suspend_no_inline__F___1();
+      __suspend_internal();
 
       struct thread* thrd = get_thread( this );
@@ -72 +70 @@
       main( this );
 
-      __signal_termination__F_P7sthread__1(thrd);
+      __thread_signal_termination(thrd);
 
       //Final suspend, should never return
-      __suspend_no_inline__F___1();
-      assertf(false, "Resumed dead thread");
+      __suspend_internal();
+      abortf("Resumed dead thread");
 }
 

src/libcfa/concurrency/kernel.c

@@ -15 +15 @@
 //
 
+//Start and stop routine for the kernel, declared first to make sure they run first
+void kernel_startup(void)  __attribute__((constructor(101)));
+void kernel_shutdown(void) __attribute__((destructor(101)));
+
 //Header
 #include "kernel"
@@ -52 +56 @@
 processor * systemProcessor;
 thread * mainThread;
-
-void kernel_startup(void)  __attribute__((constructor(101)));
-void kernel_shutdown(void) __attribute__((destructor(101)));
 
 //-----------------------------------------------------------------------------
@@ -184 +185 @@
 void main(processorCtx_t * ctx);
 thread * nextThread(cluster * this);
-void runThread(processor * this, thread * dst);
+void scheduleInternal(processor * this, thread * dst);
 void spin(processor * this, unsigned int * spin_count);
 
@@ -197 +198 @@
 
                 if(readyThread) {
-                        runThread(this, readyThread);
+                        scheduleInternal(this, readyThread);
                         spin_count = 0;
                 } else {
@@ -209 +210 @@
 }
 
-void runThread(processor * this, thread * dst) {
+//Declarations for scheduleInternal
+extern void ThreadCtxSwitch(coroutine * src, coroutine * dst);
+
+// scheduleInternal runs a thread by context switching 
+// from the processor coroutine to the target thread 
+void scheduleInternal(processor * this, thread * dst) {
+        // coroutine * proc_ctx = get_coroutine(this->ctx);
+        // coroutine * thrd_ctx = get_coroutine(dst);
+
+        // //Update global state
+        // this->current_thread = dst;
+
+        // // Context Switch to the thread
+        // ThreadCtxSwitch(proc_ctx, thrd_ctx);
+        // // when ThreadCtxSwitch returns we are back in the processor coroutine
+
         coroutine * proc_ctx = get_coroutine(this->ctx);
         coroutine * thrd_ctx = get_coroutine(dst);
-        thrd_ctx->last = proc_ctx;
-
-        // context switch to specified coroutine
-        // Which is now the current_coroutine
-        // LIB_DEBUG_PRINTF("Kernel : switching to ctx %p (from %p, current %p)\n", thrd_ctx, proc_ctx, current_coroutine);
-        this->current_thread = dst;
-        this->current_coroutine = thrd_ctx;
-        CtxSwitch( proc_ctx->stack.context, thrd_ctx->stack.context );
-        this->current_coroutine = proc_ctx;
-        // LIB_DEBUG_PRINTF("Kernel : returned from ctx %p (to %p, current %p)\n", thrd_ctx, proc_ctx, current_coroutine);
-
-        // when CtxSwitch returns we are back in the processor coroutine
-}
-
+      thrd_ctx->last = proc_ctx;
+ 
+      // context switch to specified coroutine
+      // Which is now the current_coroutine
+      LIB_DEBUG_PRINTF("Kernel : switching to ctx %p (from %p, current %p)\n", thrd_ctx, proc_ctx, this->current_coroutine);
+      this->current_thread = dst;
+      this->current_coroutine = thrd_ctx;
+      CtxSwitch( proc_ctx->stack.context, thrd_ctx->stack.context );
+      this->current_coroutine = proc_ctx;
+      LIB_DEBUG_PRINTF("Kernel : returned from ctx %p (to %p, current %p)\n", thrd_ctx, proc_ctx, this->current_coroutine);
+ 
+      // when CtxSwitch returns we are back in the processor coroutine
+}
+
+// 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)++;
 }
 
+// Context invoker for processors
+// This is the entry point for processors (kernel threads)
+// It effectively constructs a coroutine by stealing the pthread stack
 void * CtxInvokeProcessor(void * arg) {
         processor * proc = (processor *) arg;
@@ -241 +262 @@
         processorCtx_t proc_cor_storage = { proc, &info };
 
+        //Set global state
         proc->current_coroutine = &proc->ctx->c;
         proc->current_thread = NULL;
 
+        //We now have a proper context from which to schedule threads
         LIB_DEBUG_PRINTF("Kernel : core %p created (%p)\n", proc, proc->ctx);
-
-        // LIB_DEBUG_PRINTF("Kernel : core    base : %p \n", info.base );
-        // LIB_DEBUG_PRINTF("Kernel : core storage : %p \n", info.storage );
-        // LIB_DEBUG_PRINTF("Kernel : core    size : %x \n", info.size );
-        // LIB_DEBUG_PRINTF("Kernel : core   limit : %p \n", info.limit );
-        // LIB_DEBUG_PRINTF("Kernel : core context : %p \n", info.context );
-        // LIB_DEBUG_PRINTF("Kernel : core     top : %p \n", info.top );
-
-        //We now have a proper context from which to schedule threads
 
         // SKULLDUGGERY: Since the coroutine doesn't have its own stack, we can't 
@@ -264 +278 @@
       proc_cor_storage.c.notHalted = false;
 
+        // Main routine of the core returned, the core is now fully terminated
         LIB_DEBUG_PRINTF("Kernel : core %p main ended (%p)\n", proc, proc->ctx);        
 

src/libcfa/concurrency/threads

@@ -27 +27 @@
 // Anything that implements this trait can be resumed.
 // Anything that is resumed is a coroutine.
-trait is_thread(dtype T | sized(T)) {
+trait is_thread(dtype T) {
       void main(T* this);
       thread* get_thread(T* this);
 };
 
-#define DECL_THREAD(X) static inline thread* get_thread(X* this) { return &this->t; } void main(X* this);
+#define DECL_THREAD(X) thread* get_thread(X* this) { return &this->t; } void main(X* this);
 
-forall( dtype T | sized(T) | is_thread(T) )
+forall( dtype T | is_thread(T) )
 static inline coroutine* get_coroutine(T* this) {
         return &get_thread(this)->c;

src/libcfa/concurrency/threads.c

@@ -31 +31 @@
 //-----------------------------------------------------------------------------
 // Forward declarations
-forall( dtype T | sized(T) | is_thread(T) )
+forall( dtype T | is_thread(T) )
 void start( T* this );
 
-forall( dtype T | sized(T) | is_thread(T) )
+forall( dtype T | is_thread(T) )
 void stop( T* this );
 
@@ -78 +78 @@
 extern void thread_schedule( thread * );
 
-forall( dtype T | sized(T) | is_thread(T) )
+forall( dtype T | is_thread(T) )
 void start( T* this ) {
         coroutine* thrd_c = get_coroutine(this);
@@ -85 +85 @@
         get_this_processor()->current_coroutine = thrd_c;
 
-        // LIB_DEBUG_PRINTF("Thread start : %p (t %p, c %p)\n", handle, thrd_c, thrd_h);
+        LIB_DEBUG_PRINTF("Thread start : %p (t %p, c %p)\n", this, thrd_c, thrd_h);
 
         create_stack(&thrd_c->stack, thrd_c->stack.size);
@@ -91 +91 @@
         CtxSwitch( thrd_c->last->stack.context, thrd_c->stack.context );
 
+        LIB_DEBUG_PRINTF("Thread started : %p (t %p, c %p)\n", this, thrd_c, thrd_h);
+
         thread_schedule(thrd_h);
 }
 
-forall( dtype T | sized(T) | is_thread(T) )
+forall( dtype T | is_thread(T) )
 void stop( T* this ) {
         thread*  thrd = get_thread(this);
@@ -102 +104 @@
 }
 
-void signal_termination( thread * this ) {
-        this->c.state = Halt;
-      this->c.notHalted = false;
-        unlock( &this->lock );
-}
-
 void yield( void ) {
         thread_schedule( this_thread() );
         suspend();
+}
+
+void ThreadCtxSwitch(coroutine* src, coroutine* dst) {
+        dst->last = src;
+
+        // set state of current coroutine to inactive
+        src->state = Inactive;
+
+        // set new coroutine that task is executing
+        get_this_processor()->current_coroutine = dst;  
+
+        // context switch to specified coroutine
+        CtxSwitch( src->stack.context, dst->stack.context );
+        // when CtxSwitch returns we are back in the src coroutine
+
+        // set state of new coroutine to active
+        src->state = Active;
+}
+
+// C Helper to signal the termination of a thread
+// Used in invoke.c
+extern "C" {
+        void __thread_signal_termination( thread * this ) {
+                this->c.state = Halt;
+                this->c.notHalted = false;
+                unlock( &this->lock );
+        }
 }