Changeset cb0e6de for src/libcfa/concurrency

Timestamp:

Mar 21, 2017, 2:29:25 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:

9c31349

Parents:

e04b636

Message:

Threads now use monitor semantics to wait until completion

Location:

src/libcfa/concurrency

Files:

• invoke.c (modified) (3 diffs)
• invoke.h (modified) (1 diff)
• monitor (modified) (1 diff)
• monitor.c (modified) (3 diffs)
• thread (modified) (3 diffs)
• thread.c (modified) (2 diffs)

src/libcfa/concurrency/invoke.c

@@ -29 +29 @@
 
 extern void __suspend_internal(void);
-extern void __thread_signal_termination(struct thread_desc*);
+extern void __leave_monitor_desc( struct monitor_desc * this );
 
 void CtxInvokeCoroutine(
@@ -65 +65 @@
       struct thread_desc* thrd = get_thread( this );
       struct coroutine_desc* cor = &thrd->cor;
+      struct monitor_desc* mon = &thrd->mon;
       cor->state = Active;
 
@@ -70 +71 @@
       main( this );
 
-      __thread_signal_termination(thrd);
+      __leave_monitor_desc( mon );
 
       //Final suspend, should never return

src/libcfa/concurrency/invoke.h

@@ -79 +79 @@
       };
 
+      struct monitor_desc {
+            struct spinlock lock;
+            struct thread_desc * owner;
+            struct simple_thread_list entry_queue;
+            unsigned int recursion;
+      };
+
       struct thread_desc {
-            struct coroutine_desc cor;            // coroutine body used to store context
+            struct coroutine_desc cor;          // coroutine body used to store context
+            struct monitor_desc mon;
             struct signal_once terminated;      // indicate if execuation state is not halted
             struct thread_desc * next;          // instrusive link field for threads

src/libcfa/concurrency/monitor

@@ -21 +21 @@
 #include "invoke.h"
 #include "stdlib"
-
-struct monitor_desc {
-        spinlock lock;
-        thread_desc * owner;
-        simple_thread_list entry_queue;
-        unsigned int recursion;
-};
 
 static inline void ?{}(monitor_desc * this) {

src/libcfa/concurrency/monitor.c

@@ -19 +19 @@
 #include "kernel_private.h"
 
-void enter(monitor_desc * this) {
-        lock( &this->lock );
-        thread_desc * thrd = this_thread();
+extern "C" {
+        void __enter_monitor_desc(monitor_desc * this) {
+                lock( &this->lock );
+                thread_desc * thrd = this_thread();
 
-        if( !this->owner ) {
-                //No one has the monitor, just take it
-                this->owner = thrd;
-                this->recursion = 1;
-        }
-        else if( this->owner == thrd) {
-                //We already have the monitor, just not how many times we took it
-                assert( this->recursion > 0 );
-                this->recursion += 1;
-        }
-        else {
-                //Some one else has the monitor, wait in line for it
-                append( &this->entry_queue, thrd );
-                ScheduleInternal( &this->lock );
+                if( !this->owner ) {
+                        //No one has the monitor, just take it
+                        this->owner = thrd;
+                        this->recursion = 1;
+                }
+                else if( this->owner == thrd) {
+                        //We already have the monitor, just not how many times we took it
+                        assert( this->recursion > 0 );
+                        this->recursion += 1;
+                }
+                else {
+                        //Some one else has the monitor, wait in line for it
+                        append( &this->entry_queue, thrd );
+                        ScheduleInternal( &this->lock );
 
-                //ScheduleInternal will unlock spinlock, no need to unlock ourselves
-                return; 
+                        //ScheduleInternal will unlock spinlock, no need to unlock ourselves
+                        return; 
+                }
+
+                unlock( &this->lock );
         }
 
-        unlock( &this->lock );
-}
+        void __leave_monitor_desc(monitor_desc * this) {
+                lock( &this->lock );
 
-void leave(monitor_desc * this) {
-        lock( &this->lock );
+                thread_desc * thrd = this_thread();
+                assert( thrd == this->owner );
 
-        thread_desc * thrd = this_thread();
-        assert( thrd == this->owner );
+                //Leaving a recursion level, decrement the counter
+                this->recursion -= 1;
 
-        //Leaving a recursion level, decrement the counter
-        this->recursion -= 1;
+                //If we left the last level of recursion it means we are changing who owns the monitor
+                thread_desc * new_owner = 0;
+                if( this->recursion == 0) {
+                        //Get the next thread in the list
+                        new_owner = this->owner = pop_head( &this->entry_queue );
 
-        //If we left the last level of recursion it means we are changing who owns the monitor
-        thread_desc * new_owner = 0;
-        if( this->recursion == 0) {
-                //Get the next thread in the list
-                new_owner = this->owner = pop_head( &this->entry_queue );
+                        //We are passing the monitor to someone else, which means recursion level is not 0
+                        this->recursion = new_owner ? 1 : 0;
+                }       
 
-                //We are passing the monitor to someone else, which means recursion level is not 0
-                this->recursion = new_owner ? 1 : 0;
-        }       
+                unlock( &this->lock );
 
-        unlock( &this->lock );
-
-        //If we have a new owner, we need to wake-up the thread
-        if( new_owner ) {
-                ScheduleThread( new_owner );
+                //If we have a new owner, we need to wake-up the thread
+                if( new_owner ) {
+                        ScheduleThread( new_owner );
+                }
         }
 }
@@ -74 +76 @@
 void enter(monitor_desc ** monitors, int count) {
         for(int i = 0; i < count; i++) {
-                enter( monitors[i] );
+                __enter_monitor_desc( monitors[i] );
         }
 }
@@ -80 +82 @@
 void leave(monitor_desc ** monitors, int count) {
         for(int i = count - 1; i >= 0; i--) {
-                leave( monitors[i] );
+                __leave_monitor_desc( monitors[i] );
         }
 }

src/libcfa/concurrency/thread

@@ -22 +22 @@
 
 #include "coroutine"
+#include "monitor"
 
 //-----------------------------------------------------------------------------
@@ -28 +29 @@
 // Anything that is resumed is a coroutine.
 trait is_thread(dtype T) {
-      void ^?{}(T* this);
+      void ^?{}(T* mutex this);
       void main(T* this);
       thread_desc* get_thread(T* this);
@@ -40 +41 @@
 }
 
-static inline coroutine_desc* get_coroutine(thread_desc* this) {
+forall( dtype T | is_thread(T) )
+static inline monitor_desc* get_monitor(T * this) {
+        return &get_thread(this)->mon;
+}
+
+static inline coroutine_desc* get_coroutine(thread_desc * this) {
         return &this->cor;
+}
+
+static inline monitor_desc* get_monitor(thread_desc * this) {
+        return &this->mon;
 }
 

src/libcfa/concurrency/thread.c

@@ -44 +44 @@
         (&this->cor){};
         this->cor.name = "Anonymous Coroutine";
+        this->mon.owner = this;
+        this->mon.recursion = 1;
         (&this->terminated){};
         this->next = NULL;
@@ -90 +92 @@
 forall( dtype T | is_thread(T) )
 void stop( T* this ) {
-        wait( & get_thread(this)->terminated ); 
+        // wait( & get_thread(this)->terminated );      
 }