Changeset 90c4df0 for src/libcfa

Timestamp:

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

Files:

• invoke.h (modified) (3 diffs)
• kernel.c (modified) (1 diff)
• monitor (modified) (2 diffs)
• monitor.c (modified) (7 diffs)
• preemption.c (modified) (2 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();