Event_Pack.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 "Event_Pack.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "Stream_MGR.h"
#include "Posix_Wrapp.h"

struct Event_Packer _malp_e_packer;


void Event_Packer_init()
{
        Event_Packer()->pack = calloc( 1, sizeof(struct Event_Pack) );
        
        Event_Packer()->lock = 0;
        
        if( !Event_Packer()->pack )
        {
                perror("malloc");
                exit(1);
        }

        VMPI_Partition_desc *an_desc = VMPI_Get_desc_by_name("Analyzer");

        if( !an_desc)
        {
                /* If there is no analyzer lets try to find an IO_Connector
                 * in writing mode to store a post mortem trace
                 */
                an_desc = VMPI_Get_desc_by_name("IO_WRITE");

                if(!an_desc)
                {
                        printf("ERRROR : Could not locate either an IO partition or an Analyzer\n");
                        exit(1);
                }
        }
        
        int target_partition_id = an_desc->id;
        
        VMPI_Map map;
        VMPI_Map_clear( &map );

        if( VMPI_Map_partitions( target_partition_id, VMPI_MAP_ROUND_ROBIN, &map) )
        {
                printf("Failed to map Instrumentation to Analyzer or IO Connector\n");
                exit(1);
        }

        VMPI_Stream_init( &Event_Packer()->stream, sizeof(struct Event_Pack), VMPI_STREAM_BALANCE_ROUND_ROBIN );

        VMPI_Stream_open_map( &Event_Packer()->stream, &map, "w" );
}

void Event_Packer_release()
{
        Event_Packer_flush();
        VMPI_Stream_close( &Event_Packer()->stream );
        free( Event_Packer()->pack );
        Event_Packer()->pack = NULL;
}


void __Event_Packer_push(struct MALP_Trace_Event *event)
{
                /* Otherwise process the actual event */
                MALP_Spinlock_lock( &Event_Packer()->lock );

                if( Event_Packer()->pack->offset == EVENT_PACK_SIZE )
                {
                        Event_Packer_flush();
                }

                if( event->type == MALP_EVENT_META )
                {
                        Event_Packer()->pack->contains_metadata = 1;
                }

                memcpy( &Event_Packer()->pack->events[Event_Packer()->pack->offset], event, sizeof( struct MALP_Trace_Event ) );

                Event_Packer()->pack->offset++;

                MALP_Spinlock_unlock( &Event_Packer()->lock );

}


static inline void Stream_Infos_process_pending_events()
{
        if( !__trace_module )
                return;

        struct MALP_Trace_Event event;
        void *ret;

        ret = Buffered_FIFO_pop( &Stream_infos_get()->mpi_wrapped_event_buffer, &event);

        while( ret )
        {
                /* We call the "no lock" version as we are only called from locked CTX*/
                __Event_Packer_push( &event );
                ret = Buffered_FIFO_pop( &Stream_infos_get()->mpi_wrapped_event_buffer, &event);
        }
}


static inline void Stream_Infos_push_pending_event(struct MALP_Trace_Event * event )
{
        if( !__trace_module )
                return;

        if(MAX_PENDING_EVENT < Buffered_FIFO_count(&Stream_infos_get()->mpi_wrapped_event_buffer))
                return;

        Buffered_FIFO_push( &Stream_infos_get()->mpi_wrapped_event_buffer, (void *)event );
}



__thread MALP_Spinlock is_in_instrum = 0;

void Event_Packer_push(struct MALP_Trace_Event *event)
{
        if( !MALP_Spinlock_trylock(&is_in_instrum) )
        {
                return;
        }

        if( Stream_Infos_is_in_mpi() )
        {
                Stream_Infos_push_pending_event( event );
        }
        else
        {
                /* First lets try to process pending events */
                Stream_Infos_process_pending_events();

                __Event_Packer_push(event);
        }
        
        MALP_Spinlock_unlock(&is_in_instrum);
}


void Event_Packer_flush()
{
        if( Event_Packer()->pack->offset )
        {
                /* Set locus */
                Event_Packer()->pack->locus.partition = VMPI_Get_partition_id();
                Event_Packer()->pack->locus.application = 0;

                /* Send events */
                VMPI_Stream_write( &Event_Packer()->stream, (char *)Event_Packer()->pack, 1 );

                /* Clear context */
                Event_Packer()->pack->offset = 0;
                Event_Packer()->pack->contains_metadata = 0;
        }
}