Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Compile with flag NDEBUG.
[simgrid.git] / src / simdag / sd_daxloader.c
1 /* Copyright (c) 2009, 2010. The SimGrid Team.
2  * All rights reserved.                                                     */
3
4 /* This program is free software; you can redistribute it and/or modify it
5  * under the terms of the license (GNU LGPL) which comes with this package. */
6
7 #include "private.h"
8 #include "simdag/simdag.h"
9 #include "xbt/misc.h"
10 #include "xbt/log.h"
11
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(sd_daxparse, sd, "Parsing DAX files");
13
14 #undef CLEANUP
15 #include "dax_dtd.h"
16 #include "dax_dtd.c"
17
18 bool children_are_marked(SD_task_t task);
19 bool parents_are_marked(SD_task_t task);
20
21 /* Parsing helpers */
22 static void dax_parse_error(char *msg)
23 {
24   fprintf(stderr, "Parse error on line %d: %s\n", dax_lineno, msg);
25   abort();
26 }
27
28 static double dax_parse_double(const char *string)
29 {
30   int ret = 0;
31   double value;
32
33   ret = sscanf(string, "%lg", &value);
34   if (ret != 1)
35     dax_parse_error(bprintf("%s is not a double", string));
36   return value;
37 }
38
39 static int dax_parse_int(const char *string)
40 {
41   int ret = 0;
42   int value;
43
44   ret = sscanf(string, "%d", &value);
45   if (ret != 1)
46     dax_parse_error(bprintf("%s is not an integer", string));
47   return value;
48 }
49
50 /* Ensure that transfer tasks have unique names even though a file is used
51  * several times */
52
53 void uniq_transfer_task_name(SD_task_t task)
54 {
55   SD_task_t child, parent;
56   xbt_dynar_t children, parents;
57   char *new_name;
58
59   children = SD_task_get_children(task);
60   parents = SD_task_get_parents(task);
61
62   xbt_dynar_get_cpy(children, 0, &child);
63   xbt_dynar_get_cpy(parents, 0, &parent);
64
65   new_name = bprintf("%s_%s_%s",
66                      SD_task_get_name(parent),
67                      SD_task_get_name(task), SD_task_get_name(child));
68
69   SD_task_set_name(task, new_name);
70
71   xbt_dynar_free_container(&children);
72   xbt_dynar_free_container(&parents);
73   free(new_name);
74 }
75
76 bool children_are_marked(SD_task_t task){
77   SD_task_t child_task = NULL;
78   bool all_marked = true;
79   SD_dependency_t depafter = NULL;
80   unsigned int count;
81   xbt_dynar_foreach(task->tasks_after,count,depafter){
82     child_task = depafter->dst;
83     //test marked
84     if(child_task->marked == 0) {
85       all_marked = false;
86       break;
87     }
88     child_task = NULL;
89   }
90   return all_marked;
91 }
92
93 bool parents_are_marked(SD_task_t task){
94   SD_task_t parent_task = NULL;
95   bool all_marked = true;
96   SD_dependency_t depbefore = NULL;
97   unsigned int count;
98   xbt_dynar_foreach(task->tasks_before,count,depbefore){
99     parent_task = depbefore->src;
100     //test marked
101     if(parent_task->marked == 0) {
102       all_marked = false;
103       break;
104     }
105     parent_task = NULL;
106   }
107   return all_marked;
108 }
109
110 bool acyclic_graph_detail(xbt_dynar_t dag){
111   unsigned int count=0, count_current=0;
112   bool all_marked = true;
113   SD_task_t task = NULL, parent_task = NULL, child_task = NULL;
114   SD_dependency_t depbefore = NULL, depafter = NULL;
115   xbt_dynar_t next = NULL, current = xbt_dynar_new(sizeof(SD_task_t),NULL);
116
117   xbt_dynar_foreach(dag,count,task){
118     if(task->kind == SD_TASK_COMM_E2E) continue;
119     task->marked = 0;
120     if(xbt_dynar_length(task->tasks_after) == 0){
121       xbt_dynar_push(current, &task);
122     }
123   }
124   task = NULL;
125   count = 0;
126   //test if something has to be done for the next iteration
127   while(xbt_dynar_length(current) != 0){
128     next = xbt_dynar_new(sizeof(SD_task_t),NULL);
129     //test if the current iteration is done
130     count_current=0;
131     xbt_dynar_foreach(current,count_current,task){
132       if (task == NULL) continue;
133       count = 0;
134       //push task in next
135       task->marked = 1;
136       count = 0;
137       xbt_dynar_foreach(task->tasks_before,count,depbefore){
138         parent_task = depbefore->src;
139         if(parent_task->kind == SD_TASK_COMM_E2E){
140           unsigned int j=0;
141           parent_task->marked = 1;
142           SD_task_t parent_task_2 = NULL;
143           xbt_dynar_foreach(parent_task->tasks_before,j,depbefore){
144             parent_task_2 = depbefore->src;
145             if(children_are_marked(parent_task_2))
146               xbt_dynar_push(next, &parent_task_2);
147           }
148         } else{
149           if(children_are_marked(parent_task))
150             xbt_dynar_push(next, &parent_task);
151         }
152         parent_task = NULL;
153       }
154       task = NULL;
155       count = 0;
156     }
157     xbt_dynar_free(&current);
158     current = next;
159     next = NULL;
160   }
161   xbt_dynar_free(&current);
162   current = NULL;
163   all_marked = true;
164   xbt_dynar_foreach(dag,count,task){
165     if(task->kind == SD_TASK_COMM_E2E) continue;
166     //test if all tasks are marked
167     if(task->marked == 0){
168       WARN1("the task %s is not marked",task->name);
169       all_marked = false;
170       break;
171     }
172   }
173   task = NULL;
174   if(!all_marked){
175     VERB0("there is at least one cycle in your task graph");
176
177     current = xbt_dynar_new(sizeof(SD_task_t),NULL);
178     xbt_dynar_foreach(dag,count,task){
179       if(task->kind == SD_TASK_COMM_E2E) continue;
180       if(xbt_dynar_length(task->tasks_before) == 0){
181         xbt_dynar_push(current, &task);
182       }
183     }
184
185     count = 0;
186     task = NULL;
187     xbt_dynar_foreach(dag,count,task){
188       if(task->kind == SD_TASK_COMM_E2E) continue;
189       if(xbt_dynar_length(task->tasks_before) == 0){
190         task->marked = 1;
191         xbt_dynar_push(current, &task);
192       }
193     }
194     task = NULL;
195     count = 0;
196     //test if something has to be done for the next iteration
197     while(xbt_dynar_length(current) != 0){
198       next = xbt_dynar_new(sizeof(SD_task_t),NULL);
199       //test if the current iteration is done
200       count_current=0;
201       xbt_dynar_foreach(current,count_current,task){
202         if (task == NULL) continue;
203         count = 0;
204         //push task in next
205         task->marked = 1;
206         count = 0;
207         xbt_dynar_foreach(task->tasks_after,count,depafter){
208           child_task = depbefore->dst;
209           if(child_task->kind == SD_TASK_COMM_E2E){
210             unsigned int j=0;
211             child_task->marked = 1;
212             SD_task_t child_task_2 = NULL;
213             xbt_dynar_foreach(child_task->tasks_after,j,depafter){
214               child_task_2 = depbefore->dst;
215               if(parents_are_marked(child_task_2))
216                 xbt_dynar_push(next, &child_task_2);
217             }
218           } else{
219             if(parents_are_marked(child_task))
220               xbt_dynar_push(next, &child_task);
221           }
222           child_task = NULL;
223         }
224         task = NULL;
225         count = 0;
226       }
227       xbt_dynar_free(&current);
228       current = next;
229       next = NULL;
230     }
231     xbt_dynar_free(&current);
232     current = NULL;
233     all_marked = true;
234     xbt_dynar_foreach(dag,count,task){
235       if(task->kind == SD_TASK_COMM_E2E) continue;
236       //test if all tasks are marked
237       if(task->marked == 0){
238         WARN1("the task %s is in a cycle",task->name);
239         all_marked = false;
240       }
241     }
242   }
243   return all_marked;
244 }
245
246
247
248 static YY_BUFFER_STATE input_buffer;
249
250 static xbt_dynar_t result;
251 static xbt_dict_t jobs;
252 static xbt_dict_t files;
253 static SD_task_t current_job;
254 static SD_task_t root_task, end_task;
255
256 static void dump_res()
257 {
258   unsigned int cursor;
259   SD_task_t task;
260   xbt_dynar_foreach(result, cursor, task) {
261     INFO1("Task %d", cursor);
262     SD_task_dump(task);
263   }
264 }
265
266 static void dax_task_free(void *task)
267 {
268   SD_task_t t = task;
269   SD_task_destroy(t);
270 }
271
272 /** @brief loads a DAX file describing a DAG
273  * 
274  * See https://confluence.pegasus.isi.edu/display/pegasus/WorkflowGenerator
275  * for more details.
276  */
277 xbt_dynar_t SD_daxload(const char *filename)
278 {
279   xbt_dict_cursor_t cursor;
280   SD_task_t file;
281   char *name;
282   FILE *in_file = fopen(filename, "r");
283   xbt_assert1(in_file, "Unable to open \"%s\"\n", filename);
284   input_buffer = dax__create_buffer(in_file, 10);
285   dax__switch_to_buffer(input_buffer);
286   dax_lineno = 1;
287
288   result = xbt_dynar_new(sizeof(SD_task_t), dax_task_free);
289   files = xbt_dict_new();
290   jobs = xbt_dict_new();
291   root_task = SD_task_create_comp_seq("root", NULL, 0);
292   /* by design the root task is always SCHEDULABLE */
293   __SD_task_set_state(root_task, SD_SCHEDULABLE);
294
295   xbt_dynar_push(result, &root_task);
296   end_task = SD_task_create_comp_seq("end", NULL, 0);
297
298   xbt_assert2(!dax_lex(), "Parse error in %s: %s", filename,
299               dax__parse_err_msg());
300   dax__delete_buffer(input_buffer);
301   fclose(in_file);
302   xbt_dict_free(&jobs);
303
304   /* And now, post-process the files.
305    * We want a file task per pair of computation tasks exchanging the file. Duplicate on need
306    * Files not produced in the system are said to be produced by root task (top of DAG).
307    * Files not consumed in the system are said to be consumed by end task (bottom of DAG).
308    */
309
310   xbt_dict_foreach(files, cursor, name, file) {
311     unsigned int cpt1, cpt2;
312     SD_task_t newfile = NULL;
313     SD_dependency_t depbefore, depafter;
314     if (xbt_dynar_length(file->tasks_before) == 0) {
315       xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
316         SD_task_t newfile =
317             SD_task_create_comm_e2e(file->name, NULL, file->amount);
318         SD_task_dependency_add(NULL, NULL, root_task, newfile);
319         SD_task_dependency_add(NULL, NULL, newfile, depafter->dst);
320 #ifdef HAVE_TRACING
321         if (depafter->src){
322           const char *category = depafter->src->category;
323           if (category){
324             TRACE_category (category);
325             TRACE_sd_set_task_category (newfile, category);
326           }
327         }
328 #endif
329         xbt_dynar_push(result, &newfile);
330       }
331     } else if (xbt_dynar_length(file->tasks_after) == 0) {
332       xbt_dynar_foreach(file->tasks_before, cpt2, depbefore) {
333         SD_task_t newfile =
334             SD_task_create_comm_e2e(file->name, NULL, file->amount);
335         SD_task_dependency_add(NULL, NULL, depbefore->src, newfile);
336         SD_task_dependency_add(NULL, NULL, newfile, end_task);
337 #ifdef HAVE_TRACING
338         if (depbefore->src){
339           const char *category = depbefore->src->category;
340           if (category){
341             TRACE_category (category);
342             TRACE_sd_set_task_category (newfile, category);
343           }
344         }
345 #endif
346         xbt_dynar_push(result, &newfile);
347       }
348     } else {
349       xbt_dynar_foreach(file->tasks_before, cpt1, depbefore) {
350         xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
351           if (depbefore->src == depafter->dst) {
352             WARN2
353                 ("File %s is produced and consumed by task %s. This loop dependency will prevent the execution of the task.",
354                  file->name, depbefore->src->name);
355           }
356           newfile =
357               SD_task_create_comm_e2e(file->name, NULL, file->amount);
358           SD_task_dependency_add(NULL, NULL, depbefore->src, newfile);
359           SD_task_dependency_add(NULL, NULL, newfile, depafter->dst);
360 #ifdef HAVE_TRACING
361           if (depbefore->src){
362             const char *category = depbefore->src->category;
363             if (category){
364               TRACE_category (category);
365               TRACE_sd_set_task_category (newfile, category);
366             }
367           }
368 #endif
369           xbt_dynar_push(result, &newfile);
370         }
371       }
372     }
373   }
374
375   /* Push end task last */
376   xbt_dynar_push(result, &end_task);
377
378   /* Free previous copy of the files */
379   xbt_dict_free(&files);
380   unsigned int cpt;
381   xbt_dynar_foreach(result, cpt, file) {
382     if (SD_task_get_kind(file) == SD_TASK_COMM_E2E) {
383       uniq_transfer_task_name(file);
384     }
385   }
386
387   acyclic_graph_detail(result);
388   return result;
389 }
390
391 void STag_dax__adag(void)
392 {
393   double version;
394   version = dax_parse_double(A_dax__adag_version);
395
396   xbt_assert1((version == 2.1),
397               "Expected version 2.1 in <adag> tag, got %f. Fix the parser or your file",
398               version);
399 }
400
401 void STag_dax__job(void)
402 {
403   double runtime = dax_parse_double(A_dax__job_runtime);
404   char *name = bprintf("%s@%s", A_dax__job_id, A_dax__job_name);
405   runtime *= 4200000000.;       /* Assume that timings were done on a 4.2GFlops machine. I mean, why not? */
406 //  INFO3("See <job id=%s runtime=%s %.0f>",A_dax__job_id,A_dax__job_runtime,runtime);
407   current_job = SD_task_create_comp_seq(name, NULL, runtime);
408 #ifdef HAVE_TRACING
409   char *category = A_dax__job_name;
410   if (category){
411     TRACE_category (category);
412     TRACE_sd_set_task_category(current_job, category);
413   }
414 #endif
415   xbt_dict_set(jobs, A_dax__job_id, current_job, NULL);
416   free(name);
417   xbt_dynar_push(result, &current_job);
418 }
419
420 void STag_dax__uses(void)
421 {
422   SD_task_t file;
423   double size = dax_parse_double(A_dax__uses_size);
424   int is_input = (A_dax__uses_link == A_dax__uses_link_input);
425
426 //  INFO2("See <uses file=%s %s>",A_dax__uses_file,(is_input?"in":"out"));
427   file = xbt_dict_get_or_null(files, A_dax__uses_file);
428   if (file == NULL) {
429     file = SD_task_create_comm_e2e(A_dax__uses_file, NULL, size);
430     xbt_dict_set(files, A_dax__uses_file, file, &dax_task_free);
431   } else {
432     if (SD_task_get_amount(file) != size) {
433       WARN3("Ignoring file %s size redefinition from %.0f to %.0f",
434             A_dax__uses_file, SD_task_get_amount(file), size);
435     }
436   }
437   if (is_input) {
438     SD_task_dependency_add(NULL, NULL, file, current_job);
439   } else {
440     SD_task_dependency_add(NULL, NULL, current_job, file);
441     if (xbt_dynar_length(file->tasks_before) > 1) {
442       WARN1("File %s created at more than one location...", file->name);
443     }
444   }
445 }
446
447 static SD_task_t current_child;
448 void STag_dax__child(void)
449 {
450   current_child = xbt_dict_get_or_null(jobs, A_dax__child_ref);
451   if (current_child == NULL)
452     dax_parse_error(bprintf
453                     ("Asked to add dependencies to the non-existent %s task",
454                      A_dax__child_ref));
455 }
456
457 void ETag_dax__child(void)
458 {
459   current_child = NULL;
460 }
461
462 void STag_dax__parent(void)
463 {
464   SD_task_t parent = xbt_dict_get_or_null(jobs, A_dax__parent_ref);
465   if (parent == NULL)
466     dax_parse_error(bprintf
467                     ("Asked to add a dependency from %s to %s, but %s does not exist",
468                      current_child->name, A_dax__parent_ref,
469                      A_dax__parent_ref));
470   SD_task_dependency_add(NULL, NULL, parent, current_child);
471   DEBUG2("Control-flow dependency from %s to %s", current_child->name,
472          parent->name);
473 }
474
475 void ETag_dax__adag(void)
476 {
477 //  INFO0("See </adag>");
478 }
479
480 void ETag_dax__job(void)
481 {
482   current_job = NULL;
483 //  INFO0("See </job>");
484 }
485
486 void ETag_dax__parent(void)
487 {
488 //  INFO0("See </parent>");
489 }
490
491 void ETag_dax__uses(void)
492 {
493 //  INFO0("See </uses>");
494 }