Index: src/libcfa/concurrency/alarm.c =================================================================== --- src/libcfa/concurrency/alarm.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/alarm.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -16,5 +16,9 @@ extern "C" { +#include +#include +#include #include +#include #include } @@ -22,4 +26,5 @@ #include "alarm.h" #include "kernel_private.h" +#include "libhdr.h" #include "preemption.h" @@ -31,8 +36,21 @@ timespec curr; clock_gettime( CLOCK_REALTIME, &curr ); - return ((__cfa_time_t)curr.tv_sec * TIMEGRAN) + curr.tv_nsec; + __cfa_time_t curr_time = ((__cfa_time_t)curr.tv_sec * TIMEGRAN) + curr.tv_nsec; + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Kernel : current time is %lu\n", curr_time ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + return curr_time; } void __kernel_set_timer( __cfa_time_t alarm ) { + + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Kernel : set timer to %lu\n", (__cfa_time_t)alarm ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + itimerval val; val.it_value.tv_sec = alarm / TIMEGRAN; // seconds @@ -128,7 +146,12 @@ lock( &systemProcessor->alarm_lock ); { + bool first = !systemProcessor->alarms.head; + insert( &systemProcessor->alarms, this ); if( systemProcessor->pending_alarm ) { tick_preemption(); + } + if( first ) { + __kernel_set_timer( systemProcessor->alarms.head->alarm - __kernel_get_time() ); } } Index: src/libcfa/concurrency/invoke.c =================================================================== --- src/libcfa/concurrency/invoke.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/invoke.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -30,4 +30,6 @@ extern void __suspend_internal(void); extern void __leave_monitor_desc( struct monitor_desc * this ); +extern void disable_interrupts(); +extern void enable_interrupts(); void CtxInvokeCoroutine( @@ -67,8 +69,10 @@ struct monitor_desc* mon = &thrd->mon; cor->state = Active; + enable_interrupts(); // LIB_DEBUG_PRINTF("Invoke Thread : invoking main %p (args %p)\n", main, this); main( this ); + disable_interrupts(); __leave_monitor_desc( mon ); Index: src/libcfa/concurrency/kernel.c =================================================================== --- src/libcfa/concurrency/kernel.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/kernel.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -154,9 +154,12 @@ (&this->terminated){}; this->is_terminated = false; - this->disable_preempt_count = 0; + this->preemption_alarm = NULL; + this->preemption = default_preemption(); + this->disable_preempt_count = 1; this->pending_preemption = false; + this->kernel_thread = pthread_self(); this->runner = runner; - LIB_DEBUG_PRINT_SAFE("Kernel : constructing processor context %p\n", runner); + LIB_DEBUG_PRINT_SAFE("Kernel : constructing system processor context %p\n", runner); runner{ this }; } @@ -240,4 +243,6 @@ //Update global state this->current_thread = dst; + + LIB_DEBUG_PRINT_SAFE("Kernel : running %p\n", dst); // Context Switch to the thread @@ -322,6 +327,33 @@ void start(processor * this) { LIB_DEBUG_PRINT_SAFE("Kernel : Starting core %p\n", this); - + + // SIGALRM must only be caught by the system processor + sigset_t old_mask; + bool is_system_proc = this_processor == &systemProcessor->proc; + if ( is_system_proc ) { + // Child kernel-thread inherits the signal mask from the parent kernel-thread. So one special case for the + // system processor creating the user processor => toggle the blocking SIGALRM on system processor, create user + // processor, and toggle back (below) previous signal mask of the system processor. + + sigset_t new_mask; + sigemptyset( &new_mask ); + sigemptyset( &old_mask ); + sigaddset( &new_mask, SIGALRM ); + + if ( sigprocmask( SIG_BLOCK, &new_mask, &old_mask ) == -1 ) { + abortf( "internal error, sigprocmask" ); + } + + assert( ! sigismember( &old_mask, SIGALRM ) ); + } + pthread_create( &this->kernel_thread, NULL, CtxInvokeProcessor, (void*)this ); + + // Toggle back previous signal mask of system processor. + if ( is_system_proc ) { + if ( sigprocmask( SIG_SETMASK, &old_mask, NULL ) == -1 ) { + abortf( "internal error, sigprocmask" ); + } // if + } // if LIB_DEBUG_PRINT_SAFE("Kernel : core %p started\n", this); @@ -347,35 +379,46 @@ } -void ScheduleInternal() { +void BlockInternal() { + disable_interrupts(); suspend(); -} - -void ScheduleInternal( spinlock * lock ) { + enable_interrupts(); +} + +void BlockInternal( spinlock * lock ) { + disable_interrupts(); this_processor->finish.action_code = Release; this_processor->finish.lock = lock; suspend(); -} - -void ScheduleInternal( thread_desc * thrd ) { + enable_interrupts(); +} + +void BlockInternal( thread_desc * thrd ) { + disable_interrupts(); this_processor->finish.action_code = Schedule; this_processor->finish.thrd = thrd; suspend(); -} - -void ScheduleInternal( spinlock * lock, thread_desc * thrd ) { + enable_interrupts(); +} + +void BlockInternal( spinlock * lock, thread_desc * thrd ) { + disable_interrupts(); this_processor->finish.action_code = Release_Schedule; this_processor->finish.lock = lock; this_processor->finish.thrd = thrd; suspend(); -} - -void ScheduleInternal(spinlock ** locks, unsigned short count) { + enable_interrupts(); +} + +void BlockInternal(spinlock ** locks, unsigned short count) { + disable_interrupts(); this_processor->finish.action_code = Release_Multi; this_processor->finish.locks = locks; this_processor->finish.lock_count = count; suspend(); -} - -void ScheduleInternal(spinlock ** locks, unsigned short lock_count, thread_desc ** thrds, unsigned short thrd_count) { + enable_interrupts(); +} + +void BlockInternal(spinlock ** locks, unsigned short lock_count, thread_desc ** thrds, unsigned short thrd_count) { + disable_interrupts(); this_processor->finish.action_code = Release_Multi_Schedule; this_processor->finish.locks = locks; @@ -384,4 +427,5 @@ this_processor->finish.thrd_count = thrd_count; suspend(); + enable_interrupts(); } @@ -403,7 +447,4 @@ LIB_DEBUG_PRINT_SAFE("Kernel : Main thread ready\n"); - // Enable preemption - kernel_start_preemption(); - // Initialize the system cluster systemCluster = (cluster *)&systemCluster_storage; @@ -426,4 +467,7 @@ this_processor->current_coroutine = &mainThread->cor; + // Enable preemption + kernel_start_preemption(); + // SKULLDUGGERY: Force a context switch to the system processor to set the main thread's context to the current UNIX // context. Hence, the main thread does not begin through CtxInvokeThread, like all other threads. The trick here is that @@ -435,4 +479,6 @@ // THE SYSTEM IS NOW COMPLETELY RUNNING LIB_DEBUG_PRINT_SAFE("Kernel : Started\n--------------------------------------------------\n\n"); + + enable_interrupts(); } @@ -447,4 +493,7 @@ // THE SYSTEM IS NOW COMPLETELY STOPPED + + // Disable preemption + kernel_stop_preemption(); // Destroy the system processor and its context in reverse order of construction @@ -550,5 +599,5 @@ if( !this->cond ) { append( &this->blocked, this_thread() ); - ScheduleInternal( &this->lock ); + BlockInternal( &this->lock ); lock( &this->lock ); } Index: src/libcfa/concurrency/kernel_private.h =================================================================== --- src/libcfa/concurrency/kernel_private.h (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/kernel_private.h (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -28,10 +28,10 @@ thread_desc * nextThread(cluster * this); -void ScheduleInternal(void); -void ScheduleInternal(spinlock * lock); -void ScheduleInternal(thread_desc * thrd); -void ScheduleInternal(spinlock * lock, thread_desc * thrd); -void ScheduleInternal(spinlock ** locks, unsigned short count); -void ScheduleInternal(spinlock ** locks, unsigned short count, thread_desc ** thrds, unsigned short thrd_count); +void BlockInternal(void); +void BlockInternal(spinlock * lock); +void BlockInternal(thread_desc * thrd); +void BlockInternal(spinlock * lock, thread_desc * thrd); +void BlockInternal(spinlock ** locks, unsigned short count); +void BlockInternal(spinlock ** locks, unsigned short count, thread_desc ** thrds, unsigned short thrd_count); //----------------------------------------------------------------------------- @@ -60,21 +60,8 @@ extern thread_local processor * this_processor; -static inline void disable_interrupts() { - __attribute__((unused)) unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, 1, __ATOMIC_SEQ_CST ); - assert( prev != (unsigned short) -1 ); -} - -static inline void enable_interrupts_noRF() { - unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, -1, __ATOMIC_SEQ_CST ); - assert( prev != (unsigned short) 0 ); -} - -static inline void enable_interrupts() { - unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, -1, __ATOMIC_SEQ_CST ); - assert( prev != (unsigned short) 0 ); - if( prev == 1 && this_processor->pending_preemption ) { - ScheduleInternal( this_processor->current_thread ); - this_processor->pending_preemption = false; - } +extern "C" { + void disable_interrupts(); + void enable_interrupts_noRF(); + void enable_interrupts(); } Index: src/libcfa/concurrency/monitor.c =================================================================== --- src/libcfa/concurrency/monitor.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/monitor.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -63,7 +63,7 @@ append( &this->entry_queue, thrd ); LIB_DEBUG_PRINT_SAFE("%p Blocking on entry\n", thrd); - ScheduleInternal( &this->lock ); - - //ScheduleInternal will unlock spinlock, no need to unlock ourselves + BlockInternal( &this->lock ); + + //BlockInternal will unlock spinlock, no need to unlock ourselves return; } @@ -172,5 +172,5 @@ unsigned short count = this->monitor_count; unsigned int recursions[ count ]; //Save the current recursion levels to restore them later - spinlock * locks [ count ]; //We need to pass-in an array of locks to ScheduleInternal + spinlock * locks [ count ]; //We need to pass-in an array of locks to BlockInternal LIB_DEBUG_PRINT_SAFE("count %i\n", count); @@ -210,5 +210,5 @@ // Everything is ready to go to sleep - ScheduleInternal( locks, count, threads, thread_count ); + BlockInternal( locks, count, threads, thread_count ); @@ -283,5 +283,5 @@ unsigned short count = this->monitor_count; unsigned int recursions[ count ]; //Save the current recursion levels to restore them later - spinlock * locks [ count ]; //We need to pass-in an array of locks to ScheduleInternal + spinlock * locks [ count ]; //We need to pass-in an array of locks to BlockInternal lock_all( this->monitors, locks, count ); @@ -311,5 +311,5 @@ //Everything is ready to go to sleep - ScheduleInternal( locks, count, &signallee, 1 ); + BlockInternal( locks, count, &signallee, 1 ); @@ -343,5 +343,5 @@ // unsigned short count = this->current_monitor_count; // unsigned int recursions[ count ]; //Save the current recursion levels to restore them later - // spinlock * locks [ count ]; //We need to pass-in an array of locks to ScheduleInternal + // spinlock * locks [ count ]; //We need to pass-in an array of locks to BlockInternal // lock_all( this->current_monitors, locks, count ); @@ -352,5 +352,5 @@ // // // Everything is ready to go to sleep - // // ScheduleInternal( locks, count, threads, thread_count ); + // // BlockInternal( locks, count, threads, thread_count ); Index: src/libcfa/concurrency/preemption.c =================================================================== --- src/libcfa/concurrency/preemption.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/preemption.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -18,8 +18,14 @@ extern "C" { +#include #include -} - -#define __CFA_DEFAULT_PREEMPTION__ 10 +#include +#include +#include +} + +#include "libhdr.h" + +#define __CFA_DEFAULT_PREEMPTION__ 10000 __attribute__((weak)) unsigned int default_preemption() { @@ -27,7 +33,15 @@ } +#define __CFA_SIGCXT__ ucontext_t * +#define __CFA_SIGPARMS__ __attribute__((unused)) int sig, __attribute__((unused)) siginfo_t *sfp, __attribute__((unused)) __CFA_SIGCXT__ cxt + static void preempt( processor * this ); static void timeout( thread_desc * this ); +void sigHandler_ctxSwitch( __CFA_SIGPARMS__ ); +void sigHandler_alarm ( __CFA_SIGPARMS__ ); + +static void __kernel_sigaction( int sig, void (*handler)(__CFA_SIGPARMS__), int flags ); + //============================================================================================= // Kernel Preemption logic @@ -35,12 +49,32 @@ void kernel_start_preemption() { - + LIB_DEBUG_PRINT_SAFE("Kernel : Starting preemption\n"); + __kernel_sigaction( SIGUSR1, sigHandler_ctxSwitch, SA_SIGINFO ); + __kernel_sigaction( SIGALRM, sigHandler_alarm , SA_SIGINFO ); +} + +void kernel_stop_preemption() { + //Block all signals, we are no longer in a position to handle them + sigset_t mask; + sigfillset( &mask ); + sigprocmask( SIG_BLOCK, &mask, NULL ); + LIB_DEBUG_PRINT_SAFE("Kernel : Preemption stopped\n"); } void tick_preemption() { + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Ticking preemption\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + alarm_list_t * alarms = &systemProcessor->alarms; __cfa_time_t currtime = __kernel_get_time(); while( alarms->head && alarms->head->alarm < currtime ) { alarm_node_t * node = pop(alarms); + LIB_DEBUG_DO( + len = snprintf( text, 256, "Ticking %p\n", node ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); if( node->kernel_alarm ) { preempt( node->proc ); @@ -51,5 +85,5 @@ if( node->period > 0 ) { - node->alarm += node->period; + node->alarm = currtime + node->period; insert( alarms, node ); } @@ -62,9 +96,20 @@ __kernel_set_timer( alarms->head->alarm - currtime ); } + + LIB_DEBUG_DO( + len = snprintf( text, 256, "Ticking preemption done\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); } void update_preemption( processor * this, __cfa_time_t duration ) { - // assert( THREAD_GETMEM( disableInt ) && THREAD_GETMEM( disableIntCnt ) == 1 ); + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Processor : updating preemption to %lu\n", duration ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + alarm_node_t * alarm = this->preemption_alarm; + duration *= 1000; // Alarms need to be enabled @@ -94,7 +139,11 @@ this->proc->preemption_alarm = &this->alarm; update_preemption( this->proc, this->proc->preemption ); + + // enable_interrupts(); } void ^?{}( preemption_scope * this ) { + disable_interrupts(); + update_preemption( this->proc, 0 ); } @@ -103,4 +152,42 @@ // Kernel Signal logic //============================================================================================= + +extern "C" { + void disable_interrupts() { + __attribute__((unused)) unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, 1, __ATOMIC_SEQ_CST ); + assert( prev != (unsigned short) -1 ); + } + + void enable_interrupts_noRF() { + unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, -1, __ATOMIC_SEQ_CST ); + assert( prev != (unsigned short) 0 ); + } + + void enable_interrupts() { + unsigned short prev = __atomic_fetch_add_2( &this_processor->disable_preempt_count, -1, __ATOMIC_SEQ_CST ); + assert( prev != (unsigned short) 0 ); + if( prev == 1 && this_processor->pending_preemption ) { + this_processor->pending_preemption = false; + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Executing deferred CtxSwitch\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + BlockInternal( this_processor->current_thread ); + } + } +} + +static inline void signal_unblock( bool alarm ) { + sigset_t mask; + sigemptyset( &mask ); + sigaddset( &mask, SIGUSR1 ); + + if( alarm ) sigaddset( &mask, SIGALRM ); + + if ( sigprocmask( SIG_UNBLOCK, &mask, NULL ) == -1 ) { + abortf( "internal error, sigprocmask" ); + } // if +} static inline bool preemption_ready() { @@ -116,14 +203,38 @@ } -void sigHandler_ctxSwitch( __attribute__((unused)) int sig ) { +void sigHandler_ctxSwitch( __CFA_SIGPARMS__ ) { + + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Ctx Switch IRH\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + + signal_unblock( false ); if( preemption_ready() ) { - ScheduleInternal( this_processor->current_thread ); + LIB_DEBUG_DO( + len = snprintf( text, 256, "Ctx Switch IRH : Blocking thread\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + BlockInternal( this_processor->current_thread ); } else { + LIB_DEBUG_DO( + len = snprintf( text, 256, "Ctx Switch IRH : Defering\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); defer_ctxSwitch(); } } -void sigHandler_alarm( __attribute__((unused)) int sig ) { +void sigHandler_alarm( __CFA_SIGPARMS__ ) { + + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "\nAlarm IRH\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + + signal_unblock( true ); if( try_lock( &systemProcessor->alarm_lock ) ) { tick_preemption(); @@ -133,8 +244,28 @@ defer_alarm(); } + + if( preemption_ready() && this_processor->pending_preemption ) { + LIB_DEBUG_DO( + len = snprintf( text, 256, "Alarm IRH : Blocking thread\n" ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + this_processor->pending_preemption = false; + BlockInternal( this_processor->current_thread ); + } } static void preempt( processor * this ) { - pthread_kill( this->kernel_thread, SIGUSR1 ); + LIB_DEBUG_DO( + char text[256]; + __attribute__((unused)) int len = snprintf( text, 256, "Processor : signalling %p\n", this ); + LIB_DEBUG_WRITE( STDERR_FILENO, text, len ); + ); + + if( this != systemProcessor ) { + pthread_kill( this->kernel_thread, SIGUSR1 ); + } + else { + defer_ctxSwitch(); + } } @@ -142,2 +273,22 @@ //TODO : implement waking threads } + +static void __kernel_sigaction( int sig, void (*handler)(__CFA_SIGPARMS__), int flags ) { + struct sigaction act; + + act.sa_sigaction = (void (*)(int, siginfo_t *, void *))handler; + sigemptyset( &act.sa_mask ); + sigaddset( &act.sa_mask, SIGALRM ); // disabled during signal handler + sigaddset( &act.sa_mask, SIGUSR1 ); + + act.sa_flags = flags; + + if ( sigaction( sig, &act, NULL ) == -1 ) { + // THE KERNEL IS NOT STARTED SO CALL NO uC++ ROUTINES! + char helpText[256]; + __attribute__((unused)) int len = snprintf( helpText, 256, " __kernel_sigaction( sig:%d, handler:%p, flags:%d ), problem installing signal handler, error(%d) %s.\n", + sig, handler, flags, errno, strerror( errno ) ); + LIB_DEBUG_WRITE( STDERR_FILENO, helpText, len ); + _exit( EXIT_FAILURE ); + } // if +} Index: src/libcfa/concurrency/thread.c =================================================================== --- src/libcfa/concurrency/thread.c (revision c5ac6d50679bfa2c0447315ee7255a5bdfb68af2) +++ src/libcfa/concurrency/thread.c (revision 82ff5845369a0f55575cc8ec50be83e3b1c986c2) @@ -84,5 +84,5 @@ void yield( void ) { - ScheduleInternal( this_processor->current_thread ); + BlockInternal( this_processor->current_thread ); }