source: trunk/kernel/kern/process.c @ 415

/*
 * process.c - process related management
 *
 * Authors  Ghassan Almaless (2008,2009,2010,2011,2012)
 *          Mohamed Lamine Karaoui (2015)
 *          Alain Greiner (2016,2017)
 *
 * 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 <kernel_config.h>
#include <hal_types.h>
#include <hal_remote.h>
#include <hal_uspace.h>
#include <hal_irqmask.h>
#include <errno.h>
#include <printk.h>
#include <memcpy.h>
#include <bits.h>
#include <kmem.h>
#include <page.h>
#include <vmm.h>
#include <vfs.h>
#include <core.h>
#include <thread.h>
#include <list.h>
#include <string.h>
#include <scheduler.h>
#include <remote_spinlock.h>
#include <dqdt.h>
#include <cluster.h>
#include <ppm.h>
#include <boot_info.h>
#include <process.h>
#include <elf.h>
#include <syscalls.h>
#include <signal.h>

//////////////////////////////////////////////////////////////////////////////////////////
// Extern global variables
//////////////////////////////////////////////////////////////////////////////////////////

extern process_t process_zero;

//////////////////////////////////////////////////////////////////////////////////////////
// Process initialisation related functions
//////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////
process_t * process_alloc()
{
        kmem_req_t   req;

    req.type  = KMEM_PROCESS;
        req.size  = sizeof(process_t);
        req.flags = AF_KERNEL;

    return (process_t *)kmem_alloc( &req );
}

////////////////////////////////////////
void process_free( process_t * process )
{
    kmem_req_t  req;

        req.type = KMEM_PROCESS;
        req.ptr  = process;
        kmem_free( &req );
}

/////////////////////////////////////////////
void process_zero_init( process_t * process )
{
    // initialize PID, PPID anf PREF
    process->pid    = 0;
    process->ppid   = 0;
    process->ref_xp = XPTR( local_cxy , process );

    // reset th_tbl[] array as empty
    uint32_t i;
    for( i = 0 ; i < CONFIG_THREAD_MAX_PER_CLUSTER ; i++ )
        {
        process->th_tbl[i] = NULL;
    }
    process->th_nr  = 0;
    spinlock_init( &process->th_lock );

        hal_fence();

process_dmsg("\n[DBG] %s : core[%x,%d] exit for process %x\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , process->pid );

}  // end process_zero_init()

/////////////////////////////////////////////////
void process_reference_init( process_t * process,
                             pid_t       pid,
                             pid_t       ppid,
                             xptr_t      model_xp )
{
    cxy_t       model_cxy;
    process_t * model_ptr;
        error_t     error1;
        error_t     error2;
        error_t     error3;
    xptr_t      stdin_xp;
    xptr_t      stdout_xp;
    xptr_t      stderr_xp;
    uint32_t    stdin_id;
    uint32_t    stdout_id;
    uint32_t    stderr_id;
    error_t     error;

process_dmsg("\n[DBG] %s : core[%x,%d] enters for process %x / ppid = %x\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , pid , ppid );

    // get model process cluster and local pointer
    model_cxy = GET_CXY( model_xp );
    model_ptr = (process_t *)GET_PTR( model_xp );

    // initialize PID, PPID, and REF
        process->pid    = pid;
    process->ppid   = ppid;
    process->ref_xp = XPTR( local_cxy , process );

    // initialize vmm as empty 
    error = vmm_init( process );
    assert( (error == 0) , __FUNCTION__ , "cannot initialize VMM\n" );
 

process_dmsg("\n[DBG] %s : core[%x,%d] / vmm empty for process %x\n", 
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , pid );

    // initialize fd_array as empty
    process_fd_init( process );

    // create stdin / stdout / stderr pseudo-files
    if( ppid == 0 )                                       // process_init
    {
        error1 = vfs_open( process,
                           CONFIG_INIT_STDIN,
                           O_RDONLY, 
                           0,                // FIXME chmod
                           &stdin_xp, 
                           &stdin_id );

        error2 = vfs_open( process,
                           CONFIG_INIT_STDOUT,
                           O_WRONLY, 
                           0,                // FIXME chmod
                           &stdout_xp, 
                           &stdout_id );

        error3 = vfs_open( process,
                           CONFIG_INIT_STDERR,
                           O_WRONLY, 
                           0,                // FIXME chmod
                           &stderr_xp, 
                           &stderr_id );
    }
    else                                                  // any other process
    {
        error1 = vfs_open( process,
                           CONFIG_USER_STDIN,
                           O_RDONLY, 
                           0,                // FIXME chmod
                           &stdin_xp, 
                           &stdin_id );

        error2 = vfs_open( process,
                           CONFIG_USER_STDOUT,
                           O_WRONLY, 
                           0,                // FIXME chmod
                           &stdout_xp, 
                           &stdout_id );

        error3 = vfs_open( process,
                           CONFIG_USER_STDERR,
                           O_WRONLY, 
                           0,                // FIXME chmod
                           &stderr_xp, 
                           &stderr_id );
    }

    assert( ((error1 == 0) && (error2 == 0) && (error3 == 0)) , __FUNCTION__ ,
    "cannot open stdin/stdout/stderr pseudo files\n");

    assert( ((stdin_id == 0) && (stdout_id == 1) && (stderr_id == 2)) , __FUNCTION__ ,
    "bad indexes : stdin %d / stdout %d / stderr %d \n", stdin_id , stdout_id , stderr_id );

    // initialize specific inodes root and cwd
    process->vfs_root_xp = (xptr_t)hal_remote_lwd( XPTR( model_cxy,
                                                         &model_ptr->vfs_root_xp ) );
    process->vfs_cwd_xp  = (xptr_t)hal_remote_lwd( XPTR( model_cxy,
                                                         &model_ptr->vfs_cwd_xp ) );
    vfs_inode_remote_up( process->vfs_root_xp );
    vfs_inode_remote_up( process->vfs_cwd_xp );

    remote_rwlock_init( XPTR( local_cxy , &process->cwd_lock ) );

    // copy all open file descriptors (other than stdin / stdout / stderr) 
    process_fd_remote_copy( XPTR( local_cxy , &process->fd_array ),
                            XPTR( model_cxy , &model_ptr->fd_array ) );

process_dmsg("\n[DBG] %s : core[%x,%d] / fd array for process %x\n", 
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , pid );

    // reset children list root 
    xlist_root_init( XPTR( local_cxy , &process->children_root ) );
    process->children_nr     = 0;

    // reset semaphore / mutex / barrier / condvar list roots 
    xlist_root_init( XPTR( local_cxy , &process->sem_root ) );
    xlist_root_init( XPTR( local_cxy , &process->mutex_root ) );
    xlist_root_init( XPTR( local_cxy , &process->barrier_root ) );
    xlist_root_init( XPTR( local_cxy , &process->condvar_root ) );
    remote_spinlock_init( XPTR( local_cxy , &process->sync_lock ) );

    // register new process in the local cluster manager pref_tbl[]
    lpid_t lpid = LPID_FROM_PID( pid );
    LOCAL_CLUSTER->pmgr.pref_tbl[lpid] = XPTR( local_cxy , process );

    // register new process descriptor in local cluster manager local_list
    cluster_process_local_link( process );

    // register new process descriptor in local cluster manager copies_list 
    cluster_process_copies_link( process );

    // reset th_tbl[] array as empty in process descriptor
    uint32_t i;
    for( i = 0 ; i < CONFIG_THREAD_MAX_PER_CLUSTER ; i++ )
        {
        process->th_tbl[i] = NULL;
    }
    process->th_nr  = 0;
    spinlock_init( &process->th_lock );

        hal_fence();

process_dmsg("\n[DBG] %s : core[%x,%d] exit for process %x\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , pid );

}  // process_reference init()

/////////////////////////////////////////////////////
error_t process_copy_init( process_t * local_process,
                           xptr_t      reference_process_xp )
{
    error_t error;

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

    // set the pid, ppid, ref_xp fields in local process
    local_process->pid    = hal_remote_lw( XPTR( ref_cxy , &ref_ptr->pid ) );
    local_process->ppid   = hal_remote_lw( XPTR( ref_cxy , &ref_ptr->ppid ) );
    local_process->ref_xp = reference_process_xp;

process_dmsg("\n[DBG] %s : core[%x,%d] enters for process %x in cluster %x\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , local_process->pid );

    // reset local process vmm
    error = vmm_init( local_process );
    assert( (error == 0) , __FUNCTION__ , "cannot initialize VMM\n");

    // reset process file descriptors array
        process_fd_init( local_process );

    // reset vfs_root_xp / vfs_bin_xp / vfs_cwd_xp fields
    local_process->vfs_root_xp = hal_remote_lwd( XPTR( ref_cxy , &ref_ptr->vfs_root_xp ) );
    local_process->vfs_bin_xp  = hal_remote_lwd( XPTR( ref_cxy , &ref_ptr->vfs_bin_xp ) );
    local_process->vfs_cwd_xp  = XPTR_NULL;

    // reset children list root (not used in a process descriptor copy)
    xlist_root_init( XPTR( local_cxy , &local_process->children_root ) );
    local_process->children_nr   = 0;

    // reset brothers list (not used in a process descriptor copy)
    xlist_entry_init( XPTR( local_cxy , &local_process->brothers_list ) );

    // reset semaphores list root (not used in a process descriptor copy)
    xlist_root_init( XPTR( local_cxy , &local_process->sem_root ) );
    xlist_root_init( XPTR( local_cxy , &local_process->mutex_root ) );
    xlist_root_init( XPTR( local_cxy , &local_process->barrier_root ) );
    xlist_root_init( XPTR( local_cxy , &local_process->condvar_root ) );

    // reset th_tbl[] array as empty
    uint32_t i;
    for( i = 0 ; i < CONFIG_THREAD_MAX_PER_CLUSTER ; i++ )
        {
        local_process->th_tbl[i] = NULL;
    }
    local_process->th_nr  = 0;
    spinlock_init( &local_process->th_lock );

    // register new process descriptor in local cluster manager local_list
    cluster_process_local_link( local_process );

    // register new process descriptor in owner cluster manager copies_list
    cluster_process_copies_link( local_process );

        hal_fence();

process_dmsg("\n[DBG] %s : core[%x,%d] exit for process %x in cluster %x\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , local_process->pid );

    return 0;

} // end process_copy_init()

///////////////////////////////////////////
void process_destroy( process_t * process )
{
        if( process->th_nr != 0 )
    {
        panic("process %x in cluster %x has still active threads",
              process->pid , local_cxy );
    }

    // get local process manager pointer
    pmgr_t * pmgr = &LOCAL_CLUSTER->pmgr;

    // get the lock protecting the list of local process descriptors
    remote_spinlock_lock( XPTR( local_cxy , &pmgr->local_lock ) );

    // remove the process descriptor from local_list in local cluster manager
    xlist_unlink( XPTR( local_cxy , &process->local_list ) );

    // release the lock protecting the list of local process descriptors
    remote_spinlock_unlock( XPTR( local_cxy , &pmgr->local_lock ) );

    // get extended pointer on copies_lock in owner cluster manager
    cxy_t  owner_cxy    = CXY_FROM_PID( process->pid );
        lpid_t lpid         = LPID_FROM_PID( process->pid );
    xptr_t copies_lock  = XPTR( owner_cxy , &pmgr->copies_lock[lpid] );

    // remove the local process descriptor from copies_list
    remote_spinlock_lock( copies_lock );
    xlist_unlink( XPTR( local_cxy , &process->copies_list ) );
    remote_spinlock_unlock( copies_lock );

        hal_fence();

    // From this point, the process descriptor is unreachable

    // FIXME close all open files and update dirty [AG]

    // Decrease refcount for bin file, root file and cwd file
        if( process->vfs_bin_xp  != XPTR_NULL ) vfs_file_count_down( process->vfs_bin_xp );
        if( process->vfs_root_xp != XPTR_NULL ) vfs_file_count_down( process->vfs_root_xp );
        if( process->vfs_cwd_xp  != XPTR_NULL ) vfs_file_count_down( process->vfs_cwd_xp );

    // Destroy VMM
    vmm_destroy( process );

        process_dmsg("\n[DBG] %s for pid %d / page_faults = %d\n",
                 __FUNCTION__ , process->pid, process->vmm.pgfault_nr );

}  // end process_destroy()

/////////////////////////////////////////////////
char * process_action_str( uint32_t action_type )
{
    if     ( action_type == BLOCK_ALL_THREADS   ) return "BLOCK";
    else if( action_type == UNBLOCK_ALL_THREADS ) return "UNBLOCK";
    else if( action_type == DELETE_ALL_THREADS  ) return "DELETE";
    else                                          return "undefined";
}

////////////////////////////////////////////
void process_sigaction( process_t * process,
                        uint32_t    action_type )
{
    cxy_t              owner_cxy;         // owner cluster identifier
    lpid_t             lpid;              // process index in owner cluster
    cluster_t        * cluster;           // pointer on cluster manager
    xptr_t             root_xp;           // extended pointer on root of copies
    xptr_t             lock_xp;           // extended pointer on lock protecting copies
    xptr_t             client_xp;         // extended pointer on client thread
    uint32_t           rsp_count;         // number of expected responses
    xptr_t             rsp_xp;            // extended pointer on responses counter
    xptr_t             iter_xp;           // iterator on copies list
    xptr_t             process_xp;        // extended pointer on process copy
    cxy_t              process_cxy;       // process copy cluster identifier
    process_t        * process_ptr;       // local pointer on process copy

sigaction_dmsg("\n[DBG] %s : enter for signal %s to process %x in cluster %x\n",
__FUNCTION__ , process_action_str( action_type ) , process , local_cxy );

    thread_t         * this = CURRENT_THREAD;

    // get extended pointer on client thread and response counter
    client_xp = XPTR( local_cxy , this );
    rsp_xp    = XPTR( local_cxy , &rsp_count );

    // get owner cluster identifier and process lpid
    owner_cxy = CXY_FROM_PID( process->pid );
    lpid      = LPID_FROM_PID( process->pid );

    assert( (owner_cxy == local_cxy) , __FUNCTION__ , "illegal cluster\n" ); 
   
    // get local pointer on local cluster manager
    cluster = LOCAL_CLUSTER;

    // get extended pointers on copies root, copies lock, and number of copies
    root_xp   = XPTR( local_cxy , &cluster->pmgr.copies_root[lpid] );
    lock_xp   = XPTR( local_cxy , &cluster->pmgr.copies_lock[lpid] );

    // initialize responses number 
    rsp_count = cluster->pmgr.copies_nr[lpid];

    // take the lock protecting the copies
    remote_spinlock_lock( lock_xp );

    // send RPCs to all process copies
    XLIST_FOREACH( root_xp , iter_xp )
    {
        process_xp  = XLIST_ELEMENT( iter_xp , process_t , copies_list );
        process_cxy = GET_CXY( process_xp );
        process_ptr = (process_t *)GET_PTR( process_xp );

sigaction_dmsg("\n[DBG] %s : process = %x / pid = %x / ppid = %x\n",
__FUNCTION__ , process_ptr , process_ptr->pid , process_ptr->ppid );

        rpc_process_sigaction_client( process_cxy, 
                                      process_ptr,
                                      action_type,
                                      rsp_xp,
                                      client_xp );
    }
   
    // release the lock protecting process copies
    remote_spinlock_unlock( lock_xp );

    // block and deschedule to wait response
    thread_block( CURRENT_THREAD , THREAD_BLOCKED_RPC );
    sched_yield("BLOCKED on RPC");

sigaction_dmsg("\n[DBG] %s : exit for signal %s to process %x in cluster %x\n",
__FUNCTION__ , process_action_str( action_type ) , process , local_cxy );

}  // end process_sigaction()

////////////////////////////////////////
void process_block( process_t * process,
                    xptr_t      rsp_xp,
                    xptr_t      client_xp )
{
    thread_t          * target;         // pointer on target thread 
    uint32_t            ltid;           // index in process th_tbl
    thread_t          * killer;         // killer thread pointer
    uint32_t            count;          // requests counter
    volatile uint32_t   sig_rsp_count;  // responses counter
    cxy_t               client_cxy;     // client thread cluster identifier
    thread_t          * client_ptr;     // client thread pointer
    core_t            * client_core;    // client thread core pointer

    // get local killer thread pointer
    killer = CURRENT_THREAD;

sigaction_dmsg("\n[DBG] %s : enter for process %x in cluster %x\n",
__FUNCTION__ , process->pid , local_cxy );

    // get lock protecting process th_tbl[]
    spinlock_lock( &process->th_lock );

    // initialize local responses counter 
    sig_rsp_count = process->th_nr;

    // loop on process threads to block and deschedule all threads in cluster
    // we use both "ltid" and "count" because it can exist "holes" in th_tbl
    for( ltid = 0 , count = 0 ; count < process->th_nr ; ltid++ )
    {
        target = process->th_tbl[ltid];

        if( target != NULL )             // thread found
        {
            count++;

            // set signal in target thread descriptor
            thread_set_signal( target , (uint32_t *)sig_rsp_count );

            // set the global blocked bit in target thread descriptor.
            thread_block( target , THREAD_BLOCKED_GLOBAL );

            // - if the killer thread and the target thread are not on the same core
            //   we want the scheduler of target thread to acknowlege the signal
            //   to be sure that the target thread is descheduled
            // - if the killer thread and the target thread are on the same core
            //   we simply decrement the response counter.
            if( killer->core->lid != target->core->lid )
            {
                dev_pic_send_ipi( local_cxy , target->core->lid );
            }
            else                                                         
            {
                hal_atomic_add( (void *)&sig_rsp_count , -1 );
            }
        }
    }

    // poll the reponses counter 
    while( 1 )
    {
        // exit loop when all responses received
        if ( sig_rsp_count == 0 ) break;
    
        // wait 1000 cycles before retry
        hal_fixed_delay( 1000 );
    }

    // acknowledge client thread & unblock client thread if last response
    client_cxy  = GET_CXY( client_xp );
    client_ptr  = (thread_t *)GET_PTR( client_xp );
    client_core = (core_t *)hal_remote_lpt( XPTR( client_cxy , &client_ptr->core ) );
    if( hal_remote_atomic_add( rsp_xp , -1 ) == 1 ) 
    {
        thread_unblock( client_xp , THREAD_BLOCKED_RPC);
        dev_pic_send_ipi( client_cxy , client_core->lid );
    }

sigaction_dmsg("\n[DBG] %s : exit for process %x in cluster %x / %d threads blocked\n",
__FUNCTION__ , process->pid , local_cxy , count );

}  // end process_block()

//////////////////////////////////////////
void process_unblock( process_t * process,
                      xptr_t      rsp_xp,
                      xptr_t      client_xp )
{
    thread_t          * target;        // pointer on target thead 
    uint32_t            ltid;          // index in process th_tbl
    thread_t          * killer;        // killer thread pointer
    uint32_t            req_count;     // requests counter
    cxy_t               client_cxy;    // client thread cluster identifier
    thread_t          * client_ptr;    // client thread pointer
    core_t            * client_core;   // client thread core pointer

    // get local killer thread pointer
    killer = CURRENT_THREAD;

sigaction_dmsg("\n[DBG] %s : enter for process %x in cluster %x\n",
__FUNCTION__ , process->pid , local_cxy );

    // get lock protecting process th_tbl[]
    spinlock_lock( &process->th_lock );

    // loop on process threads to unblock all threads in cluster
    // we use both "ltid" and "req_count" because it can exist "holes" in th_tbl
    for( ltid = 0 , req_count = 0 ;
         req_count < process->th_nr ;
         ltid++ )
    {
        target = process->th_tbl[ltid];

        if( target != NULL )             // thread found
        {
            req_count++;

            // reset the global blocked bit in target thread descriptor.
            thread_unblock( XPTR( local_cxy , target ) , THREAD_BLOCKED_GLOBAL );
        }
    }

    // acknowledge client thread & unblock client thread if last response
    client_cxy  = GET_CXY( client_xp );
    client_ptr  = (thread_t *)GET_PTR( client_xp );
    client_core = (core_t *)hal_remote_lpt( XPTR( client_cxy , &client_ptr->core ) );
    if( hal_remote_atomic_add( rsp_xp , -1 ) == 1 ) 
    {
        thread_unblock( client_xp , THREAD_BLOCKED_RPC);
        dev_pic_send_ipi( client_cxy , client_core->lid );
    }

sigaction_dmsg("\n[DBG] %s : exit for process %x in cluster %x / %d threads blocked\n",
__FUNCTION__ , process->pid , local_cxy , req_count );

}  // end process_unblock()

/////////////////////////////////////////
void process_delete( process_t * process,
                     xptr_t      rsp_xp,
                     xptr_t      client_xp )
{
    thread_t          * thread;        // pointer on target thread 
    uint32_t            ltid;          // index in process th_tbl
    uint32_t            count;         // request counter
    pid_t               pid;           // process PID
    cxy_t               client_cxy;    // client thread cluster identifier
    thread_t          * client_ptr;    // client thread pointer
    core_t            * client_core;   // client thread core pointer

    // get process PID
    pid = process->pid;

sigaction_dmsg("\n[DBG] %s : enter for process %x in cluster %x at cycle %d\n",
__FUNCTION__ , pid , local_cxy , (uint32_t)hal_get_cycles() );

    // loop on threads to release memory allocated to threads
    for( ltid = 0 , count = 0  ; count < process->th_nr ; ltid++ )
    {
        thread = process->th_tbl[ltid];

        if( thread != NULL )             // thread found
        {
            count++;

            // detach thread from parent if attached
            if( (thread->flags & THREAD_FLAG_DETACHED) == 0 ) 
            thread_child_parent_unlink( thread->parent , XPTR( local_cxy , thread ) );

            // detach thread from process
            process_remove_thread( thread );

            // remove thread from scheduler
            sched_remove_thread( thread );

            // release memory allocated to thread
            thread_destroy( thread );
        }
    }

    // release memory allocated to process descriptors
    // for all clusters other than the owner cluster 
    if( local_cxy != CXY_FROM_PID( process->pid ) ) process_destroy( process );

    // acknowledge client thread & unblock client thread if last response
    client_cxy  = GET_CXY( client_xp );
    client_ptr  = (thread_t *)GET_PTR( client_xp );
    client_core = (core_t *)hal_remote_lpt( XPTR( client_cxy , &client_ptr->core ) );
    if( hal_remote_atomic_add( rsp_xp , -1 ) == 1 ) 
    {
        thread_unblock( client_xp , THREAD_BLOCKED_RPC);
        dev_pic_send_ipi( client_cxy , client_core->lid );
    }

sigaction_dmsg("\n[DBG] %s : exit for process %x in cluster %x at cycle %d\n",
__FUNCTION__ , pid , local_cxy , (uint32_t)hal_get_cycles() );

}  // end process_delete()

///////////////////////////////////////////////
process_t * process_get_local_copy( pid_t pid )
{
    error_t        error;
    process_t    * process_ptr;   // local pointer on process
    xptr_t         process_xp;    // extended pointer on process

    cluster_t * cluster = LOCAL_CLUSTER;

    // get lock protecting local list of processes
    remote_spinlock_lock( XPTR( local_cxy , &cluster->pmgr.local_lock ) );

    // scan the local list of process descriptors to find the process
    xptr_t  iter;
    bool_t  found = false;
    XLIST_FOREACH( XPTR( local_cxy , &cluster->pmgr.local_root ) , iter )
    {
        process_xp  = XLIST_ELEMENT( iter , process_t , local_list );
        process_ptr = (process_t *)GET_PTR( process_xp );
        if( process_ptr->pid == pid )
        {
            found = true;
            break;
        }
    }

    // release lock protecting local list of processes
    remote_spinlock_unlock( XPTR( local_cxy , &cluster->pmgr.local_lock ) );

    // allocate memory for a new local process descriptor
    // and initialise it from reference cluster if required
    if( !found )
    {
        // get extended pointer on reference process descriptor
        xptr_t ref_xp = cluster_get_reference_process_from_pid( pid );

        assert( (ref_xp != XPTR_NULL) , __FUNCTION__ , "illegal pid\n" );

        // allocate memory for local process descriptor
        process_ptr = process_alloc();
        if( process_ptr == NULL )  return NULL;

        // initialize local process descriptor copy
        error = process_copy_init( process_ptr , ref_xp );
        if( error ) return NULL;
    }

    return process_ptr;

}  // end process_get_local_copy()

//////////////////////////////////////////////////////////////////////////////////////////
// File descriptor array related functions
//////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////
void process_fd_init( process_t * process )
{
    uint32_t fd;

    remote_spinlock_init( XPTR( local_cxy , &process->fd_array.lock ) );

    process->fd_array.current = 0;

    // initialize array
    for ( fd = 0 ; fd < CONFIG_PROCESS_FILE_MAX_NR ; fd++ )
    {
        process->fd_array.array[fd] = XPTR_NULL;
    }
}

//////////////////////////////
bool_t process_fd_array_full()
{
    // get extended pointer on reference process
    xptr_t ref_xp = CURRENT_THREAD->process->ref_xp;

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

    // get number of open file descriptors from reference fd_array
    uint32_t current = hal_remote_lw( XPTR( ref_cxy , &ref_ptr->fd_array.current ) );

        return ( current >= CONFIG_PROCESS_FILE_MAX_NR );
}

/////////////////////////////////////////////////
error_t process_fd_register( process_t * process,
                             xptr_t      file_xp,
                             uint32_t  * fdid )
{
    bool_t    found;
    uint32_t  id;
    xptr_t    xp;

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

    // take lock protecting reference fd_array
        remote_spinlock_lock( XPTR( ref_cxy , &ref_ptr->fd_array.lock ) );

    found   = false;

    for ( id = 0; id < CONFIG_PROCESS_FILE_MAX_NR ; id++ )
    {
        xp = hal_remote_lwd( XPTR( ref_cxy , &ref_ptr->fd_array.array[id] ) );
        if ( xp == XPTR_NULL )
        {
            found = true;
            hal_remote_swd( XPTR( ref_cxy , &ref_ptr->fd_array.array[id] ) , file_xp );
                hal_remote_atomic_add( XPTR( ref_cxy , &ref_ptr->fd_array.current ) , 1 );
                        *fdid = id;
            break;
        }
    }

    // release lock protecting reference fd_array
        remote_spinlock_unlock( XPTR( ref_cxy , &ref_ptr->fd_array.lock ) );

    if ( !found ) return EMFILE;
    else          return 0;
}

////////////////////////////////////////////////
xptr_t process_fd_get_xptr( process_t * process,
                            uint32_t    fdid )
{
    xptr_t  file_xp;

    // access local copy of process descriptor
    file_xp = process->fd_array.array[fdid];

    if( file_xp == XPTR_NULL )
    {
        // get reference process cluster and local pointer
        xptr_t      ref_xp  = process->ref_xp;
        cxy_t       ref_cxy = GET_CXY( ref_xp );
        process_t * ref_ptr = (process_t *)GET_PTR( ref_xp );

        // access reference process descriptor
        file_xp = hal_remote_lwd( XPTR( ref_cxy , &ref_ptr->fd_array.array[fdid] ) );

        // update local fd_array if found
        if( file_xp != XPTR_NULL )
        {
            process->fd_array.array[fdid] = file_xp;
        }
    }

    return file_xp;

}  // end process_fd_get_xptr()

///////////////////////////////////////////
void process_fd_remote_copy( xptr_t dst_xp,
                             xptr_t src_xp )
{
    uint32_t fd;
    xptr_t   entry;

    // get cluster and local pointer for src fd_array
    cxy_t        src_cxy = GET_CXY( src_xp );
    fd_array_t * src_ptr = (fd_array_t *)GET_PTR( src_xp );

    // get cluster and local pointer for dst fd_array
    cxy_t        dst_cxy = GET_CXY( dst_xp );
    fd_array_t * dst_ptr = (fd_array_t *)GET_PTR( dst_xp );

    // get the remote lock protecting the src fd_array
        remote_spinlock_lock( XPTR( src_cxy , &src_ptr->lock ) );

    // loop on all entries other than
    // the three first entries: stdin/stdout/stderr
    for( fd = 3 ; fd < CONFIG_PROCESS_FILE_MAX_NR ; fd++ )
        {
                entry = (xptr_t)hal_remote_lwd( XPTR( src_cxy , &src_ptr->array[fd] ) );

                if( entry != XPTR_NULL )
                {
            // increment file descriptor ref count
            vfs_file_count_up( entry );

                        // copy entry in destination process fd_array
                        hal_remote_swd( XPTR( dst_cxy , &dst_ptr->array[fd] ) , entry );
                }
        }

    // release lock on source process fd_array
        remote_spinlock_unlock( XPTR( src_cxy , &src_ptr->lock ) );

}  // end process_fd_remote_copy()

////////////////////////////////////////////////////////////////////////////////////
//  Thread related functions
////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////
error_t process_register_thread( process_t * process,
                                 thread_t  * thread,
                                 trdid_t   * trdid )
{
    ltid_t   ltid;
    bool_t   found;

    assert( (process != NULL) , __FUNCTION__ , "process argument is NULL" );

    assert( (thread != NULL) , __FUNCTION__ , "thread argument is NULL" );

    // search a free slot in th_tbl[] 
    // 0 is not a valid ltid value
    found = false;
    for( ltid = 1 ; ltid < CONFIG_THREAD_MAX_PER_CLUSTER ; ltid++ )
    {
        if( process->th_tbl[ltid] == NULL )
        {
            found = true;
            break;
        }
    }

    if( found )
    {
        // register thread in th_tbl[]
        process->th_tbl[ltid] = thread;
        process->th_nr++;

        // returns trdid
        *trdid = TRDID( local_cxy , ltid );
    }

    return (found) ? 0 : ENOMEM;

}  // end process_register_thread()

///////////////////////////////////////////////
void process_remove_thread( thread_t * thread )
{
    assert( (thread != NULL) , __FUNCTION__ , "thread argument is NULL" );

    process_t * process = thread->process;

    // get thread local index
    ltid_t  ltid = LTID_FROM_TRDID( thread->trdid );

    // remove thread from th_tbl[]
    process->th_tbl[ltid] = NULL;
    process->th_nr--;

}  // process_remove_thread()

/////////////////////////////////////////////////////////
error_t process_make_fork( xptr_t      parent_process_xp,
                           xptr_t      parent_thread_xp,
                           pid_t     * child_pid,
                           thread_t ** child_thread )
{
    process_t * process;         // local pointer on child process descriptor
    thread_t  * thread;          // local pointer on child thread descriptor
    pid_t       new_pid;         // process identifier for child process
    pid_t       parent_pid;      // process identifier for parent process
    xptr_t      ref_xp;          // extended pointer on reference process
    error_t     error;

    // get cluster and local pointer for parent process
    cxy_t       parent_process_cxy = GET_CXY( parent_process_xp );
    process_t * parent_process_ptr = (process_t *)GET_PTR( parent_process_xp );

    // get parent process PID
    parent_pid = hal_remote_lw( XPTR( parent_process_cxy , &parent_process_ptr->pid ) );
    
    // check parent process is the reference
    ref_xp = hal_remote_lwd( XPTR( parent_process_cxy , &parent_process_ptr->ref_xp ) );
    assert( (parent_process_xp == ref_xp ) , __FUNCTION__ ,
    "parent process must be the reference process\n" );

fork_dmsg("\n[DBG] %s : core[%x,%d] enter at cycle %d\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid , (uint32_t)hal_get_cycles() );

    // allocate a process descriptor
    process = process_alloc();
    if( process == NULL )
    {
        printk("\n[ERROR] in %s : cannot get process in cluster %x\n", 
        __FUNCTION__, local_cxy ); 
        return -1;
    }

fork_dmsg("\n[DBG] %s : core[%x,%d] child process descriptor allocated at cycle %d\n",
 __FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, (uint32_t)hal_get_cycles() );

    // allocate a child PID from local cluster
    error = cluster_pid_alloc( XPTR( local_cxy , process ) , &new_pid );
    if( (error != 0) || (new_pid == 0) )
    {
        printk("\n[ERROR] in %s : cannot get PID in cluster %x\n", 
        __FUNCTION__, local_cxy ); 
        process_free( process );
        return -1;
    }

fork_dmsg("\n[DBG] %s : core[%x, %d] child process PID allocated = %x at cycle %d\n",
 __FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, new_pid , (uint32_t)hal_get_cycles() );

    // initializes child process descriptor from parent process descriptor 
    process_reference_init( process,
                            new_pid,
                            parent_pid,
                            parent_process_xp );

fork_dmsg("\n[DBG] %s : core[%x, %d] child process initialised at cycle %d\n",
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, hal_get_cycles() );

    // copy VMM from parent descriptor to child descriptor
    error = vmm_fork_copy( process,
                           parent_process_xp );
    if( error )
    {
        printk("\n[ERROR] in %s : cannot copy VMM in cluster %x\n", 
        __FUNCTION__, local_cxy ); 
        process_free( process );
        cluster_pid_release( new_pid );
        return -1;
    }

fork_dmsg("\n[DBG] %s : core[%x, %d] child process VMM copied at cycle %d\n",
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, (uint32_t)hal_get_cycles() );

    // create child thread descriptor from parent thread descriptor
    error = thread_user_fork( parent_thread_xp,
                              process,
                              &thread );
    if( error )
    {
        printk("\n[ERROR] in %s : cannot create thread in cluster %x\n",
        __FUNCTION__, local_cxy ); 
        process_free( process );
        cluster_pid_release( new_pid );
        return -1;
    }

fork_dmsg("\n[DBG] %s : core[%x,%d] child thread created at cycle %d\n", 
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, (uint32_t)hal_get_cycles() );

    // update parent process GPT to set Copy_On_Write for shared data vsegs 
    // this includes all replicated GPT copies
    if( parent_process_cxy == local_cxy )   // reference is local
    {
        vmm_set_cow( parent_process_ptr );
    }
    else                                    // reference is remote
    {
        rpc_vmm_set_cow_client( parent_process_cxy,
                                parent_process_ptr );
    }

fork_dmsg("\n[DBG] %s : core[%x,%d] COW set in parent_process at cycle %d\n",
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, (uint32_t)hal_get_cycles() );

    // update children list in parent process
        xlist_add_last( XPTR( parent_process_cxy , &parent_process_ptr->children_root ),
                    XPTR( local_cxy , &process->brothers_list ) );
        hal_remote_atomic_add( XPTR( parent_process_cxy,
                                 &parent_process_ptr->children_nr), 1 );

// vmm_display( process , true );
// vmm_display( parent_process_ptr , true );
// sched_display( 0 );

    // return success
    *child_thread = thread;
    *child_pid    = new_pid;

fork_dmsg("\n[DBG] %s : core[%x,%d] exit at cycle %d\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, (uint32_t)hal_get_cycles() );

    return 0;

}  // end process_make_fork()

/*  deprecated because we don't wand to destroy the existing process descriptor

/////////////////////////////////////////////////////
error_t process_make_exec( exec_info_t  * exec_info )
{
    char           * path;                    // pathname to .elf file
    process_t      * old;                     // local pointer on old process
    process_t      * new;                     // local pointer on new process
    pid_t            pid;                     // old process identifier
    thread_t       * thread;                  // pointer on new thread
    pthread_attr_t   attr;                    // main thread attributes
    lid_t            lid;                     // selected core local index
        error_t          error;

        // get .elf pathname and PID from exec_info
        path = exec_info->path;
    pid  = exec_info->pid;

    // check local cluster is process owner
    assert( (CXY_FROM_PID( pid ) == local_cxy), __FUNCTION__,
    "local cluster %x is not owner for process %x\n", local_cxy, pid );

exec_dmsg("\n[DBG] %s : core[%x,%d] enters for process %x / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, pid , path );

    // get old process local pointer
    old = (process_t *)cluster_get_local_process_from_pid( pid );
    
    assert( (old != NULL ) , __FUNCTION__ , 
    "process %x not found in cluster %x\n", pid , local_cxy ); 

    // allocate memory for new process descriptor
    new = process_alloc();

    // initialize new process descriptor
    process_reference_init( new,
                            old->pid,                   // same as old
                            old->ppid,                  // same as old
                            XPTR( local_cxy , old ) );

exec_dmsg("\n[DBG] %s : core[%x,%d] created new process %x / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, pid, path );

    // register "code" and "data" vsegs as well as entry-point
    // in new process VMM, using information contained in the elf file.
        if( elf_load_process( path , new ) )
        {
                printk("\n[ERROR] in %s : failed to access .elf file for process %x / path = %s\n",
                __FUNCTION__, pid , path );
        process_destroy( new );
        return -1;
        }

exec_dmsg("\n[DBG] %s : core[%x,%d] vsegs registered / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, path );

    // select a core in local cluster to execute the main thread
    lid  = cluster_select_local_core();

    // initialize pthread attributes for main thread
    attr.attributes = PT_ATTR_DETACH | PT_ATTR_CLUSTER_DEFINED | PT_ATTR_CORE_DEFINED;
    attr.cxy        = local_cxy;
    attr.lid        = lid;

    // create and initialize thread descriptor
        error = thread_user_create( pid,
                                (void *)new->vmm.entry_point,
                                exec_info->args_pointers,
                                &attr,
                                &thread );
        if( error )
        {
                printk("\n[ERROR] in %s : cannot create thread for process %x / path = %s\n",
                       __FUNCTION__, pid , path );
        process_destroy( new );
        return -1;
        }

exec_dmsg("\n[DBG] %s : core[%x,%d] created main thread %x for new process %x\n",
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, thread->trdid, pid );

    // update children list (rooted in parent process)
        xlist_replace( XPTR( local_cxy , &old->brothers_list ) ,
                   XPTR( local_cxy , &new->brothers_list ) );

    // request destruction of old process copies and threads in all clusters
    process_sigaction( old , SIGKILL );

    // activate new thread
        thread_unblock( XPTR( local_cxy , thread ) , THREAD_BLOCKED_GLOBAL );

exec_dmsg("\n[DBG] %s : core[%x,%d] exit for path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, path  );

        return 0;

}  // end process_make_exec()

*/

