X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/17de27f983314402e5bd16ca1ce8d0d3866d929b..8052f28d5b5a9d88b6724dff2e5e81dee10065d7:/src/dag/loaders.cpp diff --git a/src/dag/loaders.cpp b/src/dag/loaders.cpp index 2b9eaa2721..09591b1e83 100644 --- a/src/dag/loaders.cpp +++ b/src/dag/loaders.cpp @@ -1,5 +1,4 @@ -/* Copyright (c) 2009-2021. The SimGrid Team. - * All rights reserved. */ +/* Copyright (c) 2009-2023. The SimGrid Team. All rights reserved. */ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ @@ -7,6 +6,8 @@ #include "src/internal_config.h" #include #include +#include +#include #include #include #include @@ -19,6 +20,16 @@ #include "dax_dtd.h" #include "dax_dtd.c" +#if SIMGRID_HAVE_JSON +// Disable implicit conversions. See https://github.com/nlohmann/json#implicit-conversions +#ifdef JSON_USE_IMPLICIT_CONVERSIONS +#undef JSON_USE_IMPLICIT_CONVERSIONS +#endif +#define JSON_USE_IMPLICIT_CONVERSIONS 0 +#include +#include +#endif + #if HAVE_GRAPHVIZ #include #endif @@ -33,16 +44,16 @@ static void uniq_transfer_task_name(simgrid::s4u::Comm* comm) std::string new_name = parent->get_name() + "_" + comm->get_name() + "_" + child->get_name(); - comm->set_name(new_name)->vetoable_start(); + comm->set_name(new_name)->start(); } static bool check_for_cycle(const std::vector& dag) { std::vector current; - for (const auto& a : dag) - if (dynamic_cast(a.get()) != nullptr && a->is_waited_by() == 0) - current.push_back(a); + std::copy_if(begin(dag), end(dag), back_inserter(current), [](const auto& a) { + return dynamic_cast(a.get()) != nullptr && a->has_no_successor(); + }); while (not current.empty()) { std::vector next; @@ -72,14 +83,88 @@ static bool check_for_cycle(const std::vector& dag) static YY_BUFFER_STATE input_buffer; -namespace simgrid { -namespace s4u { +namespace simgrid::s4u { static std::vector result; static std::map> jobs; static std::map> files; static ExecPtr current_job; +/** @brief loads a JSON file describing a DAG + * + * See https://github.com/wfcommons/wfformat for more details. We support wfformat 1.4. + */ +std::vector create_DAG_from_json(const std::string& filename) +{ +#if SIMGRID_HAVE_JSON + std::ifstream f(filename); + auto data = nlohmann::json::parse(f); + std::vector dag = {}; + std::map, std::less<>> successors = {}; + std::map comms_destinations = {}; + ActivityPtr current; + + for (auto const& task: data["workflow"]["tasks"]) { + if (task["type"] == "compute") { + current = + Exec::init()->set_name(task["name"].get())->set_flops_amount(task["runtimeInSeconds"].get()); + if (task.contains("machine")) + dynamic_cast(current.get()) + ->set_host(simgrid::s4u::Engine::get_instance()->host_by_name(task["machine"].get())); + } + else if (task["type"] == "transfer"){ + current = Comm::sendto_init() + ->set_name(task["name"].get()) + ->set_payload_size(task["writtenBytes"].get()); + if (task.contains("machine")) + comms_destinations[current] = + simgrid::s4u::Engine::get_instance()->host_by_name(task["machine"].get()); + if (task["parents"].size() == 1) { + ActivityPtr parent_activity; + for (auto const& activity: dag) { + if (activity->get_name() == task["parents"][0]) { + parent_activity = activity; + break; + } + } + if (dynamic_cast(parent_activity.get()) != nullptr) + dynamic_cast(current.get())->set_source(dynamic_cast(parent_activity.get())->get_host()); + else if (dynamic_cast(parent_activity.get()) != nullptr) + dynamic_cast(current.get())->set_source(dynamic_cast(parent_activity.get())->get_destination()); + } + } else if (XBT_LOG_ISENABLED(dag_parsing, xbt_log_priority_debug)) { + std::stringstream ss; + ss << task["type"]; + XBT_DEBUG("Task type \"%s\" not supported.", ss.str().c_str()); + } + + dag.push_back(current); + for (auto const& parent : task["parents"]) + successors[parent.get()].push_back(current); + } + // Assign successors + for (auto const& [parent, successors_list] : successors) + for (auto const& activity: dag) + if (activity->get_name() == parent) { + for (auto const& successor: successors_list) + activity->add_successor(successor); + break; + } + // Assign destinations of Comms (if done before successors are assigned there is a bug) + for (auto const& [comm, destination]: comms_destinations) + dynamic_cast(comm.get())->set_destination(destination); + + // Start only Activities with dependencies solved + for (auto const& activity: dag) { + if (dynamic_cast(activity.get()) != nullptr && activity->dependencies_solved()) + activity->start(); + } + return dag; +#else + xbt_die("JSON support was not compiled in, probably because nlohmann/json was not found. Please install " + "nlohmann-json3-dev and recompile SimGrid to use this feature."); +#endif +} /** @brief loads a DAX file describing a DAG * * See https://confluence.pegasus.isi.edu/display/pegasus/WorkflowGenerator for more details. @@ -93,12 +178,12 @@ std::vector create_DAG_from_DAX(const std::string& filename) dax_lineno = 1; auto root_task = Exec::init()->set_name("root")->set_flops_amount(0); - root_task->vetoable_start(); + root_task->start(); result.push_back(root_task); auto end_task = Exec::init()->set_name("end")->set_flops_amount(0); - end_task->vetoable_start(); + end_task->start(); xbt_assert(dax_lex() == 0, "Parse error in %s: %s", filename.c_str(), dax__parse_err_msg()); dax__delete_buffer(input_buffer); @@ -110,10 +195,8 @@ std::vector create_DAG_from_DAX(const std::string& filename) * Files not produced in the system are said to be produced by root task (top of DAG). * Files not consumed in the system are said to be consumed by end task (bottom of DAG). */ - CommPtr file; - - for (auto const& elm : files) { - file = elm.second; + for (auto const& [_, elm] : files) { + CommPtr file = elm; CommPtr newfile; if (file->dependencies_solved()) { for (auto const& it : file->get_successors()) { @@ -123,7 +206,7 @@ std::vector create_DAG_from_DAX(const std::string& filename) result.push_back(newfile); } } - if (file->is_waited_by() == 0) { + if (file->has_no_successor()) { for (auto const& it : file->get_dependencies()) { newfile = Comm::sendto_init()->set_name(file->get_name())->set_payload_size(file->get_remaining()); it->add_successor(newfile); @@ -162,7 +245,7 @@ std::vector create_DAG_from_DAX(const std::string& filename) if ((a != root_task) && (a != end_task)) { if (a->dependencies_solved()) root_task->add_successor(a); - if (a->is_waited_by() == 0) + if (a->has_no_successor()) a->add_successor(end_task); } } @@ -185,7 +268,7 @@ std::vector create_DAG_from_dot(const std::string& filename) FILE* in_file = fopen(filename.c_str(), "r"); xbt_assert(in_file != nullptr, "Failed to open file: %s", filename.c_str()); - Agraph_t* dag_dot = agread(in_file, NIL(Agdisc_t*)); + Agraph_t* dag_dot = agread(in_file, nullptr); std::unordered_map activities; std::vector dag; @@ -196,27 +279,27 @@ std::vector create_DAG_from_dot(const std::string& filename) /* Create all the nodes */ Agnode_t* node = nullptr; for (node = agfstnode(dag_dot); node; node = agnxtnode(dag_dot, node)) { - char* name = agnameof(node); + const std::string name = agnameof(node); double amount = atof(agget(node, (char*)"size")); if (activities.find(name) == activities.end()) { - XBT_DEBUG("See ", name, amount); - act = Exec::init()->set_name(name)->set_flops_amount(amount)->vetoable_start(); - activities.insert({std::string(name), act}); - if (strcmp(name, "root") && strcmp(name, "end")) + XBT_DEBUG("See ", name.c_str(), amount); + act = Exec::init()->set_name(name)->set_flops_amount(amount)->start(); + activities.try_emplace(name, act); + if (name != "root" && name != "end") dag.push_back(act); } else { - XBT_WARN("Exec '%s' is defined more than once", name); + XBT_WARN("Exec '%s' is defined more than once", name.c_str()); } } /*Check if 'root' and 'end' nodes have been explicitly declared. If not, create them. */ if (activities.find("root") == activities.end()) - root = Exec::init()->set_name("root")->set_flops_amount(0)->vetoable_start(); + root = Exec::init()->set_name("root")->set_flops_amount(0)->start(); else root = activities.at("root"); if (activities.find("end") == activities.end()) - end = Exec::init()->set_name("end")->set_flops_amount(0)->vetoable_start(); + end = Exec::init()->set_name("end")->set_flops_amount(0)->start(); else end = activities.at("end"); @@ -241,10 +324,10 @@ std::vector create_DAG_from_dot(const std::string& filename) std::string name = std::string(src_name) + "->" + dst_name; XBT_DEBUG("See ", name.c_str(), size); if (activities.find(name) == activities.end()) { - act = Comm::sendto_init()->set_name(name)->set_payload_size(size)->vetoable_start(); + act = Comm::sendto_init()->set_name(name)->set_payload_size(size)->start(); src->add_successor(act); act->add_successor(dst); - activities.insert({name, act}); + activities.try_emplace(name, act); dag.push_back(act); } else { XBT_WARN("Comm '%s' is defined more than once", name.c_str()); @@ -267,7 +350,7 @@ std::vector create_DAG_from_dot(const std::string& filename) root->add_successor(a); } - if (a->is_waited_by() == 0 && a != end) { + if (a->has_no_successor() && a != end) { XBT_DEBUG("Activity '%s' has no successors. Add dependency to 'end'", a->get_cname()); a->add_successor(end); } @@ -281,6 +364,7 @@ std::vector create_DAG_from_dot(const std::string& filename) for (const auto& a : dag) a->destroy(); dag.clear(); + dag.shrink_to_fit(); } return dag; @@ -292,13 +376,12 @@ std::vector create_DAG_from_dot(const std::string& filename) "Please install graphviz, graphviz-dev, and libgraphviz-dev (and erase CMakeCache.txt) before recompiling."); } #endif -} // namespace s4u -} // namespace simgrid +} // namespace simgrid::s4u void STag_dax__adag() { try { - double version = std::stod(std::string(A_dax__adag_version)); + double version = std::stod(A_dax__adag_version); xbt_assert(version == 2.1, "Expected version 2.1 in tag, got %f. Fix the parser or your file", version); } catch (const std::invalid_argument&) { throw std::invalid_argument(std::string("Parse error: ") + A_dax__adag_version + " is not a double"); @@ -308,13 +391,13 @@ void STag_dax__adag() void STag_dax__job() { try { - double runtime = std::stod(std::string(A_dax__job_runtime)); + double runtime = std::stod(A_dax__job_runtime); std::string name = std::string(A_dax__job_id) + "@" + A_dax__job_name; runtime *= 4200000000.; /* Assume that timings were done on a 4.2GFlops machine. I mean, why not? */ XBT_DEBUG("See ", A_dax__job_id, A_dax__job_runtime, runtime); - simgrid::s4u::current_job = simgrid::s4u::Exec::init()->set_name(name)->set_flops_amount(runtime)->vetoable_start(); - simgrid::s4u::jobs.insert({A_dax__job_id, simgrid::s4u::current_job}); + simgrid::s4u::current_job = simgrid::s4u::Exec::init()->set_name(name)->set_flops_amount(runtime)->start(); + simgrid::s4u::jobs.try_emplace(A_dax__job_id, simgrid::s4u::current_job); simgrid::s4u::result.push_back(simgrid::s4u::current_job); } catch (const std::invalid_argument&) { throw std::invalid_argument(std::string("Parse error: ") + A_dax__job_runtime + " is not a double"); @@ -325,7 +408,7 @@ void STag_dax__uses() { double size; try { - size = std::stod(std::string(A_dax__uses_size)); + size = std::stod(A_dax__uses_size); } catch (const std::invalid_argument&) { throw std::invalid_argument(std::string("Parse error: ") + A_dax__uses_size + " is not a double"); } @@ -360,7 +443,7 @@ void STag_dax__child() if (job != simgrid::s4u::jobs.end()) { current_child = job->second; } else { - throw std::out_of_range(std::string("Parse error on line ") + std::to_string(dax_lineno) + + throw std::out_of_range("Parse error on line " + std::to_string(dax_lineno) + ": Asked to add dependencies to the non-existent " + A_dax__child_ref + "task"); } } @@ -378,9 +461,9 @@ void STag_dax__parent() parent->add_successor(current_child); XBT_DEBUG("Control-flow dependency from %s to %s", current_child->get_cname(), parent->get_cname()); } else { - throw std::out_of_range(std::string("Parse error on line ") + std::to_string(dax_lineno) + - ": Asked to add a dependency from " + current_child->get_name() + " to " + - A_dax__parent_ref + ", but " + A_dax__parent_ref + " does not exist"); + throw std::out_of_range("Parse error on line " + std::to_string(dax_lineno) + ": Asked to add a dependency from " + + current_child->get_name() + " to " + A_dax__parent_ref + ", but " + A_dax__parent_ref + + " does not exist"); } }