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

/*
 * 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_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 local_fifo_init( remote_fifo_t * fifo )
{
    uint32_t  slot;

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

////////////////////////////////////////////// 
error_t remote_fifo_put_item( xptr_t     fifo, 
                              uint64_t * item,
                              bool_t   * first )
{
    uint32_t        wr_id;
    uint32_t        rd_id;
    uint32_t        ptw;
    uint32_t        watchdog;
    reg_t           save_sr;
    uint32_t        nslots;

    // get remote cluster identifier and pointer on FIFO
    cxy_t           cxy = (cxy_t)GET_CXY( fifo );
    remote_fifo_t * ptr = (remote_fifo_t *)GET_PTR( fifo );

    // initialise watchdog for contention detection
    watchdog = 0;

    // no descheduling during put access
        hal_disable_irq( &save_sr );
        
    // get write slot index and atomic increment 
        wr_id = hal_remote_atomic_add( XPTR( cxy , &ptr->wr_id ) , 1 );

    // check fifo state
    while( 1 )
    {
        // return error if contention detected by watchdog
        if( watchdog > CONFIG_REMOTE_FIFO_MAX_ITERATIONS ) 
        {
            // restore scheduling
                hal_restore_irq( save_sr );

            // return error
            return EBUSY;
        }

        // read remote rd_id value
        rd_id = hal_remote_lw( 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;

        // exit if fifo not full
        if ( nslots < CONFIG_REMOTE_FIFO_SLOTS )  break;
        
        // restore interrupts
            hal_restore_irq( save_sr );

        // deschedule without blocking 
        if( thread_can_yield() ) sched_yield();

        // disable interrupts
            hal_disable_irq( &save_sr );

        // increment watchdog
        watchdog++;
    }

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

    // copy item to fifo
        hal_remote_swd( XPTR( cxy , &ptr->data[ptw] ), *item );
        hal_fence();

    // set the slot valid flag
        hal_remote_sw( XPTR( cxy , &ptr->valid[ptw] ) , 1 );
        hal_fence();

    // return the first RPC flag
    *first = ( wr_id == rd_id );
    
    // restore scheduling
        hal_restore_irq( save_sr );

    return 0;
 
} // end remote_fifo_put_item()

//////////////////////////////////////////////////
error_t local_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_REMOTE_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 local_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 = (cxy_t)GET_CXY( fifo );
    remote_fifo_t * ptr = (remote_fifo_t *)GET_PTR( fifo );
   
    // get read and write pointers
        uint32_t wr_id = hal_remote_lw( XPTR( cxy , &ptr->wr_id ) );
        uint32_t rd_id = hal_remote_lw( 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_REMOTE_FIFO_SLOTS );
}

//////////////////////////////////////////////////
bool_t local_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 = (cxy_t)GET_CXY( fifo );
    remote_fifo_t * ptr = (remote_fifo_t *)GET_PTR( fifo );
   
    // get read and write pointers
        uint32_t wr_id = hal_remote_lw( XPTR( cxy , &ptr->wr_id ) );
        uint32_t rd_id = hal_remote_lw( 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;
}