/*
 * rpc.c - RPC related operations implementation.
 * 
 * Authors Mohamed Lamine Karaoui (2015)
 *         Alain Greiner (2016)
 *
 * 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 <almos_config.h>
#include <hal_types.h>
#include <hal_atomic.h>
#include <hal_remote.h>
#include <hal_irqmask.h>
#include <hal_special.h>
#include <printk.h>
#include <remote_sem.h>
#include <core.h>
#include <mapper.h>
#include <chdev.h>
#include <bits.h>
#include <thread.h>
#include <cluster.h>
#include <process.h>
#include <vfs.h>
#include <fatfs.h>
#include <dev_icu.h>
#include <rpc.h>

/////////////////////////////////////////////////////////////////////////////////////////
//      array of function pointers  (must be consistent with enum in rpc.h) 
/////////////////////////////////////////////////////////////////////////////////////////

rpc_server_t * rpc_server[RPC_MAX_INDEX] =
{
    &rpc_pmem_get_pages_server,         // 0
    &rpc_process_pid_alloc_server,      // 1
    &rpc_process_exec_server,           // 2
    &rpc_process_kill_server,           // 3
    &rpc_thread_user_create_server,     // 4
    &rpc_thread_kernel_create_server,   // 5
    &rpc_undefined,                     // 6                       
    &rpc_undefined,                     // 7
    &rpc_undefined,                     // 8
    &rpc_undefined,                     // 9

    &rpc_vfs_inode_create_server,       // 10  
    &rpc_vfs_inode_destroy_server,      // 11  
    &rpc_vfs_dentry_create_server,      // 12  
    &rpc_vfs_dentry_destroy_server,     // 13  
    &rpc_undefined,                     // 14
    &rpc_undefined,                     // 15
    &rpc_undefined,                     // 16
    &rpc_undefined,                     // 17
    &rpc_undefined,                     // 18
    &rpc_undefined,                     // 19

    &rpc_vmm_get_ref_vseg_server,       // 20
    &rpc_vmm_get_pte_server,            // 21
    &rpc_semaphore_alloc_server,        // 22
    &rpc_semaphore_free_server,         // 23
    &rpc_mapper_move_server,            // 24
    &rpc_undefined,                     // 25
    &rpc_undefined,                     // 26
    &rpc_undefined,                     // 27
    &rpc_undefined,                     // 28
    &rpc_undefined,                     // 29

    &rpc_fatfs_get_cluster_server       // 30
};

//////////////////////////////////////////////
void __attribute__((noinline)) rpc_undefined()
{
    printk("\n[PANIC] ‰s called in cluster %x\n", __FUNCTION__ , local_cxy );
    hal_core_sleep();
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_PMEM_GET_PAGES 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////
void rpc_pmem_get_pages_client( cxy_t      cxy,
                                uint32_t   order,      // in 
                                error_t  * error,      // out 
                                uint32_t * ppn )       // out 
{
    // any RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_PMEM_GET_PAGES;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)order;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output arguments RPC descriptor
    *error  = (error_t)rpc.args[0];     
    *ppn    = (uint32_t)rpc.args[1];
}

///////////////////////////////////////////
void rpc_pmem_get_pages_server( xptr_t xp )
{
    uint32_t order;  // input
    error_t  error;  // output
    uint32_t ppn;    // output

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get input arguments from client RPC descriptor
    order = hal_remote_lw( XPTR( cxy , &desc->args[0] ) );
    
    // call local pmem allocator
    page_t * page = ppm_alloc_pages( order ); 
    error = ( page == NULL ) ? ENOMEM : 0;
    ppn   = ppm_page2ppn( page );

    // set output arguments into client RPC descriptor
    hal_remote_sw( XPTR( cxy , &desc->args[0] ) , error );
    hal_remote_sw( XPTR( cxy , &desc->args[1] ) , ppn );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_PROCESS_PID_ALLOC
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////
void rpc_process_pid_alloc_client( cxy_t       cxy, 
                                   process_t * process,  // in
                                   error_t   * error,    // out
                                   pid_t     * pid )     // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_PROCESS_PID_ALLOC;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)process;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output arguments RPC descriptor
    *pid    = (pid_t)rpc.args[1];
    *error  = (error_t)rpc.args[2];     
}

//////////////////////////////////////////////
void rpc_process_pid_alloc_server( xptr_t xp )
{
    process_t * process;   // input  : client process descriptor
    error_t     error;     // output : error status
    pid_t       pid;       // output : process identifier

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get input argument from client RPC descriptor
    process = (process_t*)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    
    // call local pid allocator 
    xptr_t xp_process = XPTR( client_cxy , process );
    error = cluster_pid_alloc( xp_process , &pid ); 

    // set output arguments into client RPC descriptor
    hal_remote_sw( XPTR( client_cxy , &desc->args[0] ) , (uint64_t)error );
    hal_remote_sw( XPTR( client_cxy , &desc->args[1] ) , (uint64_t)pid );
}


/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_PROCESS_EXEC 
/////////////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////
void rpc_process_exec_client( cxy_t         cxy,
                              exec_info_t * info,     // in
                              error_t     * error )   // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_PROCESS_EXEC;
    rpc.response = 1;

    // set input arguments in RPC descriptor  
    rpc.args[0] = (uint64_t)(intptr_t)info;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output arguments from RPC descriptor
    *error  = (error_t)rpc.args[1];     
}

/////////////////////////////////////////
void rpc_process_exec_server( xptr_t xp )
{
    exec_info_t * ptr;       // local pointer on remote exec_info structure
    exec_info_t   info;      // local copy of exec_info structure
    error_t       error;     // local error error status

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get pointer on exec_info structure in client cluster from RPC descriptor
    ptr = (exec_info_t*)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );

    // copy exec_info structure from client buffer to server buffer
    hal_remote_memcpy( XPTR( client_cxy , ptr ),
                       XPTR( local_cxy , &info ),
                       sizeof(exec_info_t) );

    // call local kernel function
    error = process_make_exec( &info ); 

    // set output argument into client RPC descriptor
    hal_remote_swd( XPTR( client_cxy , &desc->args[1] ) , (uint64_t)error );
}


/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_PROCESS_KILL 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_process_kill_client( process_t * process ) 
{
    // only reference cluster can send this RPC
    if( !process->is_ref )
    {
        printk("PANIC in %s : caller is not the reference process\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // get local process index in reference cluster
    lpid_t lpid = LPID_FROM_PID( process->pid );

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

    // get number of copies
    uint32_t copies = pmgr->copies_nr[lpid];

    // initialise RPC descriptor 
    rpc_desc_t  rpc;
    rpc.index    = RPC_PROCESS_KILL;
    rpc.response = copies;
    rpc.args[0]  = (uint64_t)process->pid;

    // loop on list of copies to send RPC
    xptr_t  iter;
    XLIST_FOREACH( XPTR( local_cxy , &pmgr->copies_root[lpid] ) , iter )
    {
        // get cluster_identifier for current copy
        cxy_t  target_cxy = GET_CXY( iter );

        // register RPC request in remote RPC fifo ... but the reference
        if( target_cxy != local_cxy ) rpc_send_sync( target_cxy , &rpc );
    }
}  

/////////////////////////////////////////
void rpc_process_kill_server( xptr_t xp )
{
    pid_t       pid;
    process_t * process;  

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get pid argument from RPC descriptor
    pid = (pid_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );

    // get process pointer to call local kernel function
    process = cluster_get_local_process_from_pid( pid );

    if( process == NULL )  // process not found => do nothing
    {
        printk("\n[WARNING] in %s : process %x not found in cluster %x\n",
               __FUNCTION__ , pid , local_cxy );
    }
    else                   // destroy process 
    {
        process_kill( process ); 
    }
} 


/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_THREAD_USER_CREATE               
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////
void rpc_thread_user_create_client( cxy_t            cxy,  
                                    pthread_attr_t * attr,        // in
                                    xptr_t         * thread_xp,   // out 
                                    error_t        * error )      // out 
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("\n[PANIC] in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_THREAD_USER_CREATE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)attr;

    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );

    // get output arguments from RPC descriptor
    *thread_xp = (xptr_t)rpc.args[1];
    *error     = (error_t)rpc.args[2];
}

///////////////////////////////////////////////
void rpc_thread_user_create_server( xptr_t xp )
{
    pthread_attr_t * attr_ptr;   // pointer on attributes structure in client cluster 
    pthread_attr_t   attr_copy;  // attributes structure  copy in server cluster
    thread_t       * thread_ptr; // local pointer on thread descriptor
    xptr_t           thread_xp;  // extended pointer on thread descriptor
    pid_t            pid;        // process identifier
    process_t      * process;    // local pointer on process descriptor
    vseg_t         * vseg;       // local pointer on stack vseg

    error_t          error = 0;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get pointer on attributes structure in client cluster from RPC descriptor
    attr_ptr = (pthread_attr_t *)(intptr_t)hal_remote_lwd( XPTR(client_cxy , &desc->args[0]) );

    // makes a local copy of attributes structure
    hal_remote_memcpy( XPTR( local_cxy , &attr_copy ),
                       XPTR( client_cxy , attr_ptr ), 
                       sizeof(pthread_attr_t) );
    
    if( attr_copy.cxy != local_cxy )
    {
        printk("\n[PANIC] in %s : target cluster = %X / local_cluster = %x\n",
               __FUNCTION__ , attr_copy.cxy , local_cxy );
        hal_core_sleep();
    }

    // get local process descriptor
    pid     = attr_copy.pid; 
    process = cluster_get_local_process_from_pid( pid );

    if( process == NULL )
    {
        printk("\n[ERROR] in %s : no process descriptor in cluster %x for pid = %x\n",
               __FUNCTION__ , local_cxy , pid );
        error = ENOMEM;
       
    }

    // allocate a stack from local VMM 
    vseg = vmm_create_vseg( process, 
                            0,
                            0,
                            VSEG_TYPE_STACK );
    if( vseg == NULL )
    {
        printk("\n[ERROR] in %s : cannot create stack vseg in cluster %x for pid %x\n",
               __FUNCTION__ , local_cxy , pid );
        error = ENOMEM;
    } 

    // create thread in local cluster
    if( thread_user_create( &thread_ptr,
                            &attr_copy,
                            vseg->min,
                            vseg->max - vseg->min ) ) error = ENOMEM;


    // set output arguments
    thread_xp = XPTR( local_cxy , thread_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[1] ) , (uint64_t)error );
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)thread_xp );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_THREAD_KERNEL_CREATE
/////////////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////////
void rpc_thread_kernel_create_client( cxy_t     cxy,
                                      uint32_t  type,        // in
                                      void    * func,        // in
                                      void    * args,        // in
                                      xptr_t  * thread_xp,   // out
                                      error_t * error )      // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_THREAD_KERNEL_CREATE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)type;
    rpc.args[1] = (uint64_t)(intptr_t)func;
    rpc.args[2] = (uint64_t)(intptr_t)args;
    
    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );

    // get output arguments from RPC descriptor
    *thread_xp = (xptr_t)rpc.args[3];
    *error     = (error_t)rpc.args[4];
}

/////////////////////////////////////////////////
void rpc_thread_kernel_create_server( xptr_t xp )
{
    thread_t       * thread_ptr;  // local pointer on thread descriptor
    xptr_t           thread_xp;   // extended pointer on thread descriptor
    lid_t            core_lid;    // core local index
    error_t          error;    

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get attributes from RPC descriptor
    uint32_t  type = (uint32_t)       hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    void    * func = (void*)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    void    * args = (void*)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );

    // select one core
    core_lid = cluster_select_local_core();

    // call local kernel function
    error = thread_kernel_create( &thread_ptr , type , func , args , core_lid );

    // set output arguments
    thread_xp = XPTR( local_cxy , thread_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[1] ) , (uint64_t)error );
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)thread_xp );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VFS_INODE_CREATE 
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////
void rpc_vfs_inode_create_client( cxy_t          cxy,     
                                  xptr_t         dentry_xp,  // in
                                  uint32_t       type,       // in
                                  uint32_t       attr,       // in
                                  uint32_t       mode,       // in
                                  uint32_t       uid,        // in
                                  uint32_t       gid,        // in
                                  xptr_t       * inode_xp,   // out
                                  error_t      * error )     // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target cluster is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VFS_INODE_CREATE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)dentry_xp;
    rpc.args[1] = (uint64_t)type;
    rpc.args[2] = (uint64_t)attr;
    rpc.args[3] = (uint64_t)mode;
    rpc.args[4] = (uint64_t)uid;
    rpc.args[5] = (uint64_t)gid;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output values from RPC descriptor
    *inode_xp = (xptr_t)rpc.args[6];
    *error    = (error_t)rpc.args[7];
}

/////////////////////////////////////////////
void rpc_vfs_inode_create_server( xptr_t xp )
{
    xptr_t           dentry_xp;
    uint32_t         type;
    uint32_t         attr;
    uint32_t         mode;
    uint32_t         uid;
    uint32_t         gid;
    xptr_t           inode_xp;
    error_t          error;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get input arguments from client rpc descriptor
    dentry_xp = (xptr_t)  hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    type      = (uint32_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    attr      = (uint32_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );
    mode      = (uint32_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[3] ) );
    uid       = (uid_t)   hal_remote_lwd( XPTR( client_cxy , &desc->args[4] ) );
    gid       = (gid_t)   hal_remote_lwd( XPTR( client_cxy , &desc->args[5] ) );

    // call local kernel function
    error = vfs_inode_create( dentry_xp,
                              type,
                              attr,
                              mode,
                              uid,
                              gid,
                              &inode_xp );

    // set output arguments
    hal_remote_swd( XPTR( client_cxy , &desc->args[6] ) , (uint64_t)inode_xp );
    hal_remote_swd( XPTR( client_cxy , &desc->args[7] ) , (uint64_t)error );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VFS_INODE_DESTROY 
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////////
void rpc_vfs_inode_destroy_client( cxy_t                cxy,
                                   struct vfs_inode_s * inode )
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target cluster is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VFS_INODE_DESTROY;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)inode;
    
    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );
}

//////////////////////////////////////////////
void rpc_vfs_inode_destroy_server( xptr_t xp )
{
    vfs_inode_t * inode;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get arguments "inode" from client RPC descriptor
    inode = (vfs_inode_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
                       
    // call local kernel function
    vfs_inode_destroy( inode );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VFS_DENTRY_CREATE
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////
void rpc_vfs_dentry_create_client( cxy_t                  cxy,
                                   uint32_t               type,         // in
                                   char                 * name,         // in
                                   struct vfs_inode_s   * parent,       // in
                                   xptr_t               * dentry_xp,    // out
                                   error_t              * error )       // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target cluster is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VFS_DENTRY_CREATE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)type;
    rpc.args[1] = (uint64_t)(intptr_t)name;
    rpc.args[2] = (uint64_t)strlen( name );
    rpc.args[3] = (uint64_t)(intptr_t)parent;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output values from RPC descriptor
    *dentry_xp = (xptr_t)rpc.args[4];
    *error     = (error_t)rpc.args[5];
}

//////////////////////////////////////////////
void rpc_vfs_dentry_create_server( xptr_t xp )
{
    uint32_t      type;
    char        * name;
    vfs_inode_t * parent;
    xptr_t        dentry_xp;
    char          name_copy[CONFIG_VFS_MAX_NAME_LENGTH];
    uint32_t      length;
    error_t       error;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get argument "name" & "length" from client RPC descriptor and makes a local copy
    name   = (char *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    length = (uint32_t)        hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );
    hal_remote_memcpy( XPTR( local_cxy , name_copy ),
                       XPTR( client_cxy , name ),
					   length + 1 );               // +1 for the NUL char

    // get arguments "type" and "parent" from client RPC descriptor
    type   = (uint32_t)               hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    parent = (vfs_inode_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[3] ) );
                       
    // call local kernel function
    error = vfs_dentry_create( type,
                               name_copy,
                               parent,
                               &dentry_xp );
 
    // set output arguments
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)dentry_xp );
    hal_remote_swd( XPTR( client_cxy , &desc->args[5] ) , (uint64_t)error );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VFS_DENTRY_DESTROY 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////
void rpc_vfs_dentry_destroy_client( cxy_t          cxy,
                                    vfs_dentry_t * dentry )
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target cluster is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VFS_DENTRY_DESTROY;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)dentry;
    
    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );
}

///////////////////////////////////////////////
void rpc_vfs_dentry_destroy_server( xptr_t xp )
{
    vfs_dentry_t * dentry;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get arguments "dentry" from client RPC descriptor
    dentry = (vfs_dentry_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
                       
    // call local kernel function
    vfs_dentry_destroy( dentry );
}


/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VMM_GET_VSEG
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////
void rpc_vmm_get_ref_vseg_client( cxy_t       cxy,      
                                  process_t * process,     // in  
                                  intptr_t    vaddr,       // in  
                                  xptr_t    * vseg_xp )    // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VMM_GET_REF_VSEG;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)process;
    rpc.args[1] = (uint64_t)vaddr;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output argument from rpc descriptor
    *vseg_xp = rpc.args[2];
}

/////////////////////////////////////////////
void rpc_vmm_get_ref_vseg_server( xptr_t xp )
{
    process_t   * process;
    intptr_t      vaddr;
    vseg_t      * vseg_ptr;
    xptr_t        vseg_xp;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get input argument from client RPC descriptor
    process = (process_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    vaddr   = (intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    
    // call local kernel function
    vseg_ptr = vmm_get_vseg( process , vaddr );

    // set output argument to client RPC descriptor
    vseg_xp = XPTR( local_cxy , vseg_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)vseg_xp );
}


/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_VMM_GET_PTE 
/////////////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////
void rpc_vmm_get_pte_client( cxy_t       cxy,   
                             process_t * process,  // in 
                             vpn_t       vpn,      // in 
                             uint32_t  * attr,     // out
                             ppn_t     * ppn,      // out
                             error_t   * error )   // out 
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VMM_GET_PTE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)process;
    rpc.args[1] = (uint64_t)vpn;

    // register RPC request in remote RPC fifo (blocking function)
    rpc_send_sync( cxy , &rpc );

    // get output argument from rpc descriptor
    *attr  = (uint32_t)rpc.args[2];
    *ppn   = (ppn_t)rpc.args[3];
    *error = (error_t)rpc.args[4];
}

////////////////////////////////////////
void rpc_vmm_get_pte_server( xptr_t xp )
{
    process_t   * process;
    vpn_t         vpn;
    uint32_t      attr;
    ppn_t         ppn;
    error_t       error;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc        = (rpc_desc_t *)GET_PTR( xp );

    // get input argument "process" & "vpn" from client RPC descriptor
    process = (process_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    vpn     = (vpn_t)                hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    
    // call local kernel function
    error = vmm_get_pte( process , vpn , &attr , &ppn ); 

    // set output argument "attr" & "ppn" to client RPC descriptor
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)attr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[3] ) , (uint64_t)ppn );
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)error );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_SEMAPHORE_ALLOC 
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////
void rpc_semaphore_alloc_client( cxy_t    cxy,
                                 xptr_t * sem_xp )     // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_THREAD_USER_CREATE;
    rpc.response = 1;

    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );

    // get output arguments from RPC descriptor
    *sem_xp = (xptr_t)rpc.args[0];
}

////////////////////////////////////////////
void rpc_semaphore_alloc_server( xptr_t xp )
{
    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // allocates memory for semaphore
    kmem_req_t  req;
    req.type  = KMEM_SEM;
    req.flags = AF_ZERO;
    remote_sem_t * sem_ptr = kmem_alloc( &req );

    // set output argument
    xptr_t sem_xp = XPTR( local_cxy , sem_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[0] ) , (uint64_t)sem_xp );
}   

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_SEMAPHORE_FREE 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_semaphore_free_client( cxy_t          cxy,
                                remote_sem_t * sem )   // in
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_THREAD_USER_CREATE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)sem;

    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );
}

///////////////////////////////////////////
void rpc_semaphore_free_server( xptr_t xp )
{
    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get input argument "sem_ptr" from client RPC descriptor
    void * sem = (void *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );

    // releases memory
    kmem_req_t  req;
    req.type = KMEM_SEM;
    req.ptr  = sem;
    kmem_free(&req);
}   

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_MAPPER_MOVE
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////
void rpc_mapper_move_client( cxy_t        cxy,
                             mapper_t   * mapper,      // in
                             bool_t       read,        // in
                             uint32_t     nb_frags,    // in
                             fragment_t * frags,       // in
                             error_t    * error )      // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_MAPPER_MOVE;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)mapper;
    rpc.args[1] = (uint64_t)read;
    rpc.args[2] = (uint64_t)nb_frags;
    rpc.args[3] = (uint64_t)(intptr_t)frags;

    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );

    // get output argument from rpc descriptor
    *error = (error_t)rpc.args[4];
}

////////////////////////////////////////
void rpc_mapper_move_server( xptr_t xp )
{
    mapper_t    * mapper;
    bool_t        read;
    uint32_t      nb;
    void        * frags;
    error_t       error;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get input arguments
    mapper = (mapper_t*)(intptr_t)hal_remote_lpt( XPTR( client_cxy , &desc->args[0] ) );
    read   = (bool_t)             hal_remote_lw ( XPTR( client_cxy , &desc->args[1] ) );
    nb     = (uint32_t)           hal_remote_lw ( XPTR( client_cxy , &desc->args[2] ) );
    frags  = (void*)(intptr_t)    hal_remote_lpt( XPTR( client_cxy , &desc->args[3] ) );

    // call the mapper_move_fragments() function
    error = mapper_move_fragments( mapper , read , nb , XPTR( client_cxy , frags ) );

    // set output argument
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)error );
}

/////////////////////////////////////////////////////////////////////////////////////////
//               Marshaling functions attached to RPC_FATFS_GET_CLUSTER 
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////
void rpc_fatfs_get_cluster_client( cxy_t      cxy,
                                   mapper_t * mapper,    // in
                                   uint32_t   first,     // in
                                   uint32_t   page,      // in
                                   uint32_t * cluster,   // out
                                   error_t  * error )    // out
{
    // RPC must be remote
    if( cxy == local_cxy ) 
    {
        printk("PANIC in %s : target is not remote\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_FATFS_GET_CLUSTER;
    rpc.response = 1;

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)mapper;
    rpc.args[1] = (uint64_t)first;
    rpc.args[2] = (uint64_t)page;

    // register RPC request in remote RPC fifo 
    rpc_send_sync( cxy , &rpc );

    // get output argument from rpc descriptor
    *cluster = (uint32_t)rpc.args[3];
    *error   = (error_t)rpc.args[4];
}

//////////////////////////////////////////////
void rpc_fatfs_get_cluster_server( xptr_t xp )
{
    mapper_t    * mapper;
    uint32_t      first;
    uint32_t      page;
    uint32_t      cluster;
    error_t       error;

    // get client cluster identifier and pointer on RPC descriptor
    cxy_t        client_cxy  = (cxy_t)GET_CXY( xp );
    rpc_desc_t * desc = (rpc_desc_t *)GET_PTR( xp );

    // get input arguments
    mapper = (mapper_t*)(intptr_t)hal_remote_lpt( XPTR( client_cxy , &desc->args[0] ) );
    first  = (uint32_t)           hal_remote_lw ( XPTR( client_cxy , &desc->args[1] ) );
    page   = (uint32_t)           hal_remote_lw ( XPTR( client_cxy , &desc->args[2] ) );

    // call the kernel function
    error = fatfs_get_cluster( mapper , first , page , &cluster );

    // set output argument
    hal_remote_swd( XPTR( client_cxy , &desc->args[3] ) , (uint64_t)cluster );
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)error );
}

/***************************************************************************************/
/************ Generic functions supporting RPCs : client side **************************/
/***************************************************************************************/

