Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Add a tag cabinet to write new kind of cluster. See meta_cluster.xml
[simgrid.git] / src / simdag / sd_dotloader.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 #include <stdbool.h>
12 #include <string.h>
13 #include <libgen.h>
14
15 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(sd_dotparse, sd, "Parsing DOT files");
16
17 #undef CLEANUP
18
19 #ifdef HAVE_CGRAPH_H
20 #include <graphviz/cgraph.h>
21 #elif HAVE_AGRAPH_H
22 #include <graphviz/agraph.h>
23 #define agnxtnode(dot, node)    agnxtnode(node)
24 #define agfstin(dot, node)      agfstin(node)
25 #define agnxtin(dot, edge)      agnxtin(edge)
26 #define agfstout(dot, node)     agfstout(node)
27 #define agnxtout(dot, edge)     agnxtout(edge)
28 #endif
29
30 void dot_add_task(Agnode_t * dag_node);
31 void dot_add_input_dependencies(SD_task_t current_job, Agedge_t * edge);
32 void dot_add_output_dependencies(SD_task_t current_job, Agedge_t * edge);
33 xbt_dynar_t SD_dotload_generic(const char * filename);
34
35 static double dot_parse_double(const char *string)
36 {
37   if (string == NULL)
38     return -1;
39   double value = -1;
40   char *err;
41
42   //ret = sscanf(string, "%lg", &value);
43   errno = 0;
44   value = strtod(string,&err);
45   if(errno)
46   {
47     XBT_WARN("Failed to convert string to double: %s\n",strerror(errno));
48     return -1;
49   }
50   return value;
51 }
52
53
54 static int dot_parse_int(const char *string)
55 {
56   if (string == NULL)
57     return -10;
58   int ret = 0;
59   int value = -1;
60
61   ret = sscanf(string, "%d", &value);
62   if (ret != 1)
63     XBT_WARN("%s is not an integer", string);
64   return value;
65 }
66
67 static xbt_dynar_t result;
68 static xbt_dict_t jobs;
69 static xbt_dict_t files;
70 static xbt_dict_t computers;
71 static SD_task_t root_task, end_task;
72 static Agraph_t *dag_dot;
73 static bool schedule = true;
74
75 static void dump_res()
76 {
77   unsigned int cursor;
78   SD_task_t task;
79   xbt_dynar_foreach(result, cursor, task) {
80     XBT_INFO("Task %d", cursor);
81     SD_task_dump(task);
82   }
83 }
84
85
86 static void dot_task_free(void *task)
87 {
88   SD_task_t t = task;
89   SD_task_destroy(t);
90 }
91
92 static void dot_task_p_free(void *task)
93 {
94   SD_task_t *t = task;
95   SD_task_destroy(*t);
96 }
97
98 static void TRACE_sd_dotloader (SD_task_t task, const char *category)
99 {
100   if (category){
101     if (strlen (category) != 0){
102       TRACE_category (category);
103       SD_task_set_category (task, category);
104     }
105   }
106 }
107
108 /** @brief loads a DOT file describing a DAG
109  * 
110  * See http://www.graphviz.org/doc/info/lang.html
111  * for more details.
112  * To obtain information about transfers and tasks, two attributes are
113  * required : size on task (execution time in Flop) and size on edge
114  * (the amount of data transfer in bit).
115  * if they aren't here, there choose to be equal to zero.
116  */
117 xbt_dynar_t SD_dotload(const char *filename)
118 {
119   SD_dotload_generic(filename);
120   xbt_dynar_t computer = NULL;
121   xbt_dict_cursor_t dict_cursor;
122   char *computer_name;
123   xbt_dict_foreach(computers,dict_cursor,computer_name,computer){
124     xbt_dynar_free(&computer);
125   }
126   xbt_dict_free(&computers);
127   return result;
128 }
129
130 xbt_dynar_t SD_dotload_with_sched(const char *filename){
131   SD_dotload_generic(filename);
132
133   if(schedule == true){
134     xbt_dynar_t computer = NULL;
135     xbt_dict_cursor_t dict_cursor;
136     char *computer_name;
137     const SD_workstation_t *workstations = SD_workstation_get_list ();
138     xbt_dict_foreach(computers,dict_cursor,computer_name,computer){
139       int count_computer = dot_parse_int(computer_name);
140       unsigned int count=0;
141       SD_task_t task;
142       SD_task_t task_previous = NULL;
143       xbt_dynar_foreach(computer,count,task){
144         // add dependency between the previous and the task to avoid
145         // parallel execution
146         if(task != NULL ){
147           if(task_previous != NULL &&
148              !SD_task_dependency_exists(task_previous, task))
149             SD_task_dependency_add(NULL, NULL, task_previous, task);
150           SD_task_schedulel(task, 1, workstations[count_computer]);
151           task_previous = task;
152         }
153       }
154       xbt_dynar_free(&computer);
155     }
156     xbt_dict_free(&computers);
157     if(acyclic_graph_detail(result))
158       return result;
159     else
160       XBT_WARN("There is at least one cycle in the provided task graph");
161   }else{
162     XBT_WARN("The scheduling is ignored");
163   }
164   return NULL;
165 }
166
167 xbt_dynar_t SD_dotload_generic(const char * filename)
168 {
169   xbt_assert(filename, "Unable to use a null file descriptor\n");
170   //dag_dot =  agopen((char*)filename,Agstrictdirected,0);
171   FILE *in_file = fopen(filename, "r");
172   dag_dot = agread(in_file, NIL(Agdisc_t *));
173
174   result = xbt_dynar_new(sizeof(SD_task_t), dot_task_p_free);
175   files = xbt_dict_new_homogeneous(&dot_task_free);
176   jobs = xbt_dict_new_homogeneous(NULL);
177   computers = xbt_dict_new_homogeneous(NULL);
178   root_task = SD_task_create_comp_seq("root", NULL, 0);
179   /* by design the root task is always SCHEDULABLE */
180   __SD_task_set_state(root_task, SD_SCHEDULABLE);
181
182   xbt_dict_set(jobs, "root", root_task, NULL);
183   xbt_dynar_push(result, &root_task);
184   end_task = SD_task_create_comp_seq("end", NULL, 0);
185   xbt_dict_set(jobs, "end", end_task, NULL);
186
187   Agnode_t *dag_node = NULL;
188   for (dag_node = agfstnode(dag_dot); dag_node; dag_node = agnxtnode(dag_dot, dag_node)) {
189     dot_add_task(dag_node);
190   }
191   agclose(dag_dot);
192   xbt_dict_free(&jobs);
193
194   /* And now, post-process the files.
195    * We want a file task per pair of computation tasks exchanging the file. Duplicate on need
196    * Files not produced in the system are said to be produced by root task (top of DAG).
197    * Files not consumed in the system are said to be consumed by end task (bottom of DAG).
198    */
199   xbt_dict_cursor_t cursor;
200   SD_task_t file;
201   char *name;
202   xbt_dict_foreach(files, cursor, name, file) {
203     unsigned int cpt1, cpt2;
204     SD_task_t newfile = NULL;
205     SD_dependency_t depbefore, depafter;
206     if (xbt_dynar_is_empty(file->tasks_before)) {
207       xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
208         SD_task_t newfile =
209             SD_task_create_comm_e2e(file->name, NULL, file->amount);
210         SD_task_dependency_add(NULL, NULL, root_task, newfile);
211         SD_task_dependency_add(NULL, NULL, newfile, depafter->dst);
212         xbt_dynar_push(result, &newfile);
213       }
214     } else if (xbt_dynar_is_empty(file->tasks_after)) {
215       xbt_dynar_foreach(file->tasks_before, cpt2, depbefore) {
216         SD_task_t newfile =
217             SD_task_create_comm_e2e(file->name, NULL, file->amount);
218         SD_task_dependency_add(NULL, NULL, depbefore->src, newfile);
219         SD_task_dependency_add(NULL, NULL, newfile, end_task);
220         xbt_dynar_push(result, &newfile);
221       }
222     } else {
223       xbt_dynar_foreach(file->tasks_before, cpt1, depbefore) {
224         xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
225           if (depbefore->src == depafter->dst) {
226             XBT_WARN
227                 ("File %s is produced and consumed by task %s. This loop dependency will prevent the execution of the task.",
228                  file->name, depbefore->src->name);
229           }
230           newfile =
231               SD_task_create_comm_e2e(file->name, NULL, file->amount);
232           SD_task_dependency_add(NULL, NULL, depbefore->src, newfile);
233           SD_task_dependency_add(NULL, NULL, newfile, depafter->dst);
234           xbt_dynar_push(result, &newfile);
235         }
236       }
237     }
238   }
239
240   /* Push end task last */
241   xbt_dynar_push(result, &end_task);
242
243   /* Free previous copy of the files */
244   xbt_dict_free(&files);
245   fclose(in_file);
246   if (!acyclic_graph_detail(result)) {
247     XBT_ERROR("The DOT described in %s is not a DAG. It contains a cycle.",
248               basename((char*)filename));
249     xbt_dynar_free(&result);
250     /* (result == NULL) here */
251   }
252   return result;
253 }
254
255 /* dot_add_task create a sd_task and all transfers required for this
256  * task. The execution time of the task is given by the attribute size.
257  * The unit of size is the Flop.*/
258 void dot_add_task(Agnode_t * dag_node)
259 {
260   char *name = agnameof(dag_node);
261   SD_task_t current_job;
262   double runtime = dot_parse_double(agget(dag_node, (char *) "size"));
263
264   XBT_DEBUG("See <job id=%s runtime=%s %.0f>", name,
265         agget(dag_node, (char *) "size"), runtime);
266   current_job = xbt_dict_get_or_null(jobs, name);
267   if (current_job == NULL) {
268     current_job =
269         SD_task_create_comp_seq(name, NULL , runtime);
270 #ifdef HAVE_TRACING
271    TRACE_sd_dotloader (current_job, agget (dag_node, (char*)"category"));
272 #endif
273     xbt_dict_set(jobs, name, current_job, NULL);
274     xbt_dynar_push(result, &current_job);
275   }
276   Agedge_t *e;
277   int count = 0;
278
279   for (e = agfstin(dag_dot, dag_node); e; e = agnxtin(dag_dot, e)) {
280     dot_add_input_dependencies(current_job, e);
281     count++;
282   }
283   if (count == 0 && current_job != root_task) {
284     SD_task_dependency_add(NULL, NULL, root_task, current_job);
285   }
286   count = 0;
287   for (e = agfstout(dag_dot, dag_node); e; e = agnxtout(dag_dot, e)) {
288     dot_add_output_dependencies(current_job, e);
289     count++;
290   }
291   if (count == 0 && current_job != end_task) {
292     SD_task_dependency_add(NULL, NULL, current_job, end_task);
293   }
294
295   if(schedule || XBT_LOG_ISENABLED(sd_dotparse, xbt_log_priority_verbose)){
296     /* try to take the information to schedule the task only if all is
297      * right*/
298     // performer is the computer which execute the task
299     unsigned long performer = -1;
300     char * char_performer = agget(dag_node, (char *) "performer");
301     if (char_performer != NULL)
302       performer = (long) dot_parse_int(char_performer);
303
304     // order is giving the task order on one computer
305     unsigned long order = -1;
306     char * char_order = agget(dag_node, (char *) "order");
307     if (char_order != NULL)
308       order = (long) dot_parse_int(char_order);
309     xbt_dynar_t computer = NULL;
310     //XBT_INFO("performer = %d, order=%d",performer,order);
311     if(performer != -1 && order != -1){
312       //necessary parameters are given
313       computer = xbt_dict_get_or_null(computers, char_performer);
314       if(computer == NULL){
315         computer = xbt_dynar_new(sizeof(SD_task_t), NULL);
316         xbt_dict_set(computers, char_performer, computer, NULL);
317       }
318       if(performer < xbt_lib_length(host_lib)){
319         // the  wanted computer is available
320         SD_task_t *task_test = NULL;
321         if(order < computer->used)
322           task_test = xbt_dynar_get_ptr(computer,order);
323         if(task_test != NULL && *task_test != NULL && *task_test != current_job){
324           /*the user gives the same order to several tasks*/
325           schedule = false;
326           XBT_VERB("The task %s starts on the computer %s at the position : %s like the task %s",
327                  (*task_test)->name, char_performer, char_order, current_job->name);
328         }else{
329           //the parameter seems to be ok
330           xbt_dynar_set_as(computer, order, SD_task_t, current_job);
331         }
332       }else{
333         /*the platform has not enough processors to schedule the DAG like
334         *the user wants*/
335         schedule = false;
336         XBT_VERB("The schedule is ignored, there are not enough computers");
337       }
338     }
339     else {
340       //one of necessary parameters are not given
341       schedule = false;
342       XBT_VERB("The schedule is ignored, the task %s is not correctly scheduled", current_job->name);
343     }
344   }
345 }
346
347 /* dot_add_output_dependencies create the dependencies between a task
348  * and a transfers. This is given by the edges in the dot file. 
349  * The amount of data transfers is given by the attribute size on the
350  * edge. */
351 void dot_add_input_dependencies(SD_task_t current_job, Agedge_t * edge)
352 {
353   SD_task_t file = NULL;
354   char *name_tail=agnameof(agtail(edge));
355   char *name_head=agnameof(aghead(edge));
356   char *name = malloc((strlen(name_head)+strlen(name_tail)+6)*sizeof(char));
357   sprintf(name, "%s->%s", name_tail, name_head);
358   double size = dot_parse_double(agget(edge, (char *) "size"));
359   XBT_DEBUG("size : %e, get size : %s", size, agget(edge, (char *) "size"));
360
361   if (size > 0) {
362     file = xbt_dict_get_or_null(files, name);
363     if (file == NULL) {
364       file = SD_task_create_comm_e2e(name, NULL, size);
365 #ifdef HAVE_TRACING
366       TRACE_sd_dotloader (file, agget (edge, (char*)"category"));
367 #endif
368       xbt_dict_set(files, name, file, NULL);
369     } else {
370       if (SD_task_get_amount(file) != size) {
371         XBT_WARN("Ignoring file %s size redefinition from %.0f to %.0f",
372               name, SD_task_get_amount(file), size);
373       }
374     }
375     SD_task_dependency_add(NULL, NULL, file, current_job);
376   } else {
377     file = xbt_dict_get_or_null(jobs, name_tail);
378     if (file != NULL) {
379       SD_task_dependency_add(NULL, NULL, file, current_job);
380     }
381   }
382   free(name);
383 }
384
385 /* dot_add_output_dependencies create the dependencies between a
386  * transfers and a task. This is given by the edges in the dot file.
387  * The amount of data transfers is given by the attribute size on the
388  * edge. */
389 void dot_add_output_dependencies(SD_task_t current_job, Agedge_t * edge)
390 {
391   SD_task_t file;
392   char *name_tail=agnameof(agtail(edge));
393   char *name_head=agnameof(aghead(edge));
394   char *name = malloc((strlen(name_head)+strlen(name_tail)+6)*sizeof(char));
395   sprintf(name, "%s->%s", name_tail, name_head);
396   double size = dot_parse_double(agget(edge, (char *) "size"));
397   XBT_DEBUG("size : %e, get size : %s", size, agget(edge, (char *) "size"));
398
399   if (size > 0) {
400     file = xbt_dict_get_or_null(files, name);
401     if (file == NULL) {
402       file = SD_task_create_comm_e2e(name, NULL, size);
403 #ifdef HAVE_TRACING
404       TRACE_sd_dotloader (file, agget (edge, (char*)"category"));
405 #endif
406       xbt_dict_set(files, name, file, NULL);
407     } else {
408       if (SD_task_get_amount(file) != size) {
409         XBT_WARN("Ignoring file %s size redefinition from %.0f to %.0f",
410               name, SD_task_get_amount(file), size);
411       }
412     }
413     SD_task_dependency_add(NULL, NULL, current_job, file);
414     if (xbt_dynar_length(file->tasks_before) > 1) {
415       XBT_WARN("File %s created at more than one location...", file->name);
416     }
417   } else {
418     file = xbt_dict_get_or_null(jobs, name_head);
419     if (file != NULL) {
420       SD_task_dependency_add(NULL, NULL, current_job, file);
421     }
422   }
423   free(name);
424 }