Index: doc/proposals/concurrency/thePlan.md =================================================================== --- doc/proposals/concurrency/thePlan.md (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ doc/proposals/concurrency/thePlan.md (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -4,14 +4,15 @@ done - SimpleBlockingLock. done - Synchronisation points in thread destructors. -Processors & SpinLock. +done - Processors & SpinLock. _Phase 2_ : Minimum Viable Product -Basic monitors for synchronisation (No internal/external scheduling). -Non-thrash scheduler. -Clusters. +Monitor type and enter/leave mutex member routines +Monitors as a language feature (not calling enter/leave by hand) +Internal scheduling _Phase 3_ : Kernel features -Threads features ex: detach -Internal scheduling +Detach thread +Cluster migration +Preemption _Phase 4_ : Monitor features Index: src/driver/cfa.cc =================================================================== --- src/driver/cfa.cc (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/driver/cfa.cc (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -267,4 +267,6 @@ } nargs += 1; + args[nargs] = "-lpthread"; + nargs += 1; } // if #endif //HAVE_LIBCFA Index: src/examples/thread.c =================================================================== --- src/examples/thread.c (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/examples/thread.c (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -1,3 +1,5 @@ +#line 1 "thread.c" #include +#include #include #include @@ -10,5 +12,8 @@ }; -// DECL_THREAD(MyThread) +DECL_THREAD(MyThread) + +void ?{}( MyThread * this ) { +} void ?{}( MyThread * this, unsigned id, unsigned count ) { @@ -17,29 +22,32 @@ } -// void main(MyThread* this) { -// sout | "Thread" | this->id | " : Suspending" | this->count | "times" | endl; -// yield(); +void main(MyThread* this) { + sout | "Thread" | this->id | " : Suspending" | this->count | "times" | endl; + yield(); -// for(int i = 0; i < this->count; i++) { -// sout | "Thread" | this->id | " : Suspend No." | i + 1 | endl; -// yield(); -// } -// } + for(int i = 0; i < this->count; i++) { + sout | "Thread" | this->id | " : Suspend No." | i + 1 | endl; + yield(); + } +} int main(int argc, char* argv[]) { - // unsigned itterations = 10u; - // if(argc == 2) { - // int val = ato(argv[1]); - // assert(val >= 0); - // itterations = val; - // } + unsigned itterations = 10u; + if(argc == 2) { + int val = ato(argv[1]); + assert(val >= 0); + itterations = val; + } sout | "User main begin" | endl; - // { - // thread(MyThread) thread1 = { 1u, itterations }; - // thread(MyThread) thread2 = { 2u, itterations }; - // } + { + processor p; + { + thread(MyThread) thread1 = { 1u, itterations }; + thread(MyThread) thread2 = { 2u, itterations }; + } + } sout | "User main end" | endl; Index: src/libcfa/concurrency/coroutines =================================================================== --- src/libcfa/concurrency/coroutines (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/coroutines (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -73,9 +73,9 @@ assertf( src->last != 0, - "Attempt to suspend coroutine %.256s (%p) that has never been resumed.\n" + "Attempt to suspend coroutine \"%.256s\" (%p) that has never been resumed.\n" "Possible cause is a suspend executed in a member called by a coroutine user rather than by the coroutine main.", src->name, src ); assertf( src->last->notHalted, - "Attempt by coroutine %.256s (%p) to suspend back to terminated coroutine %.256s (%p).\n" + "Attempt by coroutine \"%.256s\" (%p) to suspend back to terminated coroutine \"%.256s\" (%p).\n" "Possible cause is terminated coroutine's main routine has already returned.", src->name, src, src->last->name, src->last ); Index: src/libcfa/concurrency/coroutines.c =================================================================== --- src/libcfa/concurrency/coroutines.c (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/coroutines.c (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -1,2 +1,3 @@ +// -*- Mode: CFA -*- // // Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo @@ -31,5 +32,5 @@ #include "invoke.h" -/*thread_local*/ extern processor * this_processor; +extern processor * get_this_processor(); //----------------------------------------------------------------------------- @@ -110,5 +111,5 @@ // set new coroutine that task is executing - this_processor->current_coroutine = dst; + get_this_processor()->current_coroutine = dst; // context switch to specified coroutine Index: src/libcfa/concurrency/invoke.c =================================================================== --- src/libcfa/concurrency/invoke.c (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/invoke.c (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -1,2 +1,17 @@ +// -*- Mode: CFA -*- +// +// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo +// +// The contents of this file are covered under the licence agreement in the +// file "LICENCE" distributed with Cforall. +// +// invoke.c -- +// +// Author : Thierry Delisle +// Created On : Tue Jan 17 12:27:26 2016 +// Last Modified By : Thierry Delisle +// Last Modified On : -- +// Update Count : 0 +// #include Index: src/libcfa/concurrency/invoke.h =================================================================== --- src/libcfa/concurrency/invoke.h (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/invoke.h (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -1,2 +1,18 @@ +// -*- Mode: CFA -*- +// +// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo +// +// The contents of this file are covered under the licence agreement in the +// file "LICENCE" distributed with Cforall. +// +// invoke.h -- +// +// Author : Thierry Delisle +// Created On : Tue Jan 17 12:27:26 2016 +// Last Modified By : Thierry Delisle +// Last Modified On : -- +// Update Count : 0 +// + #include #include @@ -11,4 +27,5 @@ #define unlikely(x) __builtin_expect(!!(x), 0) + #define thread_local _Thread_local #define SCHEDULER_CAPACITY 10 Index: src/libcfa/concurrency/kernel =================================================================== --- src/libcfa/concurrency/kernel (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/kernel (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -22,8 +22,13 @@ #include "invoke.h" +extern "C" { +#include +} + //----------------------------------------------------------------------------- // Cluster struct cluster { simple_thread_list ready_queue; + pthread_spinlock_t lock; }; @@ -38,7 +43,10 @@ coroutine * current_coroutine; thread_h * current_thread; - bool terminated; + pthread_t kernel_thread; + simple_lock lock; + volatile bool terminated; }; +void ?{}(processor * this); void ?{}(processor * this, cluster * cltr); void ^?{}(processor * this); @@ -54,4 +62,41 @@ void unlock( simple_lock * ); +struct pthread_spinlock_guard { + pthread_spinlock_t * lock; +}; + +static inline void ?{}( pthread_spinlock_guard * this, pthread_spinlock_t * lock ) { + this->lock = lock; + pthread_spin_lock( this->lock ); +} + +static inline void ^?{}( pthread_spinlock_guard * this ) { + pthread_spin_unlock( this->lock ); +} + +// //Simple spinlock implementation from +// //http://stackoverflow.com/questions/1383363/is-my-spin-lock-implementation-correct-and-optimal +// //Not optimal but correct +// #define VOL + +// struct simple_spinlock { +// VOL int lock; +// }; + +// extern VOL int __sync_lock_test_and_set( VOL int *, VOL int); +// extern void __sync_synchronize(); + +// static inline void lock( simple_spinlock * this ) { +// while (__sync_lock_test_and_set(&this->lock, 1)) { +// // Do nothing. This GCC builtin instruction +// // ensures memory barrier. +// } +// } + +// static inline void unlock( simple_spinlock * this ) { +// __sync_synchronize(); // Memory barrier. +// this->lock = 0; +// } + #endif //KERNEL_H Index: src/libcfa/concurrency/kernel.c =================================================================== --- src/libcfa/concurrency/kernel.c (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/kernel.c (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -32,40 +32,6 @@ #include "invoke.h" -cluster * systemCluster; -processor * systemProcessor; -thread_h * mainThread; - -void kernel_startup(void) __attribute__((constructor(101))); -void kernel_shutdown(void) __attribute__((destructor(101))); - -void ?{}(processor * this, cluster * cltr) { - this->ctx = NULL; - this->cltr = cltr; - this->terminated = false; -} - -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; -} - -//----------------------------------------------------------------------------- -// Processor coroutine +//----------------------------------------------------------------------------- +// Kernel storage struct processorCtx_t { processor * proc; @@ -75,12 +41,141 @@ DECL_COROUTINE(processorCtx_t) +#define KERNEL_STORAGE(T,X) static char X##_storage[sizeof(T)] + +KERNEL_STORAGE(processorCtx_t, systemProcessorCtx); +KERNEL_STORAGE(cluster, systemCluster); +KERNEL_STORAGE(processor, systemProcessor); +KERNEL_STORAGE(thread_h, mainThread); +KERNEL_STORAGE(machine_context_t, mainThread_context); + +cluster * systemCluster; +processor * systemProcessor; +thread_h * mainThread; + +void kernel_startup(void) __attribute__((constructor(101))); +void kernel_shutdown(void) __attribute__((destructor(101))); + +//----------------------------------------------------------------------------- +// Global state + +thread_local processor * this_processor; + +processor * get_this_processor() { + return this_processor; +} + +coroutine * this_coroutine(void) { + return this_processor->current_coroutine; +} + +thread_h * this_thread(void) { + return this_processor->current_thread; +} + +//----------------------------------------------------------------------------- +// Main thread construction +struct current_stack_info_t { + machine_context_t ctx; + unsigned int size; // size of stack + void *base; // base of stack + void *storage; // pointer to stack + void *limit; // stack grows towards stack limit + void *context; // address of cfa_context_t + void *top; // address of top of storage +}; + +void ?{}( current_stack_info_t * this ) { + CtxGet( &this->ctx ); + this->base = this->ctx.FP; + this->storage = this->ctx.SP; + + rlimit r; + int ret = getrlimit( RLIMIT_STACK, &r); + this->size = r.rlim_cur; + + this->limit = (void *)(((intptr_t)this->base) - this->size); + this->context = &mainThread_context_storage; + this->top = this->base; +} + +void ?{}( coStack_t * this, current_stack_info_t * info) { + this->size = info->size; + this->storage = info->storage; + this->limit = info->limit; + this->base = info->base; + this->context = info->context; + this->top = info->top; + this->userStack = true; +} + +void ?{}( coroutine * this, current_stack_info_t * info) { + (&this->stack){ info }; + this->name = "Main Thread"; + this->errno_ = 0; + this->state = Inactive; + this->notHalted = true; +} + +void ?{}( thread_h * this, current_stack_info_t * info) { + (&this->c){ info }; +} + +//----------------------------------------------------------------------------- +// Processor coroutine void ?{}(processorCtx_t * this, processor * proc) { (&this->c){}; this->proc = proc; -} - -void CtxInvokeProcessor(processor * proc) { - processorCtx_t proc_cor_storage = {proc}; - resume( &proc_cor_storage ); + proc->ctx = this; +} + +void ?{}(processorCtx_t * this, processor * proc, current_stack_info_t * info) { + (&this->c){ info }; + this->proc = proc; + proc->ctx = this; +} + +void start(processor * this); + +void ?{}(processor * this) { + this{ systemCluster }; +} + +void ?{}(processor * this, cluster * cltr) { + this->cltr = cltr; + this->current_coroutine = NULL; + this->current_thread = NULL; + (&this->lock){}; + this->terminated = false; + + start( this ); +} + +void ?{}(processor * this, cluster * cltr, processorCtx_t * ctx) { + this->cltr = cltr; + this->current_coroutine = NULL; + this->current_thread = NULL; + (&this->lock){}; + this->terminated = false; + + this->ctx = ctx; + LIB_DEBUG_PRINTF("Kernel : constructing processor context %p\n", ctx); + ctx{ this }; +} + +void ^?{}(processor * this) { + if( ! this->terminated ) { + LIB_DEBUG_PRINTF("Kernel : core %p signaling termination\n", this); + this->terminated = true; + lock( &this->lock ); + } +} + +void ?{}(cluster * this) { + ( &this->ready_queue ){}; + pthread_spin_init( &this->lock, PTHREAD_PROCESS_PRIVATE ); +} + +void ^?{}(cluster * this) { + pthread_spin_destroy( &this->lock ); } @@ -109,4 +204,6 @@ } + LIB_DEBUG_PRINTF("Kernel : core %p unlocking thread\n", this); + unlock( &this->lock ); LIB_DEBUG_PRINTF("Kernel : core %p terminated\n", this); } @@ -133,71 +230,68 @@ } +void * CtxInvokeProcessor(void * arg) { + processor * proc = (processor *) arg; + this_processor = proc; + // SKULLDUGGERY: We want to create a context for the processor coroutine + // which is needed for the 2-step context switch. However, there is no reason + // to waste the perfectly valid stack create by pthread. + current_stack_info_t info; + machine_context_t ctx; + info.context = &ctx; + processorCtx_t proc_cor_storage = { proc, &info }; + + proc->current_coroutine = &proc->ctx->c; + proc->current_thread = NULL; + + 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 + // resume it to start it like it normally would, it will just context switch + // back to here. Instead directly call the main since we already are on the + // appropriate stack. + proc_cor_storage.c.state = Active; + main( &proc_cor_storage ); + proc_cor_storage.c.state = Halt; + proc_cor_storage.c.notHalted = false; + + LIB_DEBUG_PRINTF("Kernel : core %p main ended (%p)\n", proc, proc->ctx); + + return NULL; +} + +void start(processor * this) { + LIB_DEBUG_PRINTF("Kernel : Starting core %p\n", this); + + pthread_attr_t attributes; + pthread_attr_init( &attributes ); + + pthread_create( &this->kernel_thread, &attributes, CtxInvokeProcessor, (void*)this ); + + pthread_attr_destroy( &attributes ); + + LIB_DEBUG_PRINTF("Kernel : core %p started\n", this); +} + //----------------------------------------------------------------------------- // Scheduler routines void thread_schedule( thread_h * thrd ) { assertf( thrd->next == NULL, "Expected null got %p", thrd->next ); + + pthread_spinlock_guard guard = { &systemProcessor->cltr->lock }; append( &systemProcessor->cltr->ready_queue, thrd ); } thread_h * nextThread(cluster * this) { + pthread_spinlock_guard guard = { &this->lock }; return pop_head( &this->ready_queue ); -} - -//----------------------------------------------------------------------------- -// Kernel storage -#define KERNEL_STORAGE(T,X) static char X##_storage[sizeof(T)] - -KERNEL_STORAGE(processorCtx_t, systemProcessorCtx); -KERNEL_STORAGE(cluster, systemCluster); -KERNEL_STORAGE(processor, systemProcessor); -KERNEL_STORAGE(thread_h, mainThread); -KERNEL_STORAGE(machine_context_t, mainThread_context); - -//----------------------------------------------------------------------------- -// Main thread construction -struct mainThread_info_t { - machine_context_t ctx; - unsigned int size; // size of stack - void *base; // base of stack - void *storage; // pointer to stack - void *limit; // stack grows towards stack limit - void *context; // address of cfa_context_t - void *top; // address of top of storage -}; - -void ?{}( mainThread_info_t * this ) { - CtxGet( &this->ctx ); - this->base = this->ctx.FP; - this->storage = this->ctx.SP; - - rlimit r; - int ret = getrlimit( RLIMIT_STACK, &r); - this->size = r.rlim_cur; - - this->limit = (void *)(((intptr_t)this->base) - this->size); - this->context = &mainThread_context_storage; - this->top = this->base; -} - -void ?{}( coStack_t * this, mainThread_info_t * info) { - this->size = info->size; - this->storage = info->storage; - this->limit = info->limit; - this->base = info->base; - this->context = info->context; - this->top = info->top; - this->userStack = true; -} - -void ?{}( coroutine * this, mainThread_info_t * info) { - (&this->stack){ info }; - this->name = "Main Thread"; - this->errno_ = 0; - this->state = Inactive; - this->notHalted = true; -} - -void ?{}( thread_h * this, mainThread_info_t * info) { - (&this->c){ info }; } @@ -210,9 +304,16 @@ LIB_DEBUG_PRINTF("Kernel : Starting\n"); - mainThread_info_t ctx; + current_stack_info_t info; + + // 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 ); // Start by initializing the main thread mainThread = (thread_h *)&mainThread_storage; - mainThread{ &ctx }; + mainThread{ &info }; // Initialize the system cluster @@ -220,12 +321,8 @@ systemCluster{}; - // Initialize the system processor + // Initialize the system processor and the system processor ctx + // (the coroutine that contains the processing control flow) systemProcessor = (processor *)&systemProcessor_storage; - systemProcessor{ systemCluster }; - - // Initialize the system processor ctx - // (the coroutine that contains the processing control flow) - systemProcessor->ctx = (processorCtx_t *)&systemProcessorCtx_storage; - systemProcessor->ctx{ systemProcessor }; + systemProcessor{ systemCluster, (processorCtx_t *)&systemProcessorCtx_storage }; // Add the main thread to the ready queue @@ -286,5 +383,8 @@ void lock( simple_lock * this ) { - append( &this->blocked, this_thread() ); + { + pthread_spinlock_guard guard = { &systemCluster->lock }; //HUGE TEMP HACK which only works if we have a single cluster and is stupid + append( &this->blocked, this_thread() ); + } suspend(); } Index: src/libcfa/concurrency/threads =================================================================== --- src/libcfa/concurrency/threads (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/threads (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -27,14 +27,12 @@ // 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 | sized(T)) { void main(T* this); thread_h* get_thread(T* this); - /*void ?{}(T*); - void ^?{}(T*);*/ }; #define DECL_THREAD(X) static inline thread_h* get_thread(X* this) { return &this->t; } void main(X* this); -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) ) static inline coroutine* get_coroutine(T* this) { return &get_thread(this)->c; @@ -55,16 +53,16 @@ // thread runner // Structure that actually start and stop threads -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) ) struct thread { T handle; }; -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) | { void ?{}(T*); } ) void ?{}( thread(T)* this ); -forall(otype T, ttype P | is_thread(T) | { void ?{}(T*, P); } ) +forall( dtype T, ttype P | sized(T) | is_thread(T) | { void ?{}(T*, P); } ) void ?{}( thread(T)* this, P params ); -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) | { void ^?{}(T*); } ) void ^?{}( thread(T)* this ); Index: src/libcfa/concurrency/threads.c =================================================================== --- src/libcfa/concurrency/threads.c (revision 4a9ccc35f7b5fed64bf53eee245fc9caa37f687d) +++ src/libcfa/concurrency/threads.c (revision ad6343e4834fdb20857e63e71d2abdd16cb65a90) @@ -27,12 +27,12 @@ } -/*thread_local*/ extern processor * this_processor; +extern processor * get_this_processor(); //----------------------------------------------------------------------------- // Forward declarations -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) ) void start( T* this ); -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) ) void stop( T* this ); @@ -42,4 +42,5 @@ void ?{}(thread_h* this) { (&this->c){}; + this->c.name = "Anonymous Coroutine"; (&this->lock){}; this->next = NULL; @@ -50,5 +51,5 @@ } -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) | { void ?{}(T*); } ) void ?{}( thread(T)* this ) { (&this->handle){}; @@ -56,5 +57,5 @@ } -forall(otype T, ttype P | is_thread(T) | { void ?{}(T*, P); } ) +forall( dtype T, ttype P | sized(T) | is_thread(T) | { void ?{}(T*, P); } ) void ?{}( thread(T)* this, P params ) { (&this->handle){ params }; @@ -62,5 +63,5 @@ } -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) | { void ^?{}(T*); } ) void ^?{}( thread(T)* this ) { stop(&this->handle); @@ -77,10 +78,10 @@ extern void thread_schedule( thread_h * ); -forall(otype T | is_thread(T)) +forall( dtype T | sized(T) | is_thread(T) ) void start( T* this ) { coroutine* thrd_c = get_coroutine(this); thread_h* thrd_h = get_thread (this); thrd_c->last = this_coroutine(); - this_processor->current_coroutine = thrd_c; + get_this_processor()->current_coroutine = thrd_c; // LIB_DEBUG_PRINTF("Thread start : %p (t %p, c %p)\n", handle, thrd_c, thrd_h); @@ -93,5 +94,5 @@ } -forall(otype T | is_thread(T) ) +forall( dtype T | sized(T) | is_thread(T) ) void stop( T* this ) { thread_h* thrd = get_thread(this);