////////////////////////////////////////////
void rpc_send_sync( cxy_t        server_cxy, 
                    rpc_desc_t * rpc )
{
	thread_t * this = CURRENT_THREAD;
    uint32_t   cores;
    error_t    error;
    bool_t     first;
    reg_t      sr_save;

printk("\n@@@ coucou 0\n");

    // get client CPU and cluster coordinates
    cxy_t      client_cxy = local_cxy;   
    lid_t      client_lid = CURRENT_CORE->lid;

    // allocate and initialise an extended pointer on the RPC descriptor
	xptr_t   xp = XPTR( client_cxy , rpc );

    // get local pointer on rpc_fifo in remote cluster with the 
    // assumption that addresses are identical in all clusters
    rpc_fifo_t * rf = &LOCAL_CLUSTER->rpc_fifo;

	// try to post an item in remote fifo
    // deschedule and retry if remote fifo full
    do
    { 
        error = remote_fifo_put_item( XPTR( server_cxy , &rf->fifo ),
                                      (void *)&xp,
                                      sizeof(xptr_t), 
                                      &first );

	    if ( error ) 
        {
            printk("\n[WARNING] %s : core %d in cluster %x cannot post RPC to cluster %x\n",
                   __FUNCTION__ , client_lid , client_cxy , server_cxy );
            if( thread_can_yield() ) sched_yield();
        }
    }
    while( error );
 
printk("\n@@@ coucou 1\n");

    rpc_dmsg("\n[INFO] %s on core %d in cluster %x sent RPC %p to cluster %x\n", 
              __FUNCTION__ , client_lid , client_cxy , rpc , server_cxy );
	
    // send IPI if this is the first RPC in remote FIFO
    // and no CPU is in kernel mode in server cluster.
    // the selected CPU in server has the same lid as the client CPU.
	if( first )
	{
        // get number of cores in kernel mode in server cluster
        cores = hal_remote_lw( XPTR( server_cxy , &LOCAL_CLUSTER->cores_in_kernel ) );

		if( cores == 0 ) // no core in kernel mode in server
		{
		    dev_icu_send_ipi( server_cxy , client_lid );

		    rpc_dmsg("\n[INFO] %s : core %d in cluster %x send IPI to core %d in cluster %x\n",
                      __FUNCTION__, client_lid , client_cxy , client_lid , server_cxy );
        }
	}

printk("\n@@@ coucou 2\n");

	// activate preemption to allow incoming RPC and avoid deadlock
	if( this->type == THREAD_RPC ) hal_enable_irq( &sr_save );

    // the sending thread poll the response slot until RPC completed
	while( 1 )
    {
        if( rpc->response == 0 ) break;
    }

printk("\n@@@ coucou 3\n");

    // restore preemption
	if( this->type == THREAD_RPC ) hal_restore_irq( sr_save );

}  // end rpc_send_sync()



