examples/cpp/dag-scheduling/s4u-dag-scheduling.cpp

   1 /* Copyright (c) 2009-2022. 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 /* simple test to schedule a DAX file with the Min-Min algorithm.           */
   8 #include <math.h>
   9 #include <simgrid/host.h>
  10 #include <simgrid/s4u.hpp>
  11 #include <string.h>
  12
  13 XBT_LOG_NEW_DEFAULT_CATEGORY(dag_scheduling, "Logging specific to this example");
  14
  15 typedef struct _HostAttribute* HostAttribute;
  16 struct _HostAttribute {
  17   /* Earliest time at which a host is ready to execute a task */
  18   double available_at;
  19   simgrid::s4u::Exec* last_scheduled_task;
  20 };
  21
  22 static double sg_host_get_available_at(const simgrid::s4u::Host* host)
  23 {
  24   return static_cast<HostAttribute>(host->get_data())->available_at;
  25 }
  26
  27 static void sg_host_set_available_at(simgrid::s4u::Host* host, double time)
  28 {
  29   auto* attr         = static_cast<HostAttribute>(host->get_data());
  30   attr->available_at = time;
  31   host->set_data(attr);
  32 }
  33
  34 static simgrid::s4u::Exec* sg_host_get_last_scheduled_task(const simgrid::s4u::Host* host)
  35 {
  36   return static_cast<HostAttribute>(host->get_data())->last_scheduled_task;
  37 }
  38
  39 static void sg_host_set_last_scheduled_task(simgrid::s4u::Host* host, simgrid::s4u::ExecPtr task)
  40 {
  41   auto* attr                = static_cast<HostAttribute>(host->get_data());
  42   attr->last_scheduled_task = task.get();
  43   host->set_data(attr);
  44 }
  45
  46 static bool dependency_exists(const simgrid::s4u::Exec* src, simgrid::s4u::Exec* dst)
  47 {
  48   const auto& dependencies = src->get_dependencies();
  49   const auto& successors   = src->get_successors();
  50   return (std::find(successors.begin(), successors.end(), dst) != successors.end() ||
  51           dependencies.find(dst) != dependencies.end());
  52 }
  53
  54 static std::vector<simgrid::s4u::Exec*> get_ready_tasks(const std::vector<simgrid::s4u::ActivityPtr>& dax)
  55 {
  56   std::vector<simgrid::s4u::Exec*> ready_tasks;
  57   std::map<simgrid::s4u::Exec*, unsigned int> candidate_execs;
  58
  59   for (auto& a : dax) {
  60     // Only loot at activity that have their dependencies solved but are not assigned
  61     if (a->dependencies_solved() && not a->is_assigned()) {
  62       // if it is an exec, it's ready
  63       auto* exec = dynamic_cast<simgrid::s4u::Exec*>(a.get());
  64       if (exec != nullptr)
  65         ready_tasks.push_back(exec);
  66       // if it a comm, we consider its successor as a candidate. If a candidate solves all its dependencies,
  67       // i.e., get all its input data, it's ready
  68       const auto* comm = dynamic_cast<simgrid::s4u::Comm*>(a.get());
  69       if (comm != nullptr) {
  70         auto* next_exec = static_cast<simgrid::s4u::Exec*>(comm->get_successors().front().get());
  71         candidate_execs[next_exec]++;
  72         if (next_exec->get_dependencies().size() == candidate_execs[next_exec])
  73           ready_tasks.push_back(next_exec);
  74       }
  75     }
  76   }
  77   XBT_DEBUG("There are %zu ready tasks", ready_tasks.size());
  78   return ready_tasks;
  79 }
  80
  81 static double finish_on_at(const simgrid::s4u::ExecPtr task, const simgrid::s4u::Host* host)
  82 {
  83   double result;
  84
  85   const auto& parents = task->get_dependencies();
  86
  87   if (not parents.empty()) {
  88     double data_available = 0.;
  89     double last_data_available;
  90     /* compute last_data_available */
  91     last_data_available = -1.0;
  92     for (const auto& parent : parents) {
  93       /* normal case */
  94       const auto* comm = dynamic_cast<simgrid::s4u::Comm*>(parent.get());
  95       if (comm != nullptr) {
  96         auto source = comm->get_source();
  97         XBT_DEBUG("transfer from %s to %s", source->get_cname(), host->get_cname());
  98         /* Estimate the redistribution time from this parent */
  99         double redist_time;
 100         if (comm->get_remaining() <= 1e-6) {
 101           redist_time = 0;
 102         } else {
 103           redist_time = sg_host_get_route_latency(source, host) +
 104                         comm->get_remaining() / sg_host_get_route_bandwidth(source, host);
 105         }
 106         // We use the user data field to store the finish time of the predecessor of the comm, i.e., its potential start
 107         // time
 108         data_available = *(static_cast<double*>(comm->get_data())) + redist_time;
 109       }
 110
 111       const auto* exec = dynamic_cast<simgrid::s4u::Exec*>(parent.get());
 112       /* no transfer, control dependency */
 113       if (exec != nullptr) {
 114         data_available = exec->get_finish_time();
 115       }
 116
 117       if (last_data_available < data_available)
 118         last_data_available = data_available;
 119     }
 120
 121     result = fmax(sg_host_get_available_at(host), last_data_available) + task->get_remaining() / host->get_speed();
 122   } else
 123     result = sg_host_get_available_at(host) + task->get_remaining() / host->get_speed();
 124
 125   return result;
 126 }
 127
 128 static simgrid::s4u::Host* get_best_host(const simgrid::s4u::ExecPtr exec)
 129 {
 130   std::vector<simgrid::s4u::Host*> hosts = simgrid::s4u::Engine::get_instance()->get_all_hosts();
 131   auto best_host                         = hosts.front();
 132   double min_EFT                         = finish_on_at(exec, best_host);
 133
 134   for (const auto& h : hosts) {
 135     double EFT = finish_on_at(exec, h);
 136     XBT_DEBUG("%s finishes on %s at %f", exec->get_cname(), h->get_cname(), EFT);
 137
 138     if (EFT < min_EFT) {
 139       min_EFT   = EFT;
 140       best_host = h;
 141     }
 142   }
 143   return best_host;
 144 }
 145
 146 int main(int argc, char** argv)
 147 {
 148   double min_finish_time            = -1.0;
 149   simgrid::s4u::Exec* selected_task = nullptr;
 150   simgrid::s4u::Host* selected_host = nullptr;
 151
 152   simgrid::s4u::Engine e(&argc, argv);
 153   std::set<simgrid::s4u::Activity*> vetoed;
 154   e.track_vetoed_activities(&vetoed);
 155
 156   simgrid::s4u::Activity::on_completion_cb([](simgrid::s4u::Activity& activity) {
 157     // when an Exec completes, we need to set the potential start time of all its ouput comms
 158     const auto* exec = dynamic_cast<simgrid::s4u::Exec*>(&activity);
 159     if (exec == nullptr) // Only Execs are concerned here
 160       return;
 161     for (const auto& succ : exec->get_successors()) {
 162       auto* comm = dynamic_cast<simgrid::s4u::Comm*>(succ.get());
 163       if (comm != nullptr) {
 164         auto* finish_time = new double(exec->get_finish_time());
 165         // We use the user data field to store the finish time of the predecessor of the comm, i.e., its potential start
 166         // time
 167         comm->set_data(finish_time);
 168       }
 169     }
 170   });
 171
 172   e.load_platform(argv[1]);
 173
 174   /*  Allocating the host attribute */
 175   unsigned long total_nhosts = e.get_host_count();
 176   const auto hosts          = e.get_all_hosts();
 177
 178   for (unsigned long i = 0; i < total_nhosts; i++)
 179     hosts[i]->set_data(xbt_new0(struct _HostAttribute, 1));
 180
 181   /* load the DAX file */
 182   std::vector<simgrid::s4u::ActivityPtr> dax = simgrid::s4u::create_DAG_from_DAX(argv[2]);
 183
 184   /* Schedule the root first */
 185   auto* root = static_cast<simgrid::s4u::Exec*>(dax.front().get());
 186   auto host  = get_best_host(root);
 187   root->set_host(host);
 188   // we can also set the source of all the output comms of the root node
 189   for (const auto& succ : root->get_successors()) {
 190     auto* comm = dynamic_cast<simgrid::s4u::Comm*>(succ.get());
 191     if (comm != nullptr)
 192       comm->set_source(host);
 193   }
 194
 195   e.run();
 196
 197   while (not vetoed.empty()) {
 198     XBT_DEBUG("Start new scheduling round");
 199     /* Get the set of ready tasks */
 200     auto ready_tasks = get_ready_tasks(dax);
 201     vetoed.clear();
 202
 203     if (ready_tasks.empty()) {
 204       ready_tasks.clear();
 205       /* there is no ready task, let advance the simulation */
 206       e.run();
 207       continue;
 208     }
 209     /* For each ready task:
 210      * get the host that minimizes the completion time.
 211      * select the task that has the minimum completion time on its best host.
 212      */
 213     for (auto task : ready_tasks) {
 214       XBT_DEBUG("%s is ready", task->get_cname());
 215       host               = get_best_host(task);
 216       double finish_time = finish_on_at(task, host);
 217       if (min_finish_time < 0 || finish_time < min_finish_time) {
 218         min_finish_time = finish_time;
 219         selected_task   = task;
 220         selected_host   = host;
 221       }
 222     }
 223
 224     XBT_INFO("Schedule %s on %s", selected_task->get_cname(), selected_host->get_cname());
 225     selected_task->set_host(selected_host);
 226     // we can also set the destination of all the input comms of the selected task
 227     for (const auto& pred : selected_task->get_dependencies()) {
 228       auto* comm = dynamic_cast<simgrid::s4u::Comm*>(pred.get());
 229       if (comm != nullptr) {
 230         comm->set_destination(selected_host);
 231         delete static_cast<double*>(comm->get_data());
 232       }
 233     }
 234     // we can also set the source of all the output comms of the selected task
 235     for (const auto& succ : selected_task->get_successors()) {
 236       auto* comm = dynamic_cast<simgrid::s4u::Comm*>(succ.get());
 237       if (comm != nullptr)
 238         comm->set_source(selected_host);
 239     }
 240
 241     /*
 242      * tasks can be executed concurrently when they can by default.
 243      * Yet schedulers take decisions assuming that tasks wait for resource availability to start.
 244      * The solution (well crude hack is to keep track of the last task scheduled on a host and add a special type of
 245      * dependency if needed to force the sequential execution meant by the scheduler.
 246      * If the last scheduled task is already done, has failed or is a predecessor of the current task, no need for a
 247      * new dependency
 248      */
 249
 250     auto last_scheduled_task = sg_host_get_last_scheduled_task(selected_host);
 251     if (last_scheduled_task && (last_scheduled_task->get_state() != simgrid::s4u::Activity::State::FINISHED) &&
 252         (last_scheduled_task->get_state() != simgrid::s4u::Activity::State::FAILED) &&
 253         not dependency_exists(sg_host_get_last_scheduled_task(selected_host), selected_task))
 254       last_scheduled_task->add_successor(selected_task);
 255
 256     sg_host_set_last_scheduled_task(selected_host, selected_task);
 257     sg_host_set_available_at(selected_host, min_finish_time);
 258
 259     ready_tasks.clear();
 260     /* reset the min_finish_time for the next set of ready tasks */
 261     min_finish_time = -1.;
 262     e.run();
 263   }
 264
 265   XBT_INFO("Simulation Time: %f", simgrid_get_clock());
 266
 267   for (auto h : hosts) {
 268     xbt_free(h->get_data());
 269     h->set_data(nullptr);
 270   }
 271
 272   return 0;
 273 }