/*
 * rpc.c - RPC related operations implementation.
 * 
 * Author    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_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 <signal.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_make_exec_server,      // 1
    &rpc_process_make_fork_server,      // 2
    &rpc_process_kill_server,           // 3
    &rpc_thread_user_create_server,     // 4
    &rpc_thread_kernel_create_server,   // 5
    &rpc_signal_rise_server,            // 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_vfs_file_create_server,        // 14
    &rpc_vfs_file_destroy_server,       // 15
    &rpc_vfs_inode_load_server,         // 16
    &rpc_vfs_mapper_load_all_server,    // 17
    &rpc_fatfs_get_cluster_server,      // 18
    &rpc_undefined,                     // 19

    &rpc_vmm_get_vseg_server,           // 20
    &rpc_vmm_get_pte_server,            // 21
    &rpc_kcm_alloc_server,              // 22
    &rpc_kcm_free_server,               // 23
    &rpc_mapper_move_buffer_server,     // 24
    &rpc_mapper_get_page_server,        // 25
    &rpc_vmm_create_vseg_server,        // 26
    &rpc_sched_display_server,          // 27
    &rpc_vmm_set_cow_server,            // 28
    &rpc_undefined,                     // 29
};

//////////////////////////////////////////////
void __attribute__((noinline)) rpc_undefined()
{
	panic("called in cluster %x", local_cxy );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [0]           Marshaling functions attached to RPC_PMEM_GET_PAGES 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////
void rpc_pmem_get_pages_client( cxy_t      cxy,
                                uint32_t   order,      // in 
                                page_t  ** page )      // out 
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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 from RPC descriptor
    *page = (page_t *)(intptr_t)rpc.args[1];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

///////////////////////////////////////////
void rpc_pmem_get_pages_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    uint32_t order = hal_remote_lw( XPTR( cxy , &desc->args[0] ) );
    
    // call local pmem allocator
    page_t * page = ppm_alloc_pages( order ); 

    // set output arguments into client RPC descriptor
    hal_remote_swd( XPTR( cxy , &desc->args[1] ) , (uint64_t)(intptr_t)page );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [1]           Marshaling functions attached to RPC_PROCESS_MAKE_EXEC
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////
void rpc_process_make_exec_client( cxy_t         cxy,
                                   exec_info_t * info,     // in
                                   error_t     * error )   // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_PROCESS_MAKE_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];     

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

//////////////////////////////////////////////
void rpc_process_make_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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [2]           Marshaling functions attached to RPC_PROCESS_MAKE_FORK 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_process_make_fork_client( cxy_t       cxy,
                                   xptr_t      ref_process_xp,      // in
                                   xptr_t      parent_thread_xp,    // in
                                   pid_t     * child_pid,           // out
                                   thread_t ** child_thread_ptr,    // out     
                                   error_t   * error )              // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

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

    // get output arguments from RPC descriptor
    *child_pid         = (pid_t)rpc.args[2];
    *child_thread_ptr  = (thread_t *)(intptr_t)rpc.args[3];
    *error             = (error_t)rpc.args[4];     

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

//////////////////////////////////////////////
void rpc_process_make_fork_server( xptr_t xp )
{
    xptr_t     ref_process_xp;     // extended pointer on reference parent process
    xptr_t     parent_thread_xp;   // extended pointer on parent thread
    pid_t      child_pid;          // child process identifier
    thread_t * child_thread_ptr;   // local copy of exec_info structure
    error_t    error;              // local error status

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 cient RPC descriptor
    ref_process_xp   = (xptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    parent_thread_xp = (xptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );

    // call local kernel function
    error = process_make_fork( ref_process_xp,
                               parent_thread_xp,
                               &child_pid,
                               &child_thread_ptr ); 

    // set output argument into client RPC descriptor
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)child_pid );
    hal_remote_swd( XPTR( client_cxy , &desc->args[3] ) , (uint64_t)(intptr_t)child_thread_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)error );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [3]           Marshaling functions attached to RPC_PROCESS_KILL 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_process_kill_client( process_t * process ) 
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // only reference cluster can send this RPC
    assert( (GET_CXY( process->ref_xp ) == local_cxy) , __FUNCTION__ ,
            "caller must be reference process cluster\n");

    // 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 );
    }

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}  

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 ); 
    }

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
} 


/////////////////////////////////////////////////////////////////////////////////////////
// [4]           Marshaling functions attached to RPC_THREAD_USER_CREATE               
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////
void rpc_thread_user_create_client( cxy_t            cxy,  
                                    pid_t            pid,         // in
                                    void           * start_func,  // in
                                    void           * start_arg,   // in
                                    pthread_attr_t * attr,        // in
                                    xptr_t         * thread_xp,   // out 
                                    error_t        * error )      // out 
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)pid;
    rpc.args[1] = (uint64_t)(intptr_t)start_func;
    rpc.args[2] = (uint64_t)(intptr_t)start_arg;
    rpc.args[3] = (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[4];
    *error     = (error_t)rpc.args[5];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

///////////////////////////////////////////////
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
    void           * start_func;
    void           * start_arg;
    error_t          error;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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

    // get input arguments from RPC descriptor
    pid        = (pid_t)                     hal_remote_lwd(XPTR(client_cxy , &desc->args[0]));
    start_func = (void *)(intptr_t)          hal_remote_lwd(XPTR(client_cxy , &desc->args[1]));
    start_arg  = (void *)(intptr_t)          hal_remote_lwd(XPTR(client_cxy , &desc->args[2]));
    attr_ptr   = (pthread_attr_t *)(intptr_t)hal_remote_lwd(XPTR(client_cxy , &desc->args[3]));

    // makes a local copy of attributes structure
    hal_remote_memcpy( XPTR( local_cxy , &attr_copy ),
                       XPTR( client_cxy , attr_ptr ), 
                       sizeof(pthread_attr_t) );
    
    // call kernel function
    error = thread_user_create( pid,
                                start_func,
                                start_arg,
                                &attr_copy,
                                &thread_ptr );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [5]           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_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////
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;    

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [6]           Marshaling functions attached to RPC_SIGNAL_RISE
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////
void rpc_signal_rise_client( cxy_t       cxy,
                             process_t * process,    // in
                             uint32_t    sig_id )    // in
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////                             
void rpc_signal_rise_server( xptr_t xp )
{
    process_t  * process;  // local pointer on process descriptor
    uint32_t     sig_id;   // signal index

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    process = (process_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    sig_id  = (uint32_t)             hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );

    // call local kernel function
    signal_rise( process , sig_id );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [10]          Marshaling functions attached to RPC_VFS_INODE_CREATE 
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////
void rpc_vfs_inode_create_client( cxy_t          cxy,     
                                  xptr_t         dentry_xp,  // in
                                  uint32_t       fs_type,    // in
                                  uint32_t       inode_type, // in
                                  void         * extend,     // in
                                  uint32_t       attr,       // in
                                  uint32_t       rights,     // in
                                  uint32_t       uid,        // in
                                  uint32_t       gid,        // in
                                  xptr_t       * inode_xp,   // out
                                  error_t      * error )     // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)fs_type;
    rpc.args[2] = (uint64_t)inode_type;
    rpc.args[3] = (uint64_t)(intptr_t)extend;
    rpc.args[4] = (uint64_t)attr;
    rpc.args[5] = (uint64_t)rights;
    rpc.args[6] = (uint64_t)uid;
    rpc.args[7] = (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[8];
    *error    = (error_t)rpc.args[9];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////
void rpc_vfs_inode_create_server( xptr_t xp )
{
    xptr_t           dentry_xp;
    uint32_t         fs_type;
    uint32_t         inode_type;
    void           * extend;
    uint32_t         attr;
    uint32_t         rights;
    uint32_t         uid;
    uint32_t         gid;
    xptr_t           inode_xp;
    error_t          error;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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] ) );
    fs_type    = (uint32_t)        hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    inode_type = (uint32_t)        hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );
    extend     = (void *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[3] ) );
    attr       = (uint32_t)        hal_remote_lwd( XPTR( client_cxy , &desc->args[4] ) );
    rights     = (uint32_t)        hal_remote_lwd( XPTR( client_cxy , &desc->args[5] ) );
    uid        = (uid_t)           hal_remote_lwd( XPTR( client_cxy , &desc->args[6] ) );
    gid        = (gid_t)           hal_remote_lwd( XPTR( client_cxy , &desc->args[7] ) );

    // call local kernel function
    error = vfs_inode_create( dentry_xp,
                              fs_type,
                              inode_type,
                              extend,
                              attr,
                              rights,
                              uid,
                              gid,
                              &inode_xp );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [11]          Marshaling functions attached to RPC_VFS_INODE_DESTROY 
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////////
void rpc_vfs_inode_destroy_client( cxy_t                cxy,
                                   struct vfs_inode_s * inode )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [12]          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_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)(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[3];
    *error     = (error_t)rpc.args[4];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

//////////////////////////////////////////////
void rpc_vfs_dentry_create_server( xptr_t xp )
{
    uint32_t      type;
    char        * name;
    vfs_inode_t * parent;
    xptr_t        dentry_xp;
    error_t       error;

    char          name_copy[CONFIG_VFS_MAX_NAME_LENGTH];

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "name", "type", and "parent" from client RPC descriptor
    type   = (uint32_t)               hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    name   = (char *)(intptr_t)       hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    parent = (vfs_inode_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );

    // makes a local copy of  name
    hal_remote_strcpy( XPTR( local_cxy , name_copy ),
                       XPTR( client_cxy , name ) );

    // 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[3] ) , (uint64_t)dentry_xp );
    hal_remote_swd( XPTR( client_cxy , &desc->args[4] ) , (uint64_t)error );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [13]          Marshaling functions attached to RPC_VFS_DENTRY_DESTROY 
/////////////////////////////////////////////////////////////////////////////////////////


///////////////////////////////////////////////////////
void rpc_vfs_dentry_destroy_client( cxy_t          cxy,
                                    vfs_dentry_t * dentry )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}


/////////////////////////////////////////////////////////////////////////////////////////
// [14]          Marshaling functions attached to RPC_VFS_FILE_CREATE
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////
void rpc_vfs_file_create_client( cxy_t                  cxy,
                                 struct vfs_inode_s   * inode,       // in
                                 uint32_t               file_attr,   // in
                                 xptr_t               * file_xp,     // out
                                 error_t              * error )      // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

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

    // get output values from RPC descriptor
    *file_xp = (xptr_t)rpc.args[2];
    *error   = (error_t)rpc.args[3];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////////
void rpc_vfs_file_create_server( xptr_t xp )
{
    uint32_t      file_attr;
    vfs_inode_t * inode;
    xptr_t        file_xp;
    error_t       error;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "file_attr" and "inode" from client RPC descriptor
    inode     = (vfs_inode_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    file_attr = (uint32_t)               hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
                       
    // call local kernel function
    error = vfs_file_create( inode,
                             file_attr,
                             &file_xp );
 
    // set output arguments
    hal_remote_swd( XPTR( client_cxy , &desc->args[2] ) , (uint64_t)file_xp );
    hal_remote_swd( XPTR( client_cxy , &desc->args[3] ) , (uint64_t)error );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [15]          Marshaling functions attached to RPC_VFS_FILE_DESTROY 
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_vfs_file_destroy_client( cxy_t        cxy,
                                  vfs_file_t * file )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////
void rpc_vfs_file_destroy_server( xptr_t xp )
{
    vfs_file_t * file;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    file = (vfs_file_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
                       
    // call local kernel function
    vfs_file_destroy( file );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [16]          Marshaling functions attached to RPC_VFS_INODE_LOAD  
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////
void rpc_vfs_inode_load_client( cxy_t         cxy,
                                vfs_inode_t * parent_inode,    // in
                                char        * name,            // in
                                xptr_t        child_inode_xp,  // in
                                error_t     * error )          // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)parent_inode;
    rpc.args[1] = (uint64_t)(intptr_t)name;
    rpc.args[2] = (uint64_t)child_inode_xp;

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

    // get output values from RPC descriptor
    *error   = (error_t)rpc.args[3];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

///////////////////////////////////////////
void rpc_vfs_inode_load_server( xptr_t xp )
{
    error_t       error;
    vfs_inode_t * parent;
    xptr_t        child_xp;
    char        * name;

    char          name_copy[CONFIG_VFS_MAX_NAME_LENGTH];

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "parent", "name", and "child_xp"
    parent     = (vfs_inode_t*)(intptr_t)hal_remote_lwd(XPTR(client_cxy , &desc->args[0]));
    name       = (char*)(intptr_t)       hal_remote_lwd(XPTR(client_cxy , &desc->args[1]));
    child_xp   = (xptr_t)                hal_remote_lwd(XPTR(client_cxy , &desc->args[2]));

    // get name local copy
    hal_remote_strcpy( XPTR( local_cxy , name_copy ) ,
                       XPTR( client_cxy , name ) );

    // call the kernel function
    error = vfs_inode_load( parent , name_copy , child_xp );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [17]          Marshaling functions attached to RPC_VFS_MAPPER_LOAD_ALL
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////
void rpc_vfs_mapper_load_all_client( cxy_t         cxy,
                                     vfs_inode_t * inode,      // in
                                     error_t     * error )     // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VFS_MAPPER_LOAD_ALL;
    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 );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////////////
void rpc_vfs_mapper_load_all_server( xptr_t xp )
{
    error_t       error;
    vfs_inode_t * inode;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "parent", "name", and "child_xp"
    inode = (vfs_inode_t*)(intptr_t)hal_remote_lwd(XPTR(client_cxy , &desc->args[0]));

    // call the kernel function
    error = vfs_mapper_load_all( inode );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [18]          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   index,     // in
                                   uint32_t * cluster,   // out
                                   error_t  * error )    // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)index;

    // 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];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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] ) );
    index  = (uint32_t)            hal_remote_lw ( XPTR( client_cxy , &desc->args[2] ) );

    // call the kernel function
    error = fatfs_get_cluster( mapper , first , index , &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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [20]          Marshaling functions attached to RPC_VMM_GET_VSEG
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////
void rpc_vmm_get_vseg_client( cxy_t       cxy,      
                              process_t * process,     // in  
                              intptr_t    vaddr,       // in  
                              xptr_t    * vseg_xp,     // out
                              error_t   * error )      // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VMM_GET_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];
    *error   = (error_t)rpc.args[3];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    error = vmm_get_vseg( process , vaddr , &vseg_ptr );

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}


/////////////////////////////////////////////////////////////////////////////////////////
// [21]          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 
                             bool_t      cow,      // in
                             uint32_t  * attr,     // out
                             ppn_t     * ppn,      // out
                             error_t   * error )   // out 
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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;
    rpc.args[2] = (uint64_t)cow;

    // 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[3];
    *ppn   = (ppn_t)rpc.args[4];
    *error = (error_t)rpc.args[5];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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] ) );
    cow     = (bool_t)               hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );
    
    // call local kernel function
    error = vmm_get_pte( process , vpn , cow , &attr , &ppn ); 

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [22]          Marshaling functions attached to RPC_KCM_ALLOC 
/////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////
void rpc_kcm_alloc_client( cxy_t      cxy,
                           uint32_t   kmem_type,   // in
                           xptr_t *   buf_xp )     // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)kmem_type;

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

    // get output arguments from RPC descriptor
    *buf_xp = (xptr_t)rpc.args[1];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

//////////////////////////////////////
void rpc_kcm_alloc_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "kmem_type" from client RPC descriptor
    uint32_t kmem_type = (uint32_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );

    // allocates memory for kcm
    kmem_req_t  req;
    req.type  = kmem_type;
    req.flags = AF_ZERO;
    void * buf_ptr = kmem_alloc( &req );

    // set output argument
    xptr_t buf_xp = XPTR( local_cxy , buf_ptr );
    hal_remote_swd( XPTR( client_cxy , &desc->args[1] ) , (uint64_t)buf_xp );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}   

/////////////////////////////////////////////////////////////////////////////////////////
// [23]          Marshaling functions attached to RPC_KCM_FREE 
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////
void rpc_kcm_free_client( cxy_t      cxy,
                          void     * buf,          // in
                          uint32_t   kmem_type )   // in
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // 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)buf;
    rpc.args[1] = (uint64_t)kmem_type;

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////
void rpc_kcm_free_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 "buf" and "kmem_type" from client RPC descriptor
    void     * buf = (void *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    uint32_t   kmem_type = (uint32_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );

    // releases memory
    kmem_req_t  req;
    req.type = kmem_type;
    req.ptr  = buf;
    kmem_free( &req );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}   

/////////////////////////////////////////////////////////////////////////////////////////
// [24]          Marshaling functions attached to RPC_MAPPER_MOVE_BUFFER
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////
void rpc_mapper_move_buffer_client( cxy_t      cxy,
                                    mapper_t * mapper,        // in
                                    bool_t     to_buffer,     // in
                                    bool_t     is_user,       // in
                                    uint32_t   file_offset,   // in
                                    uint64_t   buffer,        // in
                                    uint32_t   size,          // in
                                    error_t  * error )        // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)mapper;
    rpc.args[1] = (uint64_t)to_buffer;
    rpc.args[2] = (uint64_t)is_user;
    rpc.args[3] = (uint64_t)file_offset;
    rpc.args[4] = (uint64_t)buffer;
    rpc.args[5] = (uint64_t)size;

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

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

///////////////////////////////////////////////
void rpc_mapper_move_buffer_server( xptr_t xp )
{
    mapper_t * mapper;
    bool_t     to_buffer;
    bool_t     is_user;
    uint32_t   file_offset;
    void     * user_buffer;
    xptr_t     kern_buffer;
    uint32_t   size;
    error_t    error;

    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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 from client RPC descriptor
    mapper      = (mapper_t *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[0] ) );
    to_buffer   =                       hal_remote_lwd( XPTR( client_cxy , &desc->args[1] ) );
    is_user     =                       hal_remote_lwd( XPTR( client_cxy , &desc->args[2] ) );
    file_offset =                       hal_remote_lwd( XPTR( client_cxy , &desc->args[3] ) );
    size        =                       hal_remote_lwd( XPTR( client_cxy , &desc->args[5] ) );

    // call local kernel function
    if( is_user )
    {
        user_buffer = (void *)(intptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[4] ) );

        error = mapper_move_user( mapper,
                                  to_buffer,
                                  file_offset,
                                  user_buffer,
                                  size );
    }
    else
    {
        kern_buffer = (xptr_t)hal_remote_lwd( XPTR( client_cxy , &desc->args[4] ) );

        error = mapper_move_kernel( mapper,
                                    to_buffer,
                                    file_offset,
                                    kern_buffer,
                                    size );
    }

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [25]          Marshaling functions attached to RPC_MAPPER_GET_PAGE
/////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////
void rpc_mapper_get_page_client( cxy_t             cxy,
                                 struct mapper_s * mapper,     // in
                                 uint32_t          index,      // in
                                 page_t         ** page )      // out
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

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

    // get output values from RPC descriptor
    *page = (page_t *)(intptr_t)rpc.args[2];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////////
void rpc_mapper_get_page_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    mapper_t * mapper = (mapper_t *)(intptr_t)hal_remote_lwd( XPTR( cxy , &desc->args[0] ) );
    uint32_t   index  = (uint32_t)            hal_remote_lwd( XPTR( cxy , &desc->args[1] ) );
    
    // call local pmem allocator
    page_t * page = mapper_get_page( mapper , index ); 

    // set output arguments into client RPC descriptor
    hal_remote_swd( XPTR( cxy , &desc->args[1] ) , (uint64_t)(intptr_t)page );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [26]          Marshaling functions attached to RPC_VMM_CREATE_VSEG
/////////////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////////////
void rpc_vmm_create_vseg_client( cxy_t              cxy,
                                 struct process_s * process,
                                 vseg_type_t        type,
                                 intptr_t           base,
                                 uint32_t           size,
                                 uint32_t           file_offset,
                                 uint32_t           file_size,
                                 xptr_t             mapper_xp,
                                 cxy_t              vseg_cxy,
                                 struct vseg_s   ** vseg )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

    // set input arguments in RPC descriptor 
    rpc.args[0] = (uint64_t)(intptr_t)process;
    rpc.args[1] = (uint64_t)type;
    rpc.args[2] = (uint64_t)base;
    rpc.args[3] = (uint64_t)size;
    rpc.args[4] = (uint64_t)file_offset;
    rpc.args[5] = (uint64_t)file_size;
    rpc.args[6] = (uint64_t)mapper_xp;
    rpc.args[7] = (uint64_t)vseg_cxy;

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

    // get output values from RPC descriptor
    *vseg = (vseg_t *)(intptr_t)rpc.args[8];

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////////
void rpc_vmm_create_vseg_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    process_t * process     = (process_t *)(intptr_t)hal_remote_lwd( XPTR(cxy , &desc->args[0]));
    vseg_type_t type        = (vseg_type_t)(uint32_t)hal_remote_lwd( XPTR(cxy , &desc->args[1]));
    intptr_t    base        = (intptr_t)             hal_remote_lwd( XPTR(cxy , &desc->args[2]));
    uint32_t    size        = (uint32_t)             hal_remote_lwd( XPTR(cxy , &desc->args[3]));
    uint32_t    file_offset = (uint32_t)             hal_remote_lwd( XPTR(cxy , &desc->args[4]));
    uint32_t    file_size   = (uint32_t)             hal_remote_lwd( XPTR(cxy , &desc->args[5]));
    xptr_t      mapper_xp   = (xptr_t)               hal_remote_lwd( XPTR(cxy , &desc->args[6]));
    cxy_t       vseg_cxy    = (cxy_t)(uint32_t)      hal_remote_lwd( XPTR(cxy , &desc->args[7]));
    
    // call local kernel function
    vseg_t * vseg = vmm_create_vseg( process,
                                     type,
                                     base,
                                     size,
                                     file_offset,
                                     file_size,
                                     mapper_xp,
                                     vseg_cxy ); 

    // set output arguments into client RPC descriptor
    hal_remote_swd( XPTR( cxy , &desc->args[8] ) , (uint64_t)(intptr_t)vseg );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [27]          Marshaling functions attached to RPC_SCHED_DISPLAY
/////////////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////////////
void rpc_sched_display_client( cxy_t              cxy,
                               lid_t              lid)
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

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

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

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

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

//////////////////////////////////////////
void rpc_sched_display_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    // 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
    lid_t lid = (lid_t)hal_remote_lw( XPTR(cxy , &desc->args[0]));
    
    // call local kernel function
    sched_display( lid );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

/////////////////////////////////////////////////////////////////////////////////////////
// [28]          Marshaling functions attached to RPC_VMM_SET_COW
/////////////////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////
void rpc_vmm_set_cow_client( cxy_t       cxy,
                             process_t * process )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    assert( (cxy != local_cxy) , __FUNCTION__ , "target cluster is not remote\n");

    // initialise RPC descriptor header 
    rpc_desc_t  rpc;
    rpc.index    = RPC_VMM_SET_COW;
    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 );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

////////////////////////////////////////
void rpc_vmm_set_cow_server( xptr_t xp )
{
    rpc_dmsg("\n[DBG] %s : enter / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );

    process_t * process;

    // 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
    process = (process_t *)(intptr_t)hal_remote_lpt( XPTR(cxy , &desc->args[0]));
    
    // call local kernel function
    vmm_set_cow( process );

    rpc_dmsg("\n[DBG] %s : exit / thread %x on core[%x,%d] / cycle %d\n",
    __FUNCTION__ , CURRENT_THREAD->trdid , local_cxy, 
    CURRENT_THREAD->core->lid , hal_time_stamp() );
}

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

////////////////////////////////////////////
void rpc_send_sync( cxy_t        server_cxy, 
                    rpc_desc_t * rpc )
{
    error_t    error;

    thread_t * this = CURRENT_THREAD;
    core_t   * core = this->core;

    // register client thread pointer and core lid in RPC descriptor
    rpc->thread    = this;
    rpc->lid       = core->lid;

    // build an extended pointer on the RPC descriptor
	xptr_t   desc_xp = XPTR( local_cxy , rpc );

    // get local pointer on rpc_fifo in remote cluster, with the 
    // assumption that local pointers are identical in all clusters
    remote_fifo_t * rpc_fifo = &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 , rpc_fifo ),
                                      (uint64_t )desc_xp );
	    if ( error ) 
        {
            printk("\n[WARNING] %s : cluster %x cannot post RPC to cluster %x\n",
            __FUNCTION__ , local_cxy , server_cxy );

            if( thread_can_yield() ) sched_yield("RPC fifo full");
        }
    }
    while( error );
 
    hal_fence();
	
    // send IPI to the remote core corresponding to the client core
	dev_pic_send_ipi( server_cxy , core->lid );

    // wait RPC completion:
    // - busy waiting policy during kernel_init, or if threads cannot yield
    // - block and deschedule in all other cases

    if( (this->type == THREAD_IDLE) || (thread_can_yield() == false) ) // busy waiting
    {

grpc_dmsg("\n[DBG] %s : core[%x,%d] / thread %s busy waiting after registering RPC\n"
"        rpc = %d / server = %x / cycle %d\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , thread_type_str(this->type) ,
rpc->index , server_cxy , hal_time_stamp() );

        while( rpc->response ) hal_fixed_delay( 100 );
    
grpc_dmsg("\n[DBG] %s : core[%x,%d] / thread %s exit after RPC completion\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , thread_type_str(this->type) );

    } 
    else                                                              // block & deschedule
    {

grpc_dmsg("\n[DBG] %s : core[%x,%d] / thread %s deschedule after registering RPC\n"
"        rpc = %d / server = %x / cycle %d\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , thread_type_str(this->type) ,
rpc->index , server_cxy , hal_time_stamp() );

        thread_block( this , THREAD_BLOCKED_RPC );
        sched_yield("client blocked on RPC");

grpc_dmsg("\n[DBG] %s : core[%x,%d] / thread %s resumes after RPC completion\n",
__FUNCTION__ , local_cxy , CURRENT_THREAD->core->lid , thread_type_str(this->type) );

    }

    // check response available
    assert( (rpc->response == 0) , __FUNCTION__, "illegal RPC response\n" );

    // acknowledge the IPI sent by the server
    dev_pic_ack_ipi();
    
}  // end rpc_send_sync()



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

////////////////
void rpc_check()
{
    error_t         error;
    thread_t      * thread;  
    uint32_t        sr_save;

    bool_t          found    = false;
	thread_t      * this     = CURRENT_THREAD;
    core_t        * core     = this->core;
    scheduler_t   * sched    = &core->scheduler;
	remote_fifo_t * rpc_fifo = &LOCAL_CLUSTER->rpc_fifo;

grpc_dmsg("\n[DBG] %s : core[%x,%d] / interrupted thread %s / cycle %d\n",
__FUNCTION__, local_cxy, core->lid, thread_type_str(this->type), hal_time_stamp() );

    // interrupted thread not preemptable during RPC chek
	hal_disable_irq( &sr_save );

    // check RPC FIFO not empty and no RPC thread handling it  
	if( (rpc_fifo->owner == 0) && (local_fifo_is_empty(rpc_fifo) == false) )
    {
        // search one non blocked RPC thread   
        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 == 0 ) ) 
            {
                found = true;
                break;
            }
        }

        // create new RPC thread if not found    
        if( found == false )                    
        {
            error = thread_kernel_create( &thread,
                                          THREAD_RPC, 
			                              &rpc_thread_func, 
                                          NULL,
			                              this->core->lid );
	        if( error ) 
            {
                printk("\n[WARNING] in %s : no memory for new RPC thread in cluster %x\n",
                __FUNCTION__ , local_cxy );
            }
            else
            {
                // unblock created RPC thread
                thread->blocked = 0;

                // update core descriptor counter  
	            hal_atomic_add( &LOCAL_CLUSTER->rpc_threads , 1 );

grpc_dmsg("\n[DBG] %s : core [%x,%d] creates a new RPC thread %x / cycle %d\n", 
__FUNCTION__ , local_cxy , core->lid , thread->trdid , hal_time_stamp() );

            }
        }
    }

grpc_dmsg("\n[DBG] %s : core[%x,%d] / interrupted thread %s deschedules / cycle %d\n", 
__FUNCTION__, local_cxy, core->lid, thread_type_str(this->type), hal_time_stamp() );

    // interrupted thread deschedule always           
	sched_yield("IPI received");

grpc_dmsg("\n[DBG] %s : core[%x,%d] / interrupted thread %s resume / cycle %d\n", 
__FUNCTION__, local_cxy, core->lid, thread_type_str(this->type), hal_time_stamp() );

    // interrupted thread restore IRQs after resume 
	hal_restore_irq( sr_save );

} // end rpc_check()


//////////////////////
void rpc_thread_func()
{
    uint32_t     count;       // handled RPC requests counter
    error_t      empty;       // local RPC fifo state
    xptr_t       desc_xp;     // extended pointer on RPC request
    cxy_t        desc_cxy;    // RPC request cluster (client) 
    rpc_desc_t * desc_ptr;    // RPC request local pointer 
    uint32_t     index;       // RPC request index 
    uint32_t     responses;   // number of responses received by client 
    thread_t   * thread_ptr;  // local pointer on client thread
    lid_t        core_lid;    // local index of client core
 
    // makes RPC thread not preemptable
	hal_disable_irq( NULL );
  
	thread_t      * this     = CURRENT_THREAD;
	remote_fifo_t * rpc_fifo = &LOCAL_CLUSTER->rpc_fifo;

    // two embedded loops:
    // - external loop : "infinite" RPC thread
    // - internal loop : handle up to CONFIG_RPC_PENDING_MAX RPC requests
 
	while(1)  // external loop
	{
        // try to take RPC_FIFO ownership
        if( hal_atomic_test_set( &rpc_fifo->owner , this->trdid ) )
        {
            // initializes RPC requests counter
            count = 0;

            // acknowledge local IPI 
            dev_pic_ack_ipi();

		    // exit internal loop in three cases:
            // - RPC fifo is empty
            // - ownership has been lost (because descheduling)
            // - max number of RPCs is reached
	        while( 1 )  // internal loop
            {
  	            empty = local_fifo_get_item( rpc_fifo , (uint64_t *)&desc_xp );

	            if ( empty == 0 ) // one RPC request found
                {
                    // get client cluster and pointer on RPC descriptor
                    desc_cxy = (cxy_t)GET_CXY( desc_xp );
                    desc_ptr = (rpc_desc_t *)GET_PTR( desc_xp );

                    // get rpc index from RPC descriptor
	                index = hal_remote_lw( XPTR( desc_cxy , &desc_ptr->index ) );

grpc_dmsg("\n[DBG] %s : core[%x,%d] / RPC thread %x / starts rpc %d / cycle %d\n",
__FUNCTION__ , local_cxy , this->core->lid , this->trdid , index , hal_time_stamp() );

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

grpc_dmsg("\n[DBG] %s : core[%x,%d] / RPC thread %x / completes rpc %d / cycle %d\n",
__FUNCTION__ , local_cxy , this->core->lid , this->trdid , index , hal_time_stamp() );

                    // increment handled RPC counter
  	                count++;

                    // decrement response counter in RPC descriptor
                    responses = hal_remote_atomic_add(XPTR( desc_cxy, &desc_ptr->response ), -1);

                    // unblock client thread  and send IPI to client core if last response
                    if( responses == 1 )
                    { 
                        // get pointer on client thread and unblock it
                        thread_ptr = (thread_t *)hal_remote_lpt(XPTR(desc_cxy,&desc_ptr->thread));
                        thread_unblock( XPTR(desc_cxy,thread_ptr) , THREAD_BLOCKED_RPC );

                        hal_fence();

                        // get client core lid and send IPI
                        core_lid = hal_remote_lw(XPTR(desc_cxy, &desc_ptr->lid));
	                    dev_pic_send_ipi( desc_cxy , core_lid );
                    }
		        }
	
                // chek exit condition
  		        if( local_fifo_is_empty( rpc_fifo )  || 
                    (rpc_fifo->owner != this->trdid) || 
                    (count >= CONFIG_RPC_PENDING_MAX) ) break;
         	} // end internal loop

            // release rpc_fifo ownership if not lost
            if( rpc_fifo->owner == this->trdid ) rpc_fifo->owner = 0;
        }

        // sucide if too many RPC threads in cluster
        if( LOCAL_CLUSTER->rpc_threads >= CONFIG_RPC_THREADS_MAX )
	    {

grpc_dmsg("\n[DBG] %s : core[%x,%d] (RPC thread %x) suicide at cycle %d\n", 
__FUNCTION__, local_cxy, this->core->lid, this->trdid, hal_time_stamp() );

            // update RPC threads counter
	        hal_atomic_add( &LOCAL_CLUSTER->rpc_threads , -1 );

            // suicide
	        thread_exit();
	    }

grpc_dmsg("\n[DBG] %s : core[%x,%d] (RPC thread %x) deschedules / cycle %d\n", 
__FUNCTION__, local_cxy, this->core->lid, this->trdid, hal_time_stamp() );

        // deschedule without blocking
        sched_yield("RPC fifo empty or too much work");

grpc_dmsg("\n[DBG] %s : core[%x,%d] (RPC thread %x) resumes / cycle %d\n", 
__FUNCTION__, local_cxy, this->core->lid, this->trdid, hal_time_stamp() );

	} // end external loop

} // end rpc_thread_func()


