source: libcfa/src/concurrency/locks.cfa @ c20533ea

#include "locks.hfa"
#include "kernel_private.hfa"

#include <kernel.hfa>
#include <stdlib.hfa>

///////////////////////////////////////////////////////////////////
//// info_thread
///////////////////////////////////////////////////////////////////

forall(dtype L | is_blocking_lock(L)) {
        void ?{}( info_thread(L) & this, $thread * t ) {
                ((Seqable &) this){};
                this.t = t;
                this.lock = 0p;
        }

        void ?{}( info_thread(L) & this, $thread * t, uintptr_t info ) {
                ((Seqable &) this){};
                this.t = t;
                this.info = info;
                this.lock = 0p;
        }

        void ^?{}( info_thread(L) & this ){ }

        info_thread(L) *& Back( info_thread(L) * this ) {
                return (info_thread(L) *)Back( (Seqable *)this );
        }

        info_thread(L) *& Next( info_thread(L) * this ) {
                return (info_thread(L) *)Next( (Colable *)this );
        }
}

///////////////////////////////////////////////////////////////////
//// Blocking Locks
///////////////////////////////////////////////////////////////////

void ?{}( blocking_lock & this, bool multi_acquisition, bool strict_owner ) {
        this.lock{};
        this.blocked_threads{};
        this.wait_count = 0;
        this.multi_acquisition = multi_acquisition;
        this.strict_owner = strict_owner;
        this.owner = 0p;
        this.recursion_count = 0;
}

void ^?{}( blocking_lock & this ) {}
void ?{}( single_acquisition_lock & this ) {((blocking_lock &)this){ false, false };}
void ^?{}( single_acquisition_lock & this ) {}
void ?{}( owner_lock & this ) {((blocking_lock &)this){ true, true };}
void ^?{}( owner_lock & this ) {}
void ?{}( multiple_acquisition_lock & this ) {((blocking_lock &)this){ true, false };}
void ^?{}( multiple_acquisition_lock & this ) {}

void lock( blocking_lock & this ) with( this ) {
        lock( lock __cfaabi_dbg_ctx2 );
        if ( owner == active_thread() && !multi_acquisition) { // single acquisition lock is held by current thread
                abort("A single acquisition lock holder attempted to reacquire the lock resulting in a deadlock.");
        } else if ( owner != 0p && owner != active_thread() ) { // lock is held by some other thread
                addTail( blocked_threads, *active_thread() );
                wait_count++;
                unlock( lock );
                park( );
        } else if ( owner == active_thread() && multi_acquisition ) { // multi acquisition lock is held by current thread
                recursion_count++;
                unlock( lock );
        } else { // lock isn't held
                owner = active_thread();
                recursion_count = 1;
                unlock( lock );
        }
}

bool try_lock( blocking_lock & this ) with( this ) {
        bool ret = false;
        lock( lock __cfaabi_dbg_ctx2 );
        if ( owner == 0p ) { // lock isn't held
                owner = active_thread();
                recursion_count = 1;
                ret = true;
        } else if ( owner == active_thread() && multi_acquisition ) { // multi acquisition lock is held by current thread
                recursion_count++;
                ret = true;
        }
        unlock( lock );
        return ret;
}

void unlock_error_check( blocking_lock & this ) with( this ) {
        if ( owner == 0p ){ // lock isn't held
                abort( "There was an attempt to release a lock that isn't held" ); 
        } else if ( strict_owner && owner != active_thread() ) {
                abort( "A thread other than the owner attempted to release an owner lock" ); 
        }
}

void pop_and_set_new_owner( blocking_lock & this ) with( this ) {
        $thread * t = &dropHead( blocked_threads );
        owner = t;
        recursion_count = ( t ? 1 : 0 );
        wait_count--;
        unpark( t );
}

void unlock( blocking_lock & this ) with( this ) {
        lock( lock __cfaabi_dbg_ctx2 );
        unlock_error_check( this );
        recursion_count--;
        if ( recursion_count == 0 ) { // if recursion count is zero release lock and set new owner if one is waiting
                pop_and_set_new_owner( this );
        }
        unlock( lock );
}

size_t wait_count( blocking_lock & this ) with( this ) {
        return wait_count;
}

void set_recursion_count( blocking_lock & this, size_t recursion ) with( this ) {
        recursion_count = recursion;
}

size_t get_recursion_count( blocking_lock & this ) with( this ) {
        return recursion_count;
}

void on_notify( blocking_lock & this, $thread * t ) with( this ) {
    lock( lock __cfaabi_dbg_ctx2 );
        if ( owner != 0p ) { // lock held
                addTail( blocked_threads, *t );
                wait_count++;
                unlock( lock );
        } else {        // lock not held
                owner = t;
                recursion_count = 1;
                unpark( t );
                unlock( lock );
        }
}