/***************************************************************************************/
/************ Generic functions supporting RPCs : server side **************************/
/***************************************************************************************/

///////////////////////////////////////////
void rpc_fifo_init( rpc_fifo_t * rpc_fifo )
{
	rpc_fifo->count       = 0;
	rpc_fifo->owner       = 0;
	local_fifo_init( &rpc_fifo->fifo );
}

////////////////////////////////////////////////
error_t rpc_execute_all( rpc_fifo_t * rpc_fifo )
{
	xptr_t         xp;             // extended pointer on RPC descriptor
	uint32_t       count;          // handled RPC request counter
	thread_t     * this;           // pointer on this RPC thread
    core_t       * core;           // pointer on core running this thread
    rpc_desc_t   * desc;           // pointer on RPC descriptor 
    uint32_t       index;          // RPC index 
    uint32_t       expected;       // number of expected responses 
    cxy_t          client_cxy;     // client cluster identifier
	error_t        error;
      
	this = CURRENT_THREAD; 
    core = this->core;   


    // handle up to CONFIG_RPC_PENDING_MAX requests before exit
	count = 0;
	do
    {
	    error = local_fifo_get_item( &rpc_fifo->fifo, 
                                     &xp,
                                     sizeof(xptr_t) );

		if ( error == 0 )  // One RPC request successfully extracted from RPC_FIFO
        {
            rpc_dmsg("\n[INFO] %s : RPC_THREAD %x on core %x in cluster %x handles RPC %d\n"
				     __FUNCTION__ , this->trdid , core->lid , local_cxy , count );

            // get client cluster identifier and pointer on RPC descriptor
            client_cxy = (cxy_t)GET_CXY( xp );
            desc       = (rpc_desc_t *)GET_PTR( xp );

            // get rpc index and expected responses from RPC descriptor
	        index     = hal_remote_lw( XPTR( client_cxy , &desc->index ) );
	        expected  = hal_remote_lw( XPTR( client_cxy , &desc->response ) );

            // call the relevant server function
            rpc_server[index]( xp );

            // increment handled RPC counter
  	        count++;

            // notify RPC completion as required
            if( expected == 1 ) hal_remote_sw( XPTR(client_cxy,&desc->response) , 0 );
            if( expected >  1 ) hal_remote_atomic_add( XPTR(client_cxy,&desc->response) , -1 );
		}
	
		// exit loop in three cases:
        // - fifo is empty
        // - look has been released (because descheduling)
        // - max number of RPCs has been reached
		if( error || 
            (rpc_fifo->owner != this->trdid) || 
            (count > CONFIG_RPC_PENDING_MAX) ) break;
	}
    while( 1 )

	rpc_dmsg("\n[INFO] %s running on core %d in cluster %x exit\n"
              __FUNCTION__ , CURRENT_CORE->lid , local_cxy );
		
    // update RPC_FIFO global counter
	rpc_fifo->count += count;

	return 0;
}  // end rpc_execute_all()

