source: trunk/kernel/mm/mapper.c @ 627

/*
 * mapper.c - Kernel cache for FS files or directories implementation.
 *
 * Authors   Mohamed Lamine Karaoui (2015)
 *           Alain Greiner (2016,2017,2018,2019)
 *
 * 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_kernel_types.h>
#include <hal_special.h>
#include <hal_uspace.h>
#include <grdxt.h>
#include <string.h>
#include <rwlock.h>
#include <printk.h>
#include <memcpy.h>
#include <thread.h>
#include <core.h>
#include <process.h>
#include <kmem.h>
#include <kcm.h>
#include <ppm.h>
#include <page.h>
#include <cluster.h>
#include <vfs.h>
#include <mapper.h>
#include <dev_ioc.h>


//////////////////////////////////////////////
mapper_t * mapper_create( vfs_fs_type_t type )
{
    mapper_t * mapper;
    kmem_req_t req;
    error_t    error;

    // allocate memory for mapper
    req.type  = KMEM_MAPPER;
    req.size  = sizeof(mapper_t);
    req.flags = AF_KERNEL | AF_ZERO;
    mapper    = (mapper_t *)kmem_alloc( &req );

    if( mapper == NULL )
    {
        printk("\n[ERROR] in %s : no memory for mapper descriptor\n", __FUNCTION__ );
        return NULL;
    }

    // initialize refcount & inode
    mapper->refcount = 0;
    mapper->inode    = NULL;

    // initialize radix tree
    error = grdxt_init( &mapper->rt,
                        CONFIG_MAPPER_GRDXT_W1,
                        CONFIG_MAPPER_GRDXT_W2,
                        CONFIG_MAPPER_GRDXT_W3 );

    if( error )
    {
        printk("\n[ERROR] in %s : cannot initialize radix tree\n", __FUNCTION__ );
        req.type  = KMEM_MAPPER;
        req.ptr   = mapper;
        kmem_free( &req );
        return NULL;
    }

    // initialize mapper type
    mapper->type = type;

    // initialize mapper lock
    remote_rwlock_init( XPTR( local_cxy , &mapper->lock ) , LOCK_MAPPER_STATE );

    // initialize waiting threads xlist (empty)
    xlist_root_init( XPTR( local_cxy , &mapper->wait_root ) );

    // initialize vsegs xlist (empty)
    xlist_root_init( XPTR( local_cxy , &mapper->vsegs_root ) );

    return mapper;

}  // end mapper_create()

////////////////////////////////////////
void mapper_destroy( mapper_t * mapper )
{
    page_t   * page;
    uint32_t   found_index = 0;
    uint32_t   start_index = 0;
    kmem_req_t req;

    // scan radix tree
    do
    {
        // get page from radix tree
        page = (page_t *)grdxt_get_first( &mapper->rt , start_index , &found_index );

        // release registered pages to PPM
        if( page != NULL )
        {
            // remove page from mapper and release to PPM
            mapper_release_page( mapper , page );

            // update start_key value for next page
            start_index = found_index;
        }
    }
    while( page != NULL );

    // release the memory allocated to radix tree itself
    grdxt_destroy( &mapper->rt );

    // release memory for mapper descriptor
    req.type = KMEM_MAPPER;
    req.ptr  = mapper;
    kmem_free( &req );

}  // end mapper_destroy()

////////////////////////////////////////////////////
xptr_t  mapper_remote_get_page( xptr_t    mapper_xp,
                                uint32_t  page_id )
{
    error_t       error;
    mapper_t    * mapper_ptr;
    cxy_t         mapper_cxy;
    xptr_t        lock_xp;        // extended pointer on mapper lock
    xptr_t        page_xp;        // extended pointer on searched page descriptor
    xptr_t        rt_xp;          // extended pointer on radix tree in mapper

    thread_t * this = CURRENT_THREAD;

    // get mapper cluster and local pointer
    mapper_ptr = GET_PTR( mapper_xp );
    mapper_cxy = GET_CXY( mapper_xp );

#if DEBUG_MAPPER_GET_PAGE
vfs_inode_t * inode = hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) );
uint32_t      cycle = (uint32_t)hal_get_cycles();
char          name[CONFIG_VFS_MAX_NAME_LENGTH];
if( (DEBUG_MAPPER_GET_PAGE < cycle) && (inode == NULL) )  // FAT mapper
{
    printk("\n[%s] thread[%x,%x] enter for page %d of FAT mapper / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, page_id, cycle );
}
if( (DEBUG_MAPPER_GET_PAGE < cycle) && (inode != NULL) )  // file mapper
{
    vfs_inode_get_name( XPTR( mapper_cxy , inode ) , name );
    printk("\n[%s] thread[%x,%x] enter for page %d of <%s> mapper / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, page_id, name, cycle );
}
#endif

    // check thread can yield
    thread_assert_can_yield( this , __FUNCTION__ );

    // build extended pointer on mapper lock and mapper rt
    lock_xp  = XPTR( mapper_cxy , &mapper_ptr->lock );
    rt_xp    = XPTR( mapper_cxy , &mapper_ptr->rt );

    // take mapper lock in READ_MODE
    remote_rwlock_rd_acquire( lock_xp );

    // search page in radix tree
    page_xp  = grdxt_remote_lookup( rt_xp , page_id );

    // test mapper miss 
    if( page_xp == XPTR_NULL )                  // miss => try to handle it
    {
        // release the lock in READ_MODE and take it in WRITE_MODE
        remote_rwlock_rd_release( lock_xp );
        remote_rwlock_wr_acquire( lock_xp );

        // second test on missing page because the page status can be modified
        // by another thread, when passing from READ_MODE to WRITE_MODE.
        // from this point there is no concurrent accesses to mapper.
        page_xp = grdxt_remote_lookup( rt_xp , page_id );

        if ( page_xp == XPTR_NULL )  // miss confirmed => handle it
        {

            if( mapper_cxy == local_cxy )   // mapper is local
            {

#if (DEBUG_MAPPER_GET_PAGE & 1)
if( DEBUG_MAPPER_GET_PAGE < cycle )
printk("\n[%s] missing page => load it from FS / local access \n", __FUNCTION__ );
#endif
                 error = mapper_handle_miss( mapper_ptr,
                                             page_id, 
                                             &page_xp );
            } 
            else
            {

#if (DEBUG_MAPPER_GET_PAGE & 1)
if( DEBUG_MAPPER_GET_PAGE < cycle )
printk("\n[%s] missing page => load it from FS / RPC access \n", __FUNCTION__ );
#endif
                 rpc_mapper_handle_miss_client( mapper_cxy,
                                                mapper_ptr,
                                                page_id,
                                                &page_xp,
                                                &error );
            }

            if ( error )
            {
                printk("\n[ERROR] in %s : thread[%x,%x] cannot handle mapper miss\n",
                __FUNCTION__ , this->process->pid, this->trdid );
                remote_rwlock_wr_release( lock_xp );
                return XPTR_NULL;
            }
        }
        
        // release mapper lock from WRITE_MODE
        remote_rwlock_wr_release( lock_xp );
    }
    else                                              // hit 
    {
        // release mapper lock from READ_MODE
        remote_rwlock_rd_release( lock_xp );
    }

#if DEBUG_MAPPER_GET_PAGE
cycle = (uint32_t)hal_get_cycles();
if( (DEBUG_MAPPER_GET_PAGE < cycle) && (inode != NULL) )
{
    printk("\n[%s] thread[%x,%x] exit for page %d of <%s> mapper / ppn %x / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, page_id,
    name, ppm_page2ppn(page_xp), cycle );
}
if( (DEBUG_MAPPER_GET_PAGE < cycle) && (inode == NULL) )
{
    printk("\n[%s] thread[%x,%x] exit for page %d of FAT mapper  / ppn %x / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, page_id,
    ppm_page2ppn(page_xp), cycle );
}
#endif

    return page_xp;

}  // end mapper_remote_get_page()

//////////////////////////////////////////////
error_t mapper_handle_miss( mapper_t * mapper,
                            uint32_t   page_id,
                            xptr_t   * page_xp )
{
    kmem_req_t   req;
    page_t     * page;
    error_t      error;

    thread_t * this = CURRENT_THREAD;

#if DEBUG_MAPPER_HANDLE_MISS
uint32_t      cycle = (uint32_t)hal_get_cycles();
char          name[CONFIG_VFS_MAX_NAME_LENGTH];
vfs_inode_t * inode = mapper->inode;
if( (DEBUG_MAPPER_HANDLE_MISS < cycle) && (inode != NULL) )
{
    vfs_inode_get_name( XPTR( local_cxy , inode ) , name );
    printk("\n[%s] thread[%x,%x] enter for page %d in <%s> / cycle %d",
    __FUNCTION__, this->process->pid, this->trdid, page_id, name, cycle );
    if( DEBUG_MAPPER_HANDLE_MISS & 1 ) grdxt_display( XPTR(local_cxy,&mapper->rt), name );
}
if( (DEBUG_MAPPER_HANDLE_MISS < cycle) && (inode == NULL) )
{
    printk("\n[%s] thread[%x,%x] enter for page %d in FAT / cycle %d",
    __FUNCTION__, this->process->pid, this->trdid, page_id, cycle );
    if( DEBUG_MAPPER_HANDLE_MISS & 1 ) grdxt_display( XPTR(local_cxy,&mapper->rt), "FAT" );
}
#endif

    // allocate one page from the local cluster
    req.type  = KMEM_PAGE;
    req.size  = 0;
    req.flags = AF_NONE;
    page = kmem_alloc( &req );

    if( page == NULL )
    {
        printk("\n[ERROR] in %s : thread [%x,%x] cannot allocate page in cluster %x\n",
        __FUNCTION__ , this->process->pid, this->trdid , local_cxy );
        return -1;
    }

    // initialize the page descriptor 
    page_init( page );
    page_set_flag( page , PG_INIT );
    page_refcount_up( page );
    page->mapper = mapper;
    page->index  = page_id;

    // insert page in mapper radix tree 
    error = grdxt_insert( &mapper->rt , page_id , page );

    if( error )
    {
        printk("\n[ERROR] in %s : thread[%x,%x] cannot insert page in mapper\n",
        __FUNCTION__ , this->process->pid, this->trdid );
        mapper_release_page( mapper , page );
        req.ptr  = page;
        req.type = KMEM_PAGE;
        kmem_free(&req);
        return -1;
    }

    // launch I/O operation to load page from IOC device to mapper
    error = vfs_fs_move_page( XPTR( local_cxy , page ) , IOC_SYNC_READ );

    if( error )
    {
        printk("\n[ERROR] in %s : thread[%x,%x] cannot load page from device\n",
        __FUNCTION__ , this->process->pid, this->trdid );
        mapper_release_page( mapper , page );
        req.ptr  = page;
        req.type = KMEM_PAGE;
        kmem_free( &req );
        return -1;
    }

    // set extended pointer on allocated page
    *page_xp = XPTR( local_cxy , page );

#if DEBUG_MAPPER_HANDLE_MISS
cycle = (uint32_t)hal_get_cycles();
if( (DEBUG_MAPPER_HANDLE_MISS < cycle) && (inode != NULL) )
{
    printk("\n[%s] thread[%x,%x] exit for page %d in <%s> / ppn %x / cycle %d",
    __FUNCTION__, this->process->pid, this->trdid,
    page_id, name, ppm_page2ppn( *page_xp ), cycle );
    if( DEBUG_MAPPER_HANDLE_MISS & 1 ) grdxt_display( XPTR(local_cxy,&mapper->rt) , name );
}
if( (DEBUG_MAPPER_HANDLE_MISS < cycle) && (inode == NULL) )
{
    printk("\n[%s] thread[%x,%x] exit for page %d in FAT / ppn %x / cycle %d",
    __FUNCTION__, this->process->pid, this->trdid,
    page_id, ppm_page2ppn( *page_xp ), cycle );
    if( DEBUG_MAPPER_HANDLE_MISS & 1 ) grdxt_display( XPTR(local_cxy,&mapper->rt ), "FAT" );
}
#endif

    return 0;

}  // end mapper_handle_miss()

////////////////////////////////////////////
void mapper_release_page( mapper_t * mapper,
                          page_t   * page )
{
    // build extended pointer on mapper lock
    xptr_t mapper_lock_xp = XPTR( local_cxy , &mapper->lock );

    // take mapper lock in WRITE_MODE
    remote_rwlock_wr_acquire( mapper_lock_xp );

    // remove physical page from radix tree
    grdxt_remove( &mapper->rt , page->index );

    // release mapper lock from WRITE_MODE
    remote_rwlock_wr_release( mapper_lock_xp );

    // release page to PPM
    kmem_req_t   req;
    req.type  = KMEM_PAGE;
    req.ptr   = page;
    kmem_free( &req );

}  // end mapper_release_page()

///////////////////////////////////////////////
error_t mapper_move_user( xptr_t     mapper_xp,
                          bool_t     to_buffer,
                          uint32_t   file_offset,
                          void     * buffer,
                          uint32_t   size )
{
    uint32_t   page_offset;    // first byte to move to/from a mapper page
    uint32_t   page_count;     // number of bytes to move to/from a mapper page
    uint32_t   page_id;        // current mapper page index
    uint32_t   done;           // number of moved bytes
    xptr_t     page_xp;        // extended pointer on current mapper page descriptor

#if DEBUG_MAPPER_MOVE_USER
uint32_t      cycle      = (uint32_t)hal_get_cycles();
thread_t    * this       = CURRENT_THREAD;
cxy_t         mapper_cxy = GET_CXY( mapper_xp );
mapper_t    * mapper_ptr = GET_PTR( mapper_xp );
vfs_inode_t * inode_ptr  = hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) );
xptr_t        inode_xp   = XPTR( mapper_cxy , inode_ptr );
char          name[CONFIG_VFS_MAX_NAME_LENGTH];
vfs_inode_get_name( inode_xp , name );
if( DEBUG_MAPPER_MOVE_USER < cycle )
{
    if( to_buffer )
    printk("\n[%s] thread[%x,%x] : mapper(%s) -> buffer(%x) / bytes %d / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, name, buffer, size, cycle );
    else
    printk("\n[%s] thread[%x,%x] : buffer(%x) -> mapper(%s) / bytes %d / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, buffer, name, size, cycle );
}
#endif

    // compute offsets of first and last bytes in file
    uint32_t min_byte = file_offset;
    uint32_t max_byte = file_offset + size - 1;

    // compute indexes of pages for first and last byte in mapper
    uint32_t first = min_byte >> CONFIG_PPM_PAGE_SHIFT;
    uint32_t last  = max_byte >> CONFIG_PPM_PAGE_SHIFT;

#if (DEBUG_MAPPER_MOVE_USER & 1)
if( DEBUG_MAPPER_MOVE_USER < cycle )
printk("\n[%s] thread[%x,%x] : mapper(%x,%x) / first_page %d / last_page %d\n",
__FUNCTION__, this->process->pid, this->trdid, mapper_cxy, mapper_ptr, first, last );
#endif

    done = 0;

    // loop on pages in mapper
    for( page_id = first ; page_id <= last ; page_id++ )
    {
        // compute page_offset
        if( page_id == first ) page_offset = min_byte & CONFIG_PPM_PAGE_MASK;
        else                   page_offset = 0;

        // compute number of bytes in page
        if      ( first   == last  ) page_count = size;
        else if ( page_id == first ) page_count = CONFIG_PPM_PAGE_SIZE - page_offset;
        else if ( page_id == last  ) page_count = (max_byte & CONFIG_PPM_PAGE_MASK) + 1;
        else                         page_count = CONFIG_PPM_PAGE_SIZE;

#if (DEBUG_MAPPER_MOVE_USER & 1)
if( DEBUG_MAPPER_MOVE_USER < cycle )
printk("\n[%s] thread[%x,%x] : page_id %d / page_offset %d / bytes %d\n",
__FUNCTION__, this->process->pid, this->trdid, page_id , page_offset , page_count );
#endif

        // get extended pointer on page descriptor
        page_xp = mapper_remote_get_page( mapper_xp , page_id ); 

        if ( page_xp == XPTR_NULL ) return -1;

        // compute cluster and pointers on page in mapper
        xptr_t     map_xp  = ppm_page2base( page_xp );
        uint8_t  * map_ptr = GET_PTR( map_xp );
        cxy_t      map_cxy = GET_CXY( map_xp );

#if (DEBUG_MAPPER_MOVE_USER & 1)
if( DEBUG_MAPPER_MOVE_USER < cycle )
printk("\n[%s] thread[%x,%x] : get buffer(%x,%x) in mapper\n",
__FUNCTION__, this->process->pid, this->trdid, map_cxy, map_ptr );
#endif
        // compute pointer in user buffer
        uint8_t * buf_ptr = (uint8_t *)buffer + done;

        // move fragment
        if( to_buffer )
        {
            hal_copy_to_uspace( map_cxy , map_ptr , buf_ptr , page_count ); 

#if DEBUG_MAPPER_MOVE_USER & 1
if( DEBUG_MAPPER_MOVE_USER < cycle )
printk("\n[%s] thread[%x,%x] moved %d bytes / mapper %s (%x,%x) -> user buffer(%x,%x)\n",
__FUNCTION__, this->process->pid, this->trdid, page_count,
name, map_cxy, map_ptr, local_cxy, buf_ptr );
#endif

        }
        else
        {
            ppm_page_do_dirty( page_xp ); 
            hal_copy_from_uspace( map_cxy , map_ptr , buf_ptr , page_count ); 

#if DEBUG_MAPPER_MOVE_USER & 1
if( DEBUG_MAPPER_MOVE_USER < cycle )
printk("\n[%s] thread[%x,%x] moved %d bytes / user buffer(%x,%x) -> mapper %s (%x,%x)\n",
__FUNCTION__, this->process->pid, this->trdid, page_count,
local_cxy, buf_ptr, name, map_cxy, map_ptr );
mapper_display_page(  mapper_xp , page_id, 128 );
#endif

        }

        done += page_count;
    }

#if DEBUG_MAPPER_MOVE_USER
cycle      = (uint32_t)hal_get_cycles();
if( DEBUG_MAPPER_MOVE_USER < cycle )
{
    if( to_buffer )
    printk("\n[%s] thread[%x,%x] completed mapper(%s) -> buffer(%x) / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, name, buffer, cycle );
    else
    printk("\n[%s] thread[%x,%x] completed buffer(%x) -> mapper(%s) / cycle %d\n",
    __FUNCTION__, this->process->pid, this->trdid, buffer, name, cycle );
}
#endif

    return 0;

}  // end mapper_move_user()

////////////////////////////////////////////////
error_t mapper_move_kernel( xptr_t    mapper_xp,
                            bool_t    to_buffer,
                            uint32_t  file_offset,
                            xptr_t    buffer_xp,
                            uint32_t  size )
{
    uint32_t   page_offset;    // first byte to move to/from a mapper page
    uint32_t   page_count;     // number of bytes to move to/from a mapper page
    uint32_t   page_id;        // current mapper page index
    uint32_t   done;           // number of moved bytes
    xptr_t     page_xp;        // extended pointer on current mapper page descriptor

    uint8_t  * src_ptr;        // source buffer local pointer
    cxy_t      src_cxy;        // source cluster
    uint8_t  * dst_ptr;        // destination buffer local pointer
    cxy_t      dst_cxy;        // destination cluster

    // get buffer cluster and local pointer
    cxy_t     buffer_cxy = GET_CXY( buffer_xp );
    uint8_t * buffer_ptr = GET_PTR( buffer_xp );

    // get mapper cluster
    cxy_t     mapper_cxy = GET_CXY( mapper_xp );

#if DEBUG_MAPPER_MOVE_KERNEL
char          name[CONFIG_VFS_MAX_NAME_LENGTH];
uint32_t      cycle  = (uint32_t)hal_get_cycles();
thread_t    * this   = CURRENT_THREAD;
mapper_t    * mapper = GET_PTR( mapper_xp );
vfs_inode_t * inode  = hal_remote_lpt( XPTR( mapper_cxy , &mapper->inode ) );
vfs_inode_get_name( XPTR( mapper_cxy , inode ) , name );
if( DEBUG_MAPPER_MOVE_KERNEL < cycle )
printk("\n[%s] thread[%x,%x] enter / %d bytes / offset %d / mapper <%s> / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, size, file_offset, name, cycle );
#endif

    // compute offsets of first and last bytes in file
    uint32_t min_byte = file_offset;
    uint32_t max_byte = file_offset + size -1;

    // compute indexes for first and last pages in mapper
    uint32_t first = min_byte >> CONFIG_PPM_PAGE_SHIFT;
    uint32_t last  = max_byte >> CONFIG_PPM_PAGE_SHIFT;

    // compute source and destination clusters
    if( to_buffer )
    {
        dst_cxy = buffer_cxy;
        src_cxy = mapper_cxy;
    }
    else
    {
        src_cxy = buffer_cxy;
        dst_cxy = mapper_cxy;
    }

    done = 0;

    // loop on pages in mapper
    for( page_id = first ; page_id <= last ; page_id++ )
    {
        // compute page_offset
        if( page_id == first ) page_offset = min_byte & CONFIG_PPM_PAGE_MASK;
        else                   page_offset = 0;

        // compute number of bytes to move in page
        if      ( first == last  )   page_count = size;
        else if ( page_id == first ) page_count = CONFIG_PPM_PAGE_SIZE - page_offset;
        else if ( page_id == last  ) page_count = (max_byte & CONFIG_PPM_PAGE_MASK) + 1;
        else                         page_count = CONFIG_PPM_PAGE_SIZE;

        // get extended pointer on page descriptor
        page_xp = mapper_remote_get_page( mapper_xp , page_id );

        if ( page_xp == XPTR_NULL ) return -1;

        // get page base address
        xptr_t    base_xp  = ppm_page2base( page_xp );
        uint8_t * base_ptr = (uint8_t *)GET_PTR( base_xp );

        // compute source and destination pointers
        if( to_buffer )
        {
            dst_ptr = buffer_ptr + done;
            src_ptr = base_ptr + page_offset;
        }
        else
        {
            src_ptr = buffer_ptr + done;
            dst_ptr = base_ptr + page_offset;

            ppm_page_do_dirty( page_xp );
        }

#if (DEBUG_MAPPER_MOVE_KERNEL & 1)
if( DEBUG_MAPPER_MOVE_KERNEL < cycle )
{
    if( to_buffer )
    printk("\n[%s] mapper <%s> page %d => buffer(%x,%x) / %d bytes\n",
    __FUNCTION__, name, page_id, dst_cxy, dst_ptr, page_count );
    else
    printk("\n[%s] buffer(%x,%x) => mapper <%s> page %d / %d bytes\n",
    __FUNCTION__, src_cxy, src_ptr, name, page_id, page_count );
}
#endif

        // move fragment
        hal_remote_memcpy( XPTR( dst_cxy , dst_ptr ), XPTR( src_cxy , src_ptr ), page_count );

        done += page_count;
    }

#if DEBUG_MAPPER_MOVE_KERNEL
cycle  = (uint32_t)hal_get_cycles();
if( DEBUG_MAPPER_MOVE_KERNEL < cycle )
printk("\n[%s] thread[%x,%x] exit / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, cycle );
#endif

    return 0;

}  // end mapper_move_kernel()

///////////////////////////////////////////////////
error_t mapper_remote_get_32( xptr_t     mapper_xp,
                              uint32_t   word_id,
                              uint32_t * p_value )
{
    uint32_t   page_id;      // page index in file
    uint32_t   local_id;     // word index in page
    xptr_t     page_xp;      // extended pointer on searched page descriptor
    xptr_t     base_xp;      // extended pointer on searched page base 

    
    // get page index and local word index
    page_id  = word_id >> 10;
    local_id = word_id & 0x3FF;

    // get page containing the searched word
    page_xp  = mapper_remote_get_page( mapper_xp , page_id );

    if( page_xp == XPTR_NULL )  return -1;
    
    // get page base
    base_xp = ppm_page2base( page_xp );

    // get the value from mapper
    *p_value = hal_remote_l32( base_xp + (local_id<<2) ); 

    return 0;

}  // end mapper_remote_get_32()

///////////////////////////////////////////////////
error_t mapper_remote_set_32( xptr_t     mapper_xp,
                              uint32_t   word_id,
                              uint32_t   value )
{
    
    uint32_t   page_id;      // page index in file
    uint32_t   local_id;     // word index in page
    xptr_t     page_xp;      // extended pointer on searched page descriptor
    xptr_t     base_xp;      // extended pointer on searched page base 

    // get page index and local vord index
    page_id  = word_id >> 10;
    local_id = word_id & 0x3FF;

    // get page containing the searched word
    page_xp  = mapper_remote_get_page( mapper_xp , page_id );

    if( page_xp == XPTR_NULL ) return -1;

    // get page base
    base_xp = ppm_page2base( page_xp );

    // set value to mapper
    hal_remote_s32( (base_xp + (local_id << 2)) , value );

    // set the dirty flag
    ppm_page_do_dirty( page_xp );

    return 0;

}  // end mapper_remote_set_32()

/////////////////////////////////////////
error_t mapper_sync( mapper_t *  mapper )
{
    page_t   * page;                // local pointer on current page descriptor
    xptr_t     page_xp;             // extended pointer on current page descriptor
    grdxt_t  * rt;                  // pointer on radix_tree descriptor
    uint32_t   start_key;           // start page index in mapper
    uint32_t   found_key;           // current page index in mapper
    error_t    error;

#if DEBUG_MAPPER_SYNC
thread_t * this  = CURRENT_THREAD;
uint32_t   cycle = (uint32_t)hal_get_cycles();
char       name[CONFIG_VFS_MAX_NAME_LENGTH];
vfs_inode_get_name( XPTR( local_cxy , mapper->inode ) , name );
#endif

    // get pointer on radix tree
    rt = &mapper->rt;

    // initialise loop variable 
    start_key = 0;

    // scan radix-tree until last page found
    while( 1 )
    {
        // get page descriptor from radix tree
        page = (page_t *)grdxt_get_first( rt , start_key , &found_key );
         
        if( page == NULL ) break;

assert( (page->index == found_key ), "wrong page descriptor index" );
assert( (page->order == 0),          "mapper page order must be 0" );

        // build extended pointer on page descriptor
        page_xp = XPTR( local_cxy , page );

        // synchronize page if dirty
        if( (page->flags & PG_DIRTY) != 0 )
        {

#if DEBUG_MAPPER_SYNC
if( cycle > DEBUG_MAPPER_SYNC )
printk("\n[%s] thread[%x,%x] synchonise page %d of <%s> to IOC device\n",
__FUNCTION__, this->process->pid, this->trdid, page->index, name );
#endif
            // copy page to file system
            error = vfs_fs_move_page( page_xp , IOC_WRITE );

            if( error )
            {
                printk("\n[ERROR] in %s : cannot synchonize dirty page %d\n", 
                __FUNCTION__, page->index );
                return -1;
            }

            // remove page from PPM dirty list
            ppm_page_undo_dirty( page_xp ); 
        } 
        else
        {

#if DEBUG_MAPPER_SYNC
if( cycle > DEBUG_MAPPER_SYNC )
printk("\n[%s] thread[%x,%x] skip page %d for <%s>\n",
__FUNCTION__, this->process->pid, this->trdid, page->index, name );
#endif
        }

        // update loop variable
        start_key = page->index + 1;
    }  // end while

    return 0;

}  // end mapper_sync()

//////////////////////////////////////////////////
error_t mapper_display_page( xptr_t     mapper_xp,
                             uint32_t   page_id,
                             uint32_t   nbytes )
{
    xptr_t        page_xp;        // extended pointer on page descriptor
    xptr_t        base_xp;        // extended pointer on page base
    char          buffer[4096];   // local buffer
    uint32_t    * tabi;           // pointer on uint32_t to scan buffer
    uint32_t      line;           // line index
    uint32_t      word;           // word index
    cxy_t         mapper_cxy;     // mapper cluster identifier
    mapper_t    * mapper_ptr;     // mapper local pointer
    vfs_inode_t * inode_ptr;      // inode local pointer
 
    char       name[CONFIG_VFS_MAX_NAME_LENGTH];

    if( nbytes > 4096)
    {
        printk("\n[ERROR] in %s : nbytes (%d) cannot be larger than 4096\n",
        __FUNCTION__, nbytes );
        return -1;
    }
    
    // get extended pointer on page descriptor
    page_xp = mapper_remote_get_page( mapper_xp , page_id );

    if( page_xp == XPTR_NULL)
    {
        printk("\n[ERROR] in %s : cannot access page %d in mapper\n",
        __FUNCTION__, page_id );
        return -1;
    }

    // get cluster and local pointer
    mapper_cxy = GET_CXY( mapper_xp );
    mapper_ptr = GET_PTR( mapper_xp );

    // get inode
    inode_ptr = hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) );

    // get inode name
    if( inode_ptr == NULL ) strcpy( name , "fat" );
    else  vfs_inode_get_name( XPTR( mapper_cxy , inode_ptr ) , name );
    
    // get extended pointer on page base
    base_xp = ppm_page2base( page_xp );
   
    // copy remote page to local buffer
    hal_remote_memcpy( XPTR( local_cxy , buffer ) , base_xp , nbytes );

    // display 8 words per line
    tabi = (uint32_t *)buffer;
    printk("\n***** mapper <%s> / %d bytes in page %d (%x,%x)\n",
    name, nbytes, page_id, GET_CXY(base_xp), GET_PTR(base_xp) );
    for( line = 0 ; line < (nbytes >> 5) ; line++ )
    {
        printk("%X : ", line << 5 );
        for( word = 0 ; word < 8 ; word++ ) printk("%X ", tabi[(line<<3) + word] );
        printk("\n");
    }

    return 0;

}  // end mapper_display_page