Pbb_Workers.c

Go to the documentation of this file.

/* ############################ MALP License ############################## */
/* # Fri Jan 18 14:00:00 CET 2013                                         # */
/* # Copyright or (C) or Copr. Commissariat a l'Energie Atomique          # */
/* #                                                                      # */
/* # This software is governed by the CeCILL-C license under French law   # */
/* # and abiding by the rules of distribution of free software.  You can  # */
/* # use, modify and/ or redistribute the software under the terms of     # */
/* # the CeCILL-C license as circulated by CEA, CNRS and INRIA at the     # */
/* # following URL http://www.cecill.info.                                # */
/* #                                                                      # */
/* # The fact that you are presently reading this means that you have     # */
/* # had knowledge of the CeCILL-C license and that you accept its        # */
/* # terms.                                                               # */
/* #                                                                      # */
/* # Authors:                                                             # */
/* #   - BESNARD Jean-Baptiste jean-baptiste.besnard@cea.fr               # */
/* #                                                                      # */
/* ######################################################################## */
#include "Pbb_Workers.h"
#include <unistd.h>
#include "Data_Entry.h"
#include "MALP_Blackboard.h"


#define MAX_BACK_OFF_ID 11
static const unsigned int Pbb_workers_backoff_times[MAX_BACK_OFF_ID] =
        {0,3,4,7,12,20,33,55,90,148,245};


static inline void Pbb_workers_process( struct MALP_blackboard *bb, struct Knowledge_system_job *job )
{
    //printf("Calling [process : %p default : %p entries : %d]\n", job->process, bb->default_handler, job->entries_count);
  
    
    /*int i;
    for( i = 0 ; i < job->entries_count ; i++ )
    {
        printf("[%d/%d]@[%p] KEY : %llX SIZE : %d PL : %p\n", i + 1, job->entries_count, job->entries[i],  job->entries[i]->key, job->entries[i]->size, job->entries[i]->payload);
    }
    
    
    printf("GOT process to %p KS : %lld Entries : %d pent %p  arg : %p\n", job->process, job->ks_type, job->entries_count, job->entries, job->arg);
        */

    if( !job->process )
    {
        if( bb->default_handler )
        {
            (bb->default_handler)( bb, job->ks_type, job->entries, job->entries_count, job->arg );
        }
        //else
        //{
            //printf("ERROR : There was a KS without associated process and no default KS handler\n");
            //abort();
        //}
    }
    else
    {
        (job->process)( bb, job->entries, job->entries_count, job->arg );
    }
}

void * Pbb_workers_mainloop( void *pdesc )
{
        struct Pbb_workers_desc *desc = (struct Pbb_workers_desc *)pdesc;
        struct Pbb_workers *wk = (struct Pbb_workers *)desc->wk;
        int id = desc->id;

        int had_work = 0;
        int backoff = 0;

        struct Knowledge_system_job job;

        uint8_t *to_free = malloc( 64 * sizeof( uint8_t ) );

        if( !to_free )
        {
                perror("malloc");
                abort();
        }

        while( wk->running )
        {
                memset( to_free, 0 ,  64 * sizeof( uint8_t ));
                had_work = 0;

                /* Acquire a job */

                int i = 0;

                void *got_job = NULL;

                got_job = Buffered_FIFO_pop( &wk->fifos[id], (void *)&job);

                /* We have a job */
                if( got_job )
                {
                        had_work = 1;

                        /* Process JOB */
                        if( had_work )
                        {
                                if( 64 < job.entries_count )
                                {
                                        printf("Error maximum Data entries count reached => update value\n");
                                        abort();
                                }
                                
                                
                                for( i = 0 ; i < job.entries_count ; i++ )
                                {
                                        uint64_t ret = Data_entry_dec_ref_count( job.entries[i] );
                                        if( ret == 0 )
                                        {
                                                to_free[i] = 1;
                                        }
                                }

                Pbb_workers_process( wk->bb, &job );

                                /* Update the refcount of each data entry and free when zero */
                                for( i = 0 ; i < job.entries_count ; i++ )
                                {
                                        if( job.entries[i]->been_resubmitted == 0 )
                                        {
                                                if( to_free[i] )
                                                {
                            //printf("Removing data @ %llX\n", job.entries[i]->key );
                                                        /* One data to be freed update pending */
                                                        MALP_Blackboard_remove_pending( wk->bb );
                                                        
                                                        Data_entry_release( job.entries[i] );
                                                        job.entries[i] = NULL;
                                                }
                                        }
                                        else
                                        {
                                                job.entries[i]->been_resubmitted = 0;
                                        }
                                }
                                
                                Knowledge_system_job_release( &job );
                        }
                }

                /* Back off handling */

                if( had_work )
                {
                                backoff=0;
                }
                else
                {
                                if( backoff < (MAX_BACK_OFF_ID - 1) )
                                        backoff++;
                }

        if( 5 < backoff )
                    pthread_yield();

                //printf("Th %d  BO : %d T : %d  FIFO : %lld\n", id, backoff, Pbb_workers_backoff_times[ backoff ], (long long int)Buffered_FIFO_count(&wk->fifos[id]));
                usleep( Pbb_workers_backoff_times[ backoff ] );

        }

        return NULL;
}

