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

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