Changeset 90c4df0

Timestamp:

Aug 22, 2017, 1:44:12 PM (6 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:

80c72a7

Parents:

e50e9ff

Message:

Implemented search for external scheduling

Location:

src

Files:

• libcfa/concurrency/invoke.h (modified) (3 diffs)
• libcfa/concurrency/kernel.c (modified) (1 diff)
• libcfa/concurrency/monitor (modified) (2 diffs)
• libcfa/concurrency/monitor.c (modified) (7 diffs)
• libcfa/concurrency/preemption.c (modified) (2 diffs)
• tests/sched-ext.c (modified) (5 diffs)

src/libcfa/concurrency/invoke.h

@@ -28 +28 @@
       #define thread_local _Thread_local
 
+      typedef void (*fptr_t)();
+
       struct spinlock {
             volatile int lock;
@@ -50 +52 @@
             void append( struct __thread_queue_t *, struct thread_desc * );
             struct thread_desc * pop_head( struct __thread_queue_t * );
+            struct thread_desc * remove( struct __thread_queue_t *, struct thread_desc ** );
 
             void ?{}( struct __condition_stack_t * );
@@ -91 +94 @@
             unsigned short acceptable_count;          // number of acceptable functions
             short accepted_index;                     // the index of the accepted function, -1 if none
-            void (*pre_accept)(void);                 // function to run before an accept
+            fptr_t pre_accept;                        // function to run before an accept
        };
 
       struct thread_desc {
+            // Core threading fields
             struct coroutine_desc cor;                // coroutine body used to store context
             struct monitor_desc mon;                  // monitor body used for mutual exclusion
+
+            // Link lists fields
             struct thread_desc * next;                // instrusive link field for threads
+
+            // Current status related to monitors
             struct monitor_desc ** current_monitors;  // currently held monitors
             unsigned short current_monitor_count;     // number of currently held monitors
+            fptr_t current_monitor_func;              // last function that acquired monitors
      };
 

src/libcfa/concurrency/kernel.c

@@ -668 +668 @@
 }
 
+thread_desc * remove( __thread_queue_t * this, thread_desc ** it ) {
+        thread_desc * thrd = *it;
+        verify( thrd );
+
+        (*it) = thrd->next;
+
+        if( this->tail == &thrd->next ) {
+                this->tail = it;
+        }
+
+        thrd->next = NULL;
+
+        verify( (this->head == NULL) == (&this->head == this->tail) );
+        verify( *this->tail == NULL );
+        return thrd;
+}
+
+
+
 void ?{}( __condition_stack_t * this ) {
         this->top = NULL;

src/libcfa/concurrency/monitor

@@ -39 +39 @@
         monitor_desc ** prev_mntrs;
         unsigned short  prev_count;
+        fptr_t          prev_func;
 };
 
@@ -99 +100 @@
 // External scheduling
 
-typedef void (*void_fptr_t)(void);
-
 struct __acceptable_t {
-        void_fptr_t func;
+        fptr_t func;
         unsigned short count;
         monitor_desc ** monitors;

src/libcfa/concurrency/monitor.c

@@ -73 +73 @@
                 thread_desc * thrd = this_thread;
 
+                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entering mon %p (%p)\n", thrd, this, this->owner);
+
                 this->accepted_index = -1;
                 if( !this->owner ) {
                         // No one has the monitor, just take it
                         set_owner( this, thrd );
+
+                        LIB_DEBUG_PRINT_SAFE("Kernel :  mon is free \n");
                 }
                 else if( this->owner == thrd) {
@@ -82 +86 @@
                         verify( this->recursion > 0 );
                         this->recursion += 1;
+
+                        LIB_DEBUG_PRINT_SAFE("Kernel :  mon already owned \n");
                 }
                 else if( (this->accepted_index = is_accepted( thrd, this, group, group_cnt, func)) >= 0 ) {
                         // Some one was waiting for us, enter
                         set_owner( this, thrd );
+
+                        LIB_DEBUG_PRINT_SAFE("Kernel :  mon accepts \n");
                 }
                 else {
+                        LIB_DEBUG_PRINT_SAFE("Kernel :  blocking \n");
+
                         // Some one else has the monitor, wait in line for it
                         append( &this->entry_queue, thrd );
                         BlockInternal( &this->lock );
 
+                        LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered  mon %p\n", thrd, this);
+
                         // BlockInternal will unlock spinlock, no need to unlock ourselves
                         return;
                 }
+
+                LIB_DEBUG_PRINT_SAFE("Kernel : %10p Entered  mon %p\n", thrd, this);
 
                 // Release the lock and leave
@@ -194 +208 @@
         qsort(this->m, count);
 
-        // Enter the monitors in order
-        enter( this->m, this->count, func );
-
         // Save previous thread context
         this->prev_mntrs = this_thread->current_monitors;
         this->prev_count = this_thread->current_monitor_count;
+        this->prev_func  = this_thread->current_monitor_func;
 
         // Update thread context (needed for conditions)
         this_thread->current_monitors      = m;
         this_thread->current_monitor_count = count;
+        this_thread->current_monitor_func  = func;
+
+        // Enter the monitors in order
+        enter( this->m, this->count, func );
 }
 
@@ -214 +230 @@
         this_thread->current_monitors      = this->prev_mntrs;
         this_thread->current_monitor_count = this->prev_count;