////////////////////////////////////////////////////
error_t rpc_activate_thread( rpc_fifo_t * rpc_fifo )
{
	core_t      * core;
	thread_t    * thread;
	thread_t    * this;
    scheduler_t * sched;
	error_t       error;
    bool_t        found;
    reg_t         sr_save;

	this   = CURRENT_THREAD;
    core   = this->core;
    sched  = &core->scheduler;
    found  = false;

    // calling thread must be the RPC_FIFO owner
    if( this->trdid != rpc_fifo->owner )
    {
        printk("\n[PANIC] in %s : calling thread is not RPC_FIFO owner\n", __FUNCTION__ );
        hal_core_sleep();
    }

    // makes the calling thread not preemptable
    // during activation / creation of the RPC thread
	hal_disable_irq( &sr_save );

    // search a free RPC thread (must be in THREAD_BLOCKED_IDLE state)    
    list_entry_t * iter;
    LIST_FOREACH( &sched->k_root , iter )
    {
        thread = LIST_ELEMENT( iter , thread_t , sched_list );
        if( (thread->type == THREAD_RPC) && (thread->blocked ==  THREAD_BLOCKED_IDLE ) ) 
        {
            found = true;
            break;
        }
    }

    if( found )                    // activate this idle RPC thread      
    {
        thread->blocked = 0;
    }
    else                           // create a new RPC thread
    {
        error = thread_kernel_create( &thread,
                                      THREAD_RPC, 
			                          &rpc_thread_func, 
                                      NULL,
			                          core->lid );
	    if( error ) 
        {
	        hal_restore_irq( sr_save );
            printk("\n[ERROR] in %s : no memory for new RPC thread in cluster %x\n",
                   __FUNCTION__ , local_cxy );
            return ENOMEM;
        }

        rpc_dmsg("\n[INFO] %s creates RPC thread %x on core %x in cluster %x at cycle %d\n", 
		          __FUNCTION__ , thread , core->gid , local_cxy , hal_time_stamp() );

        // update core descriptor counter  
	    core->rpc_threads++;
    }

    // update owner in rpc_fifo
    rpc_fifo->owner = thread->trdid;

    rpc_dmsg ("\n[INFO] %s activates RPC thread %x on core %x in cluster %x at cycle %d\n",
		      __FUNCTION__ , thread , core->gid , local_cxy , hal_time_stamp() );

    // current thread deschedules / RPC thread start execution 
	sched_switch_to( thread );

    // restore IRQs for the calling thread 
	hal_restore_irq( sr_save );

    // return success
	return 0;

}  // end rpc_activate_thread()

