examples/simdag/scheduling/minmin_test.c

   1 /* simple test to schedule a DAX file with the Min-Min algorithm.           */
   2
   3 /* Copyright (c) 2009-2015. The SimGrid Team.
   4  * All rights reserved.                                                     */
   5
   6 /* This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the license (GNU LGPL) which comes with this package. */
   8
   9 #include <stdlib.h>
  10 #include <stdio.h>
  11 #include "simgrid/simdag.h"
  12 #include "xbt/log.h"
  13 #include "xbt/ex.h"
  14 #include <string.h>
  15
  16 XBT_LOG_NEW_DEFAULT_CATEGORY(test,
  17                              "Logging specific to this SimDag example");
  18
  19 typedef struct _WorkstationAttribute *WorkstationAttribute;
  20 struct _WorkstationAttribute {
  21   /* Earliest time at which a workstation is ready to execute a task */
  22   double available_at;
  23   SD_task_t last_scheduled_task;
  24 };
  25
  26 static void sg_host_allocate_attribute(sg_host_t workstation)
  27 {
  28   void *data;
  29   data = calloc(1, sizeof(struct _WorkstationAttribute));
  30   sg_host_user_set(workstation, data);
  31 }
  32
  33 static void sg_host_free_attribute(sg_host_t workstation)
  34 {
  35   free(sg_host_user(workstation));
  36   sg_host_user_set(workstation, NULL);
  37 }
  38
  39 static double sg_host_get_available_at(sg_host_t workstation)
  40 {
  41   WorkstationAttribute attr =
  42       (WorkstationAttribute) sg_host_user(workstation);
  43   return attr->available_at;
  44 }
  45
  46 static void sg_host_set_available_at(sg_host_t workstation,
  47                                             double time)
  48 {
  49   WorkstationAttribute attr =
  50       (WorkstationAttribute) sg_host_user(workstation);
  51   attr->available_at = time;
  52   sg_host_user_set(workstation, attr);
  53 }
  54
  55 static SD_task_t sg_host_get_last_scheduled_task( sg_host_t workstation){
  56   WorkstationAttribute attr =
  57       (WorkstationAttribute) sg_host_user(workstation);
  58   return attr->last_scheduled_task;
  59 }
  60
  61 static void sg_host_set_last_scheduled_task(sg_host_t workstation,
  62     SD_task_t task){
  63   WorkstationAttribute attr =
  64       (WorkstationAttribute) sg_host_user(workstation);
  65   attr->last_scheduled_task=task;
  66   sg_host_user_set(workstation, attr);
  67 }
  68
  69 static xbt_dynar_t get_ready_tasks(xbt_dynar_t dax)
  70 {
  71   unsigned int i;
  72   xbt_dynar_t ready_tasks;
  73   SD_task_t task;
  74
  75   ready_tasks = xbt_dynar_new(sizeof(SD_task_t), NULL);
  76   xbt_dynar_foreach(dax, i, task) {
  77     if (SD_task_get_kind(task) == SD_TASK_COMP_SEQ &&
  78         SD_task_get_state(task) == SD_SCHEDULABLE) {
  79       xbt_dynar_push(ready_tasks, &task);
  80     }
  81   }
  82   XBT_DEBUG("There are %lu ready tasks", xbt_dynar_length(ready_tasks));
  83
  84   return ready_tasks;
  85 }
  86
  87 static double finish_on_at(SD_task_t task, sg_host_t workstation)
  88 {
  89   volatile double result;
  90   unsigned int i;
  91   double data_available = 0.;
  92   double redist_time = 0;
  93   double last_data_available;
  94   SD_task_t parent, grand_parent;
  95   xbt_dynar_t parents, grand_parents;
  96
  97   sg_host_t *grand_parent_workstation_list;
  98
  99   parents = SD_task_get_parents(task);
 100
 101   if (!xbt_dynar_is_empty(parents)) {
 102     /* compute last_data_available */
 103     last_data_available = -1.0;
 104     xbt_dynar_foreach(parents, i, parent) {
 105
 106       /* normal case */
 107       if (SD_task_get_kind(parent) == SD_TASK_COMM_E2E) {
 108         grand_parents = SD_task_get_parents(parent);
 109
 110         xbt_assert(xbt_dynar_length(grand_parents) <2,
 111                    "Error: transfer %s has 2 parents",
 112                    SD_task_get_name(parent));
 113
 114         xbt_dynar_get_cpy(grand_parents, 0, &grand_parent);
 115
 116         grand_parent_workstation_list =
 117             SD_task_get_workstation_list(grand_parent);
 118         /* Estimate the redistribution time from this parent */
 119         redist_time =
 120             SD_route_get_communication_time(grand_parent_workstation_list
 121                                             [0], workstation,
 122                                             SD_task_get_amount(parent));
 123         data_available =
 124             SD_task_get_finish_time(grand_parent) + redist_time;
 125
 126         xbt_dynar_free_container(&grand_parents);
 127       }
 128
 129       /* no transfer, control dependency */
 130       if (SD_task_get_kind(parent) == SD_TASK_COMP_SEQ) {
 131         data_available = SD_task_get_finish_time(parent);
 132       }
 133
 134       if (last_data_available < data_available)
 135         last_data_available = data_available;
 136
 137     }
 138
 139     xbt_dynar_free_container(&parents);
 140
 141     result = MAX(sg_host_get_available_at(workstation),
 142                last_data_available) +
 143              SD_task_get_amount(task)/sg_host_speed(workstation);
 144   } else {
 145     xbt_dynar_free_container(&parents);
 146
 147     result = sg_host_get_available_at(workstation) +
 148               SD_task_get_amount(task)/sg_host_speed(workstation);
 149   }
 150   return result;
 151 }
 152
 153 static sg_host_t SD_task_get_best_workstation(SD_task_t task)
 154 {
 155   int i;
 156   double EFT, min_EFT = -1.0;
 157   const sg_host_t *workstations = sg_host_list();
 158   int nworkstations = sg_host_count();
 159   sg_host_t best_workstation;
 160
 161   best_workstation = workstations[0];
 162   min_EFT = finish_on_at(task, workstations[0]);
 163
 164   for (i = 1; i < nworkstations; i++) {
 165     EFT = finish_on_at(task, workstations[i]);
 166     XBT_DEBUG("%s finishes on %s at %f",
 167            SD_task_get_name(task),
 168            sg_host_get_name(workstations[i]), EFT);
 169
 170     if (EFT < min_EFT) {
 171       min_EFT = EFT;
 172       best_workstation = workstations[i];
 173     }
 174   }
 175
 176   return best_workstation;
 177 }
 178
 179 static void output_xml(FILE * out, xbt_dynar_t dax)
 180 {
 181   unsigned int i, j, k;
 182   int current_nworkstations;
 183   const int nworkstations = sg_host_count();
 184   const sg_host_t *workstations = sg_host_list();
 185   SD_task_t task;
 186   sg_host_t *list;
 187
 188   fprintf(out, "<?xml version=\"1.0\"?>\n");
 189   fprintf(out, "<grid_schedule>\n");
 190   fprintf(out, "   <grid_info>\n");
 191   fprintf(out, "      <info name=\"nb_clusters\" value=\"1\"/>\n");
 192   fprintf(out, "         <clusters>\n");
 193   fprintf(out,
 194           "            <cluster id=\"1\" hosts=\"%d\" first_host=\"0\"/>\n",
 195           nworkstations);
 196   fprintf(out, "         </clusters>\n");
 197   fprintf(out, "      </grid_info>\n");
 198   fprintf(out, "   <node_infos>\n");
 199
 200   xbt_dynar_foreach(dax, i, task) {
 201     fprintf(out, "      <node_statistics>\n");
 202     fprintf(out, "         <node_property name=\"id\" value=\"%s\"/>\n",
 203             SD_task_get_name(task));
 204     fprintf(out, "         <node_property name=\"type\" value=\"");
 205     if (SD_task_get_kind(task) == SD_TASK_COMP_SEQ)
 206       fprintf(out, "computation\"/>\n");
 207     if (SD_task_get_kind(task) == SD_TASK_COMM_E2E)
 208       fprintf(out, "transfer\"/>\n");
 209
 210     fprintf(out,
 211             "         <node_property name=\"start_time\" value=\"%.3f\"/>\n",
 212             SD_task_get_start_time(task));
 213     fprintf(out,
 214             "         <node_property name=\"end_time\" value=\"%.3f\"/>\n",
 215             SD_task_get_finish_time(task));
 216     fprintf(out, "         <configuration>\n");
 217
 218     current_nworkstations = SD_task_get_workstation_count(task);
 219
 220     fprintf(out,
 221             "            <conf_property name=\"host_nb\" value=\"%d\"/>\n",
 222             current_nworkstations);
 223
 224     fprintf(out, "            <host_lists>\n");
 225     list = SD_task_get_workstation_list(task);
 226     for (j = 0; j < current_nworkstations; j++) {
 227       for (k = 0; k < nworkstations; k++) {
 228         if (!strcmp(sg_host_get_name(workstations[k]),
 229                     sg_host_get_name(list[j]))) {
 230           fprintf(out, "               <hosts start=\"%u\" nb=\"1\"/>\n",
 231                   k);
 232           fprintf(out,
 233                   "            <conf_property name=\"cluster_id\" value=\"0\"/>\n");
 234           break;
 235         }
 236       }
 237     }
 238     fprintf(out, "            </host_lists>\n");
 239     fprintf(out, "         </configuration>\n");
 240     fprintf(out, "      </node_statistics>\n");
 241   }
 242   fprintf(out, "   </node_infos>\n");
 243   fprintf(out, "</grid_schedule>\n");
 244 }
 245
 246 int main(int argc, char **argv)
 247 {
 248   unsigned int cursor;
 249   double finish_time, min_finish_time = -1.0;
 250   SD_task_t task, selected_task = NULL, last_scheduled_task;
 251   xbt_dynar_t ready_tasks;
 252   sg_host_t workstation, selected_workstation = NULL;
 253   int total_nworkstations = 0;
 254   const sg_host_t *workstations = NULL;
 255   xbt_dynar_t dax;
 256   FILE *out = NULL;
 257
 258   /* initialization of SD */
 259   SD_init(&argc, argv);
 260
 261   /* Check our arguments */
 262   xbt_assert(argc > 2, "Usage: %s platform_file dax_file [jedule_file]\n"
 263              "\tExample: %s simulacrum_7_hosts.xml Montage_25.xml Montage_25.jed",
 264              argv[0], argv[0]);
 265
 266   char *last = strrchr(argv[2], '.');
 267   char * tracefilename = bprintf("%.*s.jed",(int) (last == NULL ?
 268                                                    strlen(argv[2]) :
 269                                                    last - argv[2]), argv[2]);
 270   if (argc == 4)
 271     tracefilename = xbt_strdup(argv[3]);
 272
 273   /* creation of the environment */
 274   SD_create_environment(argv[1]);
 275
 276   /*  Allocating the workstation attribute */
 277   total_nworkstations = sg_host_count();
 278   workstations = sg_host_list();
 279
 280   for (cursor = 0; cursor < total_nworkstations; cursor++)
 281     sg_host_allocate_attribute(workstations[cursor]);
 282
 283
 284   /* load the DAX file */
 285   dax = SD_daxload(argv[2]);
 286
 287   xbt_dynar_foreach(dax, cursor, task) {
 288     SD_task_watch(task, SD_DONE);
 289   }
 290
 291   /* Schedule the root first */
 292   xbt_dynar_get_cpy(dax, 0, &task);
 293   workstation = SD_task_get_best_workstation(task);
 294   SD_task_schedulel(task, 1, workstation);
 295
 296   while (!xbt_dynar_is_empty(SD_simulate(-1.0))) {
 297     /* Get the set of ready tasks */
 298     ready_tasks = get_ready_tasks(dax);
 299     if (xbt_dynar_is_empty(ready_tasks)) {
 300       xbt_dynar_free_container(&ready_tasks);
 301       /* there is no ready task, let advance the simulation */
 302       continue;
 303     }
 304     /* For each ready task:
 305      * get the workstation that minimizes the completion time.
 306      * select the task that has the minimum completion time on
 307      * its best workstation.
 308      */
 309     xbt_dynar_foreach(ready_tasks, cursor, task) {
 310       XBT_DEBUG("%s is ready", SD_task_get_name(task));
 311       workstation = SD_task_get_best_workstation(task);
 312       finish_time = finish_on_at(task, workstation);
 313       if (min_finish_time == -1. || finish_time < min_finish_time) {
 314         min_finish_time = finish_time;
 315         selected_task = task;
 316         selected_workstation = workstation;
 317       }
 318     }
 319
 320     XBT_INFO("Schedule %s on %s", SD_task_get_name(selected_task),
 321           sg_host_get_name(selected_workstation));
 322
 323     SD_task_schedulel(selected_task, 1, selected_workstation);
 324
 325     /*
 326      * SimDag allows tasks to be executed concurrently when they can by default.
 327      * Yet schedulers take decisions assuming that tasks wait for resource
 328      * availability to start.
 329      * The solution (well crude hack is to keep track of the last task scheduled
 330      * on a workstation and add a special type of dependency if needed to
 331      * force the sequential execution meant by the scheduler.
 332      * If the last scheduled task is already done, has failed or is a
 333      * predecessor of the current task, no need for a new dependency
 334     */
 335
 336     last_scheduled_task =
 337       sg_host_get_last_scheduled_task(selected_workstation);
 338     if (last_scheduled_task &&
 339   (SD_task_get_state(last_scheduled_task) != SD_DONE) &&
 340   (SD_task_get_state(last_scheduled_task) != SD_FAILED) &&
 341   !SD_task_dependency_exists(
 342      sg_host_get_last_scheduled_task(selected_workstation),
 343      selected_task))
 344       SD_task_dependency_add("resource", NULL,
 345            last_scheduled_task, selected_task);
 346
 347     sg_host_set_last_scheduled_task(selected_workstation, selected_task);
 348
 349     sg_host_set_available_at(selected_workstation, min_finish_time);
 350
 351     xbt_dynar_free_container(&ready_tasks);
 352     /* reset the min_finish_time for the next set of ready tasks */
 353     min_finish_time = -1.;
 354   }
 355
 356   XBT_INFO("Simulation Time: %f", SD_get_clock());
 357
 358
 359
 360
 361   XBT_INFO
 362       ("------------------- Produce the trace file---------------------------");
 363   XBT_INFO("Producing the trace of the run into %s", tracefilename);
 364   out = fopen(tracefilename, "w");
 365   xbt_assert(out, "Cannot write to %s", tracefilename);
 366   free(tracefilename);
 367
 368   output_xml(out, dax);
 369
 370   fclose(out);
 371
 372
 373   xbt_dynar_free_container(&ready_tasks);
 374
 375   xbt_dynar_foreach(dax, cursor, task) {
 376     SD_task_destroy(task);
 377   }
 378   xbt_dynar_free_container(&dax);
 379
 380   for (cursor = 0; cursor < total_nworkstations; cursor++)
 381     sg_host_free_attribute(workstations[cursor]);
 382
 383   /* exit */
 384   SD_exit();
 385   return 0;
 386 }