Changeset 10e90cb for src/libcfa/concurrency

Timestamp:

Jun 6, 2017, 2:50:57 PM (8 years ago)

Author:

Rob Schluntz <rschlunt@…>

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:

c6d2e93

Parents:

8ca3a72 (diff), c5ac6d5 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:/u/cforall/software/cfa/cfa-cc

Location:

src/libcfa/concurrency

Files:

• alarm.c (modified) (1 diff)
• alarm.h (added)
• kernel (modified) (3 diffs)
• kernel.c (modified) (15 diffs)
• kernel_private.h (modified) (3 diffs)
• monitor (modified) (1 diff)
• monitor.c (modified) (3 diffs)
• preemption.c (added)
• preemption.h (added)
• signal.c (deleted)

src/libcfa/concurrency/alarm.c

@@ +1 @@
+//                              -*- 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.
+//
+// alarm.c --
+//
+// Author           : Thierry Delisle
+// Created On       : Fri Jun 2 11:31:25 2017
+// Last Modified By : Thierry Delisle
+// Last Modified On : --
+// Update Count     : 0
+//
+
+extern "C" {
+#include <time.h>
+#include <sys/time.h>
+}
+
+#include "alarm.h"
+#include "kernel_private.h"
+#include "preemption.h"
+
+//=============================================================================================
+// Clock logic
+//=============================================================================================
+
+__cfa_time_t __kernel_get_time() {
+        timespec curr;
+        clock_gettime( CLOCK_REALTIME, &curr );
+        return ((__cfa_time_t)curr.tv_sec * TIMEGRAN) + curr.tv_nsec;
+}
+
+void __kernel_set_timer( __cfa_time_t alarm ) {
+        itimerval val;
+        val.it_value.tv_sec = alarm / TIMEGRAN;                 // seconds
+        val.it_value.tv_usec = (alarm % TIMEGRAN) / ( TIMEGRAN / 1_000_000L ); // microseconds
+        val.it_interval.tv_sec = 0;
+        val.it_interval.tv_usec = 0;
+        setitimer( ITIMER_REAL, &val, NULL );
+}
+
+//=============================================================================================
+// Alarm logic
+//=============================================================================================
+
+void ?{}( alarm_node_t * this, thread_desc * thrd, __cfa_time_t alarm = 0, __cfa_time_t period = 0 ) {
+        this->thrd = thrd;
+        this->alarm = alarm;
+        this->period = period;
+        this->next = 0;
+        this->set = false;
+        this->kernel_alarm = false;
+}
+
+void ?{}( alarm_node_t * this, processor   * proc, __cfa_time_t alarm = 0, __cfa_time_t period = 0 ) {
+        this->proc = proc;
+        this->alarm = alarm;
+        this->period = period;
+        this->next = 0;
+        this->set = false;
+        this->kernel_alarm = true;
+}
+
+void ^?{}( alarm_node_t * this ) {
+        if( this->set ) {
+                unregister_self( this );
+        }
+}
+
+static inline void insert_at( alarm_list_t * this, alarm_node_t * n, __alarm_it_t p ) {
+        assert( !n->next );
+        if( p == this->tail ) {
+                this->tail = &n->next;
+        }
+        else {
+                n->next = *p;
+        }
+        *p = n;
+}
+
+void insert( alarm_list_t * this, alarm_node_t * n ) {
+        alarm_node_t ** it = &this->head;
+        while( (*it) && (n->alarm > (*it)->alarm) ) {
+                it = &(*it)->next;
+        }
+
+        insert_at( this, n, it );
+}
+
+alarm_node_t * pop( alarm_list_t * this ) {
+        alarm_node_t * head = this->head;
+        if( head ) {
+                this->head = head->next;
+                if( !head->next ) {
+                        this->tail = &this->head;
+                }
+                head->next = NULL;
+        }
+        return head;
+}
+
+static inline void remove_at( alarm_list_t * this, alarm_node_t * n, __alarm_it_t it ) {
+        assert( it );
+        assert( (*it)->next == n );
+
+        (*it)->next = n->next;
+        if( !n-> next ) {
+                this->tail = it;
+        }
+        n->next = NULL;
+}
+
+static inline void remove( alarm_list_t * this, alarm_node_t * n ) {
+        alarm_node_t ** it = &this->head;
+        while( (*it) && (*it)->next != n ) {
+                it = &(*it)->next;
+        }
+
+        if( *it ) { remove_at( this, n, it ); }
+}
+
+void register_self( alarm_node_t * this ) {
+        disable_interrupts();
+        assert( !systemProcessor->pending_alarm );
+        lock( &systemProcessor->alarm_lock );
+        {
+                insert( &systemProcessor->alarms, this );
+                if( systemProcessor->pending_alarm ) {
+                        tick_preemption();
+                }
+        }
+        unlock( &systemProcessor->alarm_lock );
+        this->set = true;
+        enable_interrupts();
+}
+
+void unregister_self( alarm_node_t * this ) {
+        disable_interrupts();
+        lock( &systemProcessor->alarm_lock );
+        remove( &systemProcessor->alarms, this );
+        unlock( &systemProcessor->alarm_lock );
+        disable_interrupts();
+        this->set = false;
+}