+        this_thread->current_monitor_func  = this->prev_func;
 }
 
@@ -402 +419 @@
         thread_desc * next = search_entry_queue( acceptables, acc_count, monitors, count );
 
+        LIB_DEBUG_PRINT_SAFE("Owner(s) :");
+        for(int i = 0; i < count; i++) {
+                LIB_DEBUG_PRINT_SAFE(" %p", monitors[i]->owner );
+        }
+        LIB_DEBUG_PRINT_SAFE("\n");
+
+        LIB_DEBUG_PRINT_SAFE("Passing mon to %p\n", next);
+
         if( !next ) {
                 // Update acceptables on the current monitors
@@ -409 +434 @@
                 }
         }
+        else {
+                for(int i = 0; i < count; i++) {
+                        set_owner( monitors[i], next );
+                }
+        }
+
 
         save_recursion( monitors, recursions, count );
+
 
         // Everything is ready to go to sleep
@@ -602 +634 @@
 }
 
+static inline bool match( __acceptable_t * acc, thread_desc * thrd ) {
+        verify( thrd );
+        verify( acc );
+        if( acc->func != thrd->current_monitor_func ) return false;
+
+        return true;
+}
+
 static inline thread_desc * search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count ) {
+
+        __thread_queue_t * entry_queue = &monitors[0]->entry_queue;
+
+        // For each thread in the entry-queue
+        for(    thread_desc ** thrd_it = &entry_queue->head;
+                *thrd_it;
+                thrd_it = &(*thrd_it)->next)
+        {
+                // For each acceptable check if it matches
+                __acceptable_t * acc_end = acceptables + acc_count;
+                for( __acceptable_t * acc_it = acceptables; acc_it != acc_end; acc_it++ ) {
+                        // Check if we have a match
+                        if( match( acc_it, *thrd_it ) ) {
+
+                                // If we have a match return it
+                                // after removeing it from the entry queue
+                                return remove( entry_queue, thrd_it );
+                        }
+                }
+        }
+
         return NULL;
 }

src/libcfa/concurrency/preemption.c

@@ -332 +332 @@
                 assertf(sig == SIGALRM, "Kernel Internal Error, sigwait: Unexpected signal %d (%d : %d)\n", sig, info.si_code, info.si_value.sival_int);
 
-                LIB_DEBUG_PRINT_SAFE("Kernel : Caught alarm from %d with %d\n", info.si_code, info.si_value.sival_int );
+                // LIB_DEBUG_PRINT_SAFE("Kernel : Caught alarm from %d with %d\n", info.si_code, info.si_value.sival_int );
                 // Switch on the code (a.k.a. the sender) to
                 switch( info.si_code )
@@ -340 +340 @@
                 case SI_TIMER:
                 case SI_KERNEL:
-                        LIB_DEBUG_PRINT_SAFE("Kernel : Preemption thread tick\n");
+                        // LIB_DEBUG_PRINT_SAFE("Kernel : Preemption thread tick\n");
                         lock( &event_kernel->lock DEBUG_CTX2 );
                         tick_preemption();

src/tests/sched-ext.c

@@ -7 +7 @@
 #include <time.h>
 
-static const unsigned long N = 2_500ul;
+static const unsigned long N = 500ul;
 
 #ifndef PREEMPTION_RATE
@@ -24 +24 @@
 thread Acceptee {};
 
+volatile bool done;
+
+unsigned rand10() {
+        return (unsigned)rand48() % 10;
+}
+
 //----------------------------------------------------------------------------------------------------
 // Acceptor
@@ -29 +35 @@
 
 void do_wait( global_t * mutex a ) {
-        sout | "Preparing to wait" | endl;
+        sout | "Waiting to accept" | endl;
+        yield( rand10() );
+
+        sout | "Accepting" | endl;
 
         __acceptable_t acceptable;
-        acceptable.func          = (void_fptr_t)do_notify;
+        acceptable.func          = (fptr_t)do_notify;
         acceptable.count         = 1;
         acceptable.monitors      = &a;
         acceptable.run_preaccept = false;
 
-        sout | "Waiting for notify" | endl;
+        __accept_internal( 1, &acceptable );
 
-        int ret = __accept_internal( 1, &acceptable );
-        sout | "Back from wating, accepted" | ret | endl;
+        sout | "Accepted" | endl;
+        yield( rand10() );
 }
 
 void main( Acceptor* this ) {
-        do_wait( &globalA );
+        for( int i = 0; i < N; i++ ) {
+                do_wait( &globalA );
+                sout | i | endl;
+        }
+
+        done = true;
 }
 
@@ -50 +64 @@
 // Acceptee
 void do_notify( global_t * mutex a ) {
-        sout | "Notifying" | endl;
+
 }
 
 void main( Acceptee* this ) {
-        for( volatile int i = 0; i < N; i++ );
-
-        sout | "Call Notify" | endl;
-        do_notify( &globalA );
+        while( !done ) {
+                yield( rand10() );
+                do_notify( &globalA );
+                yield( rand10() );
+        }
 }
 
@@ -63 +78 @@
 // Main
 int main(int argc, char* argv[]) {
-        processor p;
+        done = false;
+        rand48seed( time( NULL ) );
+        printf("%p\n", &globalA);
         sout | "Starting" | endl;
         {
                 Acceptor r;
-                Acceptee e;
+                Acceptee e[13];
+
         }
         sout | "Done" | endl;