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