source: trunk/kernel/libk/user_dir.c @ 665

/*
 * user_dir.c - kernel DIR related operations implementation.
 * 
 * Authors   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_irqmask.h>
#include <hal_remote.h>
#include <thread.h>
#include <xlist.h>
#include <scheduler.h>
#include <remote_queuelock.h>
#include <user_dir.h>


/////////////////////////////////////////////
xptr_t user_dir_from_ident( intptr_t  ident )
{
    // get pointer on local process_descriptor
    process_t * process = CURRENT_THREAD->process;

    // get pointers on reference process
    xptr_t      ref_xp  = process->ref_xp;
    cxy_t       ref_cxy = GET_CXY( ref_xp );
    process_t * ref_ptr = GET_PTR( ref_xp );

    // get extended pointers on open directories list and lock  
    xptr_t root_xp = XPTR( ref_cxy , &ref_ptr->dir_root );
    xptr_t lock_xp = XPTR( ref_cxy , &ref_ptr->dir_lock );

    // get lock protecting open directories list
    remote_queuelock_acquire( lock_xp );
 
    // scan reference process dir list
    xptr_t           iter_xp;
    xptr_t           dir_xp;
    cxy_t            dir_cxy;
    user_dir_t     * dir_ptr;
    intptr_t         current;
    bool_t           found = false;
            
    XLIST_FOREACH( root_xp , iter_xp )
    {
        dir_xp  = XLIST_ELEMENT( iter_xp , user_dir_t , list );
        dir_cxy = GET_CXY( dir_xp );
        dir_ptr = GET_PTR( dir_xp );
        current = (intptr_t)hal_remote_lpt( XPTR( dir_cxy , &dir_ptr->ident ) );   
        if( ident == current )
        {
            found = true;
            break;
        }
    }

    // relese lock protecting open directories list
    remote_queuelock_release( lock_xp );
 
    if( found == false )  return XPTR_NULL;
    else                  return dir_xp;

}  // end user_dir_from_ident()

//////////////////////////////////////////////////
user_dir_t * user_dir_create( vfs_inode_t * inode,
                              xptr_t        ref_xp )
{ 
    user_dir_t    * dir;               // local pointer on created user_dir_t 
    vseg_t        * vseg;              // local pointer on dirent array vseg
    uint32_t        vseg_size;         // size of vseg in bytes
    process_t     * ref_ptr;           // local pointer on reference process
    cxy_t           ref_cxy;           // reference process cluster identifier
    pid_t           ref_pid;           // reference process PID
    xptr_t          gpt_xp;            // extended pointer on reference process GPT
    uint32_t        attr;              // attributes for all GPT entries
    uint32_t        dirents_per_page;  // number of dirent descriptors per page
    page_t        * page;              // local pointer on page descriptor
    struct dirent * base;              // local pointer on physical page base
    uint32_t        total_dirents;     // total number of dirents in dirent array
    uint32_t        total_pages;       // total number of pages for dirent array
    vpn_t           vpn_base;          // first page in dirent array vseg
    vpn_t           vpn;               // current page in dirent array vseg
    ppn_t           ppn;               // ppn of currently allocated physical page
    uint32_t        entries;           // number of dirent actually comied in one page
    uint32_t        first_entry;       // index of first dentry to copy in dirent array
    bool_t          done;              // last entry found and copied when true
    list_entry_t    root;              // root of temporary list of allocated pages
    uint32_t        page_id;           // page index in list of physical pages
    kmem_req_t      req;               // kmem request descriptor
    ppn_t           fake_ppn;          // unused, but required by hal_gptlock_pte()
    uint32_t        fake_attr;         // unused, but required by hal_gptlock_pte()
    error_t         error;

    // get cluster, local pointer, and pid of reference process
    ref_cxy = GET_CXY( ref_xp );
    ref_ptr = GET_PTR( ref_xp );
    ref_pid = hal_remote_l32( XPTR( ref_cxy , &ref_ptr->pid ) );

#if DEBUG_USER_DIR
uint32_t cycle = (uint32_t)hal_get_cycles();
thread_t * this = CURRENT_THREAD;
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] enter for inode (%x,%x) and process %x / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, local_cxy, inode, ref_pid, cycle );
#endif

// check dirent size 
assert( ( sizeof(struct dirent) == 64), "sizeof(dirent) must be 64\n");

    // compute number of dirent per page
    dirents_per_page = CONFIG_PPM_PAGE_SIZE >> 6;
    
    // initialise temporary list of pages
    list_root_init( &root );

    // allocate memory for a local user_dir descriptor
    req.type  = KMEM_KCM;
    req.order = bits_log2( sizeof(user_dir_t) );
    req.flags = AF_ZERO | AF_KERNEL;
    dir       = kmem_alloc( &req );

    if( dir == NULL )
    {
        printk("\n[ERROR] in %s : cannot allocate user_dir_t in cluster %x\n",
        __FUNCTION__, local_cxy );
        return NULL;
    }

    // Build and initialize the dirent array as a list of pages.
    // For each iteration in this while loop:
    // - allocate one physical 4 Kbytes (64 dirent slots)
    // - call the relevant FS specific function to scan the directory mapper,
    //   and copy up to 64 entries in the page.
    // - register the page in a temporary list using the embedded page list_entry
    // - exit when the last entry has been found (done == true).

    // initialize loops variables 
    done          = false;
    total_dirents = 0;
    total_pages   = 0;
    first_entry   = 0;

    while( done == false )  // loop on physical pages
    {
        // allocate one physical page
        req.type  = KMEM_PPM;
        req.order = 0;
        req.flags = AF_ZERO;
        base      = kmem_alloc( &req );

        if( base == NULL )
        {
            printk("\n[ERROR] in %s : cannot allocate page in cluster %x\n",
            __FUNCTION__, ref_cxy );
            goto user_dir_create_failure;
        }

        // call the relevant FS specific function to copy dirents in page
        error = vfs_fs_get_user_dir( inode,
                                     base,
                                     dirents_per_page,
                                     first_entry,
                                     false,        // don't create missing inodes
                                     &entries,
                                     &done );
        if( error )
        {
            printk("\n[ERROR] in %s : cannot initialise dirent array in cluster %x\n",
            __FUNCTION__, ref_cxy );
            goto user_dir_create_failure;
        }

        // increment number of written dirents
        total_dirents += entries;

        // get page descriptor pointer from base
        page = GET_PTR( ppm_base2page( XPTR( local_cxy , base ) ) );

        // register page in temporary list
        list_add_last( &root , &page->list ); 
        total_pages++; 

        // set first_entry for next iteration
        first_entry = total_dirents;

    } // end while
        
#if DEBUG_USER_DIR
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] initialised dirent array / %d entries\n",
__FUNCTION__, this->process->pid, this->trdid, total_dirents, cycle );
#endif

    // compute required vseg size
    vseg_size = total_dirents << 6;

    // create an ANON vseg and register it in reference process VSL
    if( local_cxy == ref_cxy )
    {
        vseg = vmm_create_vseg( ref_ptr,
                                VSEG_TYPE_ANON,
                                0,                      // vseg base (unused)
                                vseg_size,
                                0,                      // file offset (unused)
                                0,                      // file_size (unused)
                                XPTR_NULL,              // mapper (unused)
                                local_cxy );
    }
    else
    {
        rpc_vmm_create_vseg_client( ref_cxy,
                                    ref_ptr,
                                    VSEG_TYPE_ANON,
                                    0,                     // vseg base (unused)
                                    vseg_size,
                                    0,                     // file offset (unused)
                                    0,                     // file size (unused)
                                    XPTR_NULL,             // mapper (unused)
                                    local_cxy,
                                    &vseg ); 
    }

    if( vseg == NULL )
    {
        printk("\n[ERROR] in %s : cannot create vseg for user_dir in cluster %x\n",
        __FUNCTION__, ref_cxy);
        goto user_dir_create_failure;
    }

#if DEBUG_USER_DIR
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] allocated vseg ANON / base %x / size %x\n",
__FUNCTION__, this->process->pid, this->trdid, vseg->min, vseg->max - vseg->min );
#endif

// check vseg size
assert( (total_pages == hal_remote_l32( XPTR( ref_cxy , &vseg->vpn_size ) ) ),
"unconsistent vseg size for dirent array " );

    // build extended pointer on reference process GPT
    gpt_xp         = XPTR( ref_cxy , &ref_ptr->vmm.gpt );

    // build PTE attributes
    attr = GPT_MAPPED   |
           GPT_SMALL    |
           GPT_READABLE |
           GPT_CACHABLE |
           GPT_USER     ;

    // get first vpn from vseg descriptor
    vpn_base = hal_remote_l32( XPTR( ref_cxy , &vseg->vpn_base ) );

    // scan the list of allocated physical pages to map
    // all physical pages in the reference process GPT
    page_id = 0;
    while( list_is_empty( &root ) == false )
    {
        // get pointer on first page descriptor
        page = LIST_FIRST( &root , page_t , list );

        // compute ppn
        ppn = ppm_page2ppn( XPTR( local_cxy , page ) );

        // compute vpn
        vpn = vpn_base + page_id;
        
        // lock the PTE (and create PT2 if required)
        error = hal_gpt_lock_pte( gpt_xp,
                                  vpn,
                                  &fake_attr,
                                  &fake_ppn );
        if( error )
        {
            printk("\n[ERROR] in %s : cannot map vpn %x in GPT\n",
            __FUNCTION__, vpn );

            // delete the vseg
            intptr_t base = (intptr_t)hal_remote_lpt( XPTR( ref_cxy , &vseg->min ) );
            rpc_vmm_remove_vseg_client( ref_cxy, ref_pid, base );
         
            // release the user_dir descriptor
            req.type = KMEM_KCM;
            req.ptr  = dir;
            kmem_free( &req );
            return NULL;
        }

        // set PTE in GPT                          
        hal_gpt_set_pte( gpt_xp,
                         vpn,
                         attr,
                         ppn );

#if DEBUG_USER_DIR 
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] mapped vpn %x to ppn %x\n",
__FUNCTION__, this->process->pid, this->trdid, vpn + page_id, ppn );
#endif

        // remove the page from temporary list
        list_unlink( &page->list );

        page_id++;

    }  // end map loop

// check number of pages
assert( (page_id == total_pages) , "unconsistent pages number\n" );

    // initialise user_dir_t structure
    dir->current = 0;
    dir->entries = total_dirents;
    dir->ident   = (intptr_t)(vpn_base << CONFIG_PPM_PAGE_SHIFT);

    // build extended pointers on root and lock of user_dir xlist in ref process
    xptr_t root_xp  = XPTR( ref_cxy , &ref_ptr->dir_root );
    xptr_t lock_xp  = XPTR( ref_cxy , &ref_ptr->dir_lock );

    // build extended pointer on list field in user_dir structure
    xptr_t entry_xp = XPTR( local_cxy , &dir->list );

    // get lock protecting open directories list
    remote_queuelock_acquire( lock_xp );

    // register user_dir_t in reference process  
    xlist_add_first( root_xp , entry_xp );

    // release lock protecting  open directorie list
    remote_queuelock_release( lock_xp );

#if DEBUG_USER_DIR
cycle = (uint32_t)hal_get_cycles();
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] created user_dir (%x,%x) / %d entries / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, local_cxy, dir, total_dirents, cycle );
#endif

    return dir;

user_dir_create_failure:

    // release local user_dir_t structure
    req.type = KMEM_KCM;
    req.ptr  = dir;
    kmem_free( &req );

    // release local physical pages
    while( list_is_empty( &root ) == false )
    {
        page = LIST_FIRST( &root , page_t , list );

        // get base from page descriptor pointer 
        base = GET_PTR( ppm_page2base( XPTR( local_cxy , page ) ) );
 
        req.type  = KMEM_PPM;
        req.ptr   = base;
        kmem_free( &req );
    }

    return NULL;

}  // end user_dir_create()

////////////////////////////////////////
void user_dir_destroy( user_dir_t * dir,
                       xptr_t       ref_xp )
{
    thread_t     * this;       // local pointer on calling thread
    cluster_t    * cluster;    // local pointer on local cluster
    intptr_t       ident;      // user pointer on dirent array
    xptr_t         ref_pid;    // reference process PID
    cxy_t          ref_cxy;    // reference process cluster identifier
    process_t    * ref_ptr;    // local pointer on reference process
    xptr_t         root_xp;    // root of xlist
    xptr_t         lock_xp;    // extended pointer on lock protecting xlist
    xptr_t         iter_xp;    // iteratot in xlist 
    reg_t          save_sr;    // for critical section
    cxy_t          owner_cxy;  // owner process cluster
    lpid_t         lpid;       // process local index
    rpc_desc_t     rpc;        // rpc descriptor
    uint32_t       responses;  // response counter
     
    this    = CURRENT_THREAD;
    cluster = LOCAL_CLUSTER;

    // get cluster, local pointer, and PID of reference user process
    ref_cxy = GET_CXY( ref_xp );
    ref_ptr = GET_PTR( ref_xp );
    ref_pid = hal_remote_l32( XPTR( ref_cxy , &ref_ptr->pid ) );

#if DEBUG_USER_DIR
uint32_t cycle = (uint32_t)hal_get_cycles();
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] enter for user_dir (%x,%x) and process %x / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, local_cxy, dir, ref_pid, cycle );
#endif

    // get user pointer on dirent array
    ident = dir->ident;

    // build extended pointer on lock protecting open directories list
    lock_xp = XPTR( ref_cxy , &ref_ptr->dir_lock );

    // get lock protecting open directories list
    remote_queuelock_acquire( lock_xp );

    // remove dir from reference process xlist
    xlist_unlink( XPTR( local_cxy , &dir->list ) );

    // release lock protecting open directories list
    remote_queuelock_release( lock_xp );

    // To delete all copies of the vseg containing the dirent array, the client thread
    // send parallel RPCs to all clusters containing a client process copy (including 
    // the local cluster). It blocks and deschedules when all RPCs have been sent,
    // to wait all RPC responses, and will be unblocked by the last RPC server thread.
    // It allocates a - shared - RPC descriptor in the stack,  because all parallel
    // server threads use the same input arguments, and there is no out argument.

    // get owner cluster identifier and process lpid
    owner_cxy = CXY_FROM_PID( ref_pid );
    lpid      = LPID_FROM_PID( ref_pid );

    // get root of list of copies and lock from owner cluster
    root_xp   = XPTR( owner_cxy , &cluster->pmgr.copies_root[lpid] );
    lock_xp   = XPTR( owner_cxy , &cluster->pmgr.copies_lock[lpid] );

    // mask IRQs
    hal_disable_irq( &save_sr);

    // client thread blocks itself
    thread_block( XPTR( local_cxy , this ) , THREAD_BLOCKED_RPC );

    // initialize responses counter
    responses = 0;

    // initialize a shared RPC descriptor
    rpc.rsp       = &responses;
    rpc.blocking  = false;                  // non blocking behaviour for rpc_send()
    rpc.index     = RPC_VMM_REMOVE_VSEG;
    rpc.thread    = this;
    rpc.lid       = this->core->lid;
    rpc.args[0]   = ref_pid;
    rpc.args[1]   = ident;

    // take the lock protecting process copies
    remote_queuelock_acquire( lock_xp );

    // scan list of process copies 
    XLIST_FOREACH( root_xp , iter_xp )
    {
        // get extended pointer and cluster of process
        xptr_t      process_xp  = XLIST_ELEMENT( iter_xp , process_t , copies_list );
        cxy_t       process_cxy = GET_CXY( process_xp );

        // atomically increment responses counter
        hal_atomic_add( &responses , 1 );

#if (DEBUG_USER_DIR & 1)
uint32_t cycle = (uint32_t)hal_get_cycles();
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] register RPC request in cluster %x\n",
__FUNCTION__, this->process->pid, this->trdid, process_cxy );
#endif

        // send RPC to target cluster  
        rpc_send( process_cxy , &rpc );
    }

    // release the lock protecting process copies
    remote_queuelock_release( lock_xp );

    // client thread deschedule
    sched_yield("blocked on rpc_vmm_delete_vseg");
 
    // restore IRQs
    hal_restore_irq( save_sr);

    // release local user_dir_t structure
    kmem_req_t  req;
    req.type = KMEM_KCM;
    req.ptr  = dir;
    kmem_free( &req );

#if DEBUG_USER_DIR
cycle = (uint32_t)hal_get_cycles();
if( cycle > DEBUG_USER_DIR )
printk("\n[%s] thread[%x,%x] deleted user_dir (%x,%x) / cycle %d\n",
__FUNCTION__, this->process->pid, this->trdid, local_cxy, dir, cycle );
#endif

}  // end user_dir_destroy()