lttv/lttv/sync/event_matching_distributor.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 <stdlib.h>
  23 #include <stddef.h>
  24 #include <string.h>
  25
  26 #include "event_analysis.h"
  27 #include "sync_chain.h"
  28
  29 #include "event_matching_distributor.h"
  30
  31
  32 struct InitAggregate
  33 {
  34         SyncState* syncState;
  35         GQueue* matchingModules;
  36 };
  37
  38
  39 struct GraphAggregate
  40 {
  41         /* Offset whithin Matching module of the field* containing the function
  42          * pointer */
  43         size_t offset;
  44         unsigned int i, j;
  45 };
  46
  47
  48 // Functions common to all matching modules
  49 static void initMatchingDistributor(SyncState* const syncState);
  50 static void destroyMatchingDistributor(SyncState* const syncState);
  51
  52 static void matchEventDistributor(SyncState* const syncState, Event* const
  53         event);
  54 static GArray* finalizeMatchingDistributor(SyncState* const syncState);
  55 static void printMatchingStatsDistributor(SyncState* const syncState);
  56 static void writeMatchingTraceTraceForePlotsDistributor(SyncState* const
  57         syncState, const unsigned int i, const unsigned int j);
  58 static void writeMatchingTraceTraceBackPlotsDistributor(SyncState* const
  59         syncState, const unsigned int i, const unsigned int j);
  60 static void writeMatchingTraceTraceOptionsDistributor(SyncState* const
  61         syncState, const unsigned int i, const unsigned int j);
  62 static void writeMatchingTraceTimeForePlotsDistributor(SyncState* const
  63         syncState, const unsigned int i, const unsigned int j);
  64 static void writeMatchingTraceTimeBackPlotsDistributor(SyncState* const
  65         syncState, const unsigned int i, const unsigned int j);
  66 static void writeMatchingTraceTimeOptionsDistributor(SyncState* const
  67         syncState, const unsigned int i, const unsigned int j);
  68
  69 // Functions specific to this module
  70 void gfInitModule(gpointer data, gpointer user_data);
  71 void gfDestroyModule(gpointer data, gpointer user_data);
  72 void gfMatchEvent(gpointer data, gpointer user_data);
  73 void gfFinalize(gpointer data, gpointer user_data);
  74 void gfPrintStats(gpointer data, gpointer user_data);
  75 void gfGraphFunctionCall(gpointer data, gpointer user_data);
  76
  77
  78 static MatchingModule matchingModuleDistributor = {
  79         .name= "distributor",
  80         .canMatch[TCP]= true,
  81         .canMatch[UDP]= true,
  82         .initMatching= &initMatchingDistributor,
  83         .destroyMatching= &destroyMatchingDistributor,
  84         .matchEvent= &matchEventDistributor,
  85         .finalizeMatching= &finalizeMatchingDistributor,
  86         .printMatchingStats= &printMatchingStatsDistributor,
  87         .graphFunctions= {
  88                 .writeTraceTraceForePlots=
  89                         &writeMatchingTraceTraceForePlotsDistributor,
  90                 .writeTraceTraceBackPlots=
  91                         &writeMatchingTraceTraceBackPlotsDistributor,
  92                 .writeTraceTraceOptions= &writeMatchingTraceTraceOptionsDistributor,
  93                 .writeTraceTimeForePlots= &writeMatchingTraceTimeForePlotsDistributor,
  94                 .writeTraceTimeBackPlots= &writeMatchingTraceTimeBackPlotsDistributor,
  95                 .writeTraceTimeOptions= &writeMatchingTraceTimeOptionsDistributor,
  96         },
  97 };
  98
  99
 100 /*
 101  * Matching module registering function
 102  */
 103 void registerMatchingDistributor()
 104 {
 105         g_queue_push_tail(&matchingModules, &matchingModuleDistributor);
 106 }
 107
 108
 109 /*
 110  * Matching init function
 111  *
 112  * This function is called at the beginning of a synchronization run for a set
 113  * of traces.
 114  *
 115  * Build the list and initialize other matching Modules
 116  *
 117  * Args:
 118  *   syncState     container for synchronization data.
 119  */
 120 static void initMatchingDistributor(SyncState* const syncState)
 121 {
 122         MatchingDataDistributor* matchingData;
 123
 124         matchingData= malloc(sizeof(MatchingDataDistributor));
 125         syncState->matchingData= matchingData;
 126
 127         matchingData->distributedModules= g_queue_new();
 128         g_queue_foreach(&matchingModules, &gfInitModule, &(struct InitAggregate)
 129                 {syncState, matchingData->distributedModules});
 130 }
 131
 132
 133 /*
 134  * Matching destroy function
 135  *
 136  * Destroy other modules and free the matching specific data structures
 137  *
 138  * Args:
 139  *   syncState     container for synchronization data.
 140  */
 141 static void destroyMatchingDistributor(SyncState* const syncState)
 142 {
 143         MatchingDataDistributor* matchingData= syncState->matchingData;
 144
 145         g_queue_foreach(matchingData->distributedModules, &gfDestroyModule, NULL);
 146
 147         g_queue_clear(matchingData->distributedModules);
 148         free(syncState->matchingData);
 149         syncState->matchingData= NULL;
 150 }
 151
 152
 153
 154 /*
 155  * Copy event and distribute to matching modules
 156  *
 157  * Args:
 158  *   syncState     container for synchronization data.
 159  *   event         new event to match
 160  */
 161 static void matchEventDistributor(SyncState* const syncState, Event* const event)
 162 {
 163         MatchingDataDistributor* matchingData= syncState->matchingData;
 164
 165         g_queue_foreach(matchingData->distributedModules, &gfMatchEvent, event);
 166         event->destroy(event);
 167 }
 168
 169
 170 /*
 171  * Call the distributed finalization functions and return identity factors
 172  *
 173  * Args:
 174  *   syncState     container for synchronization data.
 175  *
 176  * Returns:
 177  *   Factors[traceNb] identity factors for each trace
 178  */
 179 static GArray* finalizeMatchingDistributor(SyncState* const syncState)
 180 {
 181         GArray* factors;
 182         unsigned int i;
 183         MatchingDataDistributor* matchingData= syncState->matchingData;
 184
 185         g_queue_foreach(matchingData->distributedModules, &gfFinalize, NULL);
 186
 187         factors= g_array_sized_new(FALSE, FALSE, sizeof(Factors),
 188                 syncState->traceNb);
 189         g_array_set_size(factors, syncState->traceNb);
 190         for (i= 0; i < syncState->traceNb; i++)
 191         {
 192                 Factors* e;
 193
 194                 e= &g_array_index(factors, Factors, i);
 195                 e->drift= 1.;
 196                 e->offset= 0.;
 197         }
 198
 199         return factors;
 200 }
 201
 202
 203 /*
 204  * Call the distributed statistics functions (when they exist). Must be called
 205  * after finalizeMatching.
 206  *
 207  * Args:
 208  *   syncState     container for synchronization data.
 209  */
 210 static void printMatchingStatsDistributor(SyncState* const syncState)
 211 {
 212         MatchingDataDistributor* matchingData= syncState->matchingData;
 213
 214         g_queue_foreach(matchingData->distributedModules, &gfPrintStats, NULL);
 215 }
 216
 217
 218 /*
 219  * Call the distributed graph lines functions (when they exist).
 220  *
 221  * Args:
 222  *   syncState:    container for synchronization data
 223  *   i:            first trace number
 224  *   j:            second trace number, garanteed to be larger than i
 225  */
 226 static void writeMatchingTraceTraceForePlotsDistributor(SyncState* const
 227         syncState, const unsigned int i, const unsigned int j)
 228 {
 229         MatchingDataDistributor* matchingData= syncState->matchingData;
 230
 231         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 232                 &(struct GraphAggregate) {offsetof(MatchingModule,
 233                         graphFunctions.writeTraceTraceForePlots), i, j});
 234 }
 235
 236
 237 /*
 238  * Call the distributed graph lines functions (when they exist).
 239  *
 240  * Args:
 241  *   syncState:    container for synchronization data
 242  *   i:            first trace number
 243  *   j:            second trace number, garanteed to be larger than i
 244  */
 245 static void writeMatchingTraceTraceBackPlotsDistributor(SyncState* const
 246         syncState, const unsigned int i, const unsigned int j)
 247 {
 248         MatchingDataDistributor* matchingData= syncState->matchingData;
 249
 250         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 251                 &(struct GraphAggregate) {offsetof(MatchingModule,
 252                         graphFunctions.writeTraceTraceBackPlots), i, j});
 253 }
 254
 255
 256 /*
 257  * Call the distributed graph lines functions (when they exist).
 258  *
 259  * Args:
 260  *   syncState:    container for synchronization data
 261  *   i:            first trace number
 262  *   j:            second trace number, garanteed to be larger than i
 263  */
 264 static void writeMatchingTraceTimeForePlotsDistributor(SyncState* const
 265         syncState, const unsigned int i, const unsigned int j)
 266 {
 267         MatchingDataDistributor* matchingData= syncState->matchingData;
 268
 269         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 270                 &(struct GraphAggregate) {offsetof(MatchingModule,
 271                         graphFunctions.writeTraceTimeForePlots), i, j});
 272 }
 273
 274
 275 /*
 276  * Call the distributed graph lines functions (when they exist).
 277  *
 278  * Args:
 279  *   syncState:    container for synchronization data
 280  *   i:            first trace number
 281  *   j:            second trace number, garanteed to be larger than i
 282  */
 283 static void writeMatchingTraceTimeBackPlotsDistributor(SyncState* const
 284         syncState, const unsigned int i, const unsigned int j)
 285 {
 286         MatchingDataDistributor* matchingData= syncState->matchingData;
 287
 288         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 289                 &(struct GraphAggregate) {offsetof(MatchingModule,
 290                         graphFunctions.writeTraceTimeBackPlots), i, j});
 291 }
 292
 293
 294 /*
 295  * Call the distributed graph options functions (when they exist).
 296  *
 297  * Args:
 298  *   syncState:    container for synchronization data
 299  *   i:            first trace number
 300  *   j:            second trace number, garanteed to be larger than i
 301  */
 302 static void writeMatchingTraceTraceOptionsDistributor(SyncState* const syncState,
 303         const unsigned int i, const unsigned int j)
 304 {
 305         MatchingDataDistributor* matchingData= syncState->matchingData;
 306
 307         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 308                 &(struct GraphAggregate) {offsetof(MatchingModule,
 309                         graphFunctions.writeTraceTraceOptions), i, j});
 310 }
 311
 312
 313 /*
 314  * Call the distributed graph options functions (when they exist).
 315  *
 316  * Args:
 317  *   syncState:    container for synchronization data
 318  *   i:            first trace number
 319  *   j:            second trace number, garanteed to be larger than i
 320  */
 321 static void writeMatchingTraceTimeOptionsDistributor(SyncState* const syncState,
 322         const unsigned int i, const unsigned int j)
 323 {
 324         MatchingDataDistributor* matchingData= syncState->matchingData;
 325
 326         g_queue_foreach(matchingData->distributedModules, &gfGraphFunctionCall,
 327                 &(struct GraphAggregate) {offsetof(MatchingModule,
 328                         graphFunctions.writeTraceTimeOptions), i, j});
 329 }
 330
 331
 332 /*
 333  * A GFunc for g_queue_foreach()
 334  *
 335  * Add and initialize matching module
 336  *
 337  * Args:
 338  *   data          MatchingModule*, module to add
 339  *   user_data     InitAggregate*
 340  */
 341 void gfInitModule(gpointer data, gpointer user_data)
 342 {
 343         SyncState* parallelSS;
 344         struct InitAggregate* aggregate= user_data;
 345         MatchingModule* matchingModule= data;
 346
 347         if (strcmp(matchingModule->name, matchingModuleDistributor.name) == 0)
 348         {
 349                 return;
 350         }
 351
 352         parallelSS= malloc(sizeof(SyncState));
 353         memcpy(parallelSS, aggregate->syncState, sizeof(SyncState));
 354         g_queue_push_tail(aggregate->matchingModules, parallelSS);
 355
 356         parallelSS->matchingModule= matchingModule;
 357         parallelSS->matchingModule->initMatching(parallelSS);
 358 }
 359
 360
 361 /*
 362  * A GFunc for g_queue_foreach()
 363  *
 364  * Destroy and remove matching module
 365  *
 366  * Args:
 367  *   data          SyncState* containing the module to destroy
 368  *   user_data     NULL
 369  */
 370 void gfDestroyModule(gpointer data, gpointer user_data)
 371 {
 372         SyncState* parallelSS= data;
 373
 374         parallelSS->matchingModule->destroyMatching(parallelSS);
 375         free(parallelSS);
 376 }
 377
 378
 379 /*
 380  * A GFunc for g_queue_foreach()
 381  *
 382  * Args:
 383  *   data          SyncState* containing the distributed matching module
 384  *   user_data     Event* original event
 385  */
 386 void gfMatchEvent(gpointer data, gpointer user_data)
 387 {
 388         SyncState* parallelSS= data;
 389         const Event* event= user_data;
 390         Event* newEvent;
 391
 392         if (parallelSS->matchingModule->canMatch[event->type])
 393         {
 394                 event->copy(event, &newEvent);
 395                 parallelSS->matchingModule->matchEvent(parallelSS, newEvent);
 396         }
 397 }
 398
 399
 400 /*
 401  * A GFunc for g_queue_foreach()
 402  *
 403  * Args:
 404  *   data          SyncState* containing the distributed matching module
 405  *   user_data     NULL
 406  */
 407 void gfFinalize(gpointer data, gpointer user_data)
 408 {
 409         GArray* factors;
 410         SyncState* parallelSS= data;
 411
 412         factors= parallelSS->matchingModule->finalizeMatching(parallelSS);
 413         g_array_free(factors, TRUE);
 414 }
 415
 416
 417 /*
 418  * A GFunc for g_queue_foreach()
 419  *
 420  * Args:
 421  *   data          SyncState* containing the distributed matching module
 422  *   user_data     NULL
 423  */
 424 void gfPrintStats(gpointer data, gpointer user_data)
 425 {
 426         SyncState* parallelSS= data;
 427
 428         if (parallelSS->matchingModule->printMatchingStats != NULL)
 429         {
 430                 parallelSS->matchingModule->printMatchingStats(parallelSS);
 431         }
 432 }
 433
 434
 435 /*
 436  * A GFunc for g_queue_foreach()
 437  *
 438  * Call a certain matching function
 439  *
 440  * Args:
 441  *   data          SyncState* containing the distributed matching module
 442  *   user_data     size_t,
 443  */
 444 void gfGraphFunctionCall(gpointer data, gpointer user_data)
 445 {
 446         SyncState* parallelSS= data;
 447         struct GraphAggregate* aggregate= user_data;
 448         typedef void (*GraphFunction)(struct _SyncState*, const unsigned int,
 449                 const unsigned int);
 450         GraphFunction graphFunction= *(GraphFunction*)((void*)
 451                 parallelSS->matchingModule + (size_t) aggregate->offset);
 452
 453         if (graphFunction != NULL)
 454         {
 455                 graphFunction(parallelSS, aggregate->i, aggregate->j);
 456         }
 457 }