X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=lttv%2Flttv%2Fsync%2Fevent_analysis_eval.c;h=49f3e17700f51a0399e61f02cdd2e17f831a576c;hb=76be6fc24daf61767bf7f0c2e64f4691fbb56c63;hp=9ee0f47f08d2776d6f4923748088d420396ca9f1;hpb=cdce23b31002139312fbf30f7afa0808a825d841;p=lttv.git diff --git a/lttv/lttv/sync/event_analysis_eval.c b/lttv/lttv/sync/event_analysis_eval.c index 9ee0f47f..49f3e177 100644 --- a/lttv/lttv/sync/event_analysis_eval.c +++ b/lttv/lttv/sync/event_analysis_eval.c @@ -16,13 +16,24 @@ * MA 02111-1307, USA. */ +#define _GNU_SOURCE + #ifdef HAVE_CONFIG_H #include #endif +#include +#include +#include +#include +#include #include +#include +#include +#include -#include "sync_chain_lttv.h" +#include "lookup3.h" +#include "sync_chain.h" #include "event_analysis_eval.h" @@ -42,6 +53,15 @@ static void printAnalysisStatsEval(SyncState* const syncState); // Functions specific to this module static void registerAnalysisEval() __attribute__((constructor (102))); +static guint ghfRttKeyHash(gconstpointer key); +static gboolean gefRttKeyEqual(gconstpointer a, gconstpointer b); +static void gdnDestroyRttKey(gpointer data); +static void gdnDestroyDouble(gpointer data); +static void readRttInfo(GHashTable* rttInfo, FILE* rttFile); +static void positionStream(FILE* stream); + +static void gfSum(gpointer data, gpointer userData); +static void gfSumSquares(gpointer data, gpointer userData); static AnalysisModule analysisModuleEval= { @@ -57,6 +77,14 @@ static AnalysisModule analysisModuleEval= { .writeAnalysisGraphsOptions= NULL, }; +static ModuleOption optionEvalRttFile= { + .longName= "eval-rtt-file", + .hasArg= REQUIRED_ARG, + {.arg= NULL}, + .optionHelp= "specify the file containing rtt information", + .argHelp= "FILE", +}; + /* * Analysis module registering function @@ -64,6 +92,7 @@ static AnalysisModule analysisModuleEval= { static void registerAnalysisEval() { g_queue_push_tail(&analysisModules, &analysisModuleEval); + g_queue_push_tail(&moduleOptions, &optionEvalRttFile); } @@ -84,19 +113,39 @@ static void initAnalysisEval(SyncState* const syncState) analysisData= malloc(sizeof(AnalysisDataEval)); syncState->analysisData= analysisData; - //readRttInfo(&analysisData->rttInfo, optionEvalRttFile); + analysisData->rttInfo= g_hash_table_new_full(&ghfRttKeyHash, + &gefRttKeyEqual, &gdnDestroyRttKey, &gdnDestroyDouble); + if (optionEvalRttFile.arg) + { + FILE* rttStream; + int retval; + + rttStream= fopen(optionEvalRttFile.arg, "r"); + if (rttStream == NULL) + { + g_error(strerror(errno)); + } + + readRttInfo(analysisData->rttInfo, rttStream); + + retval= fclose(rttStream); + if (retval == EOF) + { + g_error(strerror(errno)); + } + } if (syncState->stats) { - analysisData->stats= malloc(sizeof(AnalysisStatsEval)); + analysisData->stats= calloc(1, sizeof(AnalysisStatsEval)); analysisData->stats->broadcastDiffSum= 0.; - analysisData->stats->allStats= malloc(syncState->traceNb * - sizeof(TracePairStats*)); + analysisData->stats->messageStats= malloc(syncState->traceNb * + sizeof(MessageStats*)); for (i= 0; i < syncState->traceNb; i++) { - analysisData->stats->allStats[i]= calloc(syncState->traceNb, - sizeof(TracePairStats)); + analysisData->stats->messageStats[i]= calloc(syncState->traceNb, + sizeof(MessageStats)); } } } @@ -122,16 +171,16 @@ static void destroyAnalysisEval(SyncState* const syncState) return; } - //g_hash_table_destroy(analysisData->rttInfo); + g_hash_table_destroy(analysisData->rttInfo); analysisData->rttInfo= NULL; if (syncState->stats) { for (i= 0; i < syncState->traceNb; i++) { - free(analysisData->stats->allStats[i]); + free(analysisData->stats->messageStats[i]); } - free(analysisData->stats->allStats); + free(analysisData->stats->messageStats); free(analysisData->stats); } @@ -143,6 +192,8 @@ static void destroyAnalysisEval(SyncState* const syncState) /* * Perform analysis on an event pair. * + * Check if there is message inversion or messages that are too fast. + * * Args: * syncState container for synchronization data * message structure containing the events @@ -150,14 +201,52 @@ static void destroyAnalysisEval(SyncState* const syncState) static void analyzeMessageEval(SyncState* const syncState, Message* const message) { AnalysisDataEval* analysisData; + MessageStats* messageStats; + double* rttInfo; + double tt; + struct RttKey rttKey; + + if (!syncState->stats) + { + return; + } analysisData= (AnalysisDataEval*) syncState->analysisData; + messageStats= + &analysisData->stats->messageStats[message->outE->traceNum][message->inE->traceNum]; + + messageStats->total++; + + tt= wallTimeSub(&message->inE->wallTime, &message->outE->wallTime); + if (tt <= 0) + { + messageStats->inversionNb++; + } + + g_assert(message->inE->type == UDP); + rttKey.saddr= message->inE->event.udpEvent->datagramKey->saddr; + rttKey.daddr= message->inE->event.udpEvent->datagramKey->daddr; + rttInfo= g_hash_table_lookup(analysisData->rttInfo, &rttKey); + + if (rttInfo) + { + if (tt < *rttInfo / 2.) + { + messageStats->tooFastNb++; + } + } + else + { + messageStats->noRTTInfoNb++; + } } /* * Perform analysis on multiple messages * + * Measure the RTT + * * Args: * syncState container for synchronization data * exchange structure containing the messages @@ -173,6 +262,8 @@ static void analyzeExchangeEval(SyncState* const syncState, Exchange* const exch /* * Perform analysis on muliple events * + * Sum the broadcast differential delays + * * Args: * syncState container for synchronization data * broadcast structure containing the events @@ -180,8 +271,27 @@ static void analyzeExchangeEval(SyncState* const syncState, Exchange* const exch static void analyzeBroadcastEval(SyncState* const syncState, Broadcast* const broadcast) { AnalysisDataEval* analysisData; + double sum= 0, squaresSum= 0; + double y; + + if (!syncState->stats) + { + return; + } analysisData= (AnalysisDataEval*) syncState->analysisData; + + g_queue_foreach(broadcast->events, &gfSum, &sum); + g_queue_foreach(broadcast->events, &gfSumSquares, &squaresSum); + + analysisData->stats->broadcastNb++; + // Because of numerical errors, this can at times be < 0 + y= squaresSum / g_queue_get_length(broadcast->events) - pow(sum / + g_queue_get_length(broadcast->events), 2.); + if (y > 0) + { + analysisData->stats->broadcastDiffSum+= sqrt(y); + } } @@ -240,26 +350,280 @@ static void printAnalysisStatsEval(SyncState* const syncState) printf("Synchronization evaluation analysis stats:\n"); printf("\tsum of broadcast differential delays: %g\n", analysisData->stats->broadcastDiffSum); + printf("\taverage broadcast differential delays: %g\n", + analysisData->stats->broadcastDiffSum / + analysisData->stats->broadcastNb); printf("\tIndividual evaluation:\n" - "\t\tTrace pair Inversions Too fast (No RTT info)\n"); + "\t\tTrace pair Inversions Too fast (No RTT info) Total\n"); for (i= 0; i < syncState->traceNb; i++) { for (j= i + 1; j < syncState->traceNb; j++) { - TracePairStats* tpStats; - const char* format= "\t\t%3d - %-3d %-10u %-10u %u\n"; + MessageStats* messageStats; + const char* format= "\t\t%3d - %-3d %-10u %-10u %-10u %u\n"; - tpStats= &analysisData->stats->allStats[i][j]; + messageStats= &analysisData->stats->messageStats[i][j]; - printf(format, i, j, tpStats->inversionNb, tpStats->tooFastNb, - tpStats->noRTTInfoNb); + printf(format, i, j, messageStats->inversionNb, messageStats->tooFastNb, + messageStats->noRTTInfoNb, messageStats->total); - tpStats= &analysisData->stats->allStats[j][i]; + messageStats= &analysisData->stats->messageStats[j][i]; - printf(format, j, i, tpStats->inversionNb, tpStats->tooFastNb, - tpStats->noRTTInfoNb); + printf(format, j, i, messageStats->inversionNb, messageStats->tooFastNb, + messageStats->noRTTInfoNb, messageStats->total); } } } + + +/* + * A GHashFunc for g_hash_table_new() + * + * Args: + * key struct RttKey* + */ +static guint ghfRttKeyHash(gconstpointer key) +{ + struct RttKey* rttKey; + uint32_t a, b, c; + + rttKey= (struct RttKey*) key; + + a= rttKey->saddr; + b= rttKey->daddr; + c= 0; + final(a, b, c); + + return c; +} + + +/* + * A GDestroyNotify function for g_hash_table_new_full() + * + * Args: + * data: struct RttKey* + */ +static void gdnDestroyRttKey(gpointer data) +{ + free(data); +} + + +/* + * A GDestroyNotify function for g_hash_table_new_full() + * + * Args: + * data: double* + */ +static void gdnDestroyDouble(gpointer data) +{ + free(data); +} + + +/* + * A GEqualFunc for g_hash_table_new() + * + * Args: + * a, b RttKey* + * + * Returns: + * TRUE if both values are equal + */ +static gboolean gefRttKeyEqual(gconstpointer a, gconstpointer b) +{ + const struct RttKey* rkA, * rkB; + + rkA= (struct RttKey*) a; + rkB= (struct RttKey*) b; + + if (rkA->saddr == rkB->saddr && rkA->daddr == rkB->daddr) + { + return TRUE; + } + else + { + return FALSE; + } +} + + +/* + * Read a file contain minimum round trip time values and fill an array with + * them. The file is formatted as such: + * + * ip's should be in dotted quad format + * + * Args: + * rttInfo: double* rttInfo[RttKey], empty table, will be filled + * rttStream: stream from which to read + */ +static void readRttInfo(GHashTable* rttInfo, FILE* rttStream) +{ + char* line= NULL; + size_t len; + int retval; + + positionStream(rttStream); + retval= getline(&line, &len, rttStream); + while(!feof(rttStream)) + { + struct RttKey* rttKey; + char saddrDQ[20], daddrDQ[20]; + double* rtt; + char tmp; + struct in_addr addr; + unsigned int i; + struct { + char* dq; + size_t offset; + } loopValues[] = { + {saddrDQ, offsetof(struct RttKey, saddr)}, + {daddrDQ, offsetof(struct RttKey, daddr)} + }; + + if (retval == -1 && !feof(rttStream)) + { + g_error(strerror(errno)); + } + + if (line[retval - 1] == '\n') + { + line[retval - 1]= '\0'; + } + + rtt= malloc(sizeof(double)); + retval= sscanf(line, " %19s %19s %lf %c", saddrDQ, daddrDQ, rtt, + &tmp); + if (retval == EOF) + { + g_error(strerror(errno)); + } + else if (retval != 3) + { + g_error("Error parsing RTT file, line was '%s'", line); + } + + rttKey= malloc(sizeof(struct RttKey)); + for (i= 0; i < sizeof(loopValues) / sizeof(*loopValues); i++) + { + retval= inet_aton(loopValues[i].dq, &addr); + if (retval == 0) + { + g_error("Error converting address '%s'", loopValues[i].dq); + } + *(uint32_t*) ((void*) rttKey + loopValues[i].offset)= + addr.s_addr; + } + + *rtt/= 1e3; + g_hash_table_insert(rttInfo, rttKey, rtt); + + positionStream(rttStream); + retval= getline(&line, &len, rttStream); + } + + if (line) + { + free(line); + } +} + + +/* + * Advance stream over empty space, empty lines and lines that begin with '#' + * + * Args: + * stream: stream, at exit, will be over the first non-empty character + * of a line of be at EOF + */ +static void positionStream(FILE* stream) +{ + int firstChar; + ssize_t retval; + char* line= NULL; + size_t len; + + do + { + firstChar= fgetc(stream); + if (firstChar == (int) '#') + { + retval= getline(&line, &len, stream); + if (retval == -1) + { + if (feof(stream)) + { + goto outEof; + } + else + { + g_error(strerror(errno)); + } + } + } + else if (firstChar == (int) '\n' || firstChar == (int) ' ' || + firstChar == (int) '\t') + {} + else if (firstChar == EOF) + { + goto outEof; + } + else + { + break; + } + } while (true); + retval= ungetc(firstChar, stream); + if (retval == EOF) + { + g_error("Error: ungetc()"); + } + +outEof: + if (line) + { + free(line); + } +} + + +/* + * A GFunc for g_queue_foreach() + * + * Args: + * data Event*, a UDP broadcast event + * user_data double*, the running sum + * + * Returns: + * Adds the time of the event to the sum + */ +static void gfSum(gpointer data, gpointer userData) +{ + Event* event= (Event*) data; + + *(double*) userData+= event->wallTime.seconds + event->wallTime.nanosec / + 1e9; +} + + +/* + * A GFunc for g_queue_foreach() + * + * Args: + * data Event*, a UDP broadcast event + * user_data double*, the running sum + * + * Returns: + * Adds the square of the time of the event to the sum + */ +static void gfSumSquares(gpointer data, gpointer userData) +{ + Event* event= (Event*) data; + + *(double*) userData+= pow(event->wallTime.seconds + event->wallTime.nanosec + / 1e9, 2.); +}