Changeset f7e4f8e8 for libcfa/src/concurrency

Timestamp:

Oct 30, 2020, 12:36:16 PM (5 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

0ab3b73, 36d0a80

Parents:

b9537e6 (diff), 4ae78c1 (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:software/cfa/cfa-cc

Location:

libcfa/src/concurrency

Files:

• coroutine.cfa (modified) (6 diffs)
• exception.cfa (modified) (3 diffs)
• exception.hfa (modified) (2 diffs)
• invoke.h (modified) (1 diff)
• io/setup.cfa (modified) (1 diff)
• kernel.cfa (modified) (6 diffs)
• kernel.hfa (modified) (1 diff)
• kernel/startup.cfa (modified) (2 diffs)
• kernel_private.hfa (modified) (2 diffs)
• monitor.cfa (modified) (1 diff)
• thread.cfa (modified) (4 diffs)
• thread.hfa (modified) (3 diffs)

libcfa/src/concurrency/coroutine.cfa

@@ -10 +10 @@
 // Created On       : Mon Nov 28 12:27:26 2016
 // Last Modified By : Peter A. Buhr
-// Last Modified On : Tue May 26 22:06:09 2020
-// Update Count     : 21
+// Last Modified On : Fri Oct 23 23:05:24 2020
+// Update Count     : 22
 //
 
@@ -24 +24 @@
 #include <unistd.h>
 #include <sys/mman.h>                                                                   // mprotect
-extern "C" {
-// use this define to make unwind.h play nice, definitely a hack
-#define HIDE_EXPORTS
 #include <unwind.h>
-#undef HIDE_EXPORTS
-}
 
 #include "kernel_private.hfa"
+#include "exception.hfa"
 
 #define __CFA_INVOKE_PRIVATE__
@@ -49 +45 @@
 FORALL_DATA_INSTANCE(CoroutineCancelled, (dtype coroutine_t), (coroutine_t))
 
-struct __cfaehm_node {
-        struct _Unwind_Exception unwind_exception;
-        struct __cfaehm_node * next;
-        int handler_index;
-};
-
 forall(dtype T)
 void mark_exception(CoroutineCancelled(T) *) {}
@@ -60 +50 @@
 forall(dtype T)
 void copy(CoroutineCancelled(T) * dst, CoroutineCancelled(T) * src) {
+        dst->virtual_table = src->virtual_table;
         dst->the_coroutine = src->the_coroutine;
         dst->the_exception = src->the_exception;
@@ -74 +65 @@
         verify( desc->cancellation );
         desc->state = Cancelled;
-        exception_t * except = (exception_t *)(1 + (__cfaehm_node *)desc->cancellation);
+        exception_t * except = __cfaehm_cancellation_exception( desc->cancellation );
 
         // TODO: Remove explitate vtable set once trac#186 is fixed.
@@ -217 +208 @@
                 size = libFloor(create_size - stack_data_size - diff, libAlign());
         } // if
-        assertf( size >= MinStackSize, "Stack size %zd provides less than minimum of %d bytes for a stack.", size, MinStackSize );
+        assertf( size >= MinStackSize, "Stack size %zd provides less than minimum of %zd bytes for a stack.", size, MinStackSize );
 
         this->storage = (__stack_t *)((intptr_t)storage + size);

libcfa/src/concurrency/exception.cfa

@@ -10 +10 @@
 // Created On       : Mon Aug 17 10:41:00 2020
 // Last Modified By : Andrew Beach
-// Last Modified On : Tue Aug 25 14:41:00 2020
-// Update Count     : 0
+// Last Modified On : Wed Oct 28 14:34:00 2020
+// Update Count     : 1
 //
 
-extern "C" {
-// use this define to make unwind.h play nice, definitely a hack
-#define HIDE_EXPORTS
-#include <unwind.h>
-#undef HIDE_EXPORTS
-}
+#define __cforall_thread__
 
-#include "invoke.h"
 #include "exception.hfa"
+
 #include "coroutine.hfa"
 
 extern struct $thread * mainThread;
+extern "C" {
+extern void __cfactx_thrd_leave();
+}
 
 // Common pattern for all the stop functions, wait until the end then act.
@@ -52 +50 @@
 
 STOP_AT_END_FUNCTION(thread_cancelstop,
-        // TODO: Instead pass information to the joiner.
-        abort();
+        __cfactx_thrd_leave();
+        __cabi_abort( "Resumed cancelled thread" );
 )
 
@@ -85 +83 @@
                 stop_param = (void *)0x22;
         } else {
+                this_thread->self_cor.cancellation = unwind_exception;
+
                 stop_func = thread_cancelstop;
                 stop_param = this_thread;

libcfa/src/concurrency/exception.hfa

@@ -16 +16 @@
 #pragma once
 
+// This is an internal bridge between the two modes and must be C compatable.
+
+#include <unwind.h>
 #include "bits/defs.hfa"
 #include "invoke.h"
+#include "exception.h"
 
 #ifdef __cforall
 extern "C" {
-
-#define HIDE_EXPORTS
 #endif
-#include "unwind.h"
 
 struct exception_context_t * this_exception_context(void) OPTIONAL_THREAD;
@@ -32 +33 @@
 
 #ifdef __cforall
-#undef HIDE_EXPORTS
 }
 #endif

libcfa/src/concurrency/invoke.h

@@ -157 +157 @@
 
                 // current execution status for coroutine
+                // Possible values are:
+                //    - TICKET_BLOCKED (-1) thread is blocked
+                //    - TICKET_RUNNING ( 0) thread is running
+                //    - TICKET_UNBLOCK ( 1) thread should ignore next block
                 volatile int ticket;
                 enum __Coroutine_State state:8;

libcfa/src/concurrency/io/setup.cfa

@@ -250 +250 @@
                                         // Fixup the thread state
                                         thrd.state = Blocked;
-                                        thrd.ticket = 0;
+                                        thrd.ticket = TICKET_BLOCKED;
                                         thrd.preempted = __NO_PREEMPTION;
 

libcfa/src/concurrency/kernel.cfa

@@ -252 +252 @@
                 /* paranoid */ verify( kernelTLS.this_thread == thrd_dst );
                 /* paranoid */ verify( thrd_dst->context.SP );
+                /* paranoid */ verify( thrd_dst->state != Halted );
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) < ((uintptr_t)__get_stack(thrd_dst->curr_cor)->base ) || thrd_dst->curr_cor == proc_cor, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too small.\n", thrd_dst ); // add escape condition if we are setting up the processor
                 /* paranoid */ verifyf( ((uintptr_t)thrd_dst->context.SP) > ((uintptr_t)__get_stack(thrd_dst->curr_cor)->limit) || thrd_dst->curr_cor == proc_cor, "ERROR : Destination $thread %p has been corrupted.\n StackPointer too large.\n", thrd_dst ); // add escape condition if we are setting up the processor
@@ -287 +288 @@
                 if(unlikely(thrd_dst->state == Halted)) {
                         // The thread has halted, it should never be scheduled/run again
-                        // We may need to wake someone up here since
-                        unpark( this->destroyer );
-                        this->destroyer = 0p;
+                        // finish the thread
+                        __thread_finish( thrd_dst );
                         break RUNNING;
                 }
@@ -299 +299 @@
                 int old_ticket = __atomic_fetch_sub(&thrd_dst->ticket, 1, __ATOMIC_SEQ_CST);
                 switch(old_ticket) {
-                        case 1:
+                        case TICKET_RUNNING:
                                 // This is case 1, the regular case, nothing more is needed
                                 break RUNNING;
-                        case 2:
+                        case TICKET_UNBLOCK:
                                 // This is case 2, the racy case, someone tried to run this thread before it finished blocking
                                 // In this case, just run it again.
@@ -410 +410 @@
         int old_ticket = __atomic_fetch_add(&thrd->ticket, 1, __ATOMIC_SEQ_CST);
         switch(old_ticket) {
-                case 1:
+                case TICKET_RUNNING:
                         // Wake won the race, the thread will reschedule/rerun itself
                         break;
-                case 0:
+                case TICKET_BLOCKED:
                         /* paranoid */ verify( ! thrd->preempted != __NO_PREEMPTION );
                         /* paranoid */ verify( thrd->state == Blocked );
@@ -422 +422 @@
                 default:
                         // This makes no sense, something is wrong abort
-                        abort();
+                        abort("Thread %p (%s) has mismatch park/unpark\n", thrd, thrd->self_cor.name);
         }
 }
@@ -448 +448 @@
 }
 