//////////////////
bool_t rpc_check()
{
	thread_t   * this     = CURRENT_THREAD;
	rpc_fifo_t * rpc_fifo = &LOCAL_CLUSTER->rpc_fifo;
    error_t      error;

    // calling thread does nothing if light lock already taken or FIFO empty  
	if( (rpc_fifo->owner != 0) || (local_fifo_is_empty( &rpc_fifo->fifo )) )
    {
        return false;
    }

	// calling thread tries to take the light lock,
    // and activates an RPC thread if success
    if( hal_atomic_test_set( &rpc_fifo->owner , this->trdid ) )
	{
        error = rpc_activate_thread( rpc_fifo );

        if( error )    // cannot activate an RPC_THREAD 
        {
            rpc_fifo->owner = 0;

            printk("\n[ERROR] in %s : no memory to create a RPC thread for core %d"
                   " in cluster %x => do nothing\n",
                   __FUNCTION__ , CURRENT_CORE->lid , local_cxy );
        }

        return true;
    }
    else  // light lock taken by another thread
    {
        return false;
    }
} // end rpc_check()


//////////////////////
void rpc_thread_func()
{
    // makes the calling thread not preemptable
	hal_disable_irq( NULL );
  
	thread_t   * this     = CURRENT_THREAD;
	rpc_fifo_t * rpc_fifo = &LOCAL_CLUSTER->rpc_fifo;

    rpc_dmsg("\n[INFO] RPC thread %x created on core %d in cluster %x at cycle %d\n",
             this->trdid , this->core->lid , local_cxy , hal_time_stamp() );

    // this infinite loop is not preemptable
    // the RPC thread deschedule when the RPC_FIFO is empty
	while(1)
	{
        // check fifo ownership (ownership should be given by rpc_activate()
        if( this->trdid != rpc_fifo->owner )
        {
            printk("\n[PANIC] in %s : RPC_THREAD %x not owner of RPC_FIFO in cluster %x\n",
                   __FUNCTION__ , this->trdid , local_cxy );
            hal_core_sleep();
        }
 
        // executes pending RPC(s) 
        rpc_execute_all( rpc_fifo );

        // release rpc_fifo ownership (can be lost during RPC execution)
        if( rpc_fifo->owner == this->trdid ) rpc_fifo->owner = 0;


        // suicide if too much RPC threads for this core
		if( this->core->rpc_threads > CONFIG_RPC_THREADS_MAX ) 
		{
            rpc_dmsg("\n[INFO] RPC thread %x suicide on core %d in cluster %x at cycle %d\n", 
		             this->trdid , this->core->lid , local_cxy , hal_time_stamp() );

            // update core descriptor counter
			this->core->rpc_threads--;

            // suicide
			thread_exit();
		}

        // block and deschedule
        rpc_dmsg("\n[INFO] RPC thread %x deschedule on core %d in cluster %x at cycle %d\n", 
			      this->trdid , this->core->lid , local_cxy , hal_time_stamp() ); 

		thread_block( this , THREAD_BLOCKED_IDLE );
        sched_yield();

		rpc_dmsg("\n[INFO] RPC thread %x wake up on core %d in cluster %x at cycle %d\n", 
			      this->trdid , this->core->lid , local_cxy , hal_time_stamp() );
	}
} // end rpc_thread_func()

