Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
f1a270e9690504bf7e8a5f2adb23a679e113fb98
[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
19 /* Parsing helpers */
20 static void dax_parse_error(char *msg) {
21   fprintf(stderr, "Parse error on line %d: %s\n", dax_lineno, msg);
22   abort();
23 }
24 static double dax_parse_double(const char *string) {
25   int ret = 0;
26   double value;
27
28   ret = sscanf(string, "%lg", &value);
29   if (ret != 1)
30     dax_parse_error(bprintf("%s is not a double", string));
31   return value;
32 }
33 static int dax_parse_int(const char *string) {
34   int ret = 0;
35   int value;
36
37   ret = sscanf(string, "%d", &value);
38   if (ret != 1)
39     dax_parse_error(bprintf("%s is not an integer", string));
40   return value;
41 }
42
43 static YY_BUFFER_STATE input_buffer;
44
45 static xbt_dynar_t result;
46 static xbt_dict_t files;
47 static SD_task_t current_job;
48 static SD_task_t root_task,end_task;
49
50 static void dump_res() {
51   unsigned int cursor;
52   SD_task_t task;
53   xbt_dynar_foreach(result,cursor,task) {
54     INFO1("Task %d",cursor);
55     SD_task_dump(task);
56   }
57 }
58
59 static void dax_task_free(void*task){
60   SD_task_t t=task;
61   SD_task_destroy(t);
62 }
63
64 /** @brief loads a DAX file describing a DAG
65  * 
66  * See https://confluence.pegasus.isi.edu/display/pegasus/WorkflowGenerator
67  * for more details.
68  */
69 xbt_dynar_t SD_daxload(const char*filename) {
70   FILE* in_file = fopen(filename,"r");
71   xbt_assert1(in_file, "Unable to open \"%s\"\n", filename);
72   input_buffer = dax__create_buffer(in_file, 10);
73   dax__switch_to_buffer(input_buffer);
74   dax_lineno = 1;
75
76   result = xbt_dynar_new(sizeof(SD_task_t),dax_task_free);
77   files=xbt_dict_new();
78   root_task = SD_task_create_comp_seq("root",NULL,0);
79   xbt_dynar_push(result,&root_task);
80   end_task = SD_task_create_comp_seq("end",NULL,0);
81
82   xbt_assert2(!dax_lex(),"Parse error in %s: %s",filename,dax__parse_err_msg());
83   dax__delete_buffer(input_buffer);
84   fclose(in_file);
85
86   /* And now, post-process the files.
87    * We want a file task per pair of computation tasks exchanging the file. Dupplicate on need
88    * Files not produced in the system are said to be produced by root task (top of DAG).
89    * Files not consumed in the system are said to be consumed by end task (bottom of DAG).
90    */
91   xbt_dict_cursor_t cursor;
92   SD_task_t file;
93   char *name;
94   xbt_dict_foreach(files,cursor,name,file) {
95     unsigned int cpt1,cpt2;
96     SD_dependency_t depbefore,depafter;
97     if (xbt_dynar_length(file->tasks_before) == 0) {
98       xbt_dynar_foreach(file->tasks_after,cpt2,depafter) {
99         SD_task_t newfile = SD_task_create_comm_e2e(file->name,NULL,file->amount);
100         SD_task_dependency_add(NULL,NULL,root_task,newfile);
101         SD_task_dependency_add(NULL,NULL,newfile,depafter->dst);
102         xbt_dynar_push(result,&newfile);
103       }
104     } else if (xbt_dynar_length(file->tasks_after) == 0) {
105       xbt_dynar_foreach(file->tasks_before,cpt2,depbefore) {
106         SD_task_t newfile = SD_task_create_comm_e2e(file->name,NULL,file->amount);
107         SD_task_dependency_add(NULL,NULL,depbefore->src,newfile);
108         SD_task_dependency_add(NULL,NULL,newfile,end_task);
109         xbt_dynar_push(result,&newfile);
110       }
111     } else {
112       xbt_dynar_foreach(file->tasks_before,cpt1,depbefore) {
113         xbt_dynar_foreach(file->tasks_after,cpt2,depafter) {
114           SD_task_t newfile = SD_task_create_comm_e2e(file->name,NULL,file->amount);
115           SD_task_dependency_add(NULL,NULL,depbefore->src,newfile);
116           SD_task_dependency_add(NULL,NULL,newfile,depafter->dst);
117           xbt_dynar_push(result,&newfile);
118         }
119       }
120     }
121   }
122
123   /* Push end task last */
124   xbt_dynar_push(result,&end_task);
125
126   /* Free previous copy of the files */
127   xbt_dict_free(&files);
128
129   return result;
130 }
131
132 void STag_dax__adag(void) {
133   double version = dax_parse_double(A_dax__adag_version);
134
135   xbt_assert1((version == 2.1), "Expected version 2.1 in <adag> tag, got %f. Fix the parser or your file",version);
136 }
137 void STag_dax__job(void) {
138   double runtime = dax_parse_double(A_dax__job_runtime);
139   char *name=bprintf("%s@%s",A_dax__job_id,A_dax__job_name);
140   runtime*=4200000000.; /* Assume that timings were done on a 4.2GFlops machine. I mean, why not? */
141 //  INFO3("See <job id=%s runtime=%s %.0f>",A_dax__job_id,A_dax__job_runtime,runtime);
142   current_job = SD_task_create_comp_seq(name,NULL,runtime);
143   free(name);
144   xbt_dynar_push(result,&current_job);
145
146 }
147 void STag_dax__child(void) {
148 //  INFO0("See <child>");
149 }
150 void STag_dax__parent(void) {
151 //  INFO0("See <parent>");
152 }
153 void STag_dax__uses(void) {
154   SD_task_t file;
155   double size = dax_parse_double(A_dax__uses_size);
156   int is_input = (A_dax__uses_link == A_dax__uses_link_input);
157
158 //  INFO2("See <uses file=%s %s>",A_dax__uses_file,(is_input?"in":"out"));
159   file = xbt_dict_get_or_null(files,A_dax__uses_file);
160   if (file==NULL) {
161     file = SD_task_create_comm_e2e(A_dax__uses_file,NULL,size);
162     xbt_dict_set(files,A_dax__uses_file,file,&dax_task_free);
163   } else {
164     if (SD_task_get_amount(file)!=size) {
165       WARN3("Ignoring file %s size redefinition from %.0f to %.0f",
166           A_dax__uses_file,SD_task_get_amount(file),size);
167     }
168   }
169   if (is_input) {
170     SD_task_dependency_add(NULL,NULL,file,current_job);
171   } else {
172     SD_task_dependency_add(NULL,NULL,current_job,file);
173     if (xbt_dynar_length(file->tasks_before)>1) {
174       WARN1("File %s created at more than one location...",file->name);
175     }
176   }
177 }
178 void ETag_dax__adag(void) {
179 //  INFO0("See </adag>");
180 }
181 void ETag_dax__job(void) {
182   current_job = NULL;
183 //  INFO0("See </job>");
184 }
185 void ETag_dax__child(void) {
186 //  INFO0("See </child>");
187 }
188 void ETag_dax__parent(void) {
189 //  INFO0("See </parent>");
190 }
191 void ETag_dax__uses(void) {
192 //  INFO0("See </uses>");
193 }