src/simdag/sd_task.c

   1 #include "private.h"
   2 #include "simdag/simdag.h"
   3 #include "xbt/sysdep.h"
   4 #include "xbt/dynar.h"
   5
   6 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(sd_task, sd,
   7                                 "Logging specific to SimDag (task)");
   8
   9 static void __SD_task_remove_dependencies(SD_task_t task);
  10 static void __SD_task_destroy_scheduling_data(SD_task_t task);
  11
  12 /**
  13  * \brief Creates a new task.
  14  *
  15  * \param name the name of the task (can be \c NULL)
  16  * \param data the user data you want to associate with the task (can be \c NULL)
  17  * \param amount amount of the task
  18  * \return the new task
  19  * \see SD_task_destroy()
  20  */
  21 SD_task_t SD_task_create(const char *name, void *data, double amount)
  22 {
  23
  24   SD_task_t task;
  25   SD_CHECK_INIT_DONE();
  26
  27   task = xbt_new(s_SD_task_t, 1);
  28
  29   /* general information */
  30   task->data = data;            /* user data */
  31   if (name != NULL)
  32     task->name = xbt_strdup(name);
  33   else
  34     task->name = NULL;
  35
  36   task->state_hookup.prev = NULL;
  37   task->state_hookup.next = NULL;
  38   task->state_set = sd_global->not_scheduled_task_set;
  39   task->state = SD_NOT_SCHEDULED;
  40   xbt_swag_insert(task, task->state_set);
  41
  42   task->amount = amount;
  43   task->remains = amount;
  44   task->start_time = -1.0;
  45   task->finish_time = -1.0;
  46   task->surf_action = NULL;
  47   task->watch_points = 0;
  48
  49   /* dependencies */
  50   task->tasks_before = xbt_dynar_new(sizeof(SD_dependency_t), NULL);
  51   task->tasks_after = xbt_dynar_new(sizeof(SD_dependency_t), NULL);
  52
  53   /* scheduling parameters */
  54   task->workstation_nb = 0;
  55   task->workstation_list = NULL;
  56   task->computation_amount = NULL;
  57   task->communication_amount = NULL;
  58   task->rate = 0;
  59
  60   sd_global->task_number++;
  61
  62   return task;
  63 }
  64
  65 /**
  66  * \brief Returns the user data of a task
  67  *
  68  * \param task a task
  69  * \return the user data associated with this task (can be \c NULL)
  70  * \see SD_task_set_data()
  71  */
  72 void *SD_task_get_data(SD_task_t task)
  73 {
  74   SD_CHECK_INIT_DONE();
  75   xbt_assert0(task != NULL, "Invalid parameter");
  76   return task->data;
  77 }
  78
  79 /**
  80  * \brief Sets the user data of a task
  81  *
  82  * The new data can be \c NULL. The old data should have been freed first
  83  * if it was not \c NULL.
  84  *
  85  * \param task a task
  86  * \param data the new data you want to associate with this task
  87  * \see SD_task_get_data()
  88  */
  89 void SD_task_set_data(SD_task_t task, void *data)
  90 {
  91   SD_CHECK_INIT_DONE();
  92   xbt_assert0(task != NULL, "Invalid parameter");
  93   task->data = data;
  94 }
  95
  96 /**
  97  * \brief Returns the state of a task
  98  *
  99  * \param task a task
 100  * \return the current \ref e_SD_task_state_t "state" of this task:
 101  * #SD_NOT_SCHEDULED, #SD_SCHEDULED, #SD_READY, #SD_RUNNING, #SD_DONE or #SD_FAILED
 102  * \see e_SD_task_state_t
 103  */
 104 e_SD_task_state_t SD_task_get_state(SD_task_t task)
 105 {
 106   SD_CHECK_INIT_DONE();
 107   xbt_assert0(task != NULL, "Invalid parameter");
 108   return task->state;
 109 }
 110
 111 /* Changes the state of a task. Updates the swags and the flag sd_global->watch_point_reached.
 112  */
 113 void __SD_task_set_state(SD_task_t task, e_SD_task_state_t new_state)
 114 {
 115   xbt_swag_remove(task, task->state_set);
 116   switch (new_state) {
 117   case SD_NOT_SCHEDULED:
 118     task->state_set = sd_global->not_scheduled_task_set;
 119     break;
 120   case SD_SCHEDULED:
 121     task->state_set = sd_global->scheduled_task_set;
 122     break;
 123   case SD_READY:
 124     task->state_set = sd_global->ready_task_set;
 125     break;
 126   case SD_IN_FIFO:
 127     task->state_set = sd_global->in_fifo_task_set;
 128     break;
 129   case SD_RUNNING:
 130     task->state_set = sd_global->running_task_set;
 131     task->start_time =
 132       surf_workstation_model->action_get_start_time(task->surf_action);
 133     break;
 134   case SD_DONE:
 135     task->state_set = sd_global->done_task_set;
 136     task->finish_time =
 137       surf_workstation_model->action_get_finish_time(task->surf_action);
 138     task->remains = 0;
 139     break;
 140   case SD_FAILED:
 141     task->state_set = sd_global->failed_task_set;
 142     break;
 143   default:
 144     xbt_assert0(0, "Invalid state");
 145   }
 146   xbt_swag_insert(task, task->state_set);
 147   task->state = new_state;
 148
 149   if (task->watch_points & new_state) {
 150     INFO1("Watch point reached with task '%s'!", SD_task_get_name(task));
 151     sd_global->watch_point_reached = 1;
 152     SD_task_unwatch(task, new_state);   /* remove the watch point */
 153   }
 154 }
 155
 156 /**
 157  * \brief Returns the name of a task
 158  *
 159  * \param task a task
 160  * \return the name of this task (can be \c NULL)
 161  */
 162 const char *SD_task_get_name(SD_task_t task)
 163 {
 164   SD_CHECK_INIT_DONE();
 165   xbt_assert0(task != NULL, "Invalid parameter");
 166   return task->name;
 167 }
 168
 169 /**
 170  * \brief Returns the total amount of a task
 171  *
 172  * \param task a task
 173  * \return the total amount of this task
 174  * \see SD_task_get_remaining_amount()
 175  */
 176 double SD_task_get_amount(SD_task_t task)
 177 {
 178   SD_CHECK_INIT_DONE();
 179   xbt_assert0(task != NULL, "Invalid parameter");
 180   return task->amount;
 181 }
 182
 183 /**
 184  * \brief Returns the remaining amount of a task
 185  *
 186  * \param task a task
 187  * \return the remaining amount of this task
 188  * \see SD_task_get_amount()
 189  */
 190 double SD_task_get_remaining_amount(SD_task_t task)
 191 {
 192   SD_CHECK_INIT_DONE();
 193   xbt_assert0(task != NULL, "Invalid parameter");
 194
 195   if (task->surf_action)
 196     return surf_workstation_model->get_remains(task->surf_action);
 197   else
 198     return task->remains;
 199 }
 200
 201 /* temporary function for debbuging */
 202 static void __SD_print_dependencies(SD_task_t task)
 203 {
 204   xbt_dynar_t dynar;
 205   int length;
 206   int i;
 207   SD_dependency_t dependency;
 208
 209   INFO1("The following tasks must be executed before %s:",
 210         SD_task_get_name(task));
 211   dynar = task->tasks_before;
 212   length = xbt_dynar_length(dynar);
 213
 214
 215   for (i = 0; i < length; i++) {
 216     xbt_dynar_get_cpy(dynar, i, &dependency);
 217     INFO1(" %s", SD_task_get_name(dependency->src));
 218   }
 219
 220   INFO1("The following tasks must be executed after %s:",
 221         SD_task_get_name(task));
 222
 223   dynar = task->tasks_after;
 224   length = xbt_dynar_length(dynar);
 225   for (i = 0; i < length; i++) {
 226     xbt_dynar_get_cpy(dynar, i, &dependency);
 227     INFO1(" %s", SD_task_get_name(dependency->dst));
 228   }
 229   INFO0("----------------------------");
 230 }
 231
 232 /* Destroys a dependency between two tasks.
 233  */
 234 static void __SD_task_dependency_destroy(void *dependency)
 235 {
 236   if (((SD_dependency_t) dependency)->name != NULL)
 237     xbt_free(((SD_dependency_t) dependency)->name);
 238   xbt_free(dependency);
 239 }
 240
 241 /**
 242  * \brief Adds a dependency between two tasks
 243  *
 244  * \a dst will depend on \a src, ie \a dst will not start before \a src is finished.
 245  * Their \ref e_SD_task_state_t "state" must be #SD_NOT_SCHEDULED, #SD_SCHEDULED or #SD_READY.
 246  *
 247  * \param name the name of the new dependency (can be \c NULL)
 248  * \param data the user data you want to associate with this dependency (can be \c NULL)
 249  * \param src the task which must be executed first
 250  * \param dst the task you want to make depend on \a src
 251  * \see SD_task_dependency_remove()
 252  */
 253 void SD_task_dependency_add(const char *name, void *data, SD_task_t src,
 254                             SD_task_t dst)
 255 {
 256   xbt_dynar_t dynar;
 257   int length;
 258   int found = 0;
 259   int i;
 260   SD_dependency_t dependency;
 261
 262   SD_CHECK_INIT_DONE();
 263   xbt_assert0(src != NULL && dst != NULL, "Invalid parameter");
 264
 265   dynar = src->tasks_after;
 266   length = xbt_dynar_length(dynar);
 267
 268
 269
 270   if (src == dst)
 271     THROW1(arg_error, 0,
 272            "Cannot add a dependency between task '%s' and itself",
 273            SD_task_get_name(src));
 274
 275   if (!__SD_task_is_not_scheduled(src)
 276       && !__SD_task_is_scheduled_or_ready(src))
 277     THROW1(arg_error, 0,
 278            "Task '%s' must be SD_NOT_SCHEDULED, SD_SCHEDULED or SD_READY",
 279            SD_task_get_name(src));
 280
 281   if (!__SD_task_is_not_scheduled(dst)
 282       && !__SD_task_is_scheduled_or_ready(dst))
 283     THROW1(arg_error, 0,
 284            "Task '%s' must be SD_NOT_SCHEDULED, SD_SCHEDULED or SD_READY",
 285            SD_task_get_name(dst));
 286
 287   DEBUG2("SD_task_dependency_add: src = %s, dst = %s", SD_task_get_name(src),
 288          SD_task_get_name(dst));
 289   for (i = 0; i < length && !found; i++) {
 290     xbt_dynar_get_cpy(dynar, i, &dependency);
 291     found = (dependency->dst == dst);
 292     DEBUG2("Dependency %d: dependency->dst = %s", i,
 293            SD_task_get_name(dependency->dst));
 294   }
 295
 296   if (found)
 297     THROW2(arg_error, 0,
 298            "A dependency already exists between task '%s' and task '%s'",
 299            SD_task_get_name(src), SD_task_get_name(dst));
 300
 301   dependency = xbt_new(s_SD_dependency_t, 1);
 302
 303   if (name != NULL)
 304     dependency->name = xbt_strdup(name);
 305   else
 306     dependency->name = NULL;
 307
 308   dependency->data = data;
 309   dependency->src = src;
 310   dependency->dst = dst;
 311
 312   /* src must be executed before dst */
 313   xbt_dynar_push(src->tasks_after, &dependency);
 314   xbt_dynar_push(dst->tasks_before, &dependency);
 315
 316   /* if the task was ready, then dst->tasks_before is not empty anymore,
 317      so we must go back to state SD_SCHEDULED */
 318   if (__SD_task_is_ready(dst)) {
 319     DEBUG1("SD_task_dependency_add: %s was ready and becomes scheduled!",
 320            SD_task_get_name(dst));
 321     __SD_task_set_state(dst, SD_SCHEDULED);
 322   }
 323
 324   /*  __SD_print_dependencies(src);
 325      __SD_print_dependencies(dst); */
 326 }
 327
 328 /**
 329  * \brief Indacates whether there is a dependency between two tasks.
 330  *
 331  * \param src a task
 332  * \param dst a task depending on \a src
 333  */
 334 int SD_task_dependency_exists(SD_task_t src, SD_task_t dst)
 335 {
 336   xbt_dynar_t dynar;
 337   int length;
 338   int i;
 339   SD_dependency_t dependency;
 340
 341   SD_CHECK_INIT_DONE();
 342   xbt_assert0(src != NULL && dst != NULL, "Invalid parameter");
 343
 344   dynar = src->tasks_after;
 345   length = xbt_dynar_length(dynar);
 346
 347   for (i = 0; i < length; i++) {
 348     xbt_dynar_get_cpy(dynar, i, &dependency);
 349     if (dependency->dst == dst)
 350       return 1;
 351   }
 352   return 0;
 353 }
 354
 355 /**
 356  * \brief Remove a dependency between two tasks
 357  *
 358  * \param src a task
 359  * \param dst a task depending on \a src
 360  * \see SD_task_dependency_add()
 361  */
 362 void SD_task_dependency_remove(SD_task_t src, SD_task_t dst)
 363 {
 364
 365   xbt_dynar_t dynar;
 366   int length;
 367   int found = 0;
 368   int i;
 369   SD_dependency_t dependency;
 370
 371   SD_CHECK_INIT_DONE();
 372   xbt_assert0(src != NULL && dst != NULL, "Invalid parameter");
 373
 374   /* remove the dependency from src->tasks_after */
 375   dynar = src->tasks_after;
 376   length = xbt_dynar_length(dynar);
 377
 378   for (i = 0; i < length && !found; i++) {
 379     xbt_dynar_get_cpy(dynar, i, &dependency);
 380     if (dependency->dst == dst) {
 381       xbt_dynar_remove_at(dynar, i, NULL);
 382       found = 1;
 383     }
 384   }
 385   if (!found)
 386     THROW4(arg_error, 0,
 387            "No dependency found between task '%s' and '%s': task '%s' is not a successor of task '%s'",
 388            SD_task_get_name(src), SD_task_get_name(dst),
 389            SD_task_get_name(dst), SD_task_get_name(src));
 390
 391   /* remove the dependency from dst->tasks_before */
 392   dynar = dst->tasks_before;
 393   length = xbt_dynar_length(dynar);
 394   found = 0;
 395
 396   for (i = 0; i < length && !found; i++) {
 397     xbt_dynar_get_cpy(dynar, i, &dependency);
 398     if (dependency->src == src) {
 399       xbt_dynar_remove_at(dynar, i, NULL);
 400       __SD_task_dependency_destroy(dependency);
 401       found = 1;
 402     }
 403   }
 404   /* should never happen... */
 405   xbt_assert4(found,
 406               "SimDag error: task '%s' is a successor of '%s' but task '%s' is not a predecessor of task '%s'",
 407               SD_task_get_name(dst), SD_task_get_name(src),
 408               SD_task_get_name(src), SD_task_get_name(dst));
 409
 410   /* if the task was scheduled and dst->tasks_before is empty now, we can make it ready */
 411   if (xbt_dynar_length(dst->tasks_before) == 0 && __SD_task_is_scheduled(dst))
 412     __SD_task_set_state(dst, SD_READY);
 413
 414   /*  __SD_print_dependencies(src);
 415      __SD_print_dependencies(dst); */
 416 }
 417
 418 /**
 419  * \brief Returns the user data associated with a dependency between two tasks
 420  *
 421  * \param src a task
 422  * \param dst a task depending on \a src
 423  * \return the user data associated with this dependency (can be \c NULL)
 424  * \see SD_task_dependency_add()
 425  */
 426 void *SD_task_dependency_get_data(SD_task_t src, SD_task_t dst)
 427 {
 428
 429   xbt_dynar_t dynar;
 430   int length;
 431   int found = 0;
 432   int i;
 433   SD_dependency_t dependency;
 434
 435
 436   SD_CHECK_INIT_DONE();
 437   xbt_assert0(src != NULL && dst != NULL, "Invalid parameter");
 438
 439   dynar = src->tasks_after;
 440   length = xbt_dynar_length(dynar);
 441
 442   for (i = 0; i < length && !found; i++) {
 443     xbt_dynar_get_cpy(dynar, i, &dependency);
 444     found = (dependency->dst == dst);
 445   }
 446   if (!found)
 447     THROW2(arg_error, 0, "No dependency found between task '%s' and '%s'",
 448            SD_task_get_name(src), SD_task_get_name(dst));
 449   return dependency->data;
 450 }
 451
 452 /* temporary function for debugging */
 453 static void __SD_print_watch_points(SD_task_t task)
 454 {
 455   static const int state_masks[] =
 456     { SD_SCHEDULED, SD_RUNNING, SD_READY, SD_DONE, SD_FAILED };
 457   static const char *state_names[] =
 458     { "scheduled", "running", "ready", "done", "failed" };
 459   int i;
 460
 461   INFO2("Task '%s' watch points (%x): ", SD_task_get_name(task),
 462         task->watch_points);
 463
 464
 465   for (i = 0; i < 5; i++) {
 466     if (task->watch_points & state_masks[i])
 467       INFO1("%s ", state_names[i]);
 468   }
 469 }
 470
 471 /**
 472  * \brief Adds a watch point to a task
 473  *
 474  * SD_simulate() will stop as soon as the \ref e_SD_task_state_t "state" of this
 475  * task becomes the one given in argument. The
 476  * watch point is then automatically removed.
 477  *
 478  * \param task a task
 479  * \param state the \ref e_SD_task_state_t "state" you want to watch
 480  * (cannot be #SD_NOT_SCHEDULED)
 481  * \see SD_task_unwatch()
 482  */
 483 void SD_task_watch(SD_task_t task, e_SD_task_state_t state)
 484 {
 485   SD_CHECK_INIT_DONE();
 486   xbt_assert0(task != NULL, "Invalid parameter");
 487
 488   if (state & SD_NOT_SCHEDULED)
 489     THROW0(arg_error, 0,
 490            "Cannot add a watch point for state SD_NOT_SCHEDULED");
 491
 492   task->watch_points = task->watch_points | state;
 493   /*  __SD_print_watch_points(task); */
 494 }
 495
 496 /**
 497  * \brief Removes a watch point from a task
 498  *
 499  * \param task a task
 500  * \param state the \ref e_SD_task_state_t "state" you no longer want to watch
 501  * \see SD_task_watch()
 502  */
 503 void SD_task_unwatch(SD_task_t task, e_SD_task_state_t state)
 504 {
 505   SD_CHECK_INIT_DONE();
 506   xbt_assert0(task != NULL, "Invalid parameter");
 507   xbt_assert0(state != SD_NOT_SCHEDULED,
 508               "SimDag error: Cannot have a watch point for state SD_NOT_SCHEDULED");
 509
 510   task->watch_points = task->watch_points & ~state;
 511   /*  __SD_print_watch_points(task); */
 512 }
 513
 514 /**
 515  * \brief Returns an approximative estimation of the execution time of a task.
 516  *
 517  * The estimation is very approximative because the value returned is the time
 518  * the task would take if it was executed now and if it was the only task.
 519  *
 520  * \param task the task to evaluate
 521  * \param workstation_nb number of workstations on which the task would be executed
 522  * \param workstation_list the workstations on which the task would be executed
 523  * \param computation_amount computation amount for each workstation
 524  * \param communication_amount communication amount between each pair of workstations
 525  * \param rate task execution speed rate
 526  * \see SD_schedule()
 527  */
 528 double SD_task_get_execution_time(SD_task_t task,
 529                                   int workstation_nb,
 530                                   const SD_workstation_t * workstation_list,
 531                                   const double *computation_amount,
 532                                   const double *communication_amount,
 533                                   double rate)
 534 {
 535   double time, max_time = 0.0;
 536   int i, j;
 537   SD_CHECK_INIT_DONE();
 538   xbt_assert0(task != NULL && workstation_nb > 0 && workstation_list != NULL
 539               && computation_amount != NULL
 540               && communication_amount != NULL, "Invalid parameter");
 541
 542   /* the task execution time is the maximum execution time of the parallel tasks */
 543
 544   for (i = 0; i < workstation_nb; i++) {
 545     time =
 546       SD_workstation_get_computation_time(workstation_list[i],
 547                                           computation_amount[i]);
 548
 549     for (j = 0; j < workstation_nb; j++) {
 550       time +=
 551         SD_route_get_communication_time(workstation_list[i],
 552                                         workstation_list[j],
 553                                         communication_amount[i *
 554                                                              workstation_nb +
 555                                                              j]);
 556     }
 557
 558     if (time > max_time) {
 559       max_time = time;
 560     }
 561   }
 562   return max_time * SD_task_get_amount(task);
 563 }
 564
 565 /**
 566  * \brief Schedules a task
 567  *
 568  * The task state must be #SD_NOT_SCHEDULED.
 569  * Once scheduled, a task will be executed as soon as possible in SD_simulate(),
 570  * i.e. when its dependencies are satisfied.
 571  *
 572  * \param task the task you want to schedule
 573  * \param workstation_nb number of workstations on which the task will be executed
 574  * \param workstation_list the workstations on which the task will be executed
 575  * \param computation_amount computation amount for each workstation
 576  * \param communication_amount communication amount between each pair of workstations
 577  * \param rate task execution speed rate
 578  * \see SD_task_unschedule()
 579  */
 580 void SD_task_schedule(SD_task_t task, int workstation_nb,
 581                       const SD_workstation_t * workstation_list,
 582                       const double *computation_amount,
 583                       const double *communication_amount, double rate)
 584 {
 585
 586   int communication_nb;
 587
 588   SD_CHECK_INIT_DONE();
 589   xbt_assert0(task != NULL, "Invalid parameter");
 590   xbt_assert0(workstation_nb > 0, "workstation_nb must be positive");
 591
 592   if (!__SD_task_is_not_scheduled(task))
 593     THROW1(arg_error, 0, "Task '%s' has already been scheduled",
 594            SD_task_get_name(task));
 595
 596   task->workstation_nb = workstation_nb;
 597   task->rate = rate;
 598
 599   task->computation_amount = xbt_new(double, workstation_nb);
 600   memcpy(task->computation_amount, computation_amount,
 601          sizeof(double) * workstation_nb);
 602
 603   communication_nb = workstation_nb * workstation_nb;
 604   task->communication_amount = xbt_new(double, communication_nb);
 605   memcpy(task->communication_amount, communication_amount,
 606          sizeof(double) * communication_nb);
 607
 608   task->workstation_list = xbt_new(SD_workstation_t, workstation_nb);
 609   memcpy(task->workstation_list, workstation_list,
 610          sizeof(SD_workstation_t) * workstation_nb);
 611
 612   /* update the task state */
 613   if (xbt_dynar_length(task->tasks_before) == 0)
 614     __SD_task_set_state(task, SD_READY);
 615   else
 616     __SD_task_set_state(task, SD_SCHEDULED);
 617 }
 618
 619 /**
 620  * \brief Unschedules a task
 621  *
 622  * The task state must be #SD_SCHEDULED, #SD_READY, #SD_RUNNING or #SD_FAILED.
 623  * If you call this function, the task state becomes #SD_NOT_SCHEDULED.
 624  * Call SD_task_schedule() to schedule it again.
 625  *
 626  * \param task the task you want to unschedule
 627  * \see SD_task_schedule()
 628  */
 629 void SD_task_unschedule(SD_task_t task)
 630 {
 631   SD_CHECK_INIT_DONE();
 632   xbt_assert0(task != NULL, "Invalid parameter");
 633
 634   if (task->state_set != sd_global->scheduled_task_set &&
 635       task->state_set != sd_global->ready_task_set &&
 636       task->state_set != sd_global->running_task_set &&
 637       task->state_set != sd_global->failed_task_set)
 638     THROW1(arg_error, 0,
 639            "Task %s: the state must be SD_SCHEDULED, SD_READY, SD_RUNNING or SD_FAILED",
 640            SD_task_get_name(task));
 641
 642   if (__SD_task_is_scheduled_or_ready(task))    /* if the task is scheduled or ready */
 643     __SD_task_destroy_scheduling_data(task);
 644
 645   if (__SD_task_is_running(task))       /* the task should become SD_FAILED */
 646     surf_workstation_model->action_cancel(task->surf_action);
 647   else
 648     __SD_task_set_state(task, SD_NOT_SCHEDULED);
 649   task->remains = task->amount;
 650   task->start_time = -1.0;
 651 }
 652
 653 /* Destroys the data memorised by SD_task_schedule. Task state must be SD_SCHEDULED or SD_READY.
 654  */
 655 static void __SD_task_destroy_scheduling_data(SD_task_t task)
 656 {
 657   SD_CHECK_INIT_DONE();
 658   if (!__SD_task_is_scheduled_or_ready(task) && !__SD_task_is_in_fifo(task))
 659     THROW1(arg_error, 0,
 660            "Task '%s' must be SD_SCHEDULED, SD_READY or SD_IN_FIFO",
 661            SD_task_get_name(task));
 662
 663   xbt_free(task->computation_amount);
 664   xbt_free(task->communication_amount);
 665 }
 666
 667 /* Runs a task. This function is directly called by __SD_task_try_to_run if the task
 668  * doesn't have to wait in fifos. Otherwise, it is called by __SD_task_just_done when
 669  * the task gets out of its fifos.
 670  */
 671 void __SD_task_really_run(SD_task_t task)
 672 {
 673
 674   int i;
 675   void **surf_workstations;
 676
 677   SD_CHECK_INIT_DONE();
 678   xbt_assert0(task != NULL, "Invalid parameter");
 679   xbt_assert2(__SD_task_is_ready_or_in_fifo(task),
 680               "Task '%s' is not ready or in a fifo! Task state: %d",
 681               SD_task_get_name(task), SD_task_get_state(task));
 682   xbt_assert1(task->workstation_list != NULL,
 683               "Task '%s': workstation_list is NULL!", SD_task_get_name(task));
 684
 685
 686
 687   DEBUG1("Really running task '%s'", SD_task_get_name(task));
 688
 689   /* set this task as current task for the workstations in sequential mode */
 690   for (i = 0; i < task->workstation_nb; i++) {
 691     if (SD_workstation_get_access_mode(task->workstation_list[i]) ==
 692         SD_WORKSTATION_SEQUENTIAL_ACCESS) {
 693       task->workstation_list[i]->current_task = task;
 694       xbt_assert0(__SD_workstation_is_busy(task->workstation_list[i]),
 695                   "The workstation should be busy now");
 696     }
 697   }
 698
 699   DEBUG1("Task '%s' set as current task for its workstations",
 700          SD_task_get_name(task));
 701
 702   /* start the task */
 703
 704   /* we have to create a Surf workstation array instead of the SimDag workstation array */
 705   surf_workstations = xbt_new(void *, task->workstation_nb);
 706
 707   for (i = 0; i < task->workstation_nb; i++) {
 708     surf_workstations[i] = task->workstation_list[i]->surf_workstation;
 709   }
 710
 711   task->surf_action = NULL;
 712   if ((task->workstation_nb == 1) && (task->communication_amount[0] == 0.0)) {
 713     task->surf_action =
 714       surf_workstation_model->extension.
 715       workstation.execute(surf_workstations[0], task->computation_amount[0]);
 716   } else if ((task->workstation_nb == 1)
 717              && (task->computation_amount[0] == 0.0)) {
 718     task->surf_action =
 719       surf_workstation_model->extension.
 720       workstation.communicate(surf_workstations[0], surf_workstations[0],
 721                               task->communication_amount[0], task->rate);
 722   } else if ((task->workstation_nb == 2)
 723              && (task->computation_amount[0] == 0.0)
 724              && (task->computation_amount[1] == 0.0)) {
 725     int nb = 0;
 726     double value = 0.0;
 727
 728     for (i = 0; i < task->workstation_nb * task->workstation_nb; i++) {
 729       if (task->communication_amount[i] > 0.0) {
 730         nb++;
 731         value = task->communication_amount[i];
 732       }
 733     }
 734     if (nb == 1) {
 735       task->surf_action =
 736         surf_workstation_model->extension.
 737         workstation.communicate(surf_workstations[0], surf_workstations[1],
 738                                 value, task->rate);
 739     }
 740   }
 741   if (!task->surf_action) {
 742     double *computation_amount = xbt_new(double, task->workstation_nb);
 743     double *communication_amount = xbt_new(double, task->workstation_nb *
 744                                            task->workstation_nb);
 745
 746     memcpy(computation_amount, task->computation_amount, sizeof(double) *
 747            task->workstation_nb);
 748     memcpy(communication_amount, task->communication_amount,
 749            sizeof(double) * task->workstation_nb * task->workstation_nb);
 750
 751     task->surf_action =
 752       surf_workstation_model->extension.
 753       workstation.execute_parallel_task(task->workstation_nb,
 754                                         surf_workstations, computation_amount,
 755                                         communication_amount, task->amount,
 756                                         task->rate);
 757   } else {
 758     xbt_free(surf_workstations);
 759   }
 760
 761   surf_workstation_model->action_data_set(task->surf_action, task);
 762
 763   DEBUG1("surf_action = %p", task->surf_action);
 764
 765   __SD_task_destroy_scheduling_data(task);      /* now the scheduling data are not useful anymore */
 766   __SD_task_set_state(task, SD_RUNNING);
 767   xbt_assert2(__SD_task_is_running(task), "Bad state of task '%s': %d",
 768               SD_task_get_name(task), SD_task_get_state(task));
 769
 770 }
 771
 772 /* Tries to run a task. This function is called by SD_simulate() when a scheduled task becomes SD_READY
 773  * (ie when its dependencies are satisfied).
 774  * If one of the workstations where the task is scheduled on is busy (in sequential mode),
 775  * the task doesn't start.
 776  * Returns whether the task has started.
 777  */
 778 int __SD_task_try_to_run(SD_task_t task)
 779 {
 780
 781   int can_start = 1;
 782   int i;
 783   SD_workstation_t workstation;
 784
 785   SD_CHECK_INIT_DONE();
 786   xbt_assert0(task != NULL, "Invalid parameter");
 787   xbt_assert2(__SD_task_is_ready(task),
 788               "Task '%s' is not ready! Task state: %d",
 789               SD_task_get_name(task), SD_task_get_state(task));
 790
 791
 792   for (i = 0; i < task->workstation_nb; i++) {
 793     can_start = !__SD_workstation_is_busy(task->workstation_list[i]);
 794   }
 795
 796   DEBUG2("Task '%s' can start: %d", SD_task_get_name(task), can_start);
 797
 798   if (!can_start) {             /* if the task cannot start and is not in the fifos yet */
 799     for (i = 0; i < task->workstation_nb; i++) {
 800       workstation = task->workstation_list[i];
 801       if (workstation->access_mode == SD_WORKSTATION_SEQUENTIAL_ACCESS) {
 802         DEBUG2("Pushing task '%s' in the fifo of workstation '%s'",
 803                SD_task_get_name(task), SD_workstation_get_name(workstation));
 804         xbt_fifo_push(workstation->task_fifo, task);
 805       }
 806     }
 807     __SD_task_set_state(task, SD_IN_FIFO);
 808     xbt_assert2(__SD_task_is_in_fifo(task), "Bad state of task '%s': %d",
 809                 SD_task_get_name(task), SD_task_get_state(task));
 810     DEBUG1("Task '%s' state is now SD_IN_FIFO", SD_task_get_name(task));
 811   } else {
 812     __SD_task_really_run(task);
 813   }
 814
 815   return can_start;
 816 }
 817
 818 /* This function is called by SD_simulate when a task is done.
 819  * It updates task->state and task->action and executes if necessary the tasks
 820  * which were waiting in fifos for the end of `task'
 821  */
 822 void __SD_task_just_done(SD_task_t task)
 823 {
 824   int i, j;
 825   SD_workstation_t workstation;
 826
 827   SD_task_t candidate;
 828   int candidate_nb = 0;
 829   int candidate_capacity = 8;
 830   SD_task_t *candidates;
 831   int can_start = 1;
 832
 833   SD_CHECK_INIT_DONE();
 834   xbt_assert0(task != NULL, "Invalid parameter");
 835   xbt_assert1(__SD_task_is_running(task),
 836               "The task must be running! Task state: %d",
 837               SD_task_get_state(task));
 838   xbt_assert1(task->workstation_list != NULL,
 839               "Task '%s': workstation_list is NULL!", SD_task_get_name(task));
 840
 841
 842   candidates = xbt_new(SD_task_t, 8);
 843
 844   __SD_task_set_state(task, SD_DONE);
 845   surf_workstation_model->action_unref(task->surf_action);
 846   task->surf_action = NULL;
 847
 848   DEBUG0("Looking for candidates");
 849
 850   /* if the task was executed on sequential workstations,
 851      maybe we can execute the next task of the fifo for each workstation */
 852   for (i = 0; i < task->workstation_nb; i++) {
 853     workstation = task->workstation_list[i];
 854     DEBUG2("Workstation '%s': access_mode = %d",
 855            SD_workstation_get_name(workstation), workstation->access_mode);
 856     if (workstation->access_mode == SD_WORKSTATION_SEQUENTIAL_ACCESS) {
 857       xbt_assert1(workstation->task_fifo != NULL,
 858                   "Workstation '%s' has sequential access but no fifo!",
 859                   SD_workstation_get_name(workstation));
 860       xbt_assert2(workstation->current_task =
 861                   task, "Workstation '%s': current task should be '%s'",
 862                   SD_workstation_get_name(workstation),
 863                   SD_task_get_name(task));
 864
 865       /* the task is over so we can release the workstation */
 866       workstation->current_task = NULL;
 867
 868       DEBUG0("Getting candidate in fifo");
 869       candidate =
 870         xbt_fifo_get_item_content(xbt_fifo_get_first_item
 871                                   (workstation->task_fifo));
 872
 873       if (candidate != NULL) {
 874         DEBUG1("Candidate: '%s'", SD_task_get_name(candidate));
 875         xbt_assert2(__SD_task_is_in_fifo(candidate),
 876                     "Bad state of candidate '%s': %d",
 877                     SD_task_get_name(candidate),
 878                     SD_task_get_state(candidate));
 879       }
 880
 881       DEBUG1("Candidate in fifo: %p", candidate);
 882
 883       /* if there was a task waiting for my place */
 884       if (candidate != NULL) {
 885         /* Unfortunately, we are not sure yet that we can execute the task now,
 886            because the task can be waiting more deeply in some other workstation's fifos...
 887            So we memorize all candidate tasks, and then we will check for each candidate
 888            whether or not all its workstations are available. */
 889
 890         /* realloc if necessary */
 891         if (candidate_nb == candidate_capacity) {
 892           candidate_capacity *= 2;
 893           candidates =
 894             xbt_realloc(candidates, sizeof(SD_task_t) * candidate_capacity);
 895         }
 896
 897         /* register the candidate */
 898         candidates[candidate_nb++] = candidate;
 899         candidate->fifo_checked = 0;
 900       }
 901     }
 902   }
 903
 904   DEBUG1("Candidates found: %d", candidate_nb);
 905
 906   /* now we check every candidate task */
 907   for (i = 0; i < candidate_nb; i++) {
 908     candidate = candidates[i];
 909
 910     if (candidate->fifo_checked) {
 911       continue;                 /* we have already evaluated that task */
 912     }
 913
 914     xbt_assert2(__SD_task_is_in_fifo(candidate),
 915                 "Bad state of candidate '%s': %d",
 916                 SD_task_get_name(candidate), SD_task_get_state(candidate));
 917
 918     for (j = 0; j < candidate->workstation_nb && can_start; j++) {
 919       workstation = candidate->workstation_list[j];
 920
 921       /* I can start on this workstation if the workstation is shared
 922          or if I am the first task in the fifo */
 923       can_start = workstation->access_mode == SD_WORKSTATION_SHARED_ACCESS ||
 924         candidate ==
 925         xbt_fifo_get_item_content(xbt_fifo_get_first_item
 926                                   (workstation->task_fifo));
 927     }
 928
 929     DEBUG2("Candidate '%s' can start: %d", SD_task_get_name(candidate),
 930            can_start);
 931
 932     /* now we are sure that I can start! */
 933     if (can_start) {
 934       for (j = 0; j < candidate->workstation_nb && can_start; j++) {
 935         workstation = candidate->workstation_list[j];
 936
 937         /* update the fifo */
 938         if (workstation->access_mode == SD_WORKSTATION_SEQUENTIAL_ACCESS) {
 939           candidate = xbt_fifo_shift(workstation->task_fifo);   /* the return value is stored just for debugging */
 940           DEBUG1("Head of the fifo: '%s'",
 941                  (candidate != NULL) ? SD_task_get_name(candidate) : "NULL");
 942           xbt_assert0(candidate == candidates[i],
 943                       "Error in __SD_task_just_done: bad first task in the fifo");
 944         }
 945       }                         /* for each workstation */
 946
 947       /* finally execute the task */
 948       DEBUG2("Task '%s' state: %d", SD_task_get_name(candidate),
 949              SD_task_get_state(candidate));
 950       __SD_task_really_run(candidate);
 951
 952       DEBUG4
 953         ("Calling __SD_task_is_running: task '%s', state set: %p, running_task_set: %p, is running: %d",
 954          SD_task_get_name(candidate), candidate->state_set,
 955          sd_global->running_task_set, __SD_task_is_running(candidate));
 956       xbt_assert2(__SD_task_is_running(candidate),
 957                   "Bad state of task '%s': %d", SD_task_get_name(candidate),
 958                   SD_task_get_state(candidate));
 959       DEBUG0("Okay, the task is running.");
 960
 961     }                           /* can start */
 962     candidate->fifo_checked = 1;
 963   }                             /* for each candidate */
 964
 965   xbt_free(candidates);
 966 }
 967
 968 /* Remove all dependencies associated with a task. This function is called when the task is destroyed.
 969  */
 970 static void __SD_task_remove_dependencies(SD_task_t task)
 971 {
 972   /* we must destroy the dependencies carefuly (with SD_dependency_remove)
 973      because each one is stored twice */
 974   SD_dependency_t dependency;
 975   while (xbt_dynar_length(task->tasks_before) > 0) {
 976     xbt_dynar_get_cpy(task->tasks_before, 0, &dependency);
 977     SD_task_dependency_remove(dependency->src, dependency->dst);
 978   }
 979
 980   while (xbt_dynar_length(task->tasks_after) > 0) {
 981     xbt_dynar_get_cpy(task->tasks_after, 0, &dependency);
 982     SD_task_dependency_remove(dependency->src, dependency->dst);
 983   }
 984 }
 985
 986 /**
 987  * \brief Returns the start time of a task
 988  *
 989  * The task state must be SD_RUNNING, SD_DONE or SD_FAILED.
 990  *
 991  * \param task: a task
 992  * \return the start time of this task
 993  */
 994 double SD_task_get_start_time(SD_task_t task)
 995 {
 996   SD_CHECK_INIT_DONE();
 997   xbt_assert0(task != NULL, "Invalid parameter");
 998   if (task->surf_action)
 999     return surf_workstation_model->action_get_start_time(task->surf_action);
1000   else
1001     return task->start_time;
1002 }
1003
1004 /**
1005  * \brief Returns the finish time of a task
1006  *
1007  * The task state must be SD_RUNNING, SD_DONE or SD_FAILED.
1008  * If the state is not completed yet, the returned value is an
1009  * estimation of the task finish time. This value can fluctuate
1010  * until the task is completed.
1011  *
1012  * \param task: a task
1013  * \return the start time of this task
1014  */
1015 double SD_task_get_finish_time(SD_task_t task)
1016 {
1017   SD_CHECK_INIT_DONE();
1018   xbt_assert0(task != NULL, "Invalid parameter");
1019
1020   if (task->surf_action)        /* should never happen as actions are destroyed right after their completion */
1021     return surf_workstation_model->action_get_finish_time(task->surf_action);
1022   else
1023     return task->finish_time;
1024 }
1025
1026 /**
1027  * \brief Destroys a task.
1028  *
1029  * The user data (if any) should have been destroyed first.
1030  *
1031  * \param task the task you want to destroy
1032  * \see SD_task_create()
1033  */
1034 void SD_task_destroy(SD_task_t task)
1035 {
1036   SD_CHECK_INIT_DONE();
1037   xbt_assert0(task != NULL, "Invalid parameter");
1038
1039   DEBUG1("Destroying task %s...", SD_task_get_name(task));
1040
1041   __SD_task_remove_dependencies(task);
1042
1043   /* if the task was scheduled or ready we have to free the scheduling parameters */
1044   if (__SD_task_is_scheduled_or_ready(task))
1045     __SD_task_destroy_scheduling_data(task);
1046
1047   if (task->name != NULL)
1048     xbt_free(task->name);
1049
1050   if (task->surf_action != NULL)
1051     surf_workstation_model->action_unref(task->surf_action);
1052
1053   if (task->workstation_list != NULL)
1054     xbt_free(task->workstation_list);
1055
1056   xbt_dynar_free(&task->tasks_before);
1057   xbt_dynar_free(&task->tasks_after);
1058   xbt_free(task);
1059
1060   sd_global->task_number--;
1061
1062   DEBUG0("Task destroyed.");
1063 }