src/simdag/sd_daxloader.cpp

   1 /* Copyright (c) 2009-2016. 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 "src/simdag/simdag_private.h"
   8 #include "simgrid/simdag.h"
   9 #include "xbt/misc.h"
  10 #include "xbt/log.h"
  11 #include "xbt/str.h"
  12 #include "xbt/file.h" /* xbt_basename() */
  13
  14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(sd_daxparse, sd, "Parsing DAX files");
  15
  16 extern "C" {
  17   #undef CLEANUP
  18   #include "dax_dtd.h"
  19   #define register /* g++ don't like register, so don't say it */
  20   #include "dax_dtd.c"
  21   #undef register
  22 }
  23
  24 /* Ensure that transfer tasks have unique names even though a file is used several times */
  25
  26 void uniq_transfer_task_name(SD_task_t task)
  27 {
  28   SD_task_t child;
  29   SD_task_t parent;
  30
  31   xbt_dynar_t children = SD_task_get_children(task);
  32   xbt_dynar_t parents = SD_task_get_parents(task);
  33
  34   xbt_dynar_get_cpy(children, 0, &child);
  35   xbt_dynar_get_cpy(parents, 0, &parent);
  36
  37   char *new_name = bprintf("%s_%s_%s", SD_task_get_name(parent), SD_task_get_name(task), SD_task_get_name(child));
  38
  39   SD_task_set_name(task, new_name);
  40
  41   xbt_dynar_free_container(&children);
  42   xbt_dynar_free_container(&parents);
  43   free(new_name);
  44 }
  45
  46 static bool children_are_marked(SD_task_t task){
  47   SD_dependency_t depafter = nullptr;
  48   unsigned int count;
  49
  50   xbt_dynar_foreach(task->tasks_after,count,depafter){
  51     if(depafter->dst->marked == 0)
  52       return false;
  53   }
  54   return true;
  55 }
  56
  57 static bool parents_are_marked(SD_task_t task){
  58   SD_dependency_t depbefore = nullptr;
  59   unsigned int count;
  60   xbt_dynar_foreach(task->tasks_before,count,depbefore){
  61     if(depbefore->src->marked == 0)
  62       return false;
  63   }
  64   return true;
  65 }
  66
  67 bool acyclic_graph_detail(xbt_dynar_t dag){
  68   unsigned int count, count_current=0;
  69   bool all_marked = true;
  70   SD_task_t task = nullptr, parent_task = nullptr, child_task = nullptr;
  71   SD_dependency_t depbefore = nullptr, depafter = nullptr;
  72   xbt_dynar_t next = nullptr, current = xbt_dynar_new(sizeof(SD_task_t),nullptr);
  73
  74   xbt_dynar_foreach(dag,count,task){
  75     if(task->kind == SD_TASK_COMM_E2E) continue;
  76     task->marked = 0;
  77     if(xbt_dynar_is_empty(task->tasks_after)){
  78       xbt_dynar_push(current, &task);
  79     }
  80   }
  81   //test if something has to be done for the next iteration
  82   while(!xbt_dynar_is_empty(current)){
  83     next = xbt_dynar_new(sizeof(SD_task_t),nullptr);
  84     //test if the current iteration is done
  85     xbt_dynar_foreach(current,count_current,task){
  86       if (task == nullptr)
  87         continue;
  88       //push task in next
  89       task->marked = 1;
  90       xbt_dynar_foreach(task->tasks_before,count,depbefore){
  91         parent_task = depbefore->src;
  92         if(parent_task->kind == SD_TASK_COMM_E2E){
  93           unsigned int j=0;
  94           parent_task->marked = 1;
  95           SD_task_t parent_task_2 = nullptr;
  96           xbt_dynar_foreach(parent_task->tasks_before,j,depbefore){
  97             parent_task_2 = depbefore->src;
  98             if(children_are_marked(parent_task_2))
  99               xbt_dynar_push(next, &parent_task_2);
 100           }
 101         } else{
 102           if(children_are_marked(parent_task))
 103             xbt_dynar_push(next, &parent_task);
 104         }
 105         parent_task = nullptr;
 106       }
 107     }
 108     xbt_dynar_free(&current);
 109     current = next;
 110     next = nullptr;
 111   }
 112   xbt_dynar_free(&current);
 113   all_marked = true;
 114   xbt_dynar_foreach(dag,count,task){
 115     if(task->kind == SD_TASK_COMM_E2E)
 116       continue;
 117     //test if all tasks are marked
 118     if(task->marked == 0){
 119       XBT_WARN("the task %s is not marked",task->name);
 120       all_marked = false;
 121       break;
 122     }
 123   }
 124   if(!all_marked){
 125     XBT_VERB("there is at least one cycle in your task graph");
 126
 127     current = xbt_dynar_new(sizeof(SD_task_t),nullptr);
 128     xbt_dynar_foreach(dag,count,task){
 129       if(task->kind == SD_TASK_COMM_E2E)
 130         continue;
 131       if(xbt_dynar_is_empty(task->tasks_before)){
 132         xbt_dynar_push(current, &task);
 133       }
 134     }
 135
 136     xbt_dynar_foreach(dag,count,task){
 137       if(task->kind == SD_TASK_COMM_E2E)
 138        continue;
 139       if(xbt_dynar_is_empty(task->tasks_before)){
 140         task->marked = 1;
 141         xbt_dynar_push(current, &task);
 142       }
 143     }
 144     //test if something has to be done for the next iteration
 145     while(!xbt_dynar_is_empty(current)){
 146       next = xbt_dynar_new(sizeof(SD_task_t),nullptr);
 147       //test if the current iteration is done
 148       xbt_dynar_foreach(current,count_current,task){
 149         if (task == nullptr)
 150           continue;
 151         //push task in next
 152         task->marked = 1;
 153         xbt_dynar_foreach(task->tasks_after,count,depafter){
 154           child_task = depbefore->dst;
 155           if(child_task->kind == SD_TASK_COMM_E2E){
 156             unsigned int j=0;
 157             child_task->marked = 1;
 158             SD_task_t child_task_2 = nullptr;
 159             xbt_dynar_foreach(child_task->tasks_after,j,depafter){
 160               child_task_2 = depbefore->dst;
 161               if(parents_are_marked(child_task_2))
 162                 xbt_dynar_push(next, &child_task_2);
 163             }
 164           } else{
 165             if(parents_are_marked(child_task))
 166               xbt_dynar_push(next, &child_task);
 167           }
 168           child_task = nullptr;
 169         }
 170       }
 171       xbt_dynar_free(&current);
 172       current = next;
 173       next = nullptr;
 174     }
 175     xbt_dynar_free(&current);
 176     all_marked = true;
 177     xbt_dynar_foreach(dag,count,task){
 178       if(task->kind == SD_TASK_COMM_E2E)
 179         continue;
 180       //test if all tasks are marked
 181       if(task->marked == 0){
 182         XBT_WARN("the task %s is in a cycle",task->name);
 183         all_marked = false;
 184       }
 185     }
 186   }
 187   return all_marked;
 188 }
 189
 190 static YY_BUFFER_STATE input_buffer;
 191
 192 static xbt_dynar_t result;
 193 static xbt_dict_t jobs;
 194 static xbt_dict_t files;
 195 static SD_task_t current_job;
 196 static SD_task_t root_task, end_task;
 197
 198 static void dax_task_free(void *task)
 199 {
 200   SD_task_destroy(static_cast<SD_task_t>(task));
 201 }
 202
 203 /** @brief loads a DAX file describing a DAG
 204  *
 205  * See https://confluence.pegasus.isi.edu/display/pegasus/WorkflowGenerator for more details.
 206  */
 207 xbt_dynar_t SD_daxload(const char *filename)
 208 {
 209   xbt_dict_cursor_t cursor;
 210   SD_task_t file;
 211   char *name;
 212   FILE *in_file = fopen(filename, "r");
 213   xbt_assert(in_file, "Unable to open \"%s\"\n", filename);
 214   input_buffer = dax__create_buffer(in_file, 10);
 215   dax__switch_to_buffer(input_buffer);
 216   dax_lineno = 1;
 217
 218   result = xbt_dynar_new(sizeof(SD_task_t), dax_task_free);
 219   files = xbt_dict_new_homogeneous(&dax_task_free);
 220   jobs = xbt_dict_new_homogeneous(nullptr);
 221   root_task = SD_task_create_comp_seq("root", nullptr, 0);
 222   /* by design the root task is always SCHEDULABLE */
 223   SD_task_set_state(root_task, SD_SCHEDULABLE);
 224
 225   xbt_dynar_push(result, &root_task);
 226   end_task = SD_task_create_comp_seq("end", nullptr, 0);
 227
 228   int res = dax_lex();
 229   if (res != 0)
 230     xbt_die("Parse error in %s: %s", filename, dax__parse_err_msg());
 231   dax__delete_buffer(input_buffer);
 232   fclose(in_file);
 233   dax_lex_destroy();
 234   xbt_dict_free(&jobs);
 235
 236   /* And now, post-process the files.
 237    * We want a file task per pair of computation tasks exchanging the file. Duplicate on need
 238    * Files not produced in the system are said to be produced by root task (top of DAG).
 239    * Files not consumed in the system are said to be consumed by end task (bottom of DAG).
 240    */
 241
 242   xbt_dict_foreach(files, cursor, name, file) {
 243     unsigned int cpt1;
 244     unsigned int cpt2;
 245     SD_task_t newfile;
 246     SD_dependency_t depbefore;
 247     SD_dependency_t depafter;
 248     if (xbt_dynar_is_empty(file->tasks_before)) {
 249       xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
 250         newfile = SD_task_create_comm_e2e(file->name, nullptr, file->amount);
 251         SD_task_dependency_add(nullptr, nullptr, root_task, newfile);
 252         SD_task_dependency_add(nullptr, nullptr, newfile, depafter->dst);
 253         xbt_dynar_push(result, &newfile);
 254       }
 255     } else if (xbt_dynar_is_empty(file->tasks_after)) {
 256       xbt_dynar_foreach(file->tasks_before, cpt2, depbefore) {
 257         newfile = SD_task_create_comm_e2e(file->name, nullptr, file->amount);
 258         SD_task_dependency_add(nullptr, nullptr, depbefore->src, newfile);
 259         SD_task_dependency_add(nullptr, nullptr, newfile, end_task);
 260         xbt_dynar_push(result, &newfile);
 261       }
 262     } else {
 263       xbt_dynar_foreach(file->tasks_before, cpt1, depbefore) {
 264         xbt_dynar_foreach(file->tasks_after, cpt2, depafter) {
 265           if (depbefore->src == depafter->dst) {
 266             XBT_WARN
 267                 ("File %s is produced and consumed by task %s."
 268                  "This loop dependency will prevent the execution of the task.", file->name, depbefore->src->name);
 269           }
 270           newfile = SD_task_create_comm_e2e(file->name, nullptr, file->amount);
 271           SD_task_dependency_add(nullptr, nullptr, depbefore->src, newfile);
 272           SD_task_dependency_add(nullptr, nullptr, newfile, depafter->dst);
 273           xbt_dynar_push(result, &newfile);
 274         }
 275       }
 276     }
 277   }
 278
 279   /* Push end task last */
 280   xbt_dynar_push(result, &end_task);
 281
 282   /* Free previous copy of the files */
 283   xbt_dict_free(&files);
 284   unsigned int cpt;
 285   xbt_dynar_foreach(result, cpt, file) {
 286     if (SD_task_get_kind(file) == SD_TASK_COMM_E2E) {
 287       uniq_transfer_task_name(file);
 288     } else if (SD_task_get_kind(file) == SD_TASK_COMP_SEQ){
 289       /* If some tasks do not take files as input, connect them to the root
 290        * if they don't produce files, connect them to the end node.
 291        */
 292       if ((file != root_task) && xbt_dynar_is_empty(file->tasks_before))
 293          SD_task_dependency_add(nullptr, nullptr, root_task, file);
 294       if ((file != end_task) && xbt_dynar_is_empty(file->tasks_after))
 295          SD_task_dependency_add(nullptr, nullptr, file, end_task);
 296     } else {
 297        THROW_IMPOSSIBLE;
 298     }
 299   }
 300
 301   if (!acyclic_graph_detail(result)) {
 302     char* base = xbt_basename(filename);
 303     XBT_ERROR("The DAX described in %s is not a DAG. It contains a cycle.", base);
 304     free(base);
 305     xbt_dynar_foreach(result, cpt, file)
 306       SD_task_destroy(file);
 307     xbt_dynar_free_container(&result);
 308     return nullptr;
 309   } else {
 310     return result;
 311   }
 312 }
 313
 314 void STag_dax__adag()
 315 {
 316   XBT_ATTRIB_UNUSED double version;
 317   version = xbt_str_parse_double(A_dax__adag_version, "Parse error: %s is not a double");
 318
 319   xbt_assert(version == 2.1, "Expected version 2.1 in <adag> tag, got %f. Fix the parser or your file", version);
 320 }
 321
 322 void STag_dax__job()
 323 {
 324   double runtime = xbt_str_parse_double(A_dax__job_runtime, "Parse error: %s is not a double");
 325   char *name = bprintf("%s@%s", A_dax__job_id, A_dax__job_name);
 326   runtime *= 4200000000.;       /* Assume that timings were done on a 4.2GFlops machine. I mean, why not? */
 327   XBT_DEBUG("See <job id=%s runtime=%s %.0f>",A_dax__job_id,A_dax__job_runtime,runtime);
 328   current_job = SD_task_create_comp_seq(name, nullptr, runtime);
 329   xbt_dict_set(jobs, A_dax__job_id, current_job, nullptr);
 330   free(name);
 331   xbt_dynar_push(result, &current_job);
 332 }
 333
 334 void STag_dax__uses(void)
 335 {
 336   double size = xbt_str_parse_double(A_dax__uses_size, "Parse error: %s is not a double");
 337   int is_input = (A_dax__uses_link == A_dax__uses_link_input);
 338
 339   XBT_DEBUG("See <uses file=%s %s>",A_dax__uses_file,(is_input?"in":"out"));
 340   SD_task_t file = static_cast<SD_task_t>(xbt_dict_get_or_null(files, A_dax__uses_file));
 341   if (file == nullptr) {
 342     file = SD_task_create_comm_e2e(A_dax__uses_file, nullptr, size);
 343     xbt_dynar_pop(sd_global->initial_task_set,nullptr);
 344     xbt_dict_set(files, A_dax__uses_file, file, nullptr);
 345   } else {
 346     if (SD_task_get_amount(file) != size) {
 347       XBT_WARN("Ignore file %s size redefinition from %.0f to %.0f", A_dax__uses_file, SD_task_get_amount(file), size);
 348     }
 349   }
 350   if (is_input) {
 351     SD_task_dependency_add(nullptr, nullptr, file, current_job);
 352   } else {
 353     SD_task_dependency_add(nullptr, nullptr, current_job, file);
 354     if (xbt_dynar_length(file->tasks_before) > 1) {
 355       XBT_WARN("File %s created at more than one location...", file->name);
 356     }
 357   }
 358 }
 359
 360 static SD_task_t current_child;
 361 void STag_dax__child()
 362 {
 363   current_child = static_cast<SD_task_t>(xbt_dict_get_or_null(jobs, A_dax__child_ref));
 364   xbt_assert(current_child != nullptr,"Parse error on line %d: Asked to add dependencies to the non-existent %s task",
 365              dax_lineno, A_dax__child_ref);
 366 }
 367
 368 void ETag_dax__child()
 369 {
 370   current_child = nullptr;
 371 }
 372
 373 void STag_dax__parent()
 374 {
 375   SD_task_t parent = static_cast<SD_task_t>(xbt_dict_get_or_null(jobs, A_dax__parent_ref));
 376   xbt_assert(parent != nullptr, "Parse error on line %d: Asked to add a dependency from %s to %s, but %s does not exist",
 377              dax_lineno, current_child->name, A_dax__parent_ref, A_dax__parent_ref);
 378   SD_task_dependency_add(nullptr, nullptr, parent, current_child);
 379   XBT_DEBUG("Control-flow dependency from %s to %s", current_child->name, parent->name);
 380 }
 381
 382 void ETag_dax__adag()
 383 {
 384   XBT_DEBUG("See </adag>");
 385 }
 386
 387 void ETag_dax__job(void)
 388 {
 389   current_job = nullptr;
 390   XBT_DEBUG("See </job>");
 391 }
 392
 393 void ETag_dax__parent(void)
 394 {
 395   XBT_DEBUG("See </parent>");
 396 }
 397
 398 void ETag_dax__uses(void)
 399 {
 400   XBT_DEBUG("See </uses>");
 401 }