src/libcfa/concurrency/kernel

@@ -28 +28 @@
 //-----------------------------------------------------------------------------
 // Locks
+bool try_lock( spinlock * );
 void lock( spinlock * );
 void unlock( spinlock * );
@@ -85 +86 @@
 
         struct FinishAction finish;
+
+        struct alarm_node_t * preemption_alarm;
+        unsigned int preemption;
+
+        unsigned short disable_preempt_count;
+
+        bool pending_preemption;
 };
 
@@ -94 +102 @@
 
 // Local Variables: //
-// mode: c //
-// tab-width: 4 //
+// mode: CFA //
+// tab-width: 6 //
 // End: //

src/libcfa/concurrency/kernel.c

@@ -36 +36 @@
 //CFA Includes
 #include "libhdr.h"
+#include "preemption.h"
 
 //Private includes
@@ -47 +48 @@
 KERNEL_STORAGE(processorCtx_t, systemProcessorCtx);
 KERNEL_STORAGE(cluster, systemCluster);
-KERNEL_STORAGE(processor, systemProcessor);
+KERNEL_STORAGE(system_proc_t, systemProcessor);
 KERNEL_STORAGE(thread_desc, mainThread);
 KERNEL_STORAGE(machine_context_t, mainThread_context);
 
 cluster * systemCluster;
-processor * systemProcessor;
+system_proc_t * systemProcessor;
 thread_desc * mainThread;
 
@@ -118 +119 @@
 // Processor coroutine
 void ?{}(processorCtx_t * this, processor * proc) {
-        (&this->__cor){};
+        (&this->__cor){ "Processor" };
         this->proc = proc;
         proc->runner = this;
@@ -139 +140 @@
         (&this->terminated){};
         this->is_terminated = false;
+        this->preemption_alarm = NULL;
+        this->preemption = default_preemption();
+        this->disable_preempt_count = 1;                //Start with interrupts disabled
+        this->pending_preemption = false;
 
         start( this );
@@ -149 +154 @@
         (&this->terminated){};
         this->is_terminated = false;
+        this->disable_preempt_count = 0;
+        this->pending_preemption = false;
 
         this->runner = runner;
         LIB_DEBUG_PRINT_SAFE("Kernel : constructing processor context %p\n", runner);
         runner{ this };
+}
+
+void ?{}(system_proc_t * this, cluster * cltr, processorCtx_t * runner) {
+        (&this->alarms){};
+        (&this->alarm_lock){};
+        this->pending_alarm = false;
+
+        (&this->proc){ cltr, runner };
 }
 
