Changeset ae06e0b for libcfa/src/concurrency

Timestamp:

Jun 8, 2022, 4:23:08 PM (2 years ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, master, pthread-emulation, qualifiedEnum

Children:

55422cf

Parents:

ced5e2a

Message:

added pthread_cond_var

Location:

libcfa/src/concurrency

Files:

• locks.cfa (modified) (3 diffs)
• locks.hfa (modified) (4 diffs)

libcfa/src/concurrency/locks.cfa

@@ -178 +178 @@
 
 //-----------------------------------------------------------------------------
-// simple_owner_lock
-
-static inline void lock(simple_owner_lock & this) with(this) {
-        if (owner == active_thread()) {
-                recursion_count++;
-                return;
-        }
-        lock( lock __cfaabi_dbg_ctx2 );
-
-        if (owner != 0p) {
-                insert_last( blocked_threads, *active_thread() );
-                unlock( lock );
-                park( );
-                return;
-        }
-        owner = active_thread();
-        recursion_count = 1;
-        unlock( lock );
-}
-
-void pop_and_set_new_owner( simple_owner_lock & this ) with( this ) {
-        thread$ * t = &try_pop_front( blocked_threads );
-        owner = t;
-        recursion_count = ( t ? 1 : 0 );
-        unpark( t );
-}
-
-static inline void unlock(simple_owner_lock & this) with(this) {
-        lock( lock __cfaabi_dbg_ctx2 );
-        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
-        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
-        // if recursion count is zero release lock and set new owner if one is waiting
-        recursion_count--;
-        if ( recursion_count == 0 ) {
-                pop_and_set_new_owner( this );
-        }
-        unlock( lock );
-}
-
-static inline void on_notify(simple_owner_lock & this, struct thread$ * t ) with(this) {
-        lock( lock __cfaabi_dbg_ctx2 );
-        // lock held
-        if ( owner != 0p ) {
-                insert_last( blocked_threads, *t );
-                unlock( lock );
-        }
-        // lock not held
-        else {
-                owner = t;
-                recursion_count = 1;
-                unpark( t );
-                unlock( lock );
-        }
-}
-
-static inline size_t on_wait(simple_owner_lock & this) with(this) { 
-        lock( lock __cfaabi_dbg_ctx2 );
-        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
-        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
-
-        size_t ret = recursion_count;
-
-        pop_and_set_new_owner( this );
-
-        unlock( lock );
-        return ret;
-}
-
-static inline void on_wakeup(simple_owner_lock & this, size_t recursion ) with(this) { recursion_count = recursion; }
-
-
-//-----------------------------------------------------------------------------
 // alarm node wrapper
 forall(L & | is_blocking_lock(L)) {
@@ -291 +219 @@
         // this casts the alarm node to our wrapped type since we used type erasure
         static void alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (alarm_node_wrap(L) &)a ); }