/////////////////////////////////////////////////////
error_t process_make_exec( exec_info_t  * exec_info )
{
    char           * path;                    // pathname to .elf file
    process_t      * process;                 // local pointer on old process
    pid_t            pid;                     // old process identifier
    thread_t       * thread;                  // pointer on new main thread
    pthread_attr_t   attr;                    // main thread attributes
    lid_t            lid;                     // selected core local index
        error_t          error;

        // get .elf pathname and PID from exec_info
        path = exec_info->path;
    pid  = exec_info->pid;

    // check local cluster is process owner
    assert( (CXY_FROM_PID( pid ) == local_cxy), __FUNCTION__,
    "local cluster %x is not owner for process %x\n", local_cxy, pid );

exec_dmsg("\n[DBG] %s : core[%x,%d] enters for process %x / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, pid , path );

    // get process local pointer
    process = (process_t *)cluster_get_local_process_from_pid( pid );
    
    assert( (process != NULL ) , __FUNCTION__ , 
    "process %x not found in cluster %x\n", pid , local_cxy ); 

    // reset the existing vmm
    vmm_destroy( process );

exec_dmsg("\n[DBG] %s : core[%x,%d] VMM cleared\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid );

    // block all existing process threads
    process_sigaction( process , BLOCK_ALL_THREADS );

    // kill all existing threads and process descriptors (other than owner)
    process_sigaction( process , DELETE_ALL_THREADS );

    // check no threads
    assert( (process->th_nr == 0) , __FUNCTION__ , "no threads at this point" );

exec_dmsg("\n[DBG] %s : core[%x,%d] all threads deleted\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid );

    // initialize VMM 
    error = vmm_init( process );
        {
                printk("\n[ERROR] in %s : cannot initialize VMM for process %x / path = %s\n",
                __FUNCTION__, pid , path );
        process_destroy( process );
        return -1;
        }

    if( error )

    // register "code" and "data" vsegs as well as entry-point and vfs_bin_xp
    // in VMM, using information contained in the elf file.
        error = elf_load_process( path , process );

    if( error )
        {
                printk("\n[ERROR] in %s : failed to access .elf file for process %x / path = %s\n",
                __FUNCTION__, pid , path );
        process_destroy( process );
        return -1;
        }

exec_dmsg("\n[DBG] %s : core[%x,%d] new vsegs registered / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, path );

// @@@
vmm_display( process , true );
// @@@

    // select a core in local cluster to execute the new main thread
    lid  = cluster_select_local_core();

    // initialize pthread attributes for new main thread
    attr.attributes = PT_ATTR_DETACH | PT_ATTR_CLUSTER_DEFINED | PT_ATTR_CORE_DEFINED;
    attr.cxy        = local_cxy;
    attr.lid        = lid;

    // create and initialize thread descriptor
        error = thread_user_create( pid,
                                (void *)process->vmm.entry_point,
                                exec_info->args_pointers,
                                &attr,
                                &thread );
        if( error )
        {
                printk("\n[ERROR] in %s : cannot create thread for process %x / path = %s\n",
                       __FUNCTION__, pid , path );
        process_destroy( process );
        return -1;
        }

exec_dmsg("\n[DBG] %s : core[%x,%d] created main thread %x for new process %x\n",
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid, thread->trdid, pid );

    // activate new thread
        thread_unblock( XPTR( local_cxy , thread ) , THREAD_BLOCKED_GLOBAL );

exec_dmsg("\n[DBG] %s : core[%x,%d] exit for path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, path  );

        return 0;

}  // end process_make_exec()

////////////////////////////////////////////
void process_make_kill( process_t * process,
                        uint32_t    sig_id )
{
    // this function must be executed by a thread running in owner cluster
    assert( (CXY_FROM_PID( process->pid ) == local_cxy) , __FUNCTION__ ,
    "must execute in owner cluster" );

    // analyse signal type
    switch( sig_id )
    {
        case SIGSTOP:     // block all threads
        {
            process_sigaction( process , BLOCK_ALL_THREADS );
        }
        break;
        case SIGCONT:     // unblock all threads
        {
            process_sigaction( process , UNBLOCK_ALL_THREADS );
        }
        break;
        case SIGKILL:  // block all threads, then delete all threads
        {
            process_sigaction( process , BLOCK_ALL_THREADS );
            process_sigaction( process , DELETE_ALL_THREADS );
            process_destroy( process );
        }
        break;
    }
}  // end process_make_kill()

////////////////////////////////////////////
void process_make_exit( process_t * process,
                        uint32_t    status )
{
    // this function must be executed by a thread running in owner cluster
    assert( (CXY_FROM_PID( process->pid ) == local_cxy) , __FUNCTION__ ,
    "must execute in owner cluster" );

    // block all threads in all clusters
    process_sigaction( process , BLOCK_ALL_THREADS );

    // delete all threads in all clusters
    process_sigaction( process , DELETE_ALL_THREADS );

    // delete local process descriptor
    process_destroy( process ); 

}  // end process_make_exit()

//////////////////////////
void process_init_create()
{
    process_t      * process;       // local pointer on process_init descriptor
    pid_t            pid;           // process_init identifier
    thread_t       * thread;        // local pointer on main thread
    pthread_attr_t   attr;          // main thread attributes
    lid_t            lid;           // selected core local index for main thread
    error_t          error;

kinit_dmsg("\n[DBG] %s :  core[%x,%d] enters\n", 
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid );

    // allocates memory for process descriptor from local cluster
        process = process_alloc(); 
        if( process == NULL )
    {
                printk("\n[PANIC] in %s : no memory for process descriptor in cluster %x\n",
                __FUNCTION__, local_cxy  );
    }

    // get PID from local cluster
    error = cluster_pid_alloc( XPTR( local_cxy , process ) , &pid );
    if( error )
    {
                printk("\n[PANIC] in %s : cannot allocate PID in cluster %x\n",
                __FUNCTION__, local_cxy );
        process_destroy( process );
    }

    assert( (LPID_FROM_PID(pid) == 1) , __FUNCTION__ , "LPID must be 1 for process_init" );

    // initialize process descriptor / parent is local process_zero
    process_reference_init( process,
                            pid,
                            0,
                            XPTR( local_cxy , &process_zero ) );

kinit_dmsg("\n[DBG] %s : core[%x,%d] / process initialised\n", 
__FUNCTION__ , local_cxy, CURRENT_THREAD->core->lid );

    // register "code" and "data" vsegs as well as entry-point
    // in process VMM, using information contained in the elf file.
        if( elf_load_process( CONFIG_PROCESS_INIT_PATH , process ) )
        {
                printk("\n[PANIC] in %s : cannot access .elf file / path = %s\n",
                __FUNCTION__, CONFIG_PROCESS_INIT_PATH );
        process_destroy( process );
        }

kinit_dmsg("\n[DBG] %s : core[%x,%d] vsegs registered / path = %s\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, CONFIG_PROCESS_INIT_PATH );

    // select a core in local cluster to execute the main thread
    lid  = cluster_select_local_core();

    // initialize pthread attributes for main thread
    attr.attributes = PT_ATTR_DETACH | PT_ATTR_CLUSTER_DEFINED | PT_ATTR_CORE_DEFINED;
    attr.cxy        = local_cxy;
    attr.lid        = lid;

    // create and initialize thread descriptor
        error = thread_user_create( pid,
                                (void *)process->vmm.entry_point,
                                NULL,
                                &attr,
                                &thread );
        if( error )
        {
                printk("\n[PANIC] in %s : cannot create main thread / path = %s\n",
                __FUNCTION__, CONFIG_PROCESS_INIT_PATH );
        process_destroy( process );
        }

    // activate thread
        thread_unblock( XPTR( local_cxy , thread ) , THREAD_BLOCKED_GLOBAL );

    hal_fence();

kinit_dmsg("\n[DBG] %s : core[%x,%d] exit / main thread = %x\n",
__FUNCTION__, local_cxy, CURRENT_THREAD->core->lid, thread );

}  // end process_init_create()