@@ -178 +193 @@
 void main(processorCtx_t * runner) {
         processor * this = runner->proc;
+
         LIB_DEBUG_PRINT_SAFE("Kernel : core %p starting\n", this);
 
-        thread_desc * readyThread = NULL;
-        for( unsigned int spin_count = 0; ! this->is_terminated; spin_count++ ) 
         {
-                readyThread = nextThread( this->cltr );
-
-                if(readyThread) 
+                // Setup preemption data
+                preemption_scope scope = { this };
+
+                LIB_DEBUG_PRINT_SAFE("Kernel : core %p started\n", this);
+
+                thread_desc * readyThread = NULL;
+                for( unsigned int spin_count = 0; ! this->is_terminated; spin_count++ ) 
                 {
-                        runThread(this, readyThread);
-
-                        //Some actions need to be taken from the kernel
-                        finishRunning(this);
-
-                        spin_count = 0;
-                } 
-                else 
-                {
-                        spin(this, &spin_count);
-                }               
-        }
-
-        LIB_DEBUG_PRINT_SAFE("Kernel : core %p unlocking thread\n", this);
+                        readyThread = nextThread( this->cltr );
+
+                        if(readyThread)
+                        {
+                                runThread(this, readyThread);
+
+                                //Some actions need to be taken from the kernel
+                                finishRunning(this);
+
+                                spin_count = 0;
+                        }
+                        else
+                        {
+                                spin(this, &spin_count);
+                        }
+                }
+
+                LIB_DEBUG_PRINT_SAFE("Kernel : core %p stopping\n", this);
+        }
+
         signal( &this->terminated );
         LIB_DEBUG_PRINT_SAFE("Kernel : core %p terminated\n", this);
@@ -299 +323 @@
         LIB_DEBUG_PRINT_SAFE("Kernel : Starting core %p\n", this);
         
-        // pthread_attr_t attributes;
-        // pthread_attr_init( &attributes );
-
         pthread_create( &this->kernel_thread, NULL, CtxInvokeProcessor, (void*)this );
-
-        // pthread_attr_destroy( &attributes );
 
         LIB_DEBUG_PRINT_SAFE("Kernel : core %p started\n", this);       
@@ -316 +335 @@
         assertf( thrd->next == NULL, "Expected null got %p", thrd->next );
         
-        lock( &systemProcessor->cltr->lock );
-        append( &systemProcessor->cltr->ready_queue, thrd );
-        unlock( &systemProcessor->cltr->lock );
+        lock( &systemProcessor->proc.cltr->lock );
+        append( &systemProcessor->proc.cltr->ready_queue, thrd );
+        unlock( &systemProcessor->proc.cltr->lock );
 }
 
@@ -367 +386 @@
 }
 
+//=============================================================================================
+// Kernel Setup logic
+//=============================================================================================
 //-----------------------------------------------------------------------------
 // Kernel boot procedures
@@ -379 +401 @@
         mainThread{ &info };
 
+        LIB_DEBUG_PRINT_SAFE("Kernel : Main thread ready\n");
+
+        // Enable preemption
+        kernel_start_preemption();
+
         // Initialize the system cluster
         systemCluster = (cluster *)&systemCluster_storage;
         systemCluster{};
 
+        LIB_DEBUG_PRINT_SAFE("Kernel : System cluster ready\n");
+
         // Initialize the system processor and the system processor ctx
         // (the coroutine that contains the processing control flow)
-        systemProcessor = (processor *)&systemProcessor_storage;
+        systemProcessor = (system_proc_t *)&systemProcessor_storage;
         systemProcessor{ systemCluster, (processorCtx_t *)&systemProcessorCtx_storage };
 
         // 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
+        // once resume is called on systemProcessor->runner the mainThread needs to be scheduled like any normal thread
         ScheduleThread(mainThread);
 
         //initialize the global state variables
