MALP_Blackboard.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 "MALP_Blackboard.h"

#include <string.h>

void MALP_blackboard_init( struct MALP_blackboard *bb, uint32_t workers_count, void *context )
{
        MALP_Trace_crc64_init();
        bb->context = context;
        Hash_Table_init( &bb->ks_table, sizeof( struct Ks_cell ), 4241, KS_COLL_AVOID );
        Hash_Table_init( &bb->key_table, sizeof( struct Key_cell ), 4241, KEY_COLL_AVOID );
    Hash_Table_init( &bb->de_name_table, sizeof( struct Name_cell ), 4241, NAME_COLL_AVOID );

        Hash_Table_use_lock( &bb->key_table );

        bb->cpu_count = sysconf( _SC_NPROCESSORS_ONLN );
        
        Pbb_workers_init( &bb->workers, bb, workers_count?workers_count:bb->cpu_count );
        
        bb->pending_datas = 0;
        bb->lock = 0;
    bb->default_handler = NULL;
}


void free_ks(void *pks)
{
        struct Knowledge_system *ks = (struct Knowledge_system *)pks;
        Knowledge_system_release( ks );
} 

void free_key_ent(void *pke)
{
        struct Key_entry *ke = (struct Key_entry *)pke;
        Key_entry_release( ke );
}

void free_name(void *pn)
{
        free( pn );
}  


void MALP_blackboard_release( struct MALP_blackboard *bb )
{
        Pbb_workers_release( &bb->workers );
        Hash_Table_release( &bb->ks_table, free_ks );
        Hash_Table_release( &bb->key_table, free_key_ent );
    Hash_Table_release( &bb->de_name_table, free_name );

        bb->lock = 1;
    bb->default_handler = NULL;
}

void MALP_blackboard_wait( struct MALP_blackboard *bb )
{
        Pbb_workers_wait( &bb->workers );
}


int knowledge_system_test_func(void *pks, uint64_t key, void *arg)
{
        struct Knowledge_system *ks = (struct Knowledge_system *)pks;

        if( ks->type == key )
                return 1;

        return 0;
}

int empty_knowledge_system_test_func(void *pks, uint64_t key, void *arg)
{
        /* Do we have freed KS in this bucket */
        struct Knowledge_system *ks = (struct Knowledge_system *)pks;

        if( ks->freed == 1 )
                return 1;

        return 0;
}

int key_entry_test_func(void *pke, uint64_t key, void *arg)
{
        struct Key_entry *ke = (struct Key_entry *)pke;

        if( ke->key == key )
                return 1;

        return 0;
}

int empty_key_entry_test_func(void *pke, uint64_t key, void *arg)
{
        /* Do we have freed KE in this bucket */
        struct Key_entry *ke = (struct Key_entry *)pke;

        if( ke->freed == 1 )
                return 1;

        return 0;
}

struct Key_entry* MALP_Blackboard_get_ke( struct MALP_blackboard *bb, uint64_t key )
{
        struct Key_entry* ret = Hash_Table_get( &bb->key_table, key, NULL, key_entry_test_func );
        return ret;
}

struct Key_entry* MALP_Blackboard_push_unique_ke( struct MALP_blackboard *bb, uint64_t key )
{
        struct Key_entry *ret = Hash_Table_get( &bb->key_table, key, NULL, key_entry_test_func );

        if( ret )
        {
                return ret;
        }
        else
        {
                struct Key_entry ke;
                Key_entry_init( &ke, key );
                
                ret = Hash_Table_get( &bb->key_table, key, NULL, empty_key_entry_test_func );
                if( ret == NULL )
                {
                        ret = Hash_Table_push( &bb->key_table, (void *)&ke , key );
                }
                else
                {
                        memcpy( ret, &ke, sizeof( struct Key_entry ) );
                }
        }

        return ret;
}




struct Knowledge_system * MALP_Blackboard_push_ks( struct MALP_blackboard *bb, struct Knowledge_system *ks )
{
        struct Knowledge_system *ret = Hash_Table_get( &bb->ks_table, ks->type, NULL, empty_knowledge_system_test_func );

        if( ret )
        {
                memcpy( ret, ks, sizeof( struct Knowledge_system ) );
        }
        else
        {
                ret = Hash_Table_push( &bb->ks_table, (void *)ks , ks->type );
        }

        return ret;
}

int MALP_Blackboard_ks_present( struct MALP_blackboard *bb, uint64_t type )
{
        struct Knowledge_system *existing_ks = Hash_Table_get( &bb->ks_table, type, NULL, knowledge_system_test_func );
        
        if( existing_ks )
                return 1;
        
        return 0;
}

