Changeset 454f478 for libcfa/src/concurrency

Timestamp:

Jan 20, 2021, 5:35:39 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:

9db2c92

Parents:

dafbde8

Message:

Re-arranged and commented low-level headers.
Main goal was for better support of weakso locks that are comming.

Location:

libcfa/src/concurrency

Files:

• io/types.hfa (modified) (2 diffs)
• kernel.cfa (modified) (2 diffs)
• kernel.hfa (modified) (3 diffs)
• kernel/fwd.hfa (modified) (2 diffs)
• kernel/startup.cfa (modified) (3 diffs)
• locks.cfa (modified) (1 diff)
• locks.hfa (modified) (1 diff)

libcfa/src/concurrency/io/types.hfa

@@ -5 +5 @@
 // file "LICENCE" distributed with Cforall.
 //
-// io/types.hfa --
+// io/types.hfa -- PRIVATE
+// Types used by the I/O subsystem
 //
 // Author           : Thierry Delisle
@@ -20 +21 @@
 }
 
-#include "bits/locks.hfa"
+#include "kernel/fwd.hfa"
 
 #if defined(CFA_HAVE_LINUX_IO_URING_H)

libcfa/src/concurrency/kernel.cfa

@@ -224 +224 @@
         }
 
-        V( this->terminated );
+        post( this->terminated );
 
         if(this == mainProcessor) {
@@ -665 +665 @@
 //=============================================================================================
 //-----------------------------------------------------------------------------
-// Locks
-void  ?{}( semaphore & this, int count = 1 ) {
-        (this.lock){};
-        this.count = count;
-        (this.waiting){};
-}
-void ^?{}(semaphore & this) {}
-
-bool P(semaphore & this) with( this ){
-        lock( lock __cfaabi_dbg_ctx2 );
-        count -= 1;
-        if ( count < 0 ) {
-                // queue current task
-                append( waiting, active_thread() );
-
-                // atomically release spin lock and block
-                unlock( lock );
-                park();
-                return true;
-        }
-        else {
-            unlock( lock );
-            return false;
-        }
-}
-
-bool V(semaphore & this) with( this ) {
-        $thread * thrd = 0p;
-        lock( lock __cfaabi_dbg_ctx2 );
-        count += 1;
-        if ( count <= 0 ) {
-                // remove task at head of waiting list
-                thrd = pop_head( waiting );
-        }
-
-        unlock( lock );
-
-        // make new owner
-        unpark( thrd );
-
-        return thrd != 0p;
-}
-
-bool V(semaphore & this, unsigned diff) with( this ) {
-        $thread * thrd = 0p;
-        lock( lock __cfaabi_dbg_ctx2 );
-        int release = max(-count, (int)diff);
-        count += diff;
-        for(release) {
-                unpark( pop_head( waiting ) );
-        }
-
-        unlock( lock );
-
-        return thrd != 0p;
-}
-
-//-----------------------------------------------------------------------------
 // Debug
 __cfaabi_dbg_debug_do(

libcfa/src/concurrency/kernel.hfa

@@ -5 +5 @@
 // file "LICENCE" distributed with Cforall.
 //
-// kernel --
+// kernel -- Header containing the core of the kernel API
 //
 // Author           : Thierry Delisle
@@ -24 +24 @@
 extern "C" {
         #include <bits/pthreadtypes.h>
+        #include <pthread.h>
         #include <linux/types.h>
 }
 
 //-----------------------------------------------------------------------------
-// Locks
-struct semaphore {
-        __spinlock_t lock;
-        int count;
-        __queue_t($thread) waiting;
-};
-
-void  ?{}(semaphore & this, int count = 1);
-void ^?{}(semaphore & this);
-bool   P (semaphore & this);
-bool   V (semaphore & this);
-bool   V (semaphore & this, unsigned count);
+// Underlying Locks
+#ifdef __CFA_WITH_VERIFY__
+        extern bool __cfaabi_dbg_in_kernel();
+#endif
+
+extern "C" {
+        char * strerror(int);
+}
+#define CHECKED(x) { int err = x; if( err != 0 ) abort("KERNEL ERROR: Operation \"" #x "\" return error %d - %s\n", err, strerror(err)); }
+
+struct __bin_sem_t {
+        pthread_mutex_t         lock;
+        pthread_cond_t          cond;
+        int                     val;
+};
+
+static inline void ?{}(__bin_sem_t & this) with( this ) {
+        // Create the mutex with error checking
+        pthread_mutexattr_t mattr;
+        pthread_mutexattr_init( &mattr );
+        pthread_mutexattr_settype( &mattr, PTHREAD_MUTEX_ERRORCHECK_NP);
+        pthread_mutex_init(&lock, &mattr);
+
+        pthread_cond_init (&cond, (const pthread_condattr_t *)0p);  // workaround trac#208: cast should not be required
+        val = 0;
+}
+
+static inline void ^?{}(__bin_sem_t & this) with( this ) {
+        CHECKED( pthread_mutex_destroy(&lock) );
+        CHECKED( pthread_cond_destroy (&cond) );
+}
+
+static inline void wait(__bin_sem_t & this) with( this ) {
+        verify(__cfaabi_dbg_in_kernel());
+        CHECKED( pthread_mutex_lock(&lock) );
+                while(val < 1) {
+                        pthread_cond_wait(&cond, &lock);
+                }
+                val -= 1;
+        CHECKED( pthread_mutex_unlock(&lock) );
+}
+
+static inline bool post(__bin_sem_t & this) with( this ) {
+        bool needs_signal = false;
+
+        CHECKED( pthread_mutex_lock(&lock) );
+                if(val < 1) {
+                        val += 1;
+                        pthread_cond_signal(&cond);
+                        needs_signal = true;
+                }
+        CHECKED( pthread_mutex_unlock(&lock) );
+
+        return needs_signal;
+}
+
+#undef CHECKED
 
 
@@ -91 +137 @@
 
         // Termination synchronisation (user semaphore)
-        semaphore terminated;
+        oneshot terminated;
 
         // pthread Stack

libcfa/src/concurrency/kernel/fwd.hfa

@@ -5 +5 @@
 // file "LICENCE" distributed with Cforall.
 //
-// kernel/fwd.hfa --
+// kernel/fwd.hfa -- PUBLIC
+// Fundamental code needed to implement threading M.E.S. algorithms.
 //
 // Author           : Thierry Delisle
@@ -134 +135 @@
                 extern uint64_t thread_rand();
 
+                // Semaphore which only supports a single thread
+                struct single_sem {
+                        struct $thread * volatile ptr;
+                };
+
+                static inline {
+                        void  ?{}(single_sem & this) {
+                                this.ptr = 0p;
+                        }
+
+                        void ^?{}(single_sem &) {}
+
+                        bool wait(single_sem & this) {
+                                for() {
+                                        struct $thread * expected = this.ptr;
+                                        if(expected == 1p) {
+                                                if(__atomic_compare_exchange_n(&this.ptr, &expected, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                        return false;
+                                                }
+                                        }
+                                        else {
+                                                /* paranoid */ verify( expected == 0p );
+                                                if(__atomic_compare_exchange_n(&this.ptr, &expected, active_thread(), false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                        park();
+                                                        return true;
+                                                }
+                                        }
+
+                                }
+                        }
+
+                        bool post(single_sem & this) {
+                                for() {
+                                        struct $thread * expected = this.ptr;
+                                        if(expected == 1p) return false;
+                                        if(expected == 0p) {
+                                                if(__atomic_compare_exchange_n(&this.ptr, &expected, 1p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                        return false;
+                                                }
+                                        }
+                                        else {
+                                                if(__atomic_compare_exchange_n(&this.ptr, &expected, 0p, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                        unpark( expected );
+                                                        return true;
+                                                }
+                                        }
+                                }
+                        }
+                }
+
+                // Synchronozation primitive which only supports a single thread and one post
+                // Similar to a binary semaphore with a 'one shot' semantic
+                // is expected to be discarded after each party call their side
+                struct oneshot {
+                        // Internal state :
+                        //     0p     : is initial state (wait will block)
+                        //     1p     : fulfilled (wait won't block)
+                        // any thread : a thread is currently waiting
+                        struct $thread * volatile ptr;
+                };
+
+                static inline {
+                        void  ?{}(oneshot & this) {
+                                this.ptr = 0p;
+                        }
+
+                        void ^?{}(oneshot &) {}
+
+                        // Wait for the post, return immidiately if it already happened.
+                        // return true if the thread was parked
+                        bool wait(oneshot & this) {
+                                for() {
+                                        struct $thread * expected = this.ptr;
+                                        if(expected == 1p) return false;
+                                        /* paranoid */ verify( expected == 0p );
+                                        if(__atomic_compare_exchange_n(&this.ptr, &expected, active_thread(), false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                park();
+                                                /* paranoid */ verify( this.ptr == 1p );
+                                                return true;
+                                        }
+                                }
+                        }
+
+                        // Mark as fulfilled, wake thread if needed
+                        // return true if a thread was unparked
+                        bool post(oneshot & this) {
+                                struct $thread * got = __atomic_exchange_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
+                                if( got == 0p ) return false;
+                                unpark( got );
+                                return true;
+                        }
+                }
+
+                // base types for future to build upon
+                // It is based on the 'oneshot' type to allow multiple futures
+                // to block on the same instance, permitting users to block a single
+                // thread on "any of" [a given set of] futures.
+                // does not support multiple threads waiting on the same future
+                struct future_t {
+                        // Internal state :
+                        //     0p      : is initial state (wait will block)
+                        //     1p      : fulfilled (wait won't block)
+                        //     2p      : in progress ()
+                        //     3p      : abandoned, server should delete
+                        // any oneshot : a context has been setup to wait, a thread could wait on it
+                        struct oneshot * volatile ptr;
+                };
+
+                static inline {
+                        void  ?{}(future_t & this) {
+                                this.ptr = 0p;
+                        }
+
+                        void ^?{}(future_t &) {}
+
+                        void reset(future_t & this) {
+                                // needs to be in 0p or 1p
+                                __atomic_exchange_n( &this.ptr, 0p, __ATOMIC_SEQ_CST);
+                        }
+
+                        // check if the future is available
+                        bool available( future_t & this ) {
+                                return this.ptr == 1p;
+                        }
+
+                        // Prepare the future to be waited on
+                        // intented to be use by wait, wait_any, waitfor, etc. rather than used directly
+                        bool setup( future_t & this, oneshot & wait_ctx ) {
+                                /* paranoid */ verify( wait_ctx.ptr == 0p );
+                                // The future needs to set the wait context
+                                for() {
+                                        struct oneshot * expected = this.ptr;
+                                        // Is the future already fulfilled?
+                                        if(expected == 1p) return false; // Yes, just return false (didn't block)
+
+                                        // The future is not fulfilled, try to setup the wait context
+                                        /* paranoid */ verify( expected == 0p );
+                                        if(__atomic_compare_exchange_n(&this.ptr, &expected, &wait_ctx, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                return true;
+                                        }
+                                }
+                        }
+
+                        // Stop waiting on a future
+                        // When multiple futures are waited for together in "any of" pattern
+                        // futures that weren't fulfilled before the thread woke up
+                        // should retract the wait ctx
+                        // intented to be use by wait, wait_any, waitfor, etc. rather than used directly
+                        void retract( future_t & this, oneshot & wait_ctx ) {
+                                // Remove the wait context
+                                struct oneshot * got = __atomic_exchange_n( &this.ptr, 0p, __ATOMIC_SEQ_CST);
+
+                                // got == 0p: future was never actually setup, just return
+                                if( got == 0p ) return;
+
+                                // got == wait_ctx: since fulfil does an atomic_swap,
+                                // if we got back the original then no one else saw context
+                                // It is safe to delete (which could happen after the return)
+                                if( got == &wait_ctx ) return;
+
+                                // got == 1p: the future is ready and the context was fully consumed
+                                // the server won't use the pointer again
+                                // It is safe to delete (which could happen after the return)
+                                if( got == 1p ) return;
+
+                                // got == 2p: the future is ready but the context hasn't fully been consumed
+                                // spin until it is safe to move on
+                                if( got == 2p ) {
+                                        while( this.ptr != 1p ) Pause();
+                                        return;
+                                }
+
+                                // got == any thing else, something wen't wrong here, abort
+                                abort("Future in unexpected state");
+                        }
+
+                        // Mark the future as abandoned, meaning it will be deleted by the server
+                        bool abandon( future_t & this ) {
+                                /* paranoid */ verify( this.ptr != 3p );
+
+                                // Mark the future as abandonned
+                                struct oneshot * got = __atomic_exchange_n( &this.ptr, 3p, __ATOMIC_SEQ_CST);
+
+                                // If the future isn't already fulfilled, let the server delete it
+                                if( got == 0p ) return false;
+
+                                // got == 2p: the future is ready but the context hasn't fully been consumed
+                                // spin until it is safe to move on
+                                if( got == 2p ) {
+                                        while( this.ptr != 1p ) Pause();
+                                        got = 1p;
+                                }
+
+                                // The future is completed delete it now
+                                /* paranoid */ verify( this.ptr != 1p );
+                                free( &this );
+                                return true;
+                        }
+
+                        // from the server side, mark the future as fulfilled
+                        // delete it if needed
+                        bool fulfil( future_t & this ) {
+                                for() {
+                                        struct oneshot * expected = this.ptr;
+                                        // was this abandoned?
+                                        #if defined(__GNUC__) && __GNUC__ >= 7
+                                                #pragma GCC diagnostic push
+                                                #pragma GCC diagnostic ignored "-Wfree-nonheap-object"
+                                        #endif
+                                                if( expected == 3p ) { free( &this ); return false; }
+                                        #if defined(__GNUC__) && __GNUC__ >= 7
+                                                #pragma GCC diagnostic pop
+                                        #endif
+
+                                        /* paranoid */ verify( expected != 1p ); // Future is already fulfilled, should not happen
+                                        /* paranoid */ verify( expected != 2p ); // Future is bein fulfilled by someone else, this is even less supported then the previous case.
+
+                                        // If there is a wait context, we need to consume it and mark it as consumed after
+                                        // If there is no context then we can skip the in progress phase
+                                        struct oneshot * want = expected == 0p ? 1p : 2p;
+                                        if(__atomic_compare_exchange_n(&this.ptr, &expected, want, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+                                                if( expected == 0p ) { /* paranoid */ verify( this.ptr == 1p); return false; }
+                                                bool ret = post( *expected );
+                                                __atomic_store_n( &this.ptr, 1p, __ATOMIC_SEQ_CST);
+                                                return ret;
+                                        }
+                                }
+
+                        }
+
+                        // Wait for the future to be fulfilled
+                        bool wait( future_t & this ) {
+                                oneshot temp;
+                                if( !setup(this, temp) ) return false;
+
+                                // Wait context is setup, just wait on it
+                                bool ret = wait( temp );
+
+                                // Wait for the future to tru
+                                while( this.ptr == 2p ) Pause();
+                                // Make sure the state makes sense
+                                // Should be fulfilled, could be in progress but it's out of date if so
+                                // since if that is the case, the oneshot was fulfilled (unparking this thread)
+                                // and the oneshot should not be needed any more
+                                __attribute__((unused)) struct oneshot * was = this.ptr;
+                                /* paranoid */ verifyf( was == 1p, "Expected this.ptr to be 1p, was %p\n", was );
+
+                                // Mark the future as fulfilled, to be consistent
+                                // with potential calls to avail
+                                // this.ptr = 1p;
+                                return ret;
+                        }
+                }
+
                 //-----------------------------------------------------------------------
                 // Statics call at the end of each thread to register statistics

libcfa/src/concurrency/kernel/startup.cfa

@@ -199 +199 @@
         void ?{}(processor & this) with( this ) {
                 ( this.idle ){};
-                ( this.terminated ){ 0 };
+                ( this.terminated ){};
                 ( this.runner ){};
                 init( this, "Main Processor", *mainCluster );
@@ -528 +528 @@
 void ?{}(processor & this, const char name[], cluster & _cltr) {
         ( this.idle ){};
-        ( this.terminated ){ 0 };
+        ( this.terminated ){};
         ( this.runner ){};
 
@@ -549 +549 @@
                 __wake_proc( &this );
 
-                P( terminated );
+                wait( terminated );
                 /* paranoid */ verify( active_processor() != &this);
         }

libcfa/src/concurrency/locks.cfa

@@ -356 +356 @@
         bool wait( condition_variable(L) & this, L & l, uintptr_t info, Time time         ) with(this) { WAIT_TIME( info, &l , time ) }
 }
+
+//-----------------------------------------------------------------------------
+// Semaphore
+void  ?{}( semaphore & this, int count = 1 ) {
+        (this.lock){};
+        this.count = count;
+        (this.waiting){};
+}
+void ^?{}(semaphore & this) {}
+
+bool P(semaphore & this) with( this ){
+        lock( lock __cfaabi_dbg_ctx2 );
+        count -= 1;
+        if ( count < 0 ) {
+                // queue current task
+                append( waiting, active_thread() );
+
+                // atomically release spin lock and block
+                unlock( lock );
+                park();
+                return true;
+        }
+        else {
+            unlock( lock );
+            return false;
+        }
+}
+
+bool V(semaphore & this) with( this ) {
+        $thread * thrd = 0p;
+        lock( lock __cfaabi_dbg_ctx2 );
+        count += 1;
+        if ( count <= 0 ) {
+                // remove task at head of waiting list
+                thrd = pop_head( waiting );
+        }
+
+        unlock( lock );
+
+        // make new owner
+        unpark( thrd );
+
+        return thrd != 0p;
+}
+
+bool V(semaphore & this, unsigned diff) with( this ) {
+        $thread * thrd = 0p;
+        lock( lock __cfaabi_dbg_ctx2 );
+        int release = max(-count, (int)diff);
+        count += diff;
+        for(release) {
+                unpark( pop_head( waiting ) );
+        }
+
+        unlock( lock );
+
+        return thrd != 0p;
+}

libcfa/src/concurrency/locks.hfa

@@ -157 +157 @@
         bool wait( condition_variable(L) & this, L & l, uintptr_t info, Time time );
 }
+
+//-----------------------------------------------------------------------------
+// Semaphore
+struct semaphore {
+        __spinlock_t lock;
+        int count;
+        __queue_t($thread) waiting;
+};
+
+void  ?{}(semaphore & this, int count = 1);
+void ^?{}(semaphore & this);
+bool   P (semaphore & this);
+bool   V (semaphore & this);
+bool   V (semaphore & this, unsigned count);