libtrace_parallel.h

Go to the documentation of this file.

/*
 *
 * Copyright (c) 2007-2016 The University of Waikato, Hamilton, New Zealand.
 * All rights reserved.
 *
 * This file is part of libtrace.
 *
 * This code has been developed by the University of Waikato WAND
 * research group. For further information please see http://www.wand.net.nz/
 *
 * libtrace is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * libtrace 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 Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *
 */


#ifndef LIBTRACE_PARALLEL_H
#define LIBTRACE_PARALLEL_H

#include "libtrace.h"
#include <stdio.h>

#ifdef __cplusplus
extern "C" {
#endif

typedef struct libtrace_result_t libtrace_result_t;

typedef union {
        /* Pointers */
        void *ptr;
        libtrace_packet_t *pkt;
        libtrace_result_t *res;

        /* C99 Integer types */
        /* NOTE: Standard doesn't require 64-bit
         * but x32 and x64 gcc does */
        int64_t sint64;
        uint64_t uint64;

        uint32_t uint32s[2];
        int32_t sint32s[2];
        uint32_t uint32;
        int32_t sint32;

        uint16_t uint16s[4];
        int16_t sint16s[4];
        uint16_t uint16;
        int16_t sint16;

        uint8_t uint8s[8];
        int8_t sint8s[8];
        uint8_t uint8;
        int8_t sint8;

        size_t size;

        /* C basic types - we cannot be certain of the size */
        int sint;
        unsigned int uint;

        signed char schars[8];
        unsigned char uchars[8];
        signed char schar;
        unsigned char uchar;

        /* Real numbers */
        float rfloat;
        double rdouble;
} libtrace_generic_t;
ct_assert(sizeof(libtrace_generic_t) == 8);

typedef struct libtrace_message_t {
        int code; 
        libtrace_generic_t data; 
        libtrace_thread_t *sender; 
} libtrace_message_t;

struct libtrace_result_t {
        uint64_t key;   
        libtrace_generic_t value;  
        int type; 
};

enum libtrace_messages {
        MESSAGE_PACKET,

        MESSAGE_RESULT,

        MESSAGE_STARTING,

        MESSAGE_STOPPING,

        MESSAGE_RESUMING,

        MESSAGE_PAUSING,

        MESSAGE_DO_PAUSE,

        MESSAGE_DO_STOP,

        MESSAGE_FIRST_PACKET,

        MESSAGE_POST_REPORTER,

        MESSAGE_TICK_INTERVAL,

        MESSAGE_TICK_COUNT,

        MESSAGE_USER = 1000
};

enum hasher_types {
        HASHER_BALANCE,

        HASHER_BIDIRECTIONAL,

        HASHER_UNIDIRECTIONAL,

        HASHER_CUSTOM
};

typedef struct libtrace_info_t {
        bool live;

        int max_threads;

        /* TODO hash fn supported list */

        /* TODO consider time/clock details?? */
} libtrace_info_t;

typedef struct libtrace_combine libtrace_combine_t;
struct libtrace_combine {

        int (*initialise)(libtrace_t *,libtrace_combine_t *);

        void (*destroy)(libtrace_t *, libtrace_combine_t *);

        void (*publish)(libtrace_t *, int thread_id, libtrace_combine_t *, libtrace_result_t *);

        void (*read)(libtrace_t *, libtrace_combine_t *);

        void (*read_final)(libtrace_t *, libtrace_combine_t *);

        void (*pause)(libtrace_t *, libtrace_combine_t *);

        void *queues;

        uint64_t last_count_tick;

        uint64_t last_ts_tick;

        libtrace_generic_t configuration;
};

typedef uint64_t (*fn_hasher)(const libtrace_packet_t* packet, void *data);


DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
                           libtrace_callback_set_t *per_packet_cbs,
                           libtrace_callback_set_t *reporter_cbs);

typedef void* (*fn_cb_starting)(libtrace_t *libtrace,
                                     libtrace_thread_t *t,
                                     void *global);

typedef void (*fn_cb_dataless)(libtrace_t *libtrace,
                                    libtrace_thread_t *t,
                                    void *global,
                                    void *tls);

typedef void (*fn_cb_first_packet)(libtrace_t *libtrace,
                                   libtrace_thread_t *t,
                                   void *global,
                                   void *tls,
                                   libtrace_thread_t *sender);

typedef void (*fn_cb_tick)(libtrace_t *libtrace,
                           libtrace_thread_t *t,
                           void *global,
                           void *tls,
                           uint64_t order);

typedef libtrace_packet_t* (*fn_cb_packet)(libtrace_t *libtrace,
                                           libtrace_thread_t *t,
                                           void *global,
                                           void *tls,
                                           libtrace_packet_t *packet);

typedef void (*fn_cb_result)(libtrace_t *libtrace, libtrace_thread_t *sender,
                void *global, void *tls, libtrace_result_t *result);


typedef void (*fn_cb_usermessage) (libtrace_t *libtrace, libtrace_thread_t *t,
                void *global, void *tls, int mesg, libtrace_generic_t data,
                libtrace_thread_t *sender);


DLLEXPORT int trace_set_starting_cb(libtrace_callback_set_t *cbset,
                fn_cb_starting handler);

DLLEXPORT int trace_set_stopping_cb(libtrace_callback_set_t *cbset,
                fn_cb_dataless handler);

DLLEXPORT int trace_set_resuming_cb(libtrace_callback_set_t *cbset,
                fn_cb_dataless handler);

DLLEXPORT int trace_set_pausing_cb(libtrace_callback_set_t *cbset,
                fn_cb_dataless handler);

DLLEXPORT int trace_set_packet_cb(libtrace_callback_set_t *cbset,
                fn_cb_packet handler);