struct Knowledge_system * MALP_Blackboard_new_ks( struct MALP_blackboard *bb, uint64_t type, uint64_t *sensivities,
                                                         uint32_t sensivities_count, void (*process)( struct MALP_blackboard *bb, struct Data_entry **de, uint32_t entry_count, void *arg ), void *arg )
{
        struct Knowledge_system *existing_ks = NULL;
        
        MALP_Spinlock_lock( &bb->lock );
        {
                /* Check if it is not already registered */
                existing_ks = Hash_Table_get( &bb->ks_table, type, NULL, knowledge_system_test_func );
                
                if( existing_ks != NULL )
                {
                        return existing_ks;
                }

                /* If not create a new Ks */
                struct Knowledge_system ks;
                Knowledge_system_init( &ks, type, sensivities, sensivities_count, process, arg );

                /* Push it to Ks list */
                existing_ks = MALP_Blackboard_push_ks( bb, &ks );

                /* Register in key entries */
                int i = 0;
                for( i = 0 ; i < sensivities_count ; i++ )
                {
                        struct Key_entry *ke =  MALP_Blackboard_push_unique_ke( bb, sensivities[i] );
                        Key_entry_push_ks( ke, existing_ks );
                }
        }
        MALP_Spinlock_unlock( &bb->lock );
        
        return existing_ks;
}


void __MALP_Blackboard_delete_ks(void *pke, void *pks)
{
        struct Key_entry *ke = (struct Key_entry *)pke;
        struct Knowledge_system *ks = (struct Knowledge_system *)pks;
        
        Key_entry_pop_ks( ke, ks );
}

void MALP_Blackboard_delete_ks( struct MALP_blackboard *bb, uint64_t type )
{
        MALP_Spinlock_lock( &bb->lock );
        {
                /* Check if KS exists */
                struct Knowledge_system *existing_ks = Hash_Table_get( &bb->ks_table, type, NULL, knowledge_system_test_func );
                
                if( existing_ks == NULL )
                {
                        /* No such KS */
                        return;
                }
                
                /* If it does unregister Ks in Key entries */
                Hash_Table_walkthrought_2p( &bb->key_table, (void *)existing_ks,  __MALP_Blackboard_delete_ks );
                
                
                /* Free it */
                Knowledge_system_release( existing_ks );

        }
        MALP_Spinlock_unlock( &bb->lock );
}


void MALP_Blackboard_compute_de_refcount( struct Key_entry *ke, struct Data_entry *de )
{
        /* Increment the refcount of the number of KS in this key */
        Data_entry_inc_ref_count( de, Key_entry_get_count(ke));
}

struct MALP_BB_submit
{
        struct Data_entry *de;      
        struct MALP_blackboard *bb; 
};

int ke_submit_handler( struct Knowledge_system *ks, void *pbbsb)
{
        struct MALP_BB_submit *bb_sb = (struct MALP_BB_submit *)pbbsb;
        struct Knowledge_system_job job = Knowledge_system_submit_data_entry( ks, bb_sb->de );

        Pbb_workers_submit( &bb_sb->bb->workers, &job );

        return 0;
}

void MALP_Blackboard_submit_to_ke_ks( struct MALP_blackboard *bb, struct Key_entry *ke, struct Data_entry *de )
{
        struct MALP_BB_submit bb_sb = { de , bb };
        Key_entry_walk( ke, ke_submit_handler, (void *)&bb_sb );
}


int MALP_Blackboard_resubmit_data_entry(  struct MALP_blackboard *bb, struct Data_entry *de )
{
        if( !de )
                return 1;

        if( Data_entry_get_ref_count( de ) )
        {
                printf("Cannot resubmit a data entry without a null refcount\n");
                abort();
        }


        struct Key_entry *ke = MALP_Blackboard_get_ke( bb, de->key );

    //printf("KEY : %llX Got KE : %p\n", de->key , ke );

        /* Update refcount */
        if( ke )
        {
                MALP_Blackboard_compute_de_refcount( ke, de );
                de->been_resubmitted = 1;
                /* Submit to KS now that all refcount were summed */
                MALP_Blackboard_submit_to_ke_ks( bb, ke, de );
        }

        return ( ke == NULL )?0:1;
}

int MALP_Blackboard_submit_data_entry(  struct MALP_blackboard *bb, struct Data_entry *de )
{
        if( !de )
                return 1;

        struct Key_entry *ke = MALP_Blackboard_get_ke( bb, de->key );


        /* Update refcount */
        if( ke )
        {
        de->been_resubmitted = 0;
                MALP_Blackboard_compute_de_refcount( ke, de );

                /* We have a new pending data */
                MALP_Blackboard_add_pending( bb );

        //printf("Submitting DE @ %llX\n", de->key );

                /* Submit to KS now that all refcount were summed */
                MALP_Blackboard_submit_to_ke_ks( bb, ke, de );
        }

        return ( ke == NULL )?0:1;
}



int MALP_Blackboard_submit_data( struct MALP_blackboard *bb, uint64_t key, void *payload, size_t payload_size )
{
        struct Data_entry *de = NULL;

        struct Key_entry *ke = MALP_Blackboard_get_ke( bb, key );

        /* Update refcount */
        if( ke )
        {
                de = Data_entry_init( key, payload, payload_size );

                MALP_Blackboard_compute_de_refcount( ke, de );
                
        //printf("Submitting DATA @ %llX\n", key );

                /* We have a new pending data */
                MALP_Blackboard_add_pending( bb );
                
                /* Submit to KS now that all refcount were summed */
                MALP_Blackboard_submit_to_ke_ks( bb, ke, de );
        }


        return (de == NULL)?0:1;
}