-        this_processor = systemProcessor;
+        this_processor = &systemProcessor->proc;
         this_processor->current_thread = mainThread;
         this_processor->current_coroutine = &mainThread->cor;
@@ -400 +429 @@
         // context. Hence, the main thread does not begin through CtxInvokeThread, like all other threads. The trick here is that
         // mainThread is on the ready queue when this call is made. 
-        resume(systemProcessor->runner);
+        resume( systemProcessor->proc.runner );
 
 
@@ -414 +443 @@
         // When its coroutine terminates, it return control to the mainThread
         // which is currently here
-        systemProcessor->is_terminated = true;
+        systemProcessor->proc.is_terminated = true;
         suspend();
 
@@ -421 +450 @@
         // Destroy the system processor and its context in reverse order of construction
         // These were manually constructed so we need manually destroy them
-        ^(systemProcessor->runner){};
+        ^(systemProcessor->proc.runner){};
         ^(systemProcessor){};
 
@@ -484 +513 @@
 }
 
+//=============================================================================================
+// Kernel Utilities
+//=============================================================================================
 //-----------------------------------------------------------------------------
 // Locks
@@ -491 +523 @@
 void ^?{}( spinlock * this ) {
 
+}
+
+bool try_lock( spinlock * this ) {
+        return this->lock == 0 && __sync_lock_test_and_set_4( &this->lock, 1 ) == 0;
 }
 

src/libcfa/concurrency/kernel_private.h

@@ -21 +21 @@
 #include "thread"
 
+#include "alarm.h"
+
 //-----------------------------------------------------------------------------
 // Scheduler
@@ -35 +37 @@
 //-----------------------------------------------------------------------------
 // Processor
-struct processorCtx_t {
+coroutine processorCtx_t {
         processor * proc;
-        coroutine_desc __cor;
 };
-
-DECL_COROUTINE(processorCtx_t);
 
 void main(processorCtx_t *);
@@ -47 +46 @@
 void finishRunning(processor * this);
 void spin(processor * this, unsigned int * spin_count);
+
+struct system_proc_t {
+        processor proc;
+
+        alarm_list_t alarms;
+        spinlock alarm_lock;
+
+        bool pending_alarm;
+};
+
+extern cluster * systemCluster;
+extern system_proc_t * systemProcessor;
+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;
+        }
+}
 
 //-----------------------------------------------------------------------------

src/libcfa/concurrency/monitor

@@ -92 +92 @@
 uintptr_t front( condition * this );
 
+struct __acceptable_t {
+        void (*func)(void);
+        unsigned short count;
+        monitor_desc * monitors[1];
+};
+
+void __accept_internal( unsigned short count, __acceptable_t * acceptables, void (*func)(void) );
+
 #endif //MONITOR_H

src/libcfa/concurrency/monitor.c

@@ -212 +212 @@
         ScheduleInternal( locks, count, threads, thread_count );
 
+
         //WE WOKE UP
 
@@ -312 +313 @@
         ScheduleInternal( locks, count, &signallee, 1 );
 
+
+
+
         LIB_DEBUG_PRINT_SAFE( "Back from signal block\n" );
 
@@ -330 +334 @@
         );
         return this->blocked.head->user_info;
+}
+
+//-----------------------------------------------------------------------------
+// Internal scheduling
+void __accept_internal( unsigned short count, __acceptable_t * acceptables, void (*func)(void) ) {
+        // thread_desc * this = this_thread();
+
+        // 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
+
+        // lock_all( this->current_monitors, locks, count );
+
+
+
+
+
+        // // // Everything is ready to go to sleep
+        // // ScheduleInternal( locks, count, threads, thread_count );
+
+
+        // //WE WOKE UP
+
+
+        // //We are back, restore the owners and recursions
+        // lock_all( locks, count );
+        // restore_recursion( this->monitors, recursions, count );
+        // unlock_all( locks, count );
 }