-// KERNEL ONLY
-void __leave_thread() {
-        /* paranoid */ verify( ! kernelTLS.preemption_state.enabled );
-        returnToKernel();
-        abort();
+extern "C" {
+        // Leave the thread monitor
+        // last routine called by a thread.
+        // Should never return
+        void __cfactx_thrd_leave() {
+                $thread * thrd = TL_GET( this_thread );
+                $monitor * this = &thrd->self_mon;
+
+                // Lock the monitor now
+                lock( this->lock __cfaabi_dbg_ctx2 );
+
+                disable_interrupts();
+
+                thrd->state = Halted;
+
+                if( thrd != this->owner || this->recursion != 1) { abort( "Thread internal monitor has unbalanced recursion" ); }
+
+                // Leave the thread
+                /* paranoid */ verify( ! kernelTLS.preemption_state.enabled );
+                returnToKernel();
+                abort();
+
+                // Control flow should never reach here!
+        }
 }
 

libcfa/src/concurrency/kernel.hfa

@@ -79 +79 @@
         // Handle to pthreads
         pthread_t kernel_thread;
-
-        // RunThread data
-        // Action to do after a thread is ran
-        $thread * destroyer;
 
         // Preemption data

libcfa/src/concurrency/kernel/startup.cfa

@@ -441 +441 @@
 
 static void ?{}( $thread & this, current_stack_info_t * info) with( this ) {
-        ticket = 1;
+        ticket = TICKET_RUNNING;
         state = Start;
         self_cor{ info };
@@ -474 +474 @@
         this.cltr = &_cltr;
         full_proc = true;
-        destroyer = 0p;
         do_terminate = false;
         preemption_alarm = 0p;

libcfa/src/concurrency/kernel_private.hfa

@@ -39 +39 @@
 ;
 
-//Block current thread and release/wake-up the following resources
-void __leave_thread() __attribute__((noreturn));
+//release/wake-up the following resources
+void __thread_finish( $thread * thrd );
 
 //-----------------------------------------------------------------------------
@@ -65 +65 @@
 // KERNEL ONLY unpark with out disabling interrupts
 void __unpark( struct __processor_id_t *, $thread * thrd );
+
+#define TICKET_BLOCKED (-1) // thread is blocked
+#define TICKET_RUNNING ( 0) // thread is running
+#define TICKET_UNBLOCK ( 1) // thread should ignore next block
 
 static inline bool __post(single_sem & this, struct __processor_id_t * id) {

libcfa/src/concurrency/monitor.cfa

@@ -281 +281 @@
 }
 
-extern "C" {
-        // Leave the thread monitor
-        // last routine called by a thread.
-        // Should never return
-        void __cfactx_thrd_leave() {
-                $thread * thrd = TL_GET( this_thread );
-                $monitor * this = &thrd->self_mon;
-
-                // Lock the monitor now
-                lock( this->lock __cfaabi_dbg_ctx2 );
-
-                disable_interrupts();
-
-                thrd->state = Halted;
-
-                /* paranoid */ verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", thrd, this->owner, this->recursion, this );
-
-                // Leaving a recursion level, decrement the counter
-                this->recursion -= 1;
-
-                // If we haven't left the last level of recursion
-                // it must mean there is an error
-                if( this->recursion != 0) { abort( "Thread internal monitor has unbalanced recursion" ); }
-
-                // Fetch the next thread, can be null
-                $thread * new_owner = next_thread( this );
-
-                // Release the monitor lock
-                unlock( this->lock );
-
-                // Unpark the next owner if needed
-                /* paranoid */ verifyf( !new_owner || new_owner == this->owner, "Expected owner to be %p, got %p (m: %p)", new_owner, this->owner, this );
-                /* paranoid */ verify( ! kernelTLS.preemption_state.enabled );
-                /* paranoid */ verify( ! kernelTLS.this_processor->destroyer );
-                /* paranoid */ verify( thrd->state == Halted );
-
-                kernelTLS.this_processor->destroyer = new_owner;
-
-                // Leave the thread
-                __leave_thread();
-
-                // Control flow should never reach here!
-        }
-}
-
-// Join a thread
-forall( dtype T | is_thread(T) )
-T & join( T & this ) {
-        $monitor *    m = get_monitor(this);
-        void (*dtor)(T& mutex this) = ^?{};
-        monitor_dtor_guard_t __guard = { &m, (fptr_t)dtor, true };
-        {
-                return this;
-        }
+void __thread_finish( $thread * thrd ) {
+        $monitor * this = &thrd->self_mon;
+
+        // Lock the monitor now
+        /* paranoid */ verify( this->lock.lock );
+        /* paranoid */ verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i, m: %p)", thrd, this->owner, this->recursion, this );
+        /* paranoid */ verify( ! kernelTLS.preemption_state.enabled );
+        /* paranoid */ verify( thrd->state == Halted );
+        /* paranoid */ verify( this->recursion == 1 );
+
+        // Leaving a recursion level, decrement the counter
+        this->recursion -= 1;
+        this->owner = 0p;
+
+        // Fetch the next thread, can be null
+        $thread * new_owner = next_thread( this );
+
+        // Release the monitor lock
+        unlock( this->lock );
+
+        // Unpark the next owner if needed
+        /* paranoid */ verifyf( !new_owner || new_owner == this->owner, "Expected owner to be %p, got %p (m: %p)", new_owner, this->owner, this );
+        /* paranoid */ verify( ! kernelTLS.preemption_state.enabled );
+        /* paranoid */ verify( thrd->state == Halted );
+        unpark( new_owner );
 }
 

libcfa/src/concurrency/thread.cfa

@@ -19 +19 @@
 
 #include "kernel_private.hfa"
+#include "exception.hfa"
 
 #define __CFA_INVOKE_PRIVATE__
@@ -28 +29 @@
         context{ 0p, 0p };
         self_cor{ name, storage, storageSize };
-        ticket = 1;
+        ticket = TICKET_RUNNING;
         state = Start;
         preempted = __NO_PREEMPTION;
@@ -56 +57 @@
         unregister(curr_cluster, this);
         ^self_cor{};
+}
+
+FORALL_DATA_INSTANCE(ThreadCancelled, (dtype thread_t), (thread_t))
+
+forall(dtype T)
+void copy(ThreadCancelled(T) * dst, ThreadCancelled(T) * src) {
+        dst->virtual_table = src->virtual_table;
+        dst->the_thread = src->the_thread;
+        dst->the_exception = src->the_exception;
+}
+
+forall(dtype T)
+const char * msg(ThreadCancelled(T) *) {
+        return "ThreadCancelled";
+}
+
+forall(dtype T)
+static void default_thread_cancel_handler(ThreadCancelled(T) & ) {
+        abort( "Unhandled thread cancellation.\n" );
+}
+
+forall(dtype T | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T)))
+void ?{}( thread_dtor_guard_t & this,
+                T & thrd, void(*defaultResumptionHandler)(ThreadCancelled(T) &)) {
+        $monitor * m = get_monitor(thrd);
+        void (*dtor)(T& mutex this) = ^?{};
+        bool join = defaultResumptionHandler != (void(*)(ThreadCancelled(T)&))0;
+        (this.mg){&m, (void(*)())dtor, join};
+
+        // After the guard set-up and any wait, check for cancellation.
+        $thread * desc = get_thread(thrd);
+        struct _Unwind_Exception * cancellation = desc->self_cor.cancellation;
+        if ( likely( 0p == cancellation ) ) {
+                return;
+        } else if ( Cancelled == desc->state ) {
+                return;
+        }
+        desc->state = Cancelled;
+        if (!join) {
+                defaultResumptionHandler = default_thread_cancel_handler;
+        }
+
+        ThreadCancelled(T) except;
+        // TODO: Remove explitate vtable set once trac#186 is fixed.
+        except.virtual_table = &get_exception_vtable(&except);
+        except.the_thread = &thrd;
+        except.the_exception = __cfaehm_cancellation_exception( cancellation );
+        throwResume except;
+
+        except.the_exception->virtual_table->free( except.the_exception );
+        free( cancellation );
+        desc->self_cor.cancellation = 0p;
+}
+
+void ^?{}( thread_dtor_guard_t & this ) {
+        ^(this.mg){};
 }
 
