Add a module to distribute messages to many analysis modules
[lttv.git] / lttv / lttv / sync / sync_chain_lttv.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 <fcntl.h>
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <sys/resource.h>
28 #include <sys/stat.h>
29 #include <sys/types.h>
30 #include <sys/stat.h>
31 #include <unistd.h>
32
33 #include <lttv/module.h>
34 #include <lttv/option.h>
35
36 #include "sync_chain.h"
37
38
39 #ifndef g_info
40 #define g_info(format...) g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, format)
41 #endif
42
43
44 static void init();
45 static void destroy();
46
47 static void gfAppendAnalysisName(gpointer data, gpointer user_data);
48 static void gfAddModuleOption(gpointer data, gpointer user_data);
49 static void gfRemoveModuleOption(gpointer data, gpointer user_data);
50
51 GQueue processingModules= G_QUEUE_INIT;
52 GQueue matchingModules= G_QUEUE_INIT;
53 GQueue analysisModules= G_QUEUE_INIT;
54 GQueue moduleOptions= G_QUEUE_INIT;
55
56 static char* argHelpNone= "none";
57 static ModuleOption optionSync= {
58 .longName= "sync",
59 .hasArg= NO_ARG,
60 {.present= false},
61 .optionHelp= "synchronize the time between the traces",
62 };
63 static char graphsDir[20];
64 static ModuleOption optionSyncStats= {
65 .longName= "sync-stats",
66 .hasArg= NO_ARG,
67 {.present= false},
68 .optionHelp= "print statistics about the time synchronization",
69 };
70 static ModuleOption optionSyncNull= {
71 .longName= "sync-null",
72 .hasArg= NO_ARG,
73 {.present= false},
74 .optionHelp= "read the events but do not perform any processing",
75 };
76 static GString* analysisModulesNames;
77 static ModuleOption optionSyncAnalysis= {
78 .longName= "sync-analysis",
79 .hasArg= REQUIRED_ARG,
80 .optionHelp= "specify the algorithm to use for event analysis",
81 };
82 static ModuleOption optionSyncGraphs= {
83 .longName= "sync-graphs",
84 .hasArg= NO_ARG,
85 {.present= false},
86 .optionHelp= "output gnuplot graph showing synchronization points",
87 };
88 static ModuleOption optionSyncGraphsDir= {
89 .longName= "sync-graphs-dir",
90 .hasArg= REQUIRED_ARG,
91 .optionHelp= "specify the directory where to store the graphs",
92 };
93
94 /*
95 * Module init function
96 *
97 * This function is declared to be the module initialization function. Event
98 * modules are registered with a "constructor (102)" attribute except one in
99 * each class (processing, matching, analysis) which is chosen to be the
100 * default and which is registered with a "constructor (101)" attribute.
101 * Constructors with no priority are called after constructors with
102 * priorities. The result is that the list of event modules is known when this
103 * function is executed.
104 */
105 static void init()
106 {
107 int retval;
108
109 g_debug("\t\t\tXXXX sync init\n");
110
111 g_assert(g_queue_get_length(&analysisModules) > 0);
112 optionSyncAnalysis.arg = ((AnalysisModule*)
113 g_queue_peek_head(&analysisModules))->name;
114 analysisModulesNames= g_string_new("");
115 g_queue_foreach(&analysisModules, &gfAppendAnalysisName,
116 analysisModulesNames);
117 // remove the last ", "
118 g_string_truncate(analysisModulesNames, analysisModulesNames->len - 2);
119 optionSyncAnalysis.argHelp= analysisModulesNames->str;
120
121 retval= snprintf(graphsDir, sizeof(graphsDir), "graphs-%d", getpid());
122 if (retval > sizeof(graphsDir) - 1)
123 {
124 graphsDir[sizeof(graphsDir) - 1]= '\0';
125 }
126 optionSyncGraphsDir.arg= graphsDir;
127 optionSyncGraphsDir.argHelp= graphsDir;
128
129 g_queue_push_head(&moduleOptions, &optionSyncGraphsDir);
130 g_queue_push_head(&moduleOptions, &optionSyncGraphs);
131 g_queue_push_head(&moduleOptions, &optionSyncAnalysis);
132 g_queue_push_head(&moduleOptions, &optionSyncNull);
133 g_queue_push_head(&moduleOptions, &optionSyncStats);
134 g_queue_push_head(&moduleOptions, &optionSync);
135
136 g_queue_foreach(&moduleOptions, &gfAddModuleOption, NULL);
137
138 }
139
140
141 /*
142 * Module unload function
143 */
144 static void destroy()
145 {
146 g_debug("\t\t\tXXXX sync destroy\n");
147
148 g_queue_foreach(&moduleOptions, &gfRemoveModuleOption, NULL);
149 g_string_free(analysisModulesNames, TRUE);
150
151 g_queue_clear(&processingModules);
152 g_queue_clear(&matchingModules);
153 g_queue_clear(&analysisModules);
154 g_queue_clear(&moduleOptions);
155 }
156
157
158 /*
159 * Calculate a traceset's drift and offset values based on network events
160 *
161 * The individual correction factors are written out to each trace.
162 *
163 * Args:
164 * traceSetContext: traceset
165 */
166 void syncTraceset(LttvTracesetContext* const traceSetContext)
167 {
168 SyncState* syncState;
169 struct timeval startTime, endTime;
170 struct rusage startUsage, endUsage;
171 GList* result;
172 FILE* graphsStream;
173 int retval;
174
175 if (!optionSync.present)
176 {
177 g_debug("Not synchronizing traceset because option is disabled");
178 return;
179 }
180
181 if (optionSyncStats.present)
182 {
183 gettimeofday(&startTime, 0);
184 getrusage(RUSAGE_SELF, &startUsage);
185 }
186
187 // Initialize data structures
188 syncState= malloc(sizeof(SyncState));
189 syncState->traceNb= lttv_traceset_number(traceSetContext->ts);
190
191 if (optionSyncStats.present)
192 {
193 syncState->stats= true;
194 }
195 else
196 {
197 syncState->stats= false;
198 }
199
200 if (optionSyncGraphs.present)
201 {
202 syncState->graphs= optionSyncGraphsDir.arg;
203 }
204 else
205 {
206 syncState->graphs= NULL;
207 }
208
209 graphsStream= NULL;
210 if (syncState->graphs)
211 {
212 char* cwd;
213 int graphsFp;
214
215 // Create the graph directory right away in case the module initialization
216 // functions have something to write in it.
217 cwd= changeToGraphDir(syncState->graphs);
218
219 if ((graphsFp= open("graphs.gnu", O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR |
220 S_IWUSR | S_IXUSR | S_IRGRP | S_IWGRP | S_IXGRP | S_IROTH
221 | S_IWOTH | S_IXOTH)) == -1)
222 {
223 g_error(strerror(errno));
224 }
225 if ((graphsStream= fdopen(graphsFp, "w")) == NULL)
226 {
227 g_error(strerror(errno));
228 }
229
230 retval= chdir(cwd);
231 if (retval == -1)
232 {
233 g_error(strerror(errno));
234 }
235 free(cwd);
236 }
237
238 // Identify and initialize modules
239 syncState->processingData= NULL;
240 if (optionSyncNull.present)
241 {
242 result= g_queue_find_custom(&processingModules, "LTTV-null",
243 &gcfCompareProcessing);
244 }
245 else
246 {
247 result= g_queue_find_custom(&processingModules, "LTTV-standard",
248 &gcfCompareProcessing);
249 }
250 g_assert(result != NULL);
251 syncState->processingModule= (ProcessingModule*) result->data;
252
253 syncState->matchingData= NULL;
254 result= g_queue_find_custom(&matchingModules, "TCP", &gcfCompareMatching);
255 g_assert(result != NULL);
256 syncState->matchingModule= (MatchingModule*) result->data;
257
258 syncState->analysisData= NULL;
259 result= g_queue_find_custom(&analysisModules, optionSyncAnalysis.arg,
260 &gcfCompareAnalysis);
261 if (result != NULL)
262 {
263 syncState->analysisModule= (AnalysisModule*) result->data;
264 }
265 else
266 {
267 g_error("Analysis module '%s' not found", optionSyncAnalysis.arg);
268 }
269
270 if (!optionSyncNull.present)
271 {
272 syncState->analysisModule->initAnalysis(syncState);
273 syncState->matchingModule->initMatching(syncState);
274 }
275 syncState->processingModule->initProcessing(syncState, traceSetContext);
276
277 // Process traceset
278 lttv_process_traceset_seek_time(traceSetContext, ltt_time_zero);
279 lttv_process_traceset_middle(traceSetContext, ltt_time_infinite,
280 G_MAXULONG, NULL);
281 lttv_process_traceset_seek_time(traceSetContext, ltt_time_zero);
282
283 syncState->processingModule->finalizeProcessing(syncState);
284
285 // Write graphs file
286 if (graphsStream != NULL)
287 {
288 unsigned int i, j;
289
290 fprintf(graphsStream,
291 "#!/usr/bin/gnuplot\n\n"
292 "set terminal postscript eps color size 8in,6in\n");
293
294 // Cover the upper triangular matrix, i is the reference node.
295 for (i= 0; i < syncState->traceNb; i++)
296 {
297 for (j= i + 1; j < syncState->traceNb; j++)
298 {
299 long pos;
300
301 fprintf(graphsStream,
302 "\nset output \"%03d-%03d.eps\"\n"
303 "plot \\\n", i, j);
304
305 syncState->processingModule->writeProcessingGraphsPlots(graphsStream,
306 syncState, i, j);
307
308 // Remove the ", \\\n" from the last graph plot line
309 fflush(graphsStream);
310 pos= ftell(graphsStream);
311 if (ftruncate(fileno(graphsStream), pos - 4) == -1)
312 {
313 g_error(strerror(errno));
314 }
315 if (fseek(graphsStream, 0, SEEK_END) == -1)
316 {
317 g_error(strerror(errno));
318 }
319
320 fprintf(graphsStream,
321 "\nset output \"%1$03d-%2$03d.eps\"\n"
322 "set key inside right bottom\n"
323 "set title \"\"\n"
324 "set xlabel \"Clock %1$u\"\n"
325 "set xtics nomirror\n"
326 "set ylabel \"Clock %2$u\"\n"
327 "set ytics nomirror\n", i, j);
328
329 syncState->processingModule->writeProcessingGraphsOptions(graphsStream,
330 syncState, i, j);
331
332 fprintf(graphsStream,
333 "replot\n");
334 }
335 }
336
337 if (fclose(graphsStream) != 0)
338 {
339 g_error(strerror(errno));
340 }
341 }
342
343 if (syncState->processingModule->printProcessingStats != NULL)
344 {
345 syncState->processingModule->printProcessingStats(syncState);
346 }
347
348 syncState->processingModule->destroyProcessing(syncState);
349 if (syncState->matchingModule != NULL)
350 {
351 syncState->matchingModule->destroyMatching(syncState);
352 }
353 if (syncState->analysisModule != NULL)
354 {
355 syncState->analysisModule->destroyAnalysis(syncState);
356 }
357
358 free(syncState);
359
360 if (optionSyncStats.present)
361 {
362 gettimeofday(&endTime, 0);
363 retval= getrusage(RUSAGE_SELF, &endUsage);
364
365 timeDiff(&endTime, &startTime);
366 timeDiff(&endUsage.ru_utime, &startUsage.ru_utime);
367 timeDiff(&endUsage.ru_stime, &startUsage.ru_stime);
368
369 printf("Synchronization time:\n");
370 printf("\treal time: %ld.%06ld\n", endTime.tv_sec, endTime.tv_usec);
371 printf("\tuser time: %ld.%06ld\n", endUsage.ru_utime.tv_sec,
372 endUsage.ru_utime.tv_usec);
373 printf("\tsystem time: %ld.%06ld\n", endUsage.ru_stime.tv_sec,
374 endUsage.ru_stime.tv_usec);
375 }
376 }
377
378
379 /*
380 * Calculate the elapsed time between two timeval values
381 *
382 * Args:
383 * end: end time, result is also stored in this structure
384 * start: start time
385 */
386 void timeDiff(struct timeval* const end, const struct timeval* const start)
387 {
388 if (end->tv_usec >= start->tv_usec)
389 {
390 end->tv_sec-= start->tv_sec;
391 end->tv_usec-= start->tv_usec;
392 }
393 else
394 {
395 end->tv_sec= end->tv_sec - start->tv_sec - 1;
396 end->tv_usec= end->tv_usec - start->tv_usec + 1e6;
397 }
398 }
399
400
401 /*
402 * A GCompareFunc for g_slist_find_custom()
403 *
404 * Args:
405 * a: ProcessingModule*, element's data
406 * b: char*, user data to compare against
407 *
408 * Returns:
409 * 0 if the processing module a's name is b
410 */
411 gint gcfCompareProcessing(gconstpointer a, gconstpointer b)
412 {
413 const ProcessingModule* processingModule;
414 const char* name;
415
416 processingModule= (const ProcessingModule*) a;
417 name= (const char*) b;
418
419 return strncmp(processingModule->name, name,
420 strlen(processingModule->name) + 1);
421 }
422
423
424 /*
425 * A GCompareFunc for g_slist_find_custom()
426 *
427 * Args:
428 * a: MatchingModule*, element's data
429 * b: char*, user data to compare against
430 *
431 * Returns:
432 * 0 if the matching module a's name is b
433 */
434 gint gcfCompareMatching(gconstpointer a, gconstpointer b)
435 {
436 const MatchingModule* matchingModule;
437 const char* name;
438
439 matchingModule= (const MatchingModule*) a;
440 name= (const char*) b;
441
442 return strncmp(matchingModule->name, name, strlen(matchingModule->name) +
443 1);
444 }
445
446
447 /*
448 * A GCompareFunc for g_slist_find_custom()
449 *
450 * Args:
451 * a: AnalysisModule*, element's data
452 * b: char*, user data to compare against
453 *
454 * Returns:
455 * 0 if the analysis module a's name is b
456 */
457 gint gcfCompareAnalysis(gconstpointer a, gconstpointer b)
458 {
459 const AnalysisModule* analysisModule;
460 const char* name;
461
462 analysisModule= (const AnalysisModule*) a;
463 name= (const char*) b;
464
465 return strncmp(analysisModule->name, name, strlen(analysisModule->name) +
466 1);
467 }
468
469
470 /*
471 * A GFunc for g_queue_foreach()
472 *
473 * Concatenate analysis module names.
474 *
475 * Args:
476 * data: AnalysisModule*
477 * user_data: GString*, concatenated names
478 */
479 static void gfAppendAnalysisName(gpointer data, gpointer user_data)
480 {
481 g_string_append((GString*) user_data, ((AnalysisModule*) data)->name);
482 g_string_append((GString*) user_data, ", ");
483 }
484
485
486 /*
487 * Change to the directory used to hold graphs. Create it if necessary.
488 *
489 * Args:
490 * graph: name of directory
491 *
492 * Returns:
493 * The current working directory before the execution of the function. The
494 * string must be free'd by the caller.
495 */
496 char* changeToGraphDir(const char* const graphs)
497 {
498 int retval;
499 char* cwd;
500
501 cwd= getcwd(NULL, 0);
502 if (cwd == NULL)
503 {
504 g_error(strerror(errno));
505 }
506 while ((retval= chdir(graphs)) != 0)
507 {
508 if (errno == ENOENT)
509 {
510 retval= mkdir(graphs, S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP |
511 S_IWGRP | S_IXGRP | S_IROTH | S_IWOTH | S_IXOTH);
512 if (retval != 0)
513 {
514 g_error(strerror(errno));
515 }
516 }
517 else
518 {
519 g_error(strerror(errno));
520 }
521 }
522
523 return cwd;
524 }
525
526
527 /*
528 * A GFunc for g_queue_foreach()
529 *
530 * Args:
531 * data: ModuleOption*
532 * user_data: NULL
533 */
534 static void gfAddModuleOption(gpointer data, gpointer user_data)
535 {
536 ModuleOption* option;
537 LttvOptionType conversion[]= {
538 [NO_ARG]= LTTV_OPT_NONE,
539 [REQUIRED_ARG]= LTTV_OPT_STRING,
540 };
541
542 g_assert_cmpuint(sizeof(conversion) / sizeof(*conversion), ==,
543 HAS_ARG_COUNT);
544 option= (ModuleOption*) data;
545 lttv_option_add(option->longName, '\0', option->optionHelp,
546 option->argHelp ? option->argHelp : argHelpNone,
547 conversion[option->hasArg], &option->arg, NULL, NULL);
548 }
549
550
551 /*
552 * A GFunc for g_queue_foreach()
553 *
554 * Args:
555 * data: ModuleOption*
556 * user_data: NULL
557 */
558 static void gfRemoveModuleOption(gpointer data, gpointer user_data)
559 {
560 lttv_option_remove(((ModuleOption*) data)->longName);
561 }
562
563
564 LTTV_MODULE("sync", "Synchronize traces", \
565 "Synchronizes a traceset based on the correspondance of network events", \
566 init, destroy, "option")
This page took 0.041679 seconds and 4 git commands to generate.