source: trunk/kernel/libk/remote_fifo.c @ 687

/*
 * remote_fifo.c  Implement a lock-less FIFO, multiple-remote-writers / single-local-reader
 *
 * Authors : Mohamed Lamine Karaoui (2015)
 *           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 <hal_kernel_types.h>
#include <hal_irqmask.h>
#include <hal_remote.h>
#include <hal_special.h>
#include <bits.h>
#include <memcpy.h>
#include <kmem.h>
#include <thread.h>
#include <cluster.h>
#include <remote_fifo.h>

/////////////////////////////////////////////
void remote_fifo_init( remote_fifo_t * fifo )
{
    uint32_t  slot;

    fifo->owner     = 0;
        fifo->wr_id     = 0;
        fifo->rd_id     = 0;
    for( slot = 0 ; slot < CONFIG_RPC_FIFO_SLOTS ; slot++ )
    {
        fifo->valid[slot] = 0;
    }
}

/////////////////////////////////////////////////
error_t remote_fifo_put_item( xptr_t     fifo_xp, 
                              uint64_t   item )
{
    uint32_t        wr_id;
    uint32_t        rd_id;
    uint32_t        ptw;
    uint32_t        watchdog;
    uint32_t        nslots;

    // get remote cluster identifier and pointer on FIFO
    cxy_t           fifo_cxy = GET_CXY( fifo_xp );
    remote_fifo_t * fifo_ptr = GET_PTR( fifo_xp );

    // initialise watchdog for contention detection
    watchdog = 0;

    // get write slot index with atomic increment QQQQQQQQQQ
        wr_id = hal_remote_atomic_add( XPTR( fifo_cxy , &fifo_ptr->wr_id ) , 1 );

    // wait until allocated slot is empty in remote FIFO
    // max retry = CONFIG_RPC_FIFO_MAX_ITERATIONS  
    // return error if watchdog is reached
    while( 1 )
    {
        // return error if contention detected by watchdog
        if( watchdog > CONFIG_RPC_FIFO_MAX_ITERATIONS )  return EBUSY;

        // read remote rd_id value
        rd_id = hal_remote_l32( XPTR( fifo_cxy , &fifo_ptr->rd_id ) );

        // compute number of full slots
        if( wr_id >= rd_id ) nslots = wr_id - rd_id;
        else                 nslots = (0xFFFFFFFF - rd_id) + wr_id;

        // exit waiting loop as soon as fifo not full
        if ( nslots < CONFIG_RPC_FIFO_SLOTS )  break;
        
        // retry later if fifo full:
        // - deschedule without blocking if possible
        // - wait ~1000 cycles otherwise
        if( CURRENT_THREAD->busylocks == 0 ) sched_yield( "wait RPC fifo" );
        else                                 hal_fixed_delay( 1000 );

        // increment watchdog
        watchdog++;
    }

    // compute actual write slot pointer
    ptw = wr_id % CONFIG_RPC_FIFO_SLOTS;

    // copy item to fifo
        hal_remote_s64( XPTR( fifo_cxy , &fifo_ptr->data[ptw] ), item );
        hal_fence();

    // set the slot valid flag
        hal_remote_s32( XPTR( fifo_cxy , &fifo_ptr->valid[ptw] ) , 1 );
        hal_fence();

    return 0;
 
} // end remote_fifo_put_item()

///////////////////////////////////////////////////
error_t remote_fifo_get_item( remote_fifo_t * fifo,
                              uint64_t      * item )
{
    // get fifo state
        uint32_t rd_id     = fifo->rd_id;
        uint32_t wr_id     = fifo->wr_id;

        // return if fifo empty
        if( rd_id == wr_id ) return EAGAIN;
        
    // compute actual read slot pointer 
        uint32_t ptr = rd_id % CONFIG_RPC_FIFO_SLOTS;
        
        // wait slot filled by the writer
        while( fifo->valid[ptr] == 0 ) {}
        
    // copy item from FIFO to local buffer
    *item = fifo->data[ptr];

    // reset valid slot flag
    fifo->valid[ptr] = 0;

    // increment the read index
        fifo->rd_id += 1;

        return 0;

} // end remote_fifo_get_item()

/////////////////////////////////////////
bool_t remote_fifo_is_full( xptr_t fifo )
{
    uint32_t nslots;

    // get remote cluster identifier and pointer on FIFO
    cxy_t           cxy = GET_CXY( fifo );
    remote_fifo_t * ptr = GET_PTR( fifo );
   
    // get read and write pointers
        uint32_t wr_id = hal_remote_l32( XPTR( cxy , &ptr->wr_id ) );
        uint32_t rd_id = hal_remote_l32( XPTR( cxy , &ptr->rd_id ) );

    // compute number of full slots
    if( wr_id >= rd_id ) nslots = wr_id - rd_id;
    else                 nslots = (0xFFFFFFFF - rd_id) + wr_id;

    return ( nslots >= CONFIG_RPC_FIFO_SLOTS );
}

///////////////////////////////////////////////////
bool_t remote_fifo_is_empty( remote_fifo_t * fifo )
{
    return ( fifo->wr_id == fifo->rd_id );
}

/////////////////////////////////////////
uint32_t remote_fifo_items( xptr_t fifo )
{
    uint32_t nslots;

    // get remote cluster identifier and pointer on FIFO
    cxy_t           cxy = GET_CXY( fifo );
    remote_fifo_t * ptr = GET_PTR( fifo );
   
    // get read and write pointers
        uint32_t wr_id = hal_remote_l32( XPTR( cxy , &ptr->wr_id ) );
        uint32_t rd_id = hal_remote_l32( XPTR( cxy , &ptr->rd_id ) );

    // compute number of full slots
    if( wr_id >= rd_id ) nslots = wr_id - rd_id;
    else                 nslots = (0xFFFFFFFF - rd_id) + wr_id;

    return nslots;
}