Changeset bd98b58 for src/libcfa

Timestamp:

Jan 20, 2017, 4:50:15 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:

207c7e1d

Parents:

dcb42b8

Message:

Kernel now uses intrusive lists and blocking locks for ready queue management.
Update the plan for concurrency.

Location:

src/libcfa/concurrency

Files:

• coroutines (modified) (1 diff)
• coroutines.c (modified) (4 diffs)
• invoke.c (modified) (2 diffs)
• invoke.h (modified) (2 diffs)
• kernel (modified) (1 diff)
• kernel.c (modified) (12 diffs)
• threads (modified) (2 diffs)
• threads.c (modified) (5 diffs)

src/libcfa/concurrency/coroutines ¶

@@ -62 +62 @@
 
 // Get current coroutine
-extern coroutine * current_coroutine; //PRIVATE, never use directly
-static inline coroutine * this_coroutine(void) {
-        return current_coroutine;
-}
+coroutine * this_coroutine(void);
 
 // Private wrappers for context switch and stack creation

src/libcfa/concurrency/coroutines.c ¶

@@ -14 +14 @@
 //
 
+#include "coroutines"
+
 extern "C" {
 #include <stddef.h>
@@ -23 +25 @@
 }
 
-#include "coroutines"
+#include "kernel"
 #include "libhdr.h"
 
 #define __CFA_INVOKE_PRIVATE__
 #include "invoke.h"
+
+/*thread_local*/ extern processor * this_processor;
 
 //-----------------------------------------------------------------------------
@@ -35 +39 @@
 #define MinStackSize 1000
 static size_t pageSize = 0;                             // architecture pagesize HACK, should go in proper runtime singleton
-
-//Current coroutine
-//Will need to be in TLS when multi-threading is added
-coroutine* current_coroutine;
 
 //-----------------------------------------------------------------------------
@@ -110 +110 @@
 
         // set new coroutine that task is executing
-        current_coroutine = dst;                        
+        this_processor->current_coroutine = dst;                        
 
         // context switch to specified coroutine

src/libcfa/concurrency/invoke.c ¶

@@ -14 +14 @@
 
 extern void __suspend_no_inline__F___1(void);