void on_wait( blocking_lock & this ) with( this ) {
    lock( lock __cfaabi_dbg_ctx2 );
        unlock_error_check( this );
        pop_and_set_new_owner( this );
        unlock( lock );
}

///////////////////////////////////////////////////////////////////
//// Overloaded routines for traits
///////////////////////////////////////////////////////////////////

// These routines are temporary until an inheritance bug is fixed

void lock( single_acquisition_lock & this ){ lock( (blocking_lock &)this ); }
void unlock( single_acquisition_lock & this ){ unlock( (blocking_lock &)this ); }
void on_notify( single_acquisition_lock & this, struct $thread * t ){ on_notify( (blocking_lock &)this, t ); }
void on_wait( single_acquisition_lock & this ){ on_wait( (blocking_lock &)this ); }
void set_recursion_count( single_acquisition_lock & this, size_t recursion ){ set_recursion_count( (blocking_lock &)this, recursion ); }
size_t get_recursion_count( single_acquisition_lock & this ){ return get_recursion_count( (blocking_lock &)this ); }

void lock( owner_lock & this ){ lock( (blocking_lock &)this ); }
void unlock( owner_lock & this ){ unlock( (blocking_lock &)this ); }
void on_notify( owner_lock & this, struct $thread * t ){ on_notify( (blocking_lock &)this, t ); }
void on_wait( owner_lock & this ){ on_wait( (blocking_lock &)this ); }
void set_recursion_count( owner_lock & this, size_t recursion ){ set_recursion_count( (blocking_lock &)this, recursion ); }
size_t get_recursion_count( owner_lock & this ){ return get_recursion_count( (blocking_lock &)this ); }

void lock( multiple_acquisition_lock & this ){ lock( (blocking_lock &)this ); }
void unlock( multiple_acquisition_lock & this ){ unlock( (blocking_lock &)this ); }
void on_notify( multiple_acquisition_lock & this, struct $thread * t ){ on_notify( (blocking_lock &)this, t ); }
void on_wait( multiple_acquisition_lock & this ){ on_wait( (blocking_lock &)this ); }
void set_recursion_count( multiple_acquisition_lock & this, size_t recursion ){ set_recursion_count( (blocking_lock &)this, recursion ); }
size_t get_recursion_count( multiple_acquisition_lock & this ){ return get_recursion_count( (blocking_lock &)this ); }


forall(dtype L | is_blocking_lock(L)) {
        struct alarm_node_wrap {
                alarm_node_t alarm_node;
                condition_variable(L) * cond;
                info_thread(L) * i;
        };

        void ?{}( alarm_node_wrap(L) & this, Time alarm, Duration period, Alarm_Callback callback ) {
                this.alarm_node{ callback, alarm, period };
        }

        void ^?{}( alarm_node_wrap(L) & this ) { }

        void timeout_handler ( alarm_node_wrap(L) & this ) with( this ) {
        // This condition_variable 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 ( listed(i) ) {                                                  // is thread on queue
                i->signalled = false;
                        remove( cond->blocked_threads, *i );    // remove this thread O(1)
                        cond->count--;
                        if( i->lock ) {
                                on_notify(*i->lock, i->t);                      // call lock's on_notify if a lock was passed
                } else {
                        unpark( i->t );                                         // otherwise wake thread
                }
            }
            unlock( cond->lock );
        }

        // this casts the alarm node to our wrapped type since we used type erasure
        void alarm_node_wrap_cast( alarm_node_t & a ) { timeout_handler( (alarm_node_wrap(L) &)a ); }

}

///////////////////////////////////////////////////////////////////
//// condition variable
///////////////////////////////////////////////////////////////////