@@ -93 +150 @@
 }
 
+//-----------------------------------------------------------------------------
+forall(dtype T | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T)))
+T & join( T & this ) {
+        thread_dtor_guard_t guard = { this, defaultResumptionHandler };
+        return this;
+}
+
 // Local Variables: //
 // mode: c //

libcfa/src/concurrency/thread.hfa

@@ -22 +22 @@
 #include "kernel.hfa"
 #include "monitor.hfa"
+#include "exception.hfa"
 
 //-----------------------------------------------------------------------------
 // thread trait
 trait is_thread(dtype T) {
-      void ^?{}(T& mutex this);
-      void main(T& this);
-      $thread* get_thread(T& this);
+        void ^?{}(T& mutex this);
+        void main(T& this);
+        $thread* get_thread(T& this);
 };
+
+FORALL_DATA_EXCEPTION(ThreadCancelled, (dtype thread_t), (thread_t)) (
+        thread_t * the_thread;
+        exception_t * the_exception;
+);
+
+forall(dtype T)
+void copy(ThreadCancelled(T) * dst, ThreadCancelled(T) * src);
+
+forall(dtype T)
+const char * msg(ThreadCancelled(T) *);
 
 // define that satisfies the trait without using the thread keyword
@@ -65 +77 @@
 static inline void ?{}($thread & this, const char * const name, struct cluster & cl )                   { this{ name, cl, 0p, 65000 }; }
 static inline void ?{}($thread & this, const char * const name, struct cluster & cl, size_t stackSize ) { this{ name, cl, 0p, stackSize }; }
+
+struct thread_dtor_guard_t {
+        monitor_dtor_guard_t mg;
+};
+
+forall( dtype T | is_thread(T) | IS_EXCEPTION(ThreadCancelled, (T)) )
+void ?{}( thread_dtor_guard_t & this, T & thrd, void(*)(ThreadCancelled(T) &) );
+void ^?{}( thread_dtor_guard_t & this );
 
 //-----------------------------------------------------------------------------
@@ -108 +128 @@
 //----------
 // join
-forall( dtype T | is_thread(T) )
+forall( dtype T | is_thread(T) | IS_RESUMPTION_EXCEPTION(ThreadCancelled, (T)) )
 T & join( T & this );