Subj : Re: experimental smr-based 100% lock-free eventcount schema for windows...
To   : comp.programming.threads
From : SenderX
Date : Tue Feb 01 2005 10:02 pm

Here we go... This solves the signaling race I found. I think there might be 
another racer that could leak waitsets... Humm, I need to study this issue 
further...


< extreme pseudo-code >


typedef struct ac_waitset_
{
 struct ac_waitset_ *next;
 volatile LONG refs; /* set to 0 */
 volatile LONG dtor; /* set to 0 */
 HANDLE gate; /* manual reset event, val = 0  */

} ac_waitset_t;




typedef struct ac_event_count_
{
 volatile ac_waitset_t *active_waitset;
 DWORD ec;

} ac_event_count_t;




/* smr quiescent period callback */
void ac_waitset_dtor( void *s )
{
 ac_waitset_t *_this = s;

 assert( ! _this->refs && _this->dtor );

 _this-dtor = 0;
 _this->next = 0;

 ResetEvent( _this->gate );

 ac_waitset_cache_push( _this );
}




DWORD ac_event_count_wait
( ac_event_count_t *_this,
  DWORD ec_cmp,
  DWORD timeout )
{
    int dtor = 0;
    DWORD wait_ret = WAIT_OBJECT_0;
    register ac_event_count_t cmp, xchg;
    regitser ac_waitset_t *local_waitset = 0;
    void **hazard = ac_lfgc_get_hazard( 1 );

    cmp.ec = _this->ec;
    cmp.active_waitset = _this->active_waitset;

    do
    {
        xchg.cmp = cmp.ec;

        if ( ! cmp.active_waitset &&
               cmp.ec == ec_cmp )
        {
            if ( ! local_waitset )
            {
                local_waitset =
                    ac_waitset_cache_pop
                        ( &global_cache );
            }

            xchg.active_waitset = local_waitset;
        }

        else
        {
            xchg.active_waitset = cmp.active_waitset;
        }

        *hazard = xchg.active_waitset;
    }

    while ( np_ac_atomic_dwcas_acq_dep
                ( _this,
                  &cmp,
                  &xchg ) );

    if ( local_waitset &&
         xchg.active_waitset != local_waitset )
    {
        ac_waitset_cache_push( local_waitset );
    }

    if ( xchg.cmp == ec_cmp )
    {
        InterlockedIncrement
            ( &xchg.active_waitset->refs );

        wait_ret = WaitForSingleObject
                    ( &xchg.active_waitset->queue );

        if ( ! InterlockedDecrement
                ( &xchg->active_waitset->refs ) )
        {
            if ( ! InterlockedExchange
                    ( &xchg->active_waitset->dtor,
                      1 ) )
            {
                xchg.active_waitset = 0;

                np_ac_atomic_dwcas_rel
                    ( _this,
                      &cmp,
                      &xchg );

                dtor = 1;
            }
        }

        *hazard = 0;

        if ( dtor )
        {
            ac_lfgc_collect
                ( xchg->active_waitset,
                  ac_waitset_dtor );
        }
    }

    return wait_ret;
}




VOID ac_event_count_inc
( ac_event_count_t *_this )
{
    register ac_event_count_t cmp, xchg;
    void **hazard = ac_lfgc_get_hazard( 1 );

    cmp.ec = _this->ec;
    cmp.active_waitset = _this->active_waitset;

    do
    {
        if ( ! ( cmp.ec & 0x80000000 ) )
        {
            *hazard = 0;
            return;
        }

        xchg.cmp = ( cmp.ec & ~0x80000000 ) + 1;
        xchg.active_waitset = 0;

        *hazard = cmp.active_waitset;
    }

    while ( np_ac_atomic_dwcas_rel
                ( _this,
                  &cmp,
                  &xchg ) );

    if ( cmp.active_waitset )
    {
        SetEvent( cmp.active_waitset->queue );

        *hazard = 0;
    }
}




DWORD ac_event_count_get
( ac_event_count_t *_this )
{
    register ac_event_count_t cmp, xchg;

    cmp.ec = _this->ec;
    cmp.active_waitset = _this->active_waitset;

    do
    {
        xchg.ec = cmp.ec & 0x80000000;
        xchg.active_waitset = cmp.active_waitset;
    }

    while ( np_ac_atomic_dwcas_rel
                ( _this,
                  &cmp,
                  &xchg ) );

    if ( cmp.active_waitset )
    {
        SetEvent( cmp.active_waitset->queue );

        *hazard = 0;
    }
}

.