void Pbb_workers_init( struct Pbb_workers *wk, struct MALP_blackboard *bb, uint32_t wk_count)
{
        if( !wk_count )
        {
                printf("%s has been given erroneous arguments\n", __FUNCTION__ );
                abort();
        }

        wk->bb = bb;
        wk->worker_count = wk_count;

        int i = 0;
        
        wk->running = 1;

        wk->fifos = malloc( wk_count * sizeof( struct Buffered_FIFO ) );
        for( i = 0; i < wk_count ; i++ )
                 Buffered_FIFO_init(&wk->fifos[i], 200, sizeof( struct Knowledge_system_job ));



        wk->threads = malloc( wk_count * sizeof( pthread_t ) );
        wk->descs = malloc( wk_count * sizeof( struct Pbb_workers_desc ) );

        for( i = 0; i < wk_count ; i++ )
        {
                wk->descs[i].wk = wk;
                wk->descs[i].id = i;
                pthread_create(&wk->threads[i], NULL, Pbb_workers_mainloop, (void *)&wk->descs[i] );
        }
        
}


void Pbb_workers_release( struct Pbb_workers *wk )
{
        int i = 0;

        wk->running = 0;

        for( i = 0; i < wk->worker_count ; i++ )
                pthread_join( wk->threads[i], NULL );


        free( wk->threads );
        wk->threads = NULL;

        for( i = 0; i < wk->worker_count ; i++ )
                Buffered_FIFO_release( &wk->fifos[i], NULL );

        wk->worker_count = 0;

        free( wk->fifos );
        wk->fifos = NULL;
}


void Pbb_workers_submit( struct Pbb_workers *wk, struct Knowledge_system_job *ksj )
{
        int entry_point = 0;
        
        if( ksj->is_serial )
        {
                entry_point = ksj->ks_type % wk->worker_count;
        }
        else
        {
                entry_point = rand() % wk->worker_count;
        }
        Buffered_FIFO_push(&wk->fifos[entry_point], (void *)ksj);
}

void Pbb_workers_wait( struct Pbb_workers *wk )
{
        int i = 0;
        int all_zero = 0;

        /* Check if no jobs */
        while( !all_zero )
        {

                all_zero = 1;
                for( i = 0; i < wk->worker_count ; i++ )
                {
                        if( Buffered_FIFO_count( &wk->fifos[i] ) )
                        {
                                all_zero = 0;
                                break;
                        }

                }
                
                /* If there is no job left
                 * Check if there is also no pending data in BB */
                if( all_zero )
                {
                        uint64_t pending =  MALP_Blackboard_pending( wk->bb );

                        if( pending )
                                all_zero = 0;

                }

                /* Be very nice with the other threads if we are pinned
                 * on the same core to avoid live-locking
                 */
                pthread_yield();

        }

}