lttv/lttv/sync/event_analysis_eval.c

   1 /* This file is part of the Linux Trace Toolkit viewer
   2  * Copyright (C) 2009 Benjamin Poirier <benjamin.poirier@polymtl.ca>
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License Version 2 as
   6  * published by the Free Software Foundation;
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License
  14  * along with this program; if not, write to the Free Software
  15  * Foundation, Inc., 59 Temple Place - Suite 330, Boston,
  16  * MA 02111-1307, USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include <config.h>
  23 #endif
  24
  25 #include <arpa/inet.h>
  26 #include <errno.h>
  27 #include <math.h>
  28 #include <netinet/in.h>
  29 #include <stddef.h>
  30 #include <stdlib.h>
  31 #include <stdio.h>
  32 #include <string.h>
  33 #include <sys/socket.h>
  34
  35 #include "lookup3.h"
  36 #include "sync_chain.h"
  37
  38 #include "event_analysis_eval.h"
  39
  40
  41 // Functions common to all analysis modules
  42 static void initAnalysisEval(SyncState* const syncState);
  43 static void destroyAnalysisEval(SyncState* const syncState);
  44
  45 static void analyzeMessageEval(SyncState* const syncState, Message* const
  46         message);
  47 static void analyzeExchangeEval(SyncState* const syncState, Exchange* const
  48         exchange);
  49 static void analyzeBroadcastEval(SyncState* const syncState, Broadcast* const
  50         broadcast);
  51 static GArray* finalizeAnalysisEval(SyncState* const syncState);
  52 static void printAnalysisStatsEval(SyncState* const syncState);
  53
  54 // Functions specific to this module
  55 static void registerAnalysisEval() __attribute__((constructor (102)));
  56 static guint ghfRttKeyHash(gconstpointer key);
  57 static gboolean gefRttKeyEqual(gconstpointer a, gconstpointer b);
  58 static void gdnDestroyRttKey(gpointer data);
  59 static void gdnDestroyDouble(gpointer data);
  60 static void readRttInfo(GHashTable* rttInfo, FILE* rttFile);
  61 static void positionStream(FILE* stream);
  62
  63 static void gfSum(gpointer data, gpointer userData);
  64 static void gfSumSquares(gpointer data, gpointer userData);
  65
  66
  67 static AnalysisModule analysisModuleEval= {
  68         .name= "eval",
  69         .initAnalysis= &initAnalysisEval,
  70         .destroyAnalysis= &destroyAnalysisEval,
  71         .analyzeMessage= &analyzeMessageEval,
  72         .analyzeExchange= &analyzeExchangeEval,
  73         .analyzeBroadcast= &analyzeBroadcastEval,
  74         .finalizeAnalysis= &finalizeAnalysisEval,
  75         .printAnalysisStats= &printAnalysisStatsEval,
  76         .writeAnalysisGraphsPlots= NULL,
  77         .writeAnalysisGraphsOptions= NULL,
  78 };
  79
  80 static ModuleOption optionEvalRttFile= {
  81         .longName= "eval-rtt-file",
  82         .hasArg= REQUIRED_ARG,
  83         {.arg= NULL},
  84         .optionHelp= "specify the file containing rtt information",
  85         .argHelp= "FILE",
  86 };
  87
  88
  89 /*
  90  * Analysis module registering function
  91  */
  92 static void registerAnalysisEval()
  93 {
  94         g_queue_push_tail(&analysisModules, &analysisModuleEval);
  95         g_queue_push_tail(&moduleOptions, &optionEvalRttFile);
  96 }
  97
  98
  99 /*
 100  * Analysis init function
 101  *
 102  * This function is called at the beginning of a synchronization run for a set
 103  * of traces.
 104  *
 105  * Args:
 106  *   syncState     container for synchronization data.
 107  */
 108 static void initAnalysisEval(SyncState* const syncState)
 109 {
 110         AnalysisDataEval* analysisData;
 111         unsigned int i;
 112
 113         analysisData= malloc(sizeof(AnalysisDataEval));
 114         syncState->analysisData= analysisData;
 115
 116         analysisData->rttInfo= g_hash_table_new_full(&ghfRttKeyHash,
 117                 &gefRttKeyEqual, &gdnDestroyRttKey, &gdnDestroyDouble);
 118         if (optionEvalRttFile.arg)
 119         {
 120                 FILE* rttStream;
 121                 int retval;
 122
 123                 rttStream= fopen(optionEvalRttFile.arg, "r");
 124                 if (rttStream == NULL)
 125                 {
 126                         g_error(strerror(errno));
 127                 }
 128
 129                 readRttInfo(analysisData->rttInfo, rttStream);
 130
 131                 retval= fclose(rttStream);
 132                 if (retval == EOF)
 133                 {
 134                         g_error(strerror(errno));
 135                 }
 136         }
 137
 138         if (syncState->stats)
 139         {
 140                 analysisData->stats= calloc(1, sizeof(AnalysisStatsEval));
 141                 analysisData->stats->broadcastDiffSum= 0.;
 142
 143                 analysisData->stats->messageStats= malloc(syncState->traceNb *
 144                         sizeof(MessageStats*));
 145                 for (i= 0; i < syncState->traceNb; i++)
 146                 {
 147                         analysisData->stats->messageStats[i]= calloc(syncState->traceNb,
 148                                 sizeof(MessageStats));
 149                 }
 150         }
 151 }
 152
 153
 154 /*
 155  * Analysis destroy function
 156  *
 157  * Free the analysis specific data structures
 158  *
 159  * Args:
 160  *   syncState     container for synchronization data.
 161  */
 162 static void destroyAnalysisEval(SyncState* const syncState)
 163 {
 164         unsigned int i;
 165         AnalysisDataEval* analysisData;
 166
 167         analysisData= (AnalysisDataEval*) syncState->analysisData;
 168
 169         if (analysisData == NULL || analysisData->rttInfo == NULL)
 170         {
 171                 return;
 172         }
 173
 174         g_hash_table_destroy(analysisData->rttInfo);
 175         analysisData->rttInfo= NULL;
 176
 177         if (syncState->stats)
 178         {
 179                 for (i= 0; i < syncState->traceNb; i++)
 180                 {
 181                         free(analysisData->stats->messageStats[i]);
 182                 }
 183                 free(analysisData->stats->messageStats);
 184                 free(analysisData->stats);
 185         }
 186
 187         free(syncState->analysisData);
 188         syncState->analysisData= NULL;
 189 }
 190
 191
 192 /*
 193  * Perform analysis on an event pair.
 194  *
 195  * Check if there is message inversion or messages that are too fast.
 196  *
 197  * Args:
 198  *   syncState     container for synchronization data
 199  *   message       structure containing the events
 200  */
 201 static void analyzeMessageEval(SyncState* const syncState, Message* const message)
 202 {
 203         AnalysisDataEval* analysisData;
 204         MessageStats* messageStats;
 205         double* rttInfo;
 206         double tt;
 207         struct RttKey rttKey;
 208
 209         if (!syncState->stats)
 210         {
 211                 return;
 212         }
 213
 214         analysisData= (AnalysisDataEval*) syncState->analysisData;
 215         messageStats=
 216                 &analysisData->stats->messageStats[message->outE->traceNum][message->inE->traceNum];
 217
 218         messageStats->total++;
 219
 220         tt= wallTimeSub(&message->inE->wallTime, &message->outE->wallTime);
 221         if (tt <= 0)
 222         {
 223                 messageStats->inversionNb++;
 224         }
 225
 226         g_assert(message->inE->type == UDP);
 227         rttKey.saddr= message->inE->event.udpEvent->datagramKey->saddr;
 228         rttKey.daddr= message->inE->event.udpEvent->datagramKey->daddr;
 229         rttInfo= g_hash_table_lookup(analysisData->rttInfo, &rttKey);
 230
 231         if (rttInfo)
 232         {
 233                 if (tt < *rttInfo / 2.)
 234                 {
 235                         messageStats->tooFastNb++;
 236                 }
 237         }
 238         else
 239         {
 240                 messageStats->noRTTInfoNb++;
 241         }
 242 }
 243
 244
 245 /*
 246  * Perform analysis on multiple messages
 247  *
 248  * Measure the RTT
 249  *
 250  * Args:
 251  *   syncState     container for synchronization data
 252  *   exchange      structure containing the messages
 253  */
 254 static void analyzeExchangeEval(SyncState* const syncState, Exchange* const exchange)
 255 {
 256         AnalysisDataEval* analysisData;
 257
 258         analysisData= (AnalysisDataEval*) syncState->analysisData;
 259 }
 260
 261
 262 /*
 263  * Perform analysis on muliple events
 264  *
 265  * Sum the broadcast differential delays
 266  *
 267  * Args:
 268  *   syncState     container for synchronization data
 269  *   broadcast     structure containing the events
 270  */
 271 static void analyzeBroadcastEval(SyncState* const syncState, Broadcast* const broadcast)
 272 {
 273         AnalysisDataEval* analysisData;
 274         double sum= 0, squaresSum= 0;
 275         double y;
 276
 277         if (!syncState->stats)
 278         {
 279                 return;
 280         }
 281
 282         analysisData= (AnalysisDataEval*) syncState->analysisData;
 283
 284         g_queue_foreach(broadcast->events, &gfSum, &sum);
 285         g_queue_foreach(broadcast->events, &gfSumSquares, &squaresSum);
 286
 287         analysisData->stats->broadcastNb++;
 288         // Because of numerical errors, this can at times be < 0
 289         y= squaresSum / g_queue_get_length(broadcast->events) - pow(sum /
 290                 g_queue_get_length(broadcast->events), 2.);
 291         if (y > 0)
 292         {
 293                 analysisData->stats->broadcastDiffSum+= sqrt(y);
 294         }
 295 }
 296
 297
 298 /*
 299  * Finalize the factor calculations
 300  *
 301  * Since this module does not really calculate factors, identity factors are
 302  * returned.
 303  *
 304  * Args:
 305  *   syncState     container for synchronization data.
 306  *
 307  * Returns:
 308  *   Factors[traceNb] synchronization factors for each trace
 309  */
 310 static GArray* finalizeAnalysisEval(SyncState* const syncState)
 311 {
 312         GArray* factors;
 313         unsigned int i;
 314
 315         factors= g_array_sized_new(FALSE, FALSE, sizeof(Factors),
 316                 syncState->traceNb);
 317         g_array_set_size(factors, syncState->traceNb);
 318         for (i= 0; i < syncState->traceNb; i++)
 319         {
 320                 Factors* e;
 321
 322                 e= &g_array_index(factors, Factors, i);
 323                 e->drift= 1.;
 324                 e->offset= 0.;
 325         }
 326
 327         return factors;
 328 }
 329
 330
 331 /*
 332  * Print statistics related to analysis. Must be called after
 333  * finalizeAnalysis.
 334  *
 335  * Args:
 336  *   syncState     container for synchronization data.
 337  */
 338 static void printAnalysisStatsEval(SyncState* const syncState)
 339 {
 340         AnalysisDataEval* analysisData;
 341         unsigned int i, j;
 342
 343         if (!syncState->stats)
 344         {
 345                 return;
 346         }
 347
 348         analysisData= (AnalysisDataEval*) syncState->analysisData;
 349
 350         printf("Synchronization evaluation analysis stats:\n");
 351         printf("\tsum of broadcast differential delays: %g\n",
 352                 analysisData->stats->broadcastDiffSum);
 353         printf("\taverage broadcast differential delays: %g\n",
 354                 analysisData->stats->broadcastDiffSum /
 355                 analysisData->stats->broadcastNb);
 356
 357         printf("\tIndividual evaluation:\n"
 358                 "\t\tTrace pair  Inversions Too fast   (No RTT info) Total\n");
 359
 360         for (i= 0; i < syncState->traceNb; i++)
 361         {
 362                 for (j= i + 1; j < syncState->traceNb; j++)
 363                 {
 364                         MessageStats* messageStats;
 365                         const char* format= "\t\t%3d - %-3d   %-10u %-10u %-10u    %u\n";
 366
 367                         messageStats= &analysisData->stats->messageStats[i][j];
 368
 369                         printf(format, i, j, messageStats->inversionNb, messageStats->tooFastNb,
 370                                 messageStats->noRTTInfoNb, messageStats->total);
 371
 372                         messageStats= &analysisData->stats->messageStats[j][i];
 373
 374                         printf(format, j, i, messageStats->inversionNb, messageStats->tooFastNb,
 375                                 messageStats->noRTTInfoNb, messageStats->total);
 376                 }
 377         }
 378 }
 379
 380
 381 /*
 382  * A GHashFunc for g_hash_table_new()
 383  *
 384  * Args:
 385  *    key        struct RttKey*
 386  */
 387 static guint ghfRttKeyHash(gconstpointer key)
 388 {
 389         struct RttKey* rttKey;
 390         uint32_t a, b, c;
 391
 392         rttKey= (struct RttKey*) key;
 393
 394         a= rttKey->saddr;
 395         b= rttKey->daddr;
 396         c= 0;
 397         final(a, b, c);
 398
 399         return c;
 400 }
 401
 402
 403 /*
 404  * A GDestroyNotify function for g_hash_table_new_full()
 405  *
 406  * Args:
 407  *   data:         struct RttKey*
 408  */
 409 static void gdnDestroyRttKey(gpointer data)
 410 {
 411         free(data);
 412 }
 413
 414
 415 /*
 416  * A GDestroyNotify function for g_hash_table_new_full()
 417  *
 418  * Args:
 419  *   data:         double*
 420  */
 421 static void gdnDestroyDouble(gpointer data)
 422 {
 423         free(data);
 424 }
 425
 426
 427 /*
 428  * A GEqualFunc for g_hash_table_new()
 429  *
 430  * Args:
 431  *   a, b          RttKey*
 432  *
 433  * Returns:
 434  *   TRUE if both values are equal
 435  */
 436 static gboolean gefRttKeyEqual(gconstpointer a, gconstpointer b)
 437 {
 438         const struct RttKey* rkA, * rkB;
 439
 440         rkA= (struct RttKey*) a;
 441         rkB= (struct RttKey*) b;
 442
 443         if (rkA->saddr == rkB->saddr && rkA->daddr == rkB->daddr)
 444         {
 445                 return TRUE;
 446         }
 447         else
 448         {
 449                 return FALSE;
 450         }
 451 }
 452
 453
 454 /*
 455  * Read a file contain minimum round trip time values and fill an array with
 456  * them. The file is formatted as such:
 457  * <host1 IP> <host2 IP> <RTT in milliseconds>
 458  * ip's should be in dotted quad format
 459  *
 460  * Args:
 461  *   rttInfo:      double* rttInfo[RttKey], empty table, will be filled
 462  *   rttStream:      stream from which to read
 463  */
 464 static void readRttInfo(GHashTable* rttInfo, FILE* rttStream)
 465 {
 466         char* line= NULL;
 467         size_t len;
 468         int retval;
 469
 470         positionStream(rttStream);
 471         retval= getline(&line, &len, rttStream);
 472         while(!feof(rttStream))
 473         {
 474                 struct RttKey* rttKey;
 475                 char saddrDQ[20], daddrDQ[20];
 476                 double* rtt;
 477                 char tmp;
 478                 struct in_addr addr;
 479                 unsigned int i;
 480                 struct {
 481                         char* dq;
 482                         size_t offset;
 483                 } loopValues[] = {
 484                         {saddrDQ, offsetof(struct RttKey, saddr)},
 485                         {daddrDQ, offsetof(struct RttKey, daddr)}
 486                 };
 487
 488                 if (retval == -1 && !feof(rttStream))
 489                 {
 490                         g_error(strerror(errno));
 491                 }
 492
 493                 if (line[retval - 1] == '\n')
 494                 {
 495                         line[retval - 1]= '\0';
 496                 }
 497
 498                 rtt= malloc(sizeof(double));
 499                 retval= sscanf(line, " %19s %19s %lf %c", saddrDQ, daddrDQ, rtt,
 500                         &tmp);
 501                 if (retval == EOF)
 502                 {
 503                         g_error(strerror(errno));
 504                 }
 505                 else if (retval != 3)
 506                 {
 507                         g_error("Error parsing RTT file, line was '%s'", line);
 508                 }
 509
 510                 rttKey= malloc(sizeof(struct RttKey));
 511                 for (i= 0; i < sizeof(loopValues) / sizeof(*loopValues); i++)
 512                 {
 513                         retval= inet_aton(loopValues[i].dq, &addr);
 514                         if (retval == 0)
 515                         {
 516                                 g_error("Error converting address '%s'", loopValues[i].dq);
 517                         }
 518                         *(uint32_t*) ((void*) rttKey + loopValues[i].offset)=
 519                                 addr.s_addr;
 520                 }
 521
 522                 *rtt/= 1e3;
 523                 g_hash_table_insert(rttInfo, rttKey, rtt);
 524
 525                 positionStream(rttStream);
 526                 retval= getline(&line, &len, rttStream);
 527         }
 528
 529         if (line)
 530         {
 531                 free(line);
 532         }
 533 }
 534
 535
 536 /*
 537  * Advance stream over empty space, empty lines and lines that begin with '#'
 538  *
 539  * Args:
 540  *   stream:     stream, at exit, will be over the first non-empty character
 541  *               of a line of be at EOF
 542  */
 543 static void positionStream(FILE* stream)
 544 {
 545         int firstChar;
 546         ssize_t retval;
 547         char* line= NULL;
 548         size_t len;
 549
 550         do
 551         {
 552                 firstChar= fgetc(stream);
 553                 if (firstChar == (int) '#')
 554                 {
 555                         retval= getline(&line, &len, stream);
 556                         if (retval == -1)
 557                         {
 558                                 if (feof(stream))
 559                                 {
 560                                         goto outEof;
 561                                 }
 562                                 else
 563                                 {
 564                                         g_error(strerror(errno));
 565                                 }
 566                         }
 567                 }
 568                 else if (firstChar == (int) '\n' || firstChar == (int) ' ' ||
 569                         firstChar == (int) '\t')
 570                 {}
 571                 else if (firstChar == EOF)
 572                 {
 573                         goto outEof;
 574                 }
 575                 else
 576                 {
 577                         break;
 578                 }
 579         } while (true);
 580         retval= ungetc(firstChar, stream);
 581         if (retval == EOF)
 582         {
 583                 g_error("Error: ungetc()");
 584         }
 585
 586 outEof:
 587         if (line)
 588         {
 589                 free(line);
 590         }
 591 }
 592
 593
 594 /*
 595  * A GFunc for g_queue_foreach()
 596  *
 597  * Args:
 598  *   data          Event*, a UDP broadcast event
 599  *   user_data     double*, the running sum
 600  *
 601  * Returns:
 602  *   Adds the time of the event to the sum
 603  */
 604 static void gfSum(gpointer data, gpointer userData)
 605 {
 606         Event* event= (Event*) data;
 607
 608         *(double*) userData+= event->wallTime.seconds + event->wallTime.nanosec /
 609                 1e9;
 610 }
 611
 612
 613 /*
 614  * A GFunc for g_queue_foreach()
 615  *
 616  * Args:
 617  *   data          Event*, a UDP broadcast event
 618  *   user_data     double*, the running sum
 619  *
 620  * Returns:
 621  *   Adds the square of the time of the event to the sum
 622  */
 623 static void gfSumSquares(gpointer data, gpointer userData)
 624 {
 625         Event* event= (Event*) data;
 626
 627         *(double*) userData+= pow(event->wallTime.seconds + event->wallTime.nanosec
 628                 / 1e9, 2.);
 629 }