-extern void __scheduler_remove__F_P9sthread_h__1(struct thread_h*);
+extern void __signal_termination__F_P9sthread_h__1(struct thread_h*);
 
 void CtxInvokeCoroutine(
@@ -57 +57 @@
       main( this );
 
-      cor->state = Halt;
-      cor->notHalted = false;
-      __scheduler_remove__F_P9sthread_h__1(thrd);
+      __signal_termination__F_P9sthread_h__1(thrd);
 
       //Final suspend, should never return

src/libcfa/concurrency/invoke.h ¶

@@ -11 +11 @@
 
       #define unlikely(x)    __builtin_expect(!!(x), 0)
+      #define SCHEDULER_CAPACITY 10
+
+      struct simple_thread_list {
+            struct thread_h * head;
+            struct thread_h ** tail;
+      };
+
+      #ifdef __CFORALL__
+      extern "Cforall" {
+            void ?{}( struct simple_thread_list * );
+            void append( struct simple_thread_list *, struct thread_h * );
+            struct thread_h * pop_head( struct simple_thread_list * );
+      }
+      #endif
 
       struct coStack_t {
@@ -35 +49 @@
       };
 
+      struct simple_lock {
+        struct simple_thread_list blocked;
+      };
+
       struct thread_h {
             struct coroutine c;
+            struct simple_lock lock;
+            struct thread_h * next;
       };
 

src/libcfa/concurrency/kernel ¶

@@ -20 +20 @@
 #include <stdbool.h>
 
+#include "invoke.h"
+
+//-----------------------------------------------------------------------------
+// Cluster
+struct cluster {
+        simple_thread_list ready_queue;
+};
+
+void ?{}(cluster * this);
+void ^?{}(cluster * this);
+
+//-----------------------------------------------------------------------------
+// Processor
 struct processor {
         struct processorCtx_t * ctx;
-        unsigned int thread_index;
-        unsigned int thread_count;
-        struct thread_h * threads[10];
+        cluster * cltr;
+        coroutine * current_coroutine;
+        thread_h * current_thread;
         bool terminated;
 };
 
-void ?{}(processor * this);
+void ?{}(processor * this, cluster * cltr);
 void ^?{}(processor * this);
 
-void scheduler_add( struct thread_h * thrd );
-void scheduler_remove( struct thread_h * thrd );
-void kernel_run( void );
+
+//-----------------------------------------------------------------------------
+// Locks
+
+void ?{}(simple_lock * this);
+void ^?{}(simple_lock * this);
+
+void lock( simple_lock * );
+void unlock( simple_lock * );
 
 #endif //KERNEL_H

src/libcfa/concurrency/kernel.c ¶

@@ -32 +32 @@
 #include "invoke.h"
 
+cluster * systemCluster;
 processor * systemProcessor;
 thread_h * mainThread;
@@ -38 +39 @@
 void kernel_shutdown(void) __attribute__((destructor(101)));
 
-void ?{}(processor * this) {
+void ?{}(processor * this, cluster * cltr) {
         this->ctx = NULL;
-        this->thread_index = 0;
-        this->thread_count = 10;
+        this->cltr = cltr;
         this->terminated = false;
-
-        for(int i = 0; i < 10; i++) {
-                this->threads[i] = NULL;
-        }
-
-        LIB_DEBUG_PRINTF("Processor : ctor for core %p (core spots %d)\n", this, this->thread_count);
-}
-
-void ^?{}(processor * this) {
-
+}
+
+void ^?{}(processor * this) {}
+
+void ?{}(cluster * this) {
+        ( &this->ready_queue ){};
+}
+
+void ^?{}(cluster * this) {}
+
+//-----------------------------------------------------------------------------
+// Global state
+
+/*thread_local*/ processor * this_processor;
+
+coroutine * this_coroutine(void) {
+        return this_processor->current_coroutine;
+}
+
+thread_h * this_thread(void) {
+        return this_processor->current_thread;
 }
 
@@ -77 +88 @@
 // Processor running routines
 void main(processorCtx_t * ctx);
-thread_h * nextThread(processor * this);
+thread_h * nextThread(cluster * this);
 void runThread(processor * this, thread_h * dst);
 void spin(processor * this, unsigned int * spin_count);
@@ -88 +99 @@
         for( unsigned int spin_count = 0; ! this->terminated; spin_count++ ) {
                 
-                readyThread = nextThread(this);
+                readyThread = nextThread( this->cltr );
 
                 if(readyThread) {
@@ -101 +112 @@
 }
 
-thread_h * nextThread(processor * this) {
-        for(int i = 0; i < this->thread_count; i++) {
-                this->thread_index = (this->thread_index + 1) % this->thread_count;     
-                
-                thread_h * thrd = this->threads[this->thread_index];
-                if(thrd) return thrd;
-        }
-
-        return NULL;
-}
-
 void runThread(processor * this, thread_h * dst) {
         coroutine * proc_ctx = get_coroutine(this->ctx);
@@ -120 +120 @@
         // 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);
-        current_coroutine = thrd_ctx;
+        this->current_thread = dst;
+        this->current_coroutine = thrd_ctx;
         CtxSwitch( proc_ctx->stack.context, thrd_ctx->stack.context );
-        current_coroutine = proc_ctx;
+        this->current_coroutine = proc_ctx;
         // LIB_DEBUG_PRINTF("Kernel : returned from ctx %p (to %p, current %p)\n", thrd_ctx, proc_ctx, current_coroutine);
 
@@ -133 +134 @@
 
 //-----------------------------------------------------------------------------
-// Kernel runner (Temporary)
-
-void scheduler_add( thread_h * thrd ) {
-        for(int i = 0; i < systemProcessor->thread_count; i++) {
-                if(systemProcessor->threads[i] == NULL) {
-                        systemProcessor->threads[i] = thrd;
-                        return;
-                }
-        }
-        assertf(false, "Scheduler full");
-}
-
-void scheduler_remove( thread_h * thrd ) {
-        for(int i = 0; i < systemProcessor->thread_count; i++) {
-                if(systemProcessor->threads[i] == thrd) {
-                        systemProcessor->threads[i] = NULL;
-                        return;
-                }
-        }
-        assertf(false, "Trying to unschedule unkown thread");
+// Scheduler routines
+void thread_schedule( thread_h * thrd ) {
+        assertf( thrd->next == NULL, "Expected null got %p", thrd->next );
+        append( &systemProcessor->cltr->ready_queue, thrd );
+}
+
+thread_h * nextThread(cluster * this) {
+        return pop_head( &this->ready_queue );
 }
 
@@ -160 +149 @@
 
 KERNEL_STORAGE(processorCtx_t, systemProcessorCtx);
+KERNEL_STORAGE(cluster, systemCluster);
 KERNEL_STORAGE(processor, systemProcessor);
 KERNEL_STORAGE(thread_h, mainThread);
@@ -221 +211 @@
 
         mainThread_info_t ctx;
-        // LIB_DEBUG_PRINTF("Kernel :    base : %p\n", ctx.base );
-        // LIB_DEBUG_PRINTF("Kernel :     top : %p\n", ctx.top );
-        // LIB_DEBUG_PRINTF("Kernel :   limit : %p\n", ctx.limit );
-        // LIB_DEBUG_PRINTF("Kernel :    size : %x\n", ctx.size );
-        // LIB_DEBUG_PRINTF("Kernel : storage : %p\n", ctx.storage );
-        // LIB_DEBUG_PRINTF("Kernel : context : %p\n", ctx.context );
 
         // Start by initializing the main thread
@@ -232 +216 @@
         mainThread{ &ctx };
 
-        // // Initialize the system processor
+        // Initialize the system cluster
+        systemCluster = (cluster *)&systemCluster_storage;
+        systemCluster{};
+
+        // Initialize the system processor
         systemProcessor = (processor *)&systemProcessor_storage;
-        systemProcessor{};
+        systemProcessor{ systemCluster };
 
         // Initialize the system processor ctx
@@ -243 +231 @@
         // Add the main thread to the ready queue 
         // once resume is called on systemProcessor->ctx the mainThread needs to be scheduled like any normal thread
-        scheduler_add(mainThread);
+        thread_schedule(mainThread);
 
         //initialize the global state variables
-        current_coroutine = &mainThread->c;
+        this_processor = systemProcessor;
+        this_processor->current_thread = mainThread;
+        this_processor->current_coroutine = &mainThread->c;
 
         // SKULLDUGGERY: Force a context switch to the system processor to set the main thread's context to the current UNIX
@@ -285 +275 @@
 }
 
+//-----------------------------------------------------------------------------
+// Locks
+void ?{}( simple_lock * this ) {
+        ( &this->blocked ){};
+}
+
+void ^?{}( simple_lock * this ) {
+
+}
+
+void lock( simple_lock * this ) {
+        append( &this->blocked, this_thread() );
+        suspend();
+}
+
+void unlock( simple_lock * this ) {
+        thread_h * it;
+        while( it = pop_head( &this->blocked) ) {
+                thread_schedule( it );
+        }
+}
+
+//-----------------------------------------------------------------------------
+// Queues
+void ?{}( simple_thread_list * this ) {
+        this->head = NULL;
+        this->tail = &this->head;
+}
+
+void append( simple_thread_list * this, thread_h * t ) {
+        assert( t->next == NULL );
+        *this->tail = t;
+        this->tail = &t->next;
+}
+
+thread_h * pop_head( simple_thread_list * this ) {
+        thread_h * head = this->head;
+        if( head ) {
+                this->head = head->next;
+                if( !head->next ) {
+                        this->tail = &this->head;
+                }
+                head->next = NULL;
+        }       
+        
+        return head;
+}
 // Local Variables: //
 // mode: c //

src/libcfa/concurrency/threads ¶

@@ -45 +45 @@
 }
 
+thread_h * this_thread(void);
+
 //-----------------------------------------------------------------------------
 // Ctors and dtors
@@ -67 +69 @@
 void ^?{}( thread(T)* this );
 
-//-----------------------------------------------------------------------------
-// PRIVATE exposed because of inline
+void yield();
 
 #endif //THREADS_H

src/libcfa/concurrency/threads.c ¶

@@ -23 +23 @@
 #include "invoke.h"
 
-#include <stdlib>
+extern "C" {
+        #include <stddef.h>
+}
+
+/*thread_local*/ extern processor * this_processor;
 
 //-----------------------------------------------------------------------------
 // Forward declarations
 forall(otype T | is_thread(T) )
-void start( thread(T)* this );
+void start( T* this );
 
 forall(otype T | is_thread(T) )
-void stop( thread(T)* this );
+void stop( T* this );
 
 //-----------------------------------------------------------------------------
@@ -38 +42 @@
 void ?{}(thread_h* this) {
         (&this->c){};
+        (&this->lock){};
+        this->next = NULL;
 }
 
@@ -47 +53 @@
 void ?{}( thread(T)* this ) {
         (&this->handle){};
-        start(this);
+        start(&this->handle);
 }
 
@@ -53 +59 @@
 void ?{}( thread(T)* this, P params ) {
         (&this->handle){ params };
-        start(this);
+        start(&this->handle);
 }
 
 forall(otype T | is_thread(T) )
 void ^?{}( thread(T)* this ) {
-        stop(this);
+        stop(&this->handle);
         ^(&this->handle){};
 }
@@ -69 +75 @@
 }
 
+extern void thread_schedule( thread_h * );
+
 forall(otype T | is_thread(T))
-void start( thread(T)* this ) {
-        T* handle  = &this->handle;
-        coroutine* thrd_c = get_coroutine(handle);
-        thread_h*  thrd_h = get_thread   (handle);
+void start( T* this ) {
+        coroutine* thrd_c = get_coroutine(this);
+        thread_h*  thrd_h = get_thread   (this);
         thrd_c->last = this_coroutine();
-        current_coroutine = thrd_c;
+        this_processor->current_coroutine = thrd_c;
 
         // LIB_DEBUG_PRINTF("Thread start : %p (t %p, c %p)\n", handle, thrd_c, thrd_h);
 
         create_stack(&thrd_c->stack, thrd_c->stack.size);
-        CtxStart(handle, CtxInvokeThread);
+        CtxStart(this, CtxInvokeThread);
         CtxSwitch( thrd_c->last->stack.context, thrd_c->stack.context );
 
-        scheduler_add(thrd_h);
+        thread_schedule(thrd_h);
 }
 
 forall(otype T | is_thread(T) )
-void stop( thread(T)* this ) {
-        T* handle  = &this->handle;
-        thread_h*  thrd_h = get_thread   (handle);
-        while( thrd_h->c.notHalted ) {
-                suspend();
+void stop( T* this ) {
+        thread_h*  thrd = get_thread(this);
+        if( thrd->c.notHalted ) {
+                lock( &thrd->lock );
         }
+}
+
+void signal_termination( thread_h * this ) {
+        this->c.state = Halt;
+      this->c.notHalted = false;
+        unlock( &this->lock );
+}
+
+void yield( void ) {
+        thread_schedule( this_thread() );
+        suspend();
 }