lttv/lttv/sync/event_matching_tcp.c

   1 /* This file is part of the Linux Trace Toolkit viewer
   2  * Copyright (C) 2009, 2010 Benjamin Poirier <benjamin.poirier@polymtl.ca>
   3  *
   4  * This program is free software: you can redistribute it and/or modify it
   5  * under the terms of the GNU Lesser General Public License as published by
   6  * the Free Software Foundation, either version 2.1 of the License, or (at
   7  * your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
  12  * License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public License
  15  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16  */
  17
  18 #ifdef HAVE_CONFIG_H
  19 #include <config.h>
  20 #endif
  21
  22 #include <errno.h>
  23 #include <inttypes.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <unistd.h>
  27
  28 #include "event_analysis.h"
  29 #include "sync_chain.h"
  30
  31 #include "event_matching_tcp.h"
  32
  33
  34 // Functions common to all matching modules
  35 static void initMatchingTCP(SyncState* const syncState);
  36 static void destroyMatchingTCP(SyncState* const syncState);
  37
  38 static void matchEventTCP(SyncState* const syncState, Event* const event);
  39 static AllFactors* finalizeMatchingTCP(SyncState* const syncState);
  40 static void printMatchingStatsTCP(SyncState* const syncState);
  41 static void writeMatchingGraphsPlotsTCPMessages(SyncState* const syncState,
  42         const unsigned int i, const unsigned int j);
  43
  44 // Functions specific to this module
  45 static void matchEvents(SyncState* const syncState, Event* const event,
  46         GHashTable* const unMatchedList, GHashTable* const
  47         unMatchedOppositeList, const size_t fieldOffset, const size_t
  48         oppositeFieldOffset);
  49 static void partialDestroyMatchingTCP(SyncState* const syncState);
  50
  51 static bool isAck(const Message* const message);
  52 static bool needsAck(const Message* const message);
  53 static void buildReversedConnectionKey(ConnectionKey* const
  54         reversedConnectionKey, const ConnectionKey* const connectionKey);
  55
  56 static void openGraphDataFiles(SyncState* const syncState);
  57 static void closeGraphDataFiles(SyncState* const syncState);
  58 static void writeMessagePoint(FILE* stream, const Message* const message);
  59
  60
  61 static MatchingModule matchingModuleTCP = {
  62         .name= "TCP",
  63         .canMatch[TCP]= true,
  64         .canMatch[UDP]= false,
  65         .initMatching= &initMatchingTCP,
  66         .destroyMatching= &destroyMatchingTCP,
  67         .matchEvent= &matchEventTCP,
  68         .finalizeMatching= &finalizeMatchingTCP,
  69         .printMatchingStats= &printMatchingStatsTCP,
  70         .graphFunctions= {
  71                 .writeTraceTraceForePlots= &writeMatchingGraphsPlotsTCPMessages,
  72         }
  73 };
  74
  75
  76 /*
  77  * Matching module registering function
  78  */
  79 void registerMatchingTCP()
  80 {
  81         g_queue_push_tail(&matchingModules, &matchingModuleTCP);
  82 }
  83
  84
  85 /*
  86  * Matching init function
  87  *
  88  * This function is called at the beginning of a synchronization run for a set
  89  * of traces.
  90  *
  91  * Allocate the matching specific data structures
  92  *
  93  * Args:
  94  *   syncState     container for synchronization data.
  95  *                 This function allocates these matchingData members:
  96  *                 unMatchedInE
  97  *                 unMatchedOutE
  98  *                 unAcked
  99  *                 stats
 100  */
 101 static void initMatchingTCP(SyncState* const syncState)
 102 {
 103         MatchingDataTCP* matchingData;
 104
 105         matchingData= malloc(sizeof(MatchingDataTCP));
 106         syncState->matchingData= matchingData;
 107
 108         matchingData->unMatchedInE= g_hash_table_new_full(&ghfSegmentKeyHash,
 109                 &gefSegmentKeyEqual, NULL, &gdnDestroyEvent);
 110         matchingData->unMatchedOutE= g_hash_table_new_full(&ghfSegmentKeyHash,
 111                 &gefSegmentKeyEqual, NULL, &gdnDestroyEvent);
 112         matchingData->unAcked= g_hash_table_new_full(&ghfConnectionKeyHash,
 113                 &gefConnectionKeyEqual, &gdnConnectionKeyDestroy,
 114                 &gdnTCPSegmentListDestroy);
 115
 116         if (syncState->stats)
 117         {
 118                 unsigned int i;
 119
 120                 matchingData->stats= calloc(1, sizeof(MatchingStatsTCP));
 121                 matchingData->stats->totMessageArray= malloc(syncState->traceNb *
 122                         sizeof(unsigned int*));
 123                 for (i= 0; i < syncState->traceNb; i++)
 124                 {
 125                         matchingData->stats->totMessageArray[i]=
 126                                 calloc(syncState->traceNb, sizeof(unsigned int));
 127                 }
 128         }
 129         else
 130         {
 131                 matchingData->stats= NULL;
 132         }
 133
 134         if (syncState->graphsStream)
 135         {
 136                 openGraphDataFiles(syncState);
 137         }
 138         else
 139         {
 140                 matchingData->messagePoints= NULL;
 141         }
 142 }
 143
 144
 145 /*
 146  * Matching destroy function
 147  *
 148  * Free the matching specific data structures
 149  *
 150  * Args:
 151  *   syncState     container for synchronization data.
 152  *                 This function deallocates these matchingData members:
 153  *                 stats
 154  */
 155 static void destroyMatchingTCP(SyncState* const syncState)
 156 {
 157         MatchingDataTCP* matchingData;
 158
 159         matchingData= (MatchingDataTCP*) syncState->matchingData;
 160
 161         if (matchingData == NULL)
 162         {
 163                 return;
 164         }
 165
 166         partialDestroyMatchingTCP(syncState);
 167
 168         if (syncState->stats)
 169         {
 170                 unsigned int i;
 171
 172                 for (i= 0; i < syncState->traceNb; i++)
 173                 {
 174                         free(matchingData->stats->totMessageArray[i]);
 175                 }
 176                 free(matchingData->stats->totMessageArray);
 177                 free(matchingData->stats);
 178         }
 179
 180         free(syncState->matchingData);
 181         syncState->matchingData= NULL;
 182 }
 183
 184
 185 /*
 186  * Free some of the matching specific data structures
 187  *
 188  * This function can be called right after the events have been processed to
 189  * free some data structures that are not needed for finalization.
 190  *
 191  * Args:
 192  *   syncState     container for synchronization data.
 193  *                 This function deallocates these matchingData members:
 194  *                 unMatchedInE
 195  *                 unMatchedOut
 196  *                 unAcked
 197  */
 198 static void partialDestroyMatchingTCP(SyncState* const syncState)
 199 {
 200         MatchingDataTCP* matchingData;
 201
 202         matchingData= (MatchingDataTCP*) syncState->matchingData;
 203
 204         if (matchingData == NULL || matchingData->unMatchedInE == NULL)
 205         {
 206                 return;
 207         }
 208
 209         g_hash_table_destroy(matchingData->unMatchedInE);
 210         matchingData->unMatchedInE= NULL;
 211         g_hash_table_destroy(matchingData->unMatchedOutE);
 212         g_hash_table_destroy(matchingData->unAcked);
 213
 214         if (syncState->graphsStream && matchingData->messagePoints)
 215         {
 216                 closeGraphDataFiles(syncState);
 217         }
 218 }
 219
 220
 221 /*
 222  * Try to match one event from a trace with the corresponding event from
 223  * another trace.
 224  *
 225  * Args:
 226  *   syncState     container for synchronization data.
 227  *   event         new event to match
 228  */
 229 static void matchEventTCP(SyncState* const syncState, Event* const event)
 230 {
 231         MatchingDataTCP* matchingData;
 232
 233         g_assert(event->type == TCP);
 234
 235         matchingData= (MatchingDataTCP*) syncState->matchingData;
 236
 237         if (event->event.tcpEvent->direction == IN)
 238         {
 239                 matchEvents(syncState, event, matchingData->unMatchedInE,
 240                         matchingData->unMatchedOutE, offsetof(Message, inE),
 241                         offsetof(Message, outE));
 242         }
 243         else
 244         {
 245                 matchEvents(syncState, event, matchingData->unMatchedOutE,
 246                         matchingData->unMatchedInE, offsetof(Message, outE),
 247                         offsetof(Message, inE));
 248         }
 249 }
 250
 251
 252 /*
 253  * Call the partial matching destroyer and obtain the factors from downstream
 254  *
 255  * Args:
 256  *   syncState     container for synchronization data.
 257  *
 258  * Returns:
 259  *   AllFactors*   synchronization factors for each trace pair
 260  */
 261 static AllFactors* finalizeMatchingTCP(SyncState* const syncState)
 262 {
 263         partialDestroyMatchingTCP(syncState);
 264
 265         return syncState->analysisModule->finalizeAnalysis(syncState);
 266 }
 267
 268
 269 /*
 270  * Print statistics related to matching. Must be called after
 271  * finalizeMatching.
 272  *
 273  * Args:
 274  *   syncState     container for synchronization data.
 275  */
 276 static void printMatchingStatsTCP(SyncState* const syncState)
 277 {
 278         unsigned int i, j;
 279         MatchingDataTCP* matchingData;
 280
 281         if (!syncState->stats)
 282         {
 283                 return;
 284         }
 285
 286         matchingData= (MatchingDataTCP*) syncState->matchingData;
 287
 288         printf("TCP matching stats:\n");
 289         printf("\ttotal input and output events matched together to form a packet: %u\n",
 290                 matchingData->stats->totPacket);
 291
 292         printf("\tMessage traffic:\n");
 293
 294         for (i= 0; i < syncState->traceNb; i++)
 295         {
 296                 for (j= i + 1; j < syncState->traceNb; j++)
 297                 {
 298                         printf("\t\t%3d - %-3d: sent %-10u received %-10u\n", i, j,
 299                                 matchingData->stats->totMessageArray[j][i],
 300                                 matchingData->stats->totMessageArray[i][j]);
 301                 }
 302         }
 303
 304         if (syncState->analysisModule->analyzeExchange != NULL)
 305         {
 306                 printf("\ttotal packets identified needing an acknowledge: %u\n",
 307                         matchingData->stats->totPacketNeedAck);
 308                 printf("\ttotal exchanges (four events matched together): %u\n",
 309                         matchingData->stats->totExchangeEffective);
 310                 printf("\ttotal synchronization exchanges: %u\n",
 311                         matchingData->stats->totExchangeSync);
 312         }
 313 }
 314
 315
 316 /*
 317  * Implementation of a packet matching algorithm for TCP
 318  *
 319  * Args:
 320  *   event:        new event to match
 321  *   unMatchedList: list of unmatched events of the same type (send or
 322  *                 receive) as event
 323  *   unMatchedOppositeList: list of unmatched events of the opposite type of
 324  *                 event
 325  *   fieldOffset:  offset of the Event field in the Message struct for the
 326  *                 field of the type of event
 327  *   oppositeFieldOffset: offset of the Event field in the Message struct
 328  *                 for the field of the opposite type of event
 329  */
 330 static void matchEvents(SyncState* const syncState, Event* const event,
 331         GHashTable* const unMatchedList, GHashTable* const unMatchedOppositeList,
 332         const size_t fieldOffset, const size_t oppositeFieldOffset)
 333 {
 334         Event* companionEvent;
 335         Message* packet;
 336         MatchingDataTCP* matchingData;
 337         GQueue* conUnAcked;
 338
 339         matchingData= (MatchingDataTCP*) syncState->matchingData;
 340
 341         companionEvent= g_hash_table_lookup(unMatchedOppositeList, event->event.tcpEvent->segmentKey);
 342         if (companionEvent != NULL)
 343         {
 344                 g_debug("Found matching companion event, ");
 345
 346                 // If it's there, remove it and create a Message
 347                 g_hash_table_steal(unMatchedOppositeList, event->event.tcpEvent->segmentKey);
 348                 packet= malloc(sizeof(Message));
 349                 *((Event**) ((void*) packet + fieldOffset))= event;
 350                 *((Event**) ((void*) packet + oppositeFieldOffset))= companionEvent;
 351                 packet->print= &printTCPSegment;
 352                 // Both events can now share the same segmentKey
 353                 free(packet->outE->event.tcpEvent->segmentKey);
 354                 packet->outE->event.tcpEvent->segmentKey= packet->inE->event.tcpEvent->segmentKey;
 355
 356                 if (syncState->stats)
 357                 {
 358                         matchingData->stats->totPacket++;
 359                         matchingData->stats->totMessageArray[packet->inE->traceNum][packet->outE->traceNum]++;
 360                 }
 361
 362                 // Discard loopback traffic
 363                 if (packet->inE->traceNum == packet->outE->traceNum)
 364                 {
 365                         destroyTCPSegment(packet);
 366                         return;
 367                 }
 368
 369                 if (syncState->graphsStream)
 370                 {
 371                         writeMessagePoint(matchingData->messagePoints[packet->inE->traceNum][packet->outE->traceNum],
 372                                 packet);
 373                 }
 374
 375                 if (syncState->analysisModule->analyzeMessage != NULL)
 376                 {
 377                         syncState->analysisModule->analyzeMessage(syncState, packet);
 378                 }
 379
 380                 // We can skip the rest of the algorithm if the analysis module is not
 381                 // interested in exchanges
 382                 if (syncState->analysisModule->analyzeExchange == NULL)
 383                 {
 384                         destroyTCPSegment(packet);
 385                         return;
 386                 }
 387
 388                 // If this packet acknowleges some data ...
 389                 if (isAck(packet))
 390                 {
 391                         ConnectionKey oppositeConnectionKey;
 392
 393                         buildReversedConnectionKey(&oppositeConnectionKey,
 394                                 &event->event.tcpEvent->segmentKey->connectionKey);
 395                         conUnAcked= g_hash_table_lookup(matchingData->unAcked,
 396                                 &oppositeConnectionKey);
 397                         if (conUnAcked != NULL)
 398                         {
 399                                 Message* ackedPacket;
 400                                 GList* result;
 401                                 Exchange* exchange;
 402
 403                                 exchange= NULL;
 404
 405                                 result= g_queue_find_custom(conUnAcked, packet, &gcfTCPSegmentAckCompare);
 406
 407                                 while (result != NULL)
 408                                 {
 409                                         // Remove the acknowledged packet from the unAcked list
 410                                         // and keep it for later offset calculations
 411                                         g_debug("Found matching unAcked packet, ");
 412
 413                                         ackedPacket= (Message*) result->data;
 414                                         g_queue_delete_link(conUnAcked, result);
 415
 416                                         if (syncState->stats)
 417                                         {
 418                                                 matchingData->stats->totExchangeEffective++;
 419                                         }
 420
 421                                         if (exchange == NULL)
 422                                         {
 423                                                 exchange= malloc(sizeof(Exchange));
 424                                                 exchange->message= packet;
 425                                                 exchange->acks= g_queue_new();
 426                                         }
 427
 428                                         g_queue_push_tail(exchange->acks, ackedPacket);
 429
 430                                         result= g_queue_find_custom(conUnAcked, packet,
 431                                                 &gcfTCPSegmentAckCompare);
 432                                 }
 433
 434                                 // It might be possible to do an offset calculation
 435                                 if (exchange != NULL)
 436                                 {
 437                                         ackedPacket= g_queue_peek_tail(exchange->acks);
 438                                         if (ackedPacket->outE->traceNum != packet->inE->traceNum
 439                                                 || ackedPacket->inE->traceNum !=
 440                                                 packet->outE->traceNum || packet->inE->traceNum ==
 441                                                 packet->outE->traceNum)
 442                                         {
 443                                                 ackedPacket->print(ackedPacket);
 444                                                 packet->print(packet);
 445                                                 g_error("Disorganized exchange encountered during "
 446                                                         "synchronization");
 447                                         }
 448                                         else
 449                                         {
 450                                                 if (syncState->stats)
 451                                                 {
 452                                                         matchingData->stats->totExchangeSync++;
 453                                                 }
 454
 455                                                 syncState->analysisModule->analyzeExchange(syncState,
 456                                                         exchange);
 457                                         }
 458
 459                                         exchange->message= NULL;
 460                                         destroyTCPExchange(exchange);
 461                                 }
 462                         }
 463                 }
 464
 465                 if (needsAck(packet))
 466                 {
 467                         if (syncState->stats)
 468                         {
 469                                 matchingData->stats->totPacketNeedAck++;
 470                         }
 471
 472                         // If this packet will generate an ack, add it to the unAcked list
 473                         g_debug("Adding to unAcked, ");
 474                         conUnAcked= g_hash_table_lookup(matchingData->unAcked,
 475                                 &event->event.tcpEvent->segmentKey->connectionKey);
 476                         if (conUnAcked == NULL)
 477                         {
 478                                 ConnectionKey* connectionKey;
 479
 480                                 connectionKey= malloc(sizeof(ConnectionKey));
 481                                 memcpy(connectionKey, &event->event.tcpEvent->segmentKey->connectionKey,
 482                                         sizeof(ConnectionKey));
 483                                 g_hash_table_insert(matchingData->unAcked, connectionKey,
 484                                         conUnAcked= g_queue_new());
 485                         }
 486                         g_queue_push_tail(conUnAcked, packet);
 487                 }
 488                 else
 489                 {
 490                         destroyTCPSegment(packet);
 491                 }
 492         }
 493         else
 494         {
 495                 // If there's no corresponding event, add the event to the unmatched
 496                 // list for this type of event
 497                 g_debug("Adding to unmatched event list, ");
 498                 g_hash_table_replace(unMatchedList, event->event.tcpEvent->segmentKey, event);
 499         }
 500 }
 501
 502
 503 /*
 504  * Check if a packet is an acknowledge
 505  *
 506  * Args:
 507  *   packet        TCP Message
 508  *
 509  * Returns:
 510  *   true if it is,
 511  *   false otherwise
 512  */
 513 static bool isAck(const Message* const packet)
 514 {
 515         if (packet->inE->event.tcpEvent->segmentKey->ack == 1)
 516         {
 517                 return true;
 518         }
 519         else
 520         {
 521                 return false;
 522         }
 523 }
 524
 525
 526 /*
 527  * Check if a packet will increment the sequence number, thus needing an
 528  * acknowledge
 529  *
 530  * Args:
 531  *   packet        TCP Message
 532  *
 533  * Returns:
 534  *   true if the packet will need an acknowledge
 535  *   false otherwise
 536  */
 537 static bool needsAck(const Message* const packet)
 538 {
 539         if (packet->inE->event.tcpEvent->segmentKey->syn || packet->inE->event.tcpEvent->segmentKey->fin ||
 540                 packet->inE->event.tcpEvent->segmentKey->tot_len - packet->inE->event.tcpEvent->segmentKey->ihl * 4 -
 541                 packet->inE->event.tcpEvent->segmentKey->doff * 4 > 0)
 542         {
 543                 return true;
 544         }
 545         else
 546         {
 547                 return false;
 548         }
 549 }
 550
 551
 552 /*
 553  * Populate a connection key structure for the opposite direction of a
 554  * connection
 555  *
 556  * Args:
 557  *   reversedConnectionKey the result, must be pre-allocated
 558  *   connectionKey the connection key to reverse
 559  */
 560 static void buildReversedConnectionKey(ConnectionKey* const
 561         reversedConnectionKey, const ConnectionKey* const connectionKey)
 562 {
 563         reversedConnectionKey->saddr= connectionKey->daddr;
 564         reversedConnectionKey->daddr= connectionKey->saddr;
 565         reversedConnectionKey->source= connectionKey->dest;
 566         reversedConnectionKey->dest= connectionKey->source;
 567 }
 568
 569
 570 /*
 571  * Create and open files used to store message points to genereate
 572  * graphs. Allocate and populate array to store file pointers.
 573  *
 574  * Args:
 575  *   syncState:    container for synchronization data
 576  */
 577 static void openGraphDataFiles(SyncState* const syncState)
 578 {
 579         unsigned int i, j;
 580         int retval;
 581         char* cwd;
 582         char name[29];
 583         MatchingDataTCP* matchingData;
 584
 585         matchingData= (MatchingDataTCP*) syncState->matchingData;
 586
 587         cwd= changeToGraphsDir(syncState->graphsDir);
 588
 589         matchingData->messagePoints= malloc(syncState->traceNb * sizeof(FILE**));
 590         for (i= 0; i < syncState->traceNb; i++)
 591         {
 592                 matchingData->messagePoints[i]= malloc(syncState->traceNb *
 593                         sizeof(FILE*));
 594                 for (j= 0; j < syncState->traceNb; j++)
 595                 {
 596                         if (i != j)
 597                         {
 598                                 retval= snprintf(name, sizeof(name),
 599                                         "matching_tcp-%03u_to_%03u.data", j, i);
 600                                 if (retval > sizeof(name) - 1)
 601                                 {
 602                                         name[sizeof(name) - 1]= '\0';
 603                                 }
 604                                 if ((matchingData->messagePoints[i][j]= fopen(name, "w")) ==
 605                                         NULL)
 606                                 {
 607                                         g_error("%s", strerror(errno));
 608                                 }
 609                         }
 610                 }
 611         }
 612
 613         retval= chdir(cwd);
 614         if (retval == -1)
 615         {
 616                 g_error("%s", strerror(errno));
 617         }
 618         free(cwd);
 619 }
 620
 621
 622 /*
 623  * Write a message point to a file used to generate graphs
 624  *
 625  * Args:
 626  *   stream:       FILE*, file pointer where to write the point
 627  *   message:      message for which to write the point
 628  */
 629 static void writeMessagePoint(FILE* stream, const Message* const message)
 630 {
 631         uint64_t x, y;
 632
 633         if (message->inE->traceNum < message->outE->traceNum)
 634         {
 635                 // CA is inE->traceNum
 636                 x= message->inE->cpuTime;
 637                 y= message->outE->cpuTime;
 638         }
 639         else
 640         {
 641                 // CA is outE->traceNum
 642                 x= message->outE->cpuTime;
 643                 y= message->inE->cpuTime;
 644         }
 645
 646         fprintf(stream, "%20" PRIu64 " %20" PRIu64 "\n", x, y);
 647 }
 648
 649
 650 /*
 651  * Close files used to store convex hull points to genereate graphs.
 652  * Deallocate array to store file pointers.
 653  *
 654  * Args:
 655  *   syncState:    container for synchronization data
 656  */
 657 static void closeGraphDataFiles(SyncState* const syncState)
 658 {
 659         unsigned int i, j;
 660         MatchingDataTCP* matchingData;
 661         int retval;
 662
 663         matchingData= (MatchingDataTCP*) syncState->matchingData;
 664
 665         if (matchingData->messagePoints == NULL)
 666         {
 667                 return;
 668         }
 669
 670         for (i= 0; i < syncState->traceNb; i++)
 671         {
 672                 for (j= 0; j < syncState->traceNb; j++)
 673                 {
 674                         if (i != j)
 675                         {
 676                                 retval= fclose(matchingData->messagePoints[i][j]);
 677                                 if (retval != 0)
 678                                 {
 679                                         g_error("%s", strerror(errno));
 680                                 }
 681                         }
 682                 }
 683                 free(matchingData->messagePoints[i]);
 684         }
 685         free(matchingData->messagePoints);
 686
 687         matchingData->messagePoints= NULL;
 688 }
 689
 690
 691 /*
 692  * Write the matching-specific graph lines in the gnuplot script.
 693  *
 694  * Args:
 695  *   syncState:    container for synchronization data
 696  *   i:            first trace number
 697  *   j:            second trace number, garanteed to be larger than i
 698  */
 699 static void writeMatchingGraphsPlotsTCPMessages(SyncState* const syncState,
 700         const unsigned int i, const unsigned int j)
 701 {
 702         fprintf(syncState->graphsStream,
 703                 "\t\"matching_tcp-%1$03d_to_%2$03d.data\" "
 704                         "title \"Sent messages\" with points linetype 4 "
 705                         "linecolor rgb \"#98fc66\" pointtype 9 pointsize 2, \\\n"
 706                 "\t\"matching_tcp-%2$03d_to_%1$03d.data\" "
 707                         "title \"Received messages\" with points linetype 4 "
 708                         "linecolor rgb \"#6699cc\" pointtype 11 pointsize 2, \\\n", i, j);
 709 }