forall(dtype L | is_blocking_lock(L)) {

        void ?{}( condition_variable(L) & this ){
                this.lock{};
                this.blocked_threads{};
                this.count = 0;
        }

        void ^?{}( condition_variable(L) & this ){ }

        void process_popped( condition_variable(L) & this, info_thread(L) & popped ) with( this ) {
                if(&popped != 0p) {
                        popped.signalled = true;
                        count--;
                        if (popped.lock) {
                                on_notify(*popped.lock, popped.t); // if lock passed call on_notify
                        } else {
                                unpark(popped.t); // otherwise wake thread
                        }
                }
        }

        bool notify_one( condition_variable(L) & this ) with( this ) {
                lock( lock __cfaabi_dbg_ctx2 );
                bool ret = !empty(blocked_threads);
                process_popped(this, dropHead( blocked_threads ));
                unlock( lock );
                return ret;
        }

        bool notify_all( condition_variable(L) & this ) with(this) {
                lock( lock __cfaabi_dbg_ctx2 );
                bool ret = !empty(blocked_threads);
                while( !empty(blocked_threads) ) {
                        process_popped(this, dropHead( blocked_threads ));
                }
                unlock( lock );
                return ret;
        }

        uintptr_t front( condition_variable(L) & this ) with(this) {
                return empty(blocked_threads) ? NULL : head(blocked_threads).info;
        }

        bool empty( condition_variable(L) & this ) with(this) { return empty(blocked_threads); }

        int counter( condition_variable(L) & this ) with(this) { return count; }

        size_t queue_and_get_recursion( condition_variable(L) & this, info_thread(L) * i ) with(this) {
                addTail( blocked_threads, *i ); // add info_thread to waiting queue
                count++;
                size_t recursion_count = 0;
                if (i->lock) { // if lock was passed get recursion count to reset to after waking thread
                        recursion_count = get_recursion_count(*i->lock);
                        on_wait( *i->lock );
                }
                return recursion_count;
        }

        // helper for wait()'s' with no timeout
        void queue_info_thread( condition_variable(L) & this, info_thread(L) & i ) with(this) {
                lock( lock __cfaabi_dbg_ctx2 );
                size_t recursion_count = queue_and_get_recursion(this, &i);
                unlock( lock );
                park( ); // blocks here
                if (i.lock) set_recursion_count(*i.lock, recursion_count); // resets recursion count here after waking
        }

        // helper for wait()'s' with a timeout
        void queue_info_thread_timeout( condition_variable(L) & this, info_thread(L) & info, Time t ) with(this) {
                lock( lock __cfaabi_dbg_ctx2 );
                size_t recursion_count = queue_and_get_recursion(this, &info);
                alarm_node_wrap(L) node_wrap = { t, 0`s, alarm_node_wrap_cast };
                node_wrap.cond = &this;
                node_wrap.i = &info;
                register_self( &node_wrap.alarm_node );
                unlock( lock );
                park();
                unregister_self( &node_wrap.alarm_node ); // unregisters alarm so it doesn't go off if this happens first
                if (info.lock) set_recursion_count(*info.lock, recursion_count); // resets recursion count here after waking
        }

        void wait( condition_variable(L) & this ) with(this) {
                info_thread( L ) i = { active_thread() };
                queue_info_thread( this, i );
        }

        void wait( condition_variable(L) & this, uintptr_t info ) with(this) {
                info_thread( L ) i = { active_thread(), info };
                queue_info_thread( this, i );
        }
        
        bool wait( condition_variable(L) & this, Duration duration ) with(this) {
                info_thread( L ) i = { active_thread() };
                queue_info_thread_timeout(this, i, __kernel_get_time() + duration );
                return i.signalled;
        }

        bool wait( condition_variable(L) & this, uintptr_t info, Duration duration ) with(this) { 
                info_thread( L ) i = { active_thread(), info };
                queue_info_thread_timeout(this, i, __kernel_get_time() + duration );
                return i.signalled;
        }

        bool wait( condition_variable(L) & this, Time time ) with(this) {
                info_thread( L ) i = { active_thread() };
                queue_info_thread_timeout(this, i, time);
                return i.signalled;
        }

        bool wait( condition_variable(L) & this, uintptr_t info, Time time ) with(this) {
                info_thread( L ) i = { active_thread(), info };
                queue_info_thread_timeout(this, i, time);
                return i.signalled;
        }

        void wait( condition_variable(L) & this, L & l ) with(this) {
                info_thread(L) i = { active_thread() };
                i.lock = &l;
                queue_info_thread( this, i );
        }

        void wait( condition_variable(L) & this, L & l, uintptr_t info ) with(this) {
                info_thread(L) i = { active_thread(), info };
                i.lock = &l;
                queue_info_thread( this, i );
        }
        
        bool wait( condition_variable(L) & this, L & l, Duration duration ) with(this) {
                info_thread(L) i = { active_thread() };
                i.lock = &l;
                queue_info_thread_timeout(this, i, __kernel_get_time() + duration );
                return i.signalled;
        }
        
        bool wait( condition_variable(L) & this, L & l, uintptr_t info, Duration duration ) with(this) {
                info_thread(L) i = { active_thread(), info };
                i.lock = &l;
                queue_info_thread_timeout(this, i, __kernel_get_time() + duration );
                return i.signalled;
        }
        
        bool wait( condition_variable(L) & this, L & l, Time time ) with(this) {
                info_thread(L) i = { active_thread() };
                i.lock = &l;
                queue_info_thread_timeout(this, i, time );
                return i.signalled;
        }
        
        bool wait( condition_variable(L) & this, L & l, uintptr_t info, Time time ) with(this) {
                info_thread(L) i = { active_thread(), info };
                i.lock = &l;
                queue_info_thread_timeout(this, i, time );
                return i.signalled;
        }
}