DLLEXPORT int trace_set_first_packet_cb(libtrace_callback_set_t *cbset,
                fn_cb_first_packet handler);

DLLEXPORT int trace_set_result_cb(libtrace_callback_set_t *cbset,
                fn_cb_result handler);

DLLEXPORT int trace_set_tick_count_cb(libtrace_callback_set_t *cbset,
                fn_cb_tick handler);

DLLEXPORT int trace_set_tick_interval_cb(libtrace_callback_set_t *cbset,
                fn_cb_tick handler);

DLLEXPORT int trace_set_user_message_cb(libtrace_callback_set_t *cbset,
                fn_cb_usermessage handler);

DLLEXPORT libtrace_callback_set_t *trace_create_callback_set(void);

DLLEXPORT void trace_destroy_callback_set(libtrace_callback_set_t *cbset);


DLLEXPORT int trace_ppause(libtrace_t *libtrace);

DLLEXPORT int trace_pstop(libtrace_t *libtrace);

DLLEXPORT void trace_join(libtrace_t * trace);


DLLEXPORT int trace_set_perpkt_threads(libtrace_t *trace, int nb);

DLLEXPORT int trace_set_tick_interval(libtrace_t *trace, size_t millisec);

DLLEXPORT int trace_set_tick_count(libtrace_t *trace, size_t count);

DLLEXPORT int trace_set_tracetime(libtrace_t *trace, bool tracetime);

DLLEXPORT int trace_set_cache_size(libtrace_t *trace, size_t size);

DLLEXPORT int trace_set_thread_cache_size(libtrace_t *trace, size_t size);

DLLEXPORT int trace_set_fixed_count(libtrace_t *trace, bool fixed);

DLLEXPORT int trace_set_burst_size(libtrace_t *trace, size_t size);

DLLEXPORT int trace_set_hasher_queue_size(libtrace_t *trace, size_t size);

DLLEXPORT int trace_set_hasher_polling(libtrace_t *trace, bool polling);

DLLEXPORT int trace_set_reporter_polling(libtrace_t *trace, bool polling);

DLLEXPORT int trace_set_reporter_thold(libtrace_t *trace, size_t thold);

DLLEXPORT int trace_set_debug_state(libtrace_t *trace, bool debug_state);

DLLEXPORT int trace_set_hasher(libtrace_t *trace, enum hasher_types type,
                               fn_hasher hasher, void *data);



enum result_types {
        RESULT_PACKET,

        RESULT_TICK_INTERVAL,

        RESULT_TICK_COUNT,

        RESULT_USER = 1000

};

DLLEXPORT void trace_publish_result(libtrace_t *libtrace,
                                    libtrace_thread_t *t,
                                    uint64_t key,
                                    libtrace_generic_t value,
                                    int type);

DLLEXPORT bool trace_has_dedicated_hasher(libtrace_t * libtrace);

DLLEXPORT bool trace_has_reporter(libtrace_t * libtrace);

DLLEXPORT int trace_post_reporter(libtrace_t *libtrace);

DLLEXPORT int libtrace_thread_get_message_count(libtrace_t * libtrace,
                                                libtrace_thread_t *t);

DLLEXPORT int libtrace_thread_get_message(libtrace_t * libtrace,
                                          libtrace_thread_t *t,
                                          libtrace_message_t * message);

DLLEXPORT int libtrace_thread_try_get_message(libtrace_t * libtrace,
                                              libtrace_thread_t *t,
                                              libtrace_message_t * message);

DLLEXPORT int trace_message_reporter(libtrace_t * libtrace,
                                     libtrace_message_t * message);

DLLEXPORT int trace_message_perpkts(libtrace_t * libtrace,
                                    libtrace_message_t * message);

DLLEXPORT int trace_message_thread(libtrace_t * libtrace,
                                   libtrace_thread_t *t,
                                   libtrace_message_t * message);

DLLEXPORT bool trace_has_finished(libtrace_t * libtrace);


DLLEXPORT bool trace_is_parallel(libtrace_t * libtrace);

DLLEXPORT uint64_t trace_packet_get_order(libtrace_packet_t * packet);

DLLEXPORT uint64_t trace_packet_get_hash(libtrace_packet_t * packet);

DLLEXPORT void trace_packet_set_order(libtrace_packet_t * packet, uint64_t order);

DLLEXPORT void trace_packet_set_hash(libtrace_packet_t * packet, uint64_t hash);


DLLEXPORT int trace_get_first_packet(libtrace_t *libtrace,
                                     libtrace_thread_t *t,
                                     const libtrace_packet_t **packet,
                                     const struct timeval **tv);

DLLEXPORT void libtrace_make_packet_safe(libtrace_packet_t *pkt);

DLLEXPORT void libtrace_make_result_safe(libtrace_result_t *res);

DLLEXPORT void trace_free_packet(libtrace_t * libtrace, libtrace_packet_t * packet);

DLLEXPORT void trace_increment_packet_refcount(libtrace_packet_t *packet);

DLLEXPORT void trace_decrement_packet_refcount(libtrace_packet_t *packet);


DLLEXPORT libtrace_info_t *trace_get_information(libtrace_t * libtrace);

DLLEXPORT int trace_set_configuration(libtrace_t *trace, const char * str);

DLLEXPORT int trace_set_configuration_file(libtrace_t *trace, FILE *file);

DLLEXPORT int trace_get_perpkt_threads(libtrace_t* t); 

DLLEXPORT int trace_get_perpkt_thread_id(libtrace_thread_t *thread);

DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_t config);

extern const libtrace_combine_t combiner_unordered;

extern const libtrace_combine_t combiner_ordered;

extern const libtrace_combine_t combiner_sorted;

#ifdef __cplusplus
}
#endif

#endif // LIBTRACE_PARALLEL_H