Stream_MGR.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 "Posix_Wrapp.h"

#include "Stream_MGR.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>


#include "Timer.h"



__thread struct Stream_Infos * __trace_module = NULL;

struct Stream_Manager __stream_Stream_Manager;


uint64_t Stream_MGR_gen_id()
{
        uint64_t ret = 0;
        
        MALP_Spinlock_lock( &Stream_Manager()->lock );

        uint64_t my_rank = Stream_Manager()->rank;
        
        ret |= Stream_Manager()->partition_id << 56;
        ret |= (my_rank & 0x00FFFFFFFFFFFFFFllu) << 32;
        ret |= (Stream_Manager()->current_id++ & 0x00000000FFFFFFFFllu);
        
        MALP_Spinlock_unlock( &Stream_Manager()->lock );

        return ret;
}


void Stream_Manager_init()
{
        Chained_List_init( &Stream_Manager()->streams, 64, sizeof( struct Stream_Infos ) );
        Stream_Manager()->current_id = 1;
        Stream_Manager()->lock = 0;
        PMPI_Comm_rank( VMPI_Get_partition_comm(), &Stream_Manager()->rank );
        Stream_Manager()->partition_id = VMPI_Get_partition_id();
        
}

void Stream_Manager_release()
{
        Chained_List_release( &Stream_Manager()->streams );
        Stream_Manager()->current_id = 1;
        Stream_Manager()->lock = 0;
        Stream_Manager()->rank = -1;
        Stream_Manager()->partition_id = 0;
}


void Stream_Infos_init( struct Stream_Infos *infos )
{
        memset( infos, 0, sizeof( struct Stream_Infos ) );
        Buffered_FIFO_init( &infos->mpi_wrapped_event_buffer, 32, sizeof( struct MALP_Trace_Event ) );
}

void Stream_Infos_release( struct Stream_Infos *infos )
{
        Buffered_FIFO_release( &infos->mpi_wrapped_event_buffer, NULL );
}



struct Stream_Infos * Stream_MGR_new_stream( struct Stream_Descriptor *parent, Stream_type type )
{
        
        struct MALP_Trace_Event tmp;
        struct Stream_Infos infos;

        Stream_Infos_init( &infos );

        tmp.type = MALP_EVENT_META;
        tmp.timestamp = Timer_tsc();
        tmp.duration = 0;
        tmp.tid = parent?parent->tid:0;
        tmp.partition_id = Stream_Manager()->partition_id;


        tmp.event.stream_descriptor.tid = Stream_MGR_gen_id( Stream_Manager() );
        
        if( parent )
        {
                tmp.event.stream_descriptor.parent_tid = parent->tid;
        }
        else
        {
                tmp.event.stream_descriptor.parent_tid = 0;
        }

        tmp.event.stream_descriptor.partition_id = Stream_Manager()->partition_id;
        tmp.event.stream_descriptor.desc_type = DESC_BEGIN;
        tmp.event.stream_descriptor.type = type;
        
        tmp.event.stream_descriptor.start_time = Timer_tsc();
        tmp.event.stream_descriptor.end_time = 0;
        
        tmp.event.stream_descriptor.mpi_rank = Stream_Manager()->rank;
        tmp.event.stream_descriptor.omp_thread = 0;

        if( gethostname(tmp.event.stream_descriptor.hostname, 32) < 0 )
                sprintf( tmp.event.stream_descriptor.hostname, "Unknown");

        Event_Packer_push( &tmp );

        // Copy the infos back tot he stream descriptor
        infos.infos = tmp.event.stream_descriptor;

        void *pinfos = Chained_List_push( &Stream_Manager()->streams, (void *)&infos );
        
        return (struct Stream_Infos *)pinfos;
}

void Stream_MGR_end_stream()
{

        struct Stream_Infos *infos = Stream_infos_get();
        struct Stream_Descriptor *desc = &infos->infos;
        
        struct MALP_Trace_Event tmp;


        tmp.type = MALP_EVENT_META;
        tmp.timestamp = Timer_tsc();
        tmp.duration = 0;
        tmp.tid = desc->tid;
        tmp.partition_id = Stream_Manager()->partition_id;


        tmp.event.stream_descriptor.tid = desc->tid;
        tmp.event.stream_descriptor.parent_tid = desc->parent_tid;
        tmp.event.stream_descriptor.partition_id = Stream_Manager()->partition_id;
        tmp.event.stream_descriptor.desc_type = DESC_END;
        tmp.event.stream_descriptor.type = desc->type;
        
        tmp.event.stream_descriptor.start_time = 0;
        tmp.event.stream_descriptor.end_time = Timer_tsc();
        
        tmp.event.stream_descriptor.mpi_rank = Stream_Manager()->rank;
        tmp.event.stream_descriptor.omp_thread = 0;

        TEST_LOAD_SYMBOL(gethostname);

        if( REAL(gethostname)(tmp.event.stream_descriptor.hostname, 32) < 0 )
                sprintf( tmp.event.stream_descriptor.hostname, "Unknown");


        Event_Packer_push( &tmp );

        Stream_Infos_release( infos );

        Stream_infos_set(NULL);
}