source: trunk/kernel/libk/remote_sem.c @ 582

/*
 * remote_sem.c - POSIX unnamed semaphore implementation.
 * 
 * Author   Alain Greiner  (2016,2017,2018)
 *
 * Copyright (c) UPMC Sorbonne Universites
 *
 * This file is part of ALMOS-MKH.
 *
 * ALMOS-MKH is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2.0 of the License.
 *
 * ALMOS-MKH is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with ALMOS-MKH; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include <hal_kernel_types.h>
#include <hal_remote.h>
#include <thread.h>
#include <kmem.h>
#include <printk.h>
#include <process.h>
#include <vmm.h>
#include <remote_sem.h>


///////////////////////////////////////////////
xptr_t remote_sem_from_ident( intptr_t  ident )
{
    // get pointer on local process_descriptor
    process_t * process = CURRENT_THREAD->process;

    // get extended pointer on reference process
    xptr_t      ref_xp = process->ref_xp;

    // get cluster and local pointer on reference process 
    cxy_t          ref_cxy = GET_CXY( ref_xp );
    process_t    * ref_ptr = GET_PTR( ref_xp );

    // get extended pointer on semaphores list 
    xptr_t root_xp = XPTR( ref_cxy , &ref_ptr->sem_root );
    xptr_t lock_xp = XPTR( ref_cxy , &ref_ptr->sync_lock );
   
    // get lock protecting synchro lists
    remote_queuelock_acquire( lock_xp );
 
    // scan reference process semaphores list
    xptr_t         iter_xp;
    xptr_t         sem_xp;
    cxy_t          sem_cxy;
    remote_sem_t * sem_ptr;
    intptr_t       current;
    bool_t         found = false;
            
    XLIST_FOREACH( root_xp , iter_xp )
    {
        sem_xp  = XLIST_ELEMENT( iter_xp , remote_sem_t , list );
        sem_cxy = GET_CXY( sem_xp );
        sem_ptr = GET_PTR( sem_xp );
        current = (intptr_t)hal_remote_lpt( XPTR( sem_cxy , &sem_ptr->ident ) );   

        if( current == ident )
        {
            found = true;
            break;
        }
    }

    // relese lock protecting synchros lists
    remote_queuelock_release( lock_xp );
 
    if( found == false )  return XPTR_NULL;
    else                  return sem_xp;

}  // end remote_sem_from_ident()

///////////////////////////////////////////
error_t remote_sem_create( intptr_t   vaddr,
                           uint32_t   value )
{
    remote_sem_t * sem_ptr;
    xptr_t         sem_xp;

    // get pointer on local process descriptor
    process_t * process = CURRENT_THREAD->process;

    // get extended pointer on reference process
    xptr_t      ref_xp = process->ref_xp;

    // get reference process cluster and local pointer
    cxy_t       ref_cxy = GET_CXY( ref_xp );
    process_t * ref_ptr = (process_t *)GET_PTR( ref_xp );

    // allocate memory for new semaphore in reference cluster
    if( ref_cxy == local_cxy )  // local cluster is the reference 
    {
        kmem_req_t req;   
        req.type  = KMEM_SEM;
        req.flags = AF_ZERO;
        sem_ptr   = kmem_alloc( &req );
        sem_xp    = XPTR( local_cxy , sem_ptr );
    }
    else                         // reference is remote
    {
        rpc_kcm_alloc_client( ref_cxy , KMEM_SEM , &sem_xp );
        sem_ptr = GET_PTR( sem_xp );
    }

    if( sem_xp == XPTR_NULL ) return 0xFFFFFFFF;

    // initialise semaphore 
    hal_remote_s32 ( XPTR( ref_cxy , &sem_ptr->count ) , value );
        hal_remote_spt( XPTR( ref_cxy , &sem_ptr->ident ) , (void *)vaddr );
        xlist_root_init( XPTR( ref_cxy , &sem_ptr->root ) );
        xlist_entry_init( XPTR( ref_cxy , &sem_ptr->list ) );
    remote_busylock_init( XPTR( ref_cxy , &sem_ptr->lock ), LOCK_SEM_STATE );

    xptr_t root_xp = XPTR( ref_cxy , &ref_ptr->sem_root );
    xptr_t list_xp = XPTR( ref_cxy , &sem_ptr->list );

    // get lock protecting user synchro lists
    remote_queuelock_acquire( XPTR( ref_cxy , &ref_ptr->sync_lock ) );

    // register semaphore in reference process list of semaphores
    xlist_add_first( root_xp , list_xp );

    // release lock protecting user synchro lists
    remote_queuelock_release( XPTR( ref_cxy , &ref_ptr->sync_lock ) );

#if DEBUG_SEM
thread_t * this = CURRENT_THREAD;
if( (uint32_t)hal_get_cycles() > DEBUG_SEM )
printk("\n[DBG] %s : thread %x in process %x INITIALIZE sem(%x,%x) / value %d\n",
__FUNCTION__, this->trdid, this->process->pid, local_cxy, sem_ptr, value );
#endif

    return 0;

}  // end remote_sem_create()
 
////////////////////////////////////////
void remote_sem_destroy( xptr_t sem_xp )
{
    // get pointer on local process descriptor
    process_t * process = CURRENT_THREAD->process;

    // get extended pointer on reference process
    xptr_t      ref_xp = process->ref_xp;

    // get reference process cluster and local pointer
    cxy_t       ref_cxy = GET_CXY( ref_xp );
    process_t * ref_ptr = GET_PTR( ref_xp );

    // get semaphore cluster and local pointer
    cxy_t          sem_cxy = GET_CXY( sem_xp );
    remote_sem_t * sem_ptr = GET_PTR( sem_xp );

    // get remote pointer on waiting queue root
    xptr_t root_xp = XPTR( sem_cxy , &sem_ptr->root );
  
    if( !xlist_is_empty( root_xp ) )   // user error
    {
        printk("WARNING in %s for thread %x in process %x : "
               "destroy semaphore, but  waiting threads queue not empty\n", 
               __FUNCTION__ , CURRENT_THREAD->trdid , CURRENT_THREAD->process->pid );
    }

    // remove semaphore from reference process xlist
    remote_queuelock_acquire( XPTR( ref_cxy , &ref_ptr->sync_lock ) );
    xlist_unlink( XPTR( sem_cxy , &sem_ptr->list ) );
    remote_queuelock_release( XPTR( ref_cxy , &ref_ptr->sync_lock ) );

    // release memory allocated for semaphore descriptor
    if( sem_cxy == local_cxy )                            // reference is local
    {
        kmem_req_t  req;
        req.type = KMEM_SEM;
        req.ptr  = sem_ptr;
        kmem_free( &req );
    }
    else                                                  // reference is remote
    {
        rpc_kcm_free_client( sem_cxy , sem_ptr , KMEM_SEM );
    }

}  // end remote_sem_destroy()

/////////////////////////////////////
void remote_sem_wait( xptr_t sem_xp )
{ 
    thread_t * this = CURRENT_THREAD;

// check calling thread can yield
assert( (this->busylocks == 0),
"cannot yield : busylocks = %d\n", this->busylocks );


    // get semaphore cluster and local pointer
    cxy_t          sem_cxy = GET_CXY( sem_xp );
    remote_sem_t * sem_ptr = GET_PTR( sem_xp );

    // get extended pointers on sem fields 
    xptr_t           count_xp = XPTR( sem_cxy , &sem_ptr->count );
    xptr_t           root_xp  = XPTR( sem_cxy , &sem_ptr->root );
    xptr_t           lock_xp  = XPTR( sem_cxy , &sem_ptr->lock );

    while( 1 )
    {
        // get busylock protecting semaphore     
            remote_busylock_acquire( lock_xp );
  
        // get semaphore current value
        uint32_t count = hal_remote_l32( count_xp );

            if( count > 0 )                     // success
            {
            // decrement semaphore value
            hal_remote_s32( count_xp , count - 1 );

#if DEBUG_SEM
if( (uint32_t)hal_get_cycles() > DEBUG_SEM )
printk("\n[DBG] %s : thread %x in process %x DECREMENT sem(%x,%x) / value %d\n",
__FUNCTION__, this->trdid, this->process->pid, sem_cxy, sem_ptr, count-1 );
#endif
            // release busylock protecting semaphore
                remote_busylock_release( XPTR( sem_cxy , &sem_ptr->lock ) );

            return;
        }
            else                               // failure
            {
            // get cluster and pointers on calling thread
            cxy_t            caller_cxy = local_cxy;
            thread_t       * caller_ptr = CURRENT_THREAD;
            xptr_t           caller_xp  = XPTR( caller_cxy , caller_ptr );

            // block the calling thread
            thread_block( caller_xp , THREAD_BLOCKED_SEM );  

            // register calling thread in waiting queue
            xptr_t entry_xp = XPTR( caller_cxy , &caller_ptr->wait_xlist );
                    xlist_add_last( root_xp , entry_xp );

#if DEBUG_SEM
if( (uint32_t)hal_get_cycles() > DEBUG_SEM )
printk("\n[DBG] %s : thread %x in process %x BLOCK on sem(%x,%x) / value %d\n",
__FUNCTION__, this->trdid, this->process->pid, sem_cxy, sem_ptr, count );
#endif
            // release busylock protecting semaphore
                remote_busylock_release( XPTR( sem_cxy , &sem_ptr->lock ) );

            // deschedule calling thread
            sched_yield("blocked on semaphore");
        }
        }
}  // end remote_sem_wait()

/////////////////////////////////////
void remote_sem_post( xptr_t sem_xp )
{
    // memory barrier before sem release
    hal_fence();

    // get semaphore cluster and local pointer
    cxy_t          sem_cxy = GET_CXY( sem_xp );
    remote_sem_t * sem_ptr = GET_PTR( sem_xp );

    // get extended pointers on sem fields 
    xptr_t           count_xp = XPTR( sem_cxy , &sem_ptr->count );
    xptr_t           root_xp  = XPTR( sem_cxy , &sem_ptr->root );
    xptr_t           lock_xp  = XPTR( sem_cxy , &sem_ptr->lock );

    // get busylock protecting semaphore
        remote_busylock_acquire( lock_xp );
  
    // increment semaphore value
    hal_remote_atomic_add( count_xp , 1 );

#if DEBUG_SEM
uint32_t count = hal_remote_l32( count_xp );
thread_t * this = CURRENT_THREAD;
if( (uint32_t)hal_get_cycles() > DEBUG_SEM )
printk("\n[DBG] %s : thread %x in process %x INCREMENT sem(%x,%x) / value %d\n",
__FUNCTION__, this->trdid, this->process->pid, sem_cxy, sem_ptr, count );
#endif

    // scan waiting queue to unblock all waiting threads
        while( xlist_is_empty( root_xp ) == false )   // waiting queue non empty
    {
        // get first waiting thread from queue
        xptr_t     thread_xp  = XLIST_FIRST( root_xp , thread_t , wait_xlist );
        cxy_t      thread_cxy = GET_CXY( thread_xp );
        thread_t * thread_ptr = GET_PTR( thread_xp );

        // remove this thread from the waiting queue
        xlist_unlink( XPTR( thread_cxy , &thread_ptr->wait_xlist ) );

        // unblock this waiting thread
                thread_unblock( thread_xp , THREAD_BLOCKED_SEM );

#if DEBUG_SEM
if( (uint32_t)hal_get_cycles() > DEBUG_SEM )
{
trdid_t     trdid   = hal_remote_l32( XPTR( thread_cxy , &thread_ptr->trdid ) );
process_t * process = hal_remote_lpt( XPTR( thread_cxy , &thread_ptr->process ) );
pid_t       pid     = hal_remote_l32( XPTR( thread_cxy , &process->pid ) );
printk("\n[DBG] %s : thread %x in process %x UNBLOCK thread %x in process %x / sem(%x,%x)\n",
__FUNCTION__, this->trdid, this->process->pid, trdid, pid, sem_cxy, sem_ptr );
}
#endif

    }

    // release busylock protecting the semaphore
        remote_busylock_release( XPTR( sem_cxy , &sem_ptr->lock ) );

}  // end remote_sem_post()


//////////////////////////////////////////////
void remote_sem_get_value( xptr_t      sem_xp,
                           uint32_t  * data )
{
    // get semaphore cluster and local pointer
    cxy_t          sem_cxy = GET_CXY( sem_xp );
    remote_sem_t * sem_ptr = GET_PTR( sem_xp );

    *data = hal_remote_l32( XPTR( sem_cxy , &sem_ptr->count ) );

}  // end remote_sem_get_value()