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

/*
 * mapper.c - Kernel cache for FS files or directories implementation.
 *
 * Authors   Mohamed Lamine Karaoui (2015)
 *           Alain Greiner (2016,2017,2018,2019,2020)
 *
 * 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 <hal_vmm.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>


/////////////////////////////////////
xptr_t  mapper_create( cxy_t     cxy,
                       uint32_t  type )
{
    mapper_t * mapper_ptr;
    kmem_req_t req;
    error_t    error;

    // allocate memory for mapper descriptor
    req.type    = KMEM_KCM;
    req.order   = bits_log2( sizeof(mapper_t) );
    req.flags   = AF_KERNEL | AF_ZERO;
    mapper_ptr  = kmem_remote_alloc( cxy , &req );

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

    // initialize refcount and type
    hal_remote_s32( XPTR( cxy , &mapper_ptr->refcount ) , 0 );
    hal_remote_s32( XPTR( cxy , &mapper_ptr->fs_type )  , type );

    // initialize radix tree
    error = grdxt_remote_init( XPTR( cxy , &mapper_ptr->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_KCM;
        req.ptr   = mapper_ptr;
        kmem_remote_free( cxy , &req );
        return XPTR_NULL;
    }

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

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

    // initialize vsegs xlist (empty)
    xlist_root_init( XPTR( cxy , &mapper_ptr->vsegs_root ) );

    return XPTR( cxy , mapper_ptr );

}  // end mapper_create()

////////////////////////////////////////
void mapper_destroy( xptr_t  mapper_xp )
{
    xptr_t     page_xp;
    page_t   * page;
    uint32_t   found_index = 0;
    uint32_t   start_index = 0;
    kmem_req_t req;

    cxy_t      mapper_cxy = GET_CXY( mapper_xp ); 
    mapper_t * mapper_ptr = GET_PTR( mapper_xp );

    // build extended pointer on radix tree
    xptr_t rt_xp = XPTR( mapper_cxy , &mapper_ptr->rt );

    // scan radix tree
    do
    {
        // get page from radix tree
        page_xp = grdxt_remote_get_first( rt_xp,
                                          start_index , 
                                          &found_index );
        page = GET_PTR( page_xp );
        
        // release registered pages to PPM
        if( page != NULL )
        {
            // remove page from mapper and release to PPM
            mapper_remote_release_page( mapper_xp , page );

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

    // release the memory allocated to radix tree itself
    grdxt_remote_destroy( rt_xp );

    // release memory for mapper descriptor
    req.type = KMEM_KCM;
    req.ptr  = mapper_ptr;
    kmem_remote_free( mapper_cxy , &req );

}  // end mapper_destroy()

/////////////////////////////////////////////////
error_t mapper_handle_miss( xptr_t     mapper_xp,
                            uint32_t   page_id,
                            xptr_t   * page_xp_ptr )
{
    error_t    error;

    uint32_t   inode_size = 0;    
    uint32_t   inode_type = 0;

    thread_t * this = CURRENT_THREAD;

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

    // get inode pointer
    vfs_inode_t * inode = hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) );

    // get inode size and type if relevant
    if( inode != NULL )
    {
        inode_size = hal_remote_l32( XPTR( mapper_cxy , &inode->size ) );
        inode_type = hal_remote_l32( XPTR( mapper_cxy , &inode->type ) );
    }

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

#if( DEBUG_MAPPER_HANDLE_MISS & 2 ) 
if( DEBUG_MAPPER_HANDLE_MISS < cycle ) 
{
    if (inode != NULL) grdxt_remote_display( XPTR( mapper_cxy , &mapper_ptr->rt ) , name );
    else               grdxt_remote_display( XPTR( mapper_cxy , &mapper_ptr->rt ) , "FAT" );
}
#endif

    // allocate one 4 Kbytes page from the remote mapper cluster
    xptr_t page_xp = ppm_remote_alloc_pages( mapper_cxy , 0 );
    page_t * page_ptr = GET_PTR( page_xp );
                            
    if( page_xp == XPTR_NULL )
    {
        printk("\n[ERROR] in %s : thread [%x,%x] cannot allocate page in cluster %x\n",
        __FUNCTION__ , this->process->pid, this->trdid , mapper_cxy );
        return -1;
    }

    // initialize the page descriptor 
    page_remote_init( page_xp );

    // initialize specific page descriptor fields
    hal_remote_s32( XPTR( mapper_cxy , &page_ptr->refcount ) , 1          );
    hal_remote_s32( XPTR( mapper_cxy , &page_ptr->index )    , page_id    );
    hal_remote_spt( XPTR( mapper_cxy , &page_ptr->mapper )   , mapper_ptr );
    hal_remote_s32( XPTR( mapper_cxy , &page_ptr->flags )    , PG_INIT    );

    // insert page in mapper radix tree 
    error = grdxt_remote_insert( XPTR( mapper_cxy , &mapper_ptr->rt),
                                 page_id,
                                 page_ptr );

    if( error )
    {
        printk("\n[ERROR] in %s : thread[%x,%x] cannot insert page in mapper\n",
        __FUNCTION__ , this->process->pid, this->trdid );
        ppm_remote_free_pages( mapper_cxy , page_ptr );
        return -1;
    }

    // launch I/O operation to load page from IOC device when required:
    // - it is the FAT mapper
    // - it is a directory mapper
    // - it is a file mapper, and it exist data on IOC device for this page
    if( (inode == NULL) || (inode_type == INODE_TYPE_DIR) || (inode_size > (page_id << 10) ) )
    {
        error = vfs_fs_move_page( page_xp , 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_remote_release_page( mapper_xp , page_ptr );
            return -1;
         }
    }

    // return extended pointer on allocated page
    *page_xp_ptr = page_xp;

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

#if( DEBUG_MAPPER_HANDLE_MISS & 2 ) 
if( DEBUG_MAPPER_HANDLE_MISS < cycle ) 
{
    if (inode != NULL) grdxt_remote_display( XPTR( mapper_cxy , &mapper_ptr->rt ) , name );
    else               grdxt_remote_display( XPTR( mapper_cxy , &mapper_ptr->rt ) , "FAT" );
}
#endif

    return 0;

}  // end mapper_handle_miss()

/////////////////////////////////////////////
xptr_t  mapper_get_page( xptr_t    mapper_xp,
                         uint32_t  page_id )
{
    error_t       error;

    thread_t * this = CURRENT_THREAD;

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

assert( (hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) ) != NULL ),
"should not be used for the FAT mapper");

#if DEBUG_MAPPER_GET_PAGE
uint32_t      cycle = (uint32_t)hal_get_cycles();
char          name[CONFIG_VFS_MAX_NAME_LENGTH];
if( DEBUG_MAPPER_GET_PAGE < cycle )  
{
    vfs_inode_t * inode = hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) );
    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

#if( DEBUG_MAPPER_GET_PAGE & 2 )
if( DEBUG_MAPPER_GET_PAGE < cycle ) 
ppm_remote_display( local_cxy );
#endif

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

    // build extended pointer on mapper lock and mapper rt
    xptr_t lock_xp  = XPTR( mapper_cxy , &mapper_ptr->lock );
    xptr_t 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
    xptr_t page_xp  = grdxt_remote_lookup( rt_xp , page_id );

    // test mapper miss 
    if( page_xp == XPTR_NULL )                  // miss => 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
        {
            error = mapper_handle_miss( mapper_xp,
                                        page_id,
                                        &page_xp );
            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;
            }
        }

#if (DEBUG_MAPPER_GET_PAGE & 1)
if( DEBUG_MAPPER_GET_PAGE < cycle )
printk("\n[%s] thread[%x,%x] introduced missing page %d in <%s> mapper / ppn %x\n",
__FUNCTION__, this->process->pid, this->trdid, page_id, name, ppm_page2ppn(page_xp) );
#endif
        
        // 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
if( DEBUG_MAPPER_GET_PAGE < cycle )
printk("\n[%s] thread[%x,%x] exit for page %d of <%s> mapper / ppn %x\n",
__FUNCTION__, this->process->pid, this->trdid, page_id, name, ppm_page2ppn(page_xp) );
#endif

#if( DEBUG_MAPPER_GET_PAGE & 2)
if( DEBUG_MAPPER_GET_PAGE < cycle ) 
ppm_remote_display( local_cxy );
#endif

    return page_xp;

}  // end mapper_get_page()

/////////////////////////////////////////////////
xptr_t  mapper_get_fat_page( xptr_t    mapper_xp,
                             uint32_t  page_id )
{
    error_t       error;

    thread_t * this = CURRENT_THREAD;

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

assert( (hal_remote_lpt( XPTR( mapper_cxy , &mapper_ptr->inode ) ) == NULL ),
"should be used for the FAT mapper");

#if DEBUG_MAPPER_GET_FAT_PAGE
uint32_t      cycle = (uint32_t)hal_get_cycles();
if( DEBUG_MAPPER_GET_FAT_PAGE < cycle ) 
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 );
#endif

#if( DEBUG_MAPPER_GET_FAT_PAGE & 2 )
if( DEBUG_MAPPER_GET_FAT_PAGE < cycle ) 
ppm_remote_display( local_cxy );
#endif

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

    // build extended pointer on mapper lock and mapper rt
    xptr_t lock_xp  = XPTR( mapper_cxy , &mapper_ptr->lock );
    xptr_t 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
    xptr_t page_xp  = grdxt_remote_lookup( rt_xp , page_id );

    // test mapper miss 
    if( page_xp == XPTR_NULL )                  // miss => 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
        {
            error = mapper_handle_miss( mapper_xp,
                                        page_id,
                                        &page_xp );
            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;
            }
        }

#if (DEBUG_MAPPER_GET_FAT_PAGE & 1)
if( DEBUG_MAPPER_GET_FAT_PAGE < cycle )
printk("\n[%s] thread[%x,%x] introduced missing page %d in FAT mapper / ppn %x\n",
__FUNCTION__, this->process->pid, this->trdid, page_id, ppm_page2ppn(page_xp) );
#endif
        
        // 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_FAT_PAGE
if( DEBUG_MAPPER_GET_FAT_PAGE < cycle )
printk("\n[%s] thread[%x,%x] exit for page %d of FAT mapper  / ppn %x\n",
__FUNCTION__, this->process->pid, this->trdid, page_id, ppm_page2ppn(page_xp) );
#endif

#if( DEBUG_MAPPER_GET_FAT_PAGE & 2)
if( DEBUG_MAPPER_GET_FAT_PAGE < cycle ) 
ppm_remote_display( local_cxy );
#endif

    return page_xp;

}  // end mapper_get_fat_page()

////////////////////////////////////////////////////
void mapper_remote_release_page( xptr_t   mapper_xp,
                                 page_t * page )
{
    // get mapper cluster an local pointer
    cxy_t      mapper_cxy = GET_CXY( mapper_xp );
    mapper_t * mapper_ptr = GET_PTR( mapper_xp );

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

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

    // remove physical page from radix tree
    grdxt_remote_remove( XPTR( mapper_cxy , &mapper_ptr->rt ) , page->index );

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

    // release page to PPM
    ppm_remote_free_pages( mapper_cxy , page );
                           
}  // 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_bytes;     // 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 indexes 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_bytes = size;
        else if ( page_id == first ) page_bytes = CONFIG_PPM_PAGE_SIZE - page_offset;
        else if ( page_id == last  ) page_bytes = (max_byte & CONFIG_PPM_PAGE_MASK) + 1;
        else                         page_bytes = 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_bytes );
#endif

        // get extended pointer on page descriptor in mapper
        page_xp = mapper_get_page( mapper_xp , page_id ); 

        if ( page_xp == XPTR_NULL ) return -1;

        // compute extended pointer on kernel mapper
        xptr_t     map_xp  = ppm_page2base( page_xp ) + page_offset;

#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, GET_CXY(map_xp), GET_PTR(map_xp) );
#endif
        // compute pointer in user buffer
        uint8_t * buf_ptr = (uint8_t *)buffer + done;

        // move fragment
        if( to_buffer )
        {
            hal_copy_to_uspace( buf_ptr , map_xp , page_bytes ); 

#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_bytes,
name, GET_CXY(map_xp), GET_PTR(map_xp), local_cxy, buf_ptr );
#endif

        }
        else
        {
            ppm_page_do_dirty( page_xp ); 
            hal_copy_from_uspace( map_xp , buf_ptr , page_bytes ); 

#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_bytes,
local_cxy, buf_ptr, name, GET_CXY(map_xp), GET_PTR(map_xp) );
mapper_display_page(  mapper_xp , page_id , 128 );
#endif

        }

        done += page_bytes;
    }

#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_bytes;     // 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_bytes = size;
        else if ( page_id == first ) page_bytes = CONFIG_PPM_PAGE_SIZE - page_offset;
        else if ( page_id == last  ) page_bytes = (max_byte & CONFIG_PPM_PAGE_MASK) + 1;
        else                         page_bytes = CONFIG_PPM_PAGE_SIZE;

        // get extended pointer on page descriptor
        page_xp = mapper_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_bytes );
    else
    printk("\n[%s] buffer (%x,%x) => mapper <%s> page %d / %d bytes\n",
    __FUNCTION__, src_cxy, src_ptr, name, page_id, page_bytes );
}
#endif

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

        done += page_bytes;
    }

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

    return 0;

}  // end mapper_move_kernel()

///////////////////////////////////////////////////
error_t mapper_remote_get_32( xptr_t     mapper_xp,
                              uint32_t   page_id,
                              uint32_t   word_id,
                              uint32_t * value )
{
    xptr_t     page_xp;      // extended pointer on searched page descriptor
    xptr_t     base_xp;      // extended pointer on searched page base 
    
    // get page containing the searched word
    page_xp  = mapper_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
    *value = hal_remote_l32( base_xp + (word_id<<2) ); 

    return 0;

}  // end mapper_remote_get_32()

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

    // get page containing the searched word
    page_xp  = mapper_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 + (word_id << 2)) , value );

    // set the dirty flag in page descriptor
    ppm_page_do_dirty( page_xp );

    return 0;

}  // end mapper_remote_set_32()

////////////////////////////////////////
error_t mapper_sync( xptr_t  mapper_xp )
{
    uint32_t   found_key;           // unused, required by grdxt_remote_get_first()
    error_t    error;

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

#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( mapper_cxy , &mapper_ptr->inode ) , name );
#endif

    // build extended pointer on radix tree
    xptr_t   rt_xp = XPTR( mapper_cxy , &mapper_ptr->rt );

    // initialise loop variable 
    uint32_t start_key = 0;

    // scan radix-tree until last page found
    while( 1 )
    {
        // get page descriptor from radix tree
        xptr_t page_xp = grdxt_remote_get_first( rt_xp , start_key , &found_key );
         
        page_t * page_ptr = GET_PTR( page_xp );

        // exit loop when last page found
        if( page_ptr == NULL ) break;

        // get page flags & index fields
        uint32_t flags = hal_remote_l32( XPTR( mapper_cxy , &page_ptr->flags ) );
        uint32_t index = hal_remote_l32( XPTR( mapper_cxy , &page_ptr->index ) );

        // synchronize page if dirty
        if( flags & PG_DIRTY )
        {

#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_ptr->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_ptr->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_ptr->index, name );
#endif
        }

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

    return 0;

}  // end mapper_sync()

///////////////////////////////////////////////
void mapper_display_page( xptr_t     mapper_xp,
                          uint32_t   page_id,
                          uint32_t   nbytes )
{
    char          buffer[4096];   // local buffer
    uint32_t      line;           // line index
    uint32_t      word;           // word index
 
    char       name[CONFIG_VFS_MAX_NAME_LENGTH];

assert( (nbytes <= 4096)         , "nbytes cannot be larger than 4096");
assert( (mapper_xp != XPTR_NULL) , "mapper_xp argument cannot be null");

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

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

    // get page cluster and local pointer
    cxy_t    page_cxy = GET_CXY( page_xp );
    page_t * page_ptr = GET_PTR( page_xp );

    // get page_id and mapper from page descriptor
    uint32_t   index   = hal_remote_l32( XPTR( page_cxy , &page_ptr->index ) );
    mapper_t * mapper  = hal_remote_lpt( XPTR( page_cxy , &page_ptr->mapper ) );

assert( (mapper_cxy == page_cxy ) , "mapper and page must be in same cluster");
assert( (mapper_ptr == mapper   ) , "unconsistent mapper field in page descriptor");
assert( (page_id    == index    ) , "unconsistent index  field in page descriptor");

    // get inode
    vfs_inode_t * 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
    xptr_t base_xp = ppm_page2base( page_xp );
   
    // copy remote page to local buffer
    hal_remote_memcpy( XPTR( local_cxy , buffer ) , base_xp , nbytes );

    // display header
    uint32_t * tabi = (uint32_t *)buffer;
    printk("\n***** mapper <%s> / page_id %d / cxy %x / mapper %x / buffer %x\n",
    name, page_id, mapper_cxy, mapper_ptr, GET_PTR( base_xp ) );

    // display 8 words per line
    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");
    }

}  // end mapper_display_page()