src/simdag/sd_daxloader.c

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