+
+        struct pthread_alarm_node_wrap {
+                alarm_node_t alarm_node;
+                pthread_cond_var(L) * cond;
+                info_thread(L) * info_thd;
+        };
+
+        void ?{}( pthread_alarm_node_wrap(L) & this, Duration alarm, Duration period, Alarm_Callback callback, pthread_cond_var(L) * c, info_thread(L) * i ) {
+                this.alarm_node{ callback, alarm, period };
+                this.cond = c;
+                this.info_thd = i;
+        }
+
+        void ^?{}( pthread_alarm_node_wrap(L) & this ) { }
+
+        static void timeout_handler ( pthread_alarm_node_wrap(L) & this ) with( this ) {
+                // This pthread_cond_var member is called from the kernel, and therefore, cannot block, but it can spin.
+                lock( cond->lock __cfaabi_dbg_ctx2 );
+
+                // this check is necessary to avoid a race condition since this timeout handler
+                //      may still be called after a thread has been removed from the queue but
+                //      before the alarm is unregistered
+                if ( (*info_thd)`isListed ) {   // is thread on queue
+                        info_thd->signalled = false;
+                        // remove this thread O(1)
+                        remove( *info_thd );
+                        on_notify(*info_thd->lock, info_thd->t);
+                }
+                unlock( cond->lock );
+        }
+
+        // this casts the alarm node to our wrapped type since we used type erasure
+        static void pthread_alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (pthread_alarm_node_wrap(L) &)a ); }
 }
 
@@ -460 +421 @@
                 on_wakeup(*i.lock, recursion_count);
         }
-}
-
+
+        //-----------------------------------------------------------------------------
+        // pthread_cond_var
+
+        void  ?{}( pthread_cond_var(L) & this ) with(this) {
+                blocked_threads{};
+                lock{};
+        }
+
+        void ^?{}( pthread_cond_var(L) & this ) { }
+
+        bool notify_one( pthread_cond_var(L) & this ) with(this) { 
+                lock( lock __cfaabi_dbg_ctx2 );
+                bool ret = ! blocked_threads`isEmpty;
+                if ( ret ) {
+                        info_thread(L) & popped = try_pop_front( blocked_threads );
+                        on_notify(*popped.lock, popped.t);
+                }
+                unlock( lock );
+                return ret;
+        }
+
+        bool notify_all( pthread_cond_var(L) & this ) with(this) { 
+                lock( lock __cfaabi_dbg_ctx2 );
+                bool ret = ! blocked_threads`isEmpty;
+                while( ! blocked_threads`isEmpty ) {
+                        info_thread(L) & popped = try_pop_front( blocked_threads );
+                        on_notify(*popped.lock, popped.t);
+                }
+                unlock( lock );
+                return ret;
+        }
+
+        uintptr_t front( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty ? NULL : blocked_threads`first.info; }
+        bool empty ( pthread_cond_var(L) & this ) with(this) { return blocked_threads`isEmpty; }
+
+        static size_t queue_and_get_recursion( pthread_cond_var(L) & this, info_thread(L) * i ) with(this) {
+                // add info_thread to waiting queue
+                insert_last( blocked_threads, *i );
+                size_t recursion_count = 0;
+                recursion_count = on_wait( *i->lock );
+                return recursion_count;
+        }
+        
+        static void queue_info_thread_timeout( pthread_cond_var(L) & this, info_thread(L) & info, Duration t, Alarm_Callback callback ) with(this) {
+                lock( lock __cfaabi_dbg_ctx2 );
+                size_t recursion_count = queue_and_get_recursion(this, &info);
+                pthread_alarm_node_wrap(L) node_wrap = { t, 0`s, callback, &this, &info };
+                register_self( &node_wrap.alarm_node );
+                unlock( lock );
+
+                // blocks here
+                park();
+
+                // unregisters alarm so it doesn't go off if this happens first
+                unregister_self( &node_wrap.alarm_node );
+
+                // resets recursion count here after waking
+                if (info.lock) on_wakeup(*info.lock, recursion_count);
+        }
+
+        void wait( pthread_cond_var(L) & this, L & l ) with(this) {
+                wait( this, l, 0 );
+        }
+
+        void wait( pthread_cond_var(L) & this, L & l, uintptr_t info ) with(this) {
+                lock( lock __cfaabi_dbg_ctx2 );
+                info_thread( L ) i = { active_thread(), info, &l };
+                size_t recursion_count = queue_and_get_recursion(this, &i);
+                unlock( lock );
+                park( );
+                on_wakeup(*i.lock, recursion_count);
+        }
+
+        #define PTHREAD_WAIT_TIME( u, l, t ) \
+                info_thread( L ) i = { active_thread(), u, l }; \
+                queue_info_thread_timeout(this, i, t, pthread_alarm_node_wrap_cast ); \
+                return i.signalled;
+
+        bool wait( pthread_cond_var(L) & this, L & l, timespec t ) {
+                Duration d = { t };
+                WAIT_TIME( 0, &l , d )
+        }
+        
+        bool wait( pthread_cond_var(L) & this, L & l, uintptr_t info, timespec t  ) {
+                Duration d = { t };
+                WAIT_TIME( info, &l , d )
+        }
+}
 //-----------------------------------------------------------------------------
 // Semaphore

libcfa/src/concurrency/locks.hfa

@@ -366 +366 @@
 static inline void ?=?( simple_owner_lock & this, simple_owner_lock this2 ) = void;
 
+static inline void lock(simple_owner_lock & this) with(this) {
+        if (owner == active_thread()) {
+                recursion_count++;
+                return;
+        }
+        lock( lock __cfaabi_dbg_ctx2 );
+
+        if (owner != 0p) {
+                insert_last( blocked_threads, *active_thread() );
+                unlock( lock );
+                park( );
+                return;
+        }
+        owner = active_thread();
+        recursion_count = 1;
+        unlock( lock );
+}
+
+// TODO: fix duplicate def issue and bring this back
+// void pop_and_set_new_owner( simple_owner_lock & this ) with( this ) {
+        // thread$ * t = &try_pop_front( blocked_threads );
+        // owner = t;
+        // recursion_count = ( t ? 1 : 0 );
+        // unpark( t );
+// }
+
+static inline void unlock(simple_owner_lock & this) with(this) {
+        lock( lock __cfaabi_dbg_ctx2 );
+        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
+        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+        // if recursion count is zero release lock and set new owner if one is waiting
+        recursion_count--;
+        if ( recursion_count == 0 ) {
+                // pop_and_set_new_owner( this );
+                thread$ * t = &try_pop_front( blocked_threads );
+                owner = t;
+                recursion_count = ( t ? 1 : 0 );
+                unpark( t );
+        }
+        unlock( lock );
+}
+
+static inline void on_notify(simple_owner_lock & this, struct thread$ * t ) with(this) {
+        lock( lock __cfaabi_dbg_ctx2 );
+        // lock held
+        if ( owner != 0p ) {
+                insert_last( blocked_threads, *t );
+                unlock( lock );
+        }
+        // lock not held
+        else {
+                owner = t;
+                recursion_count = 1;
+                unpark( t );
+                unlock( lock );
+        }
+}
+
+static inline size_t on_wait(simple_owner_lock & this) with(this) { 
+        lock( lock __cfaabi_dbg_ctx2 );
+        /* paranoid */ verifyf( owner != 0p, "Attempt to release lock %p that isn't held", &this );
+        /* paranoid */ verifyf( owner == active_thread(), "Thread %p other than the owner %p attempted to release owner lock %p", owner, active_thread(), &this );
+
+        size_t ret = recursion_count;
+
+        // pop_and_set_new_owner( this );
+
+        thread$ * t = &try_pop_front( blocked_threads );
+        owner = t;
+        recursion_count = ( t ? 1 : 0 );
+        unpark( t );
+
+        unlock( lock );
+        return ret;
+}
+
+static inline void on_wakeup(simple_owner_lock & this, size_t recursion ) with(this) { recursion_count = recursion; }
+
 //-----------------------------------------------------------------------------
 // Spin Queue Lock
@@ -606 +684 @@
         // - signalling without holding branded lock is UNSAFE!
         // - only allows usage of one lock, cond var is branded after usage
+
         struct fast_cond_var {
                 // List of blocked threads
                 dlist( info_thread(L) ) blocked_threads;
-
                 #ifdef __CFA_DEBUG__
                 L * lock_used;
@@ -615 +693 @@
         };
 
-
         void  ?{}( fast_cond_var(L) & this );
         void ^?{}( fast_cond_var(L) & this );
@@ -623 +700 @@
 
         uintptr_t front( fast_cond_var(L) & this );
-
         bool empty  ( fast_cond_var(L) & this );
 
         void wait( fast_cond_var(L) & this, L & l );
         void wait( fast_cond_var(L) & this, L & l, uintptr_t info );
-}
+
+
+        //-----------------------------------------------------------------------------
+        // pthread_cond_var
+        //
+        // - cond var with minimal footprint
+        // - supports operations needed for phthread cond
+
+        struct pthread_cond_var {
+                dlist( info_thread(L) ) blocked_threads;
+                __spinlock_t lock;
+        };
+
+        void  ?{}( pthread_cond_var(L) & this );
+        void ^?{}( pthread_cond_var(L) & this );
+
+        bool notify_one( pthread_cond_var(L) & this );
+        bool notify_all( pthread_cond_var(L) & this );
+
+        uintptr_t front( pthread_cond_var(L) & this );
+        bool empty ( pthread_cond_var(L) & this );
+
+        void wait( pthread_cond_var(L) & this, L & l );
+        void wait( pthread_cond_var(L) & this, L & l, uintptr_t info );
+        bool wait( pthread_cond_var(L) & this, L & l, timespec t );
+        bool wait( pthread_cond_var(L) & this, L & l, uintptr_t info, timespec t );
+}