Changeset 82ff5845

Timestamp:

Jun 8, 2017, 1:48:12 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:

a724ac1

Parents:

8b8152e

Message:

First implementation of preemption, does not appear to work with scheduling

Location:

src/libcfa/concurrency

Files:

• alarm.c (modified) (4 diffs)
• invoke.c (modified) (2 diffs)
• kernel.c (modified) (10 diffs)
• kernel_private.h (modified) (2 diffs)
• monitor.c (modified) (7 diffs)
• preemption.c (modified) (10 diffs)
• thread.c (modified) (1 diff)

src/libcfa/concurrency/alarm.c ¶

@@ -16 +16 @@
 
 extern "C" {
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
 #include <time.h>
+#include <unistd.h>
 #include <sys/time.h>
 }
@@ -22 +26 @@
 #include "alarm.h"
 #include "kernel_private.h"
+#include "libhdr.h"
 #include "preemption.h"
 
@@ -31 +36 @@
         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 +146 @@
         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() );
                 }
         }

src/libcfa/concurrency/invoke.c ¶

@@ -30 +30 @@
 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 +69 @@
       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 );
 

src/libcfa/concurrency/kernel.c ¶

@@ -154 +154 @@
         (&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 +243 @@
         //Update global state
         this->current_thread = dst;
+
+        LIB_DEBUG_PRINT_SAFE("Kernel : running %p\n", dst);
 
         // Context Switch to the thread
@@ -322 +327 @@
 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 +379 @@
 }
 
-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 +427 @@
         this_processor->finish.thrd_count = thrd_count;
         suspend();
+        enable_interrupts();
 }
 
@@ -403 +447 @@
         LIB_DEBUG_PRINT_SAFE("Kernel : Main thread ready\n");
 
-        // Enable preemption
-        kernel_start_preemption();
-
         // Initialize the system cluster
         systemCluster = (cluster *)&systemCluster_storage;
@@ -426 +467 @@
         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 +479 @@
         // THE SYSTEM IS NOW COMPLETELY RUNNING
         LIB_DEBUG_PRINT_SAFE("Kernel : Started\n--------------------------------------------------\n\n");
+
+        enable_interrupts();
 }
 
@@ -447 +493 @@
 
         // THE SYSTEM IS NOW COMPLETELY STOPPED
+
+        // Disable preemption
+        kernel_stop_preemption();
 
         // Destroy the system processor and its context in reverse order of construction
@@ -550 +599 @@
         if( !this->cond ) {
                 append( &this->blocked, this_thread() );
-                ScheduleInternal( &this->lock );
+                BlockInternal( &this->lock );
                 lock( &this->lock );
         }

src/libcfa/concurrency/kernel_private.h ¶

@@ -28 +28 @@
 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 +60 @@
 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();
 }
 

src/libcfa/concurrency/monitor.c ¶

@@ -63 +63 @@
                         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 +172 @@
         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 +210 @@
 
         // Everything is ready to go to sleep
-        ScheduleInternal( locks, count, threads, thread_count );
+        BlockInternal( locks, count, threads, thread_count );
 
 
@@ -283 +283 @@
         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 +311 @@
 
         //Everything is ready to go to sleep
-        ScheduleInternal( locks, count, &signallee, 1 );
+        BlockInternal( locks, count, &signallee, 1 );
 
 
@@ -343 +343 @@
         // 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 +352 @@
 
         // // // Everything is ready to go to sleep
-        // // ScheduleInternal( locks, count, threads, thread_count );
+        // // BlockInternal( locks, count, threads, thread_count );
 
 

src/libcfa/concurrency/preemption.c ¶

@@ -18 +18 @@
 
 extern "C" {
+#include <errno.h>
 #include <signal.h>
-}
-
-#define __CFA_DEFAULT_PREEMPTION__ 10
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+}
+
+#include "libhdr.h"
+
+#define __CFA_DEFAULT_PREEMPTION__ 10000
 
 __attribute__((weak)) unsigned int default_preemption() {
@@ -27 +33 @@
 }
 
+#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 +49 @@
 
 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 +85 @@
 
                 if( node->period > 0 ) {
-                        node->alarm += node->period;
+                        node->alarm = currtime + node->period;
                         insert( alarms, node );
                 }
@@ -62 +96 @@
                 __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 +139 @@
         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 +152 @@
 // 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 +203 @@
 }
 
-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 +244 @@
                 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 +273 @@
         //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
+}

src/libcfa/concurrency/thread.c ¶

@@ -84 +84 @@
 
 void yield( void ) {
-        ScheduleInternal( this_processor->current_thread );
+        BlockInternal( this_processor->current_thread );
 }