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