X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/43793104b1630504dfaa7486958884a214896049..c10c571a634491a941c1b39c50020da8d709a940:/src/s4u/s4u_Task.cpp diff --git a/src/s4u/s4u_Task.cpp b/src/s4u/s4u_Task.cpp index 9419363f1d..80756d6cfa 100644 --- a/src/s4u/s4u_Task.cpp +++ b/src/s4u/s4u_Task.cpp @@ -1,10 +1,14 @@ #include #include -#include +#include #include +#include #include #include +#include #include +#include +#include #include "src/simgrid/module.hpp" @@ -31,9 +35,9 @@ Task::Task(const std::string& name) : name_(name) {} * @brief Return True if the Task can start a new Activity. * @note The Task is ready if not already doing something and there is at least one execution waiting in queue. */ -bool Task::ready_to_run() const +bool Task::ready_to_run(std::string instance) { - return not working_ && queued_firings_ > 0; + return running_instances_[instance] < parallelism_degree_[instance] && queued_firings_[instance] > 0; } /** @@ -45,11 +49,8 @@ bool Task::ready_to_run() const void Task::receive(Task* source) { XBT_DEBUG("Task %s received a token from %s", name_.c_str(), source->name_.c_str()); - auto source_count = predecessors_[source]; predecessors_[source]++; - if (tokens_received_.size() <= queued_firings_ + source_count) - tokens_received_.emplace_back(); - tokens_received_[queued_firings_ + source_count][source] = source->token_; + tokens_received_[source].push_back(source->token_); bool enough_tokens = true; for (auto const& [key, val] : predecessors_) if (val < 1) { @@ -72,19 +73,64 @@ void Task::receive(Task* source) * Send a token to each of its successors. * Start a new execution if possible. */ -void Task::complete() +void Task::complete(std::string instance) { xbt_assert(Actor::is_maestro()); - working_ = false; - count_++; - on_this_completion(this); - on_completion(this); - if (current_activity_) - previous_activity_ = std::move(current_activity_); - for (auto const& t : successors_) - t->receive(this); - if (ready_to_run()) - fire(); + running_instances_[instance] = running_instances_[instance] - 1; + count_[instance] = count_[instance] + 1; + if (instance == "collector") { + // XBT_INFO("Trigger on completion: %s - %s", get_cname(), instance.c_str()); + on_this_completion(this); + on_completion(this); + for (auto const& t : successors_) + t->receive(this); + } else if (instance == "dispatcher") { + auto next_instance = load_balancing_function_(); + xbt_assert(next_instance != "dispatcher" and next_instance != "collector", "Invalid instance selected: %s", + next_instance.c_str()); + queued_firings_[next_instance] = queued_firings_.at(next_instance) + 1; + while (ready_to_run(next_instance)) + fire(next_instance); + } else { + queued_firings_["collector"] = queued_firings_["collector"] + 1; + while (ready_to_run("collector")) + fire("collector"); + } + if (ready_to_run(instance)) + fire(instance); +} + +/** @param n The new parallelism degree of the Task. + * @brief Set the parallelism degree of the Task to inscrease or decrease horizontal scaling. + * @note When increasing the degree the function starts new instances if there is queued firings. + * When decreasing the degree the function does NOT stop running instances. + */ +void Task::set_parallelism_degree(int n, std::string instance) +{ + xbt_assert(n > 0, "Parallelism degree must be above 0."); + simgrid::kernel::actor::simcall_answered([this, n, &instance] { + if (instance == "all") { + for (auto& [key, value] : parallelism_degree_) { + parallelism_degree_[key] = n; + while (ready_to_run(key)) + fire(key); + } + } else { + parallelism_degree_[instance] = n; + while (ready_to_run(instance)) + fire(instance); + } + }); +} + +void Task::set_internal_bytes(int bytes, std::string instance) +{ + simgrid::kernel::actor::simcall_answered([this, bytes, &instance] { internal_bytes_to_send_[instance] = bytes; }); +} + +void Task::set_load_balancing_function(std::function func) +{ + simgrid::kernel::actor::simcall_answered([this, func] { load_balancing_function_ = func; }); } /** @param n The number of firings to enqueue. @@ -94,45 +140,54 @@ void Task::complete() void Task::enqueue_firings(int n) { simgrid::kernel::actor::simcall_answered([this, n] { - queued_firings_ += n; - if (ready_to_run()) - fire(); + queued_firings_["dispatcher"] += n; + while (ready_to_run("dispatcher")) + fire("dispatcher"); }); } +/** @param name The new name to set. + * @brief Set the name of the Task. + */ +void Task::set_name(std::string name) +{ + name_ = name; +} + /** @param amount The amount to set. * @brief Set the amout of work to do. * @note Amount in flop for ExecTask and in bytes for CommTask. */ -void Task::set_amount(double amount) +void Task::set_amount(double amount, std::string instance) { - simgrid::kernel::actor::simcall_answered([this, amount] { amount_ = amount; }); + simgrid::kernel::actor::simcall_answered([this, amount, &instance] { amount_[instance] = amount; }); } /** @param token The token to set. * @brief Set the token to send to successors. - * @note The token is passed to each successor after the task end, i.e., after the on_end callback. + * @note The token is passed to each successor after the task end, i.e., after the on_completion callback. */ void Task::set_token(std::shared_ptr token) { simgrid::kernel::actor::simcall_answered([this, token] { token_ = token; }); } -/** @return Map of tokens received for the next execution. - * @note If there is no queued execution for this task the map might not exist or be partially empty. - */ -std::shared_ptr Task::get_next_token_from(TaskPtr t) +void Task::deque_token_from(TaskPtr t) { - return tokens_received_.front()[t]; + simgrid::kernel::actor::simcall_answered([this, &t] { tokens_received_.at(t).pop_front(); }); } -void Task::fire() { - on_this_start(this); - on_start(this); - working_ = true; - queued_firings_ = std::max(queued_firings_ - 1, 0); - if (not tokens_received_.empty()) - tokens_received_.pop_front(); +void Task::fire(std::string instance) +{ + if ((int)current_activities_[instance].size() > parallelism_degree_[instance]) { + current_activities_[instance].pop_front(); + } + if (instance != "dispatcher" and instance != "collector") { + on_this_start(this); + on_start(this); + } + running_instances_[instance]++; + queued_firings_[instance] = std::max(queued_firings_[instance] - 1, 0); } /** @param successor The Task to add. @@ -159,6 +214,9 @@ void Task::remove_successor(TaskPtr successor) }); } +/** + * @brief TODO + */ void Task::remove_all_successors() { simgrid::kernel::actor::simcall_answered([this] { @@ -170,12 +228,56 @@ void Task::remove_all_successors() }); } +/** + * @brief TODO + */ +void Task::add_instances(int n) +{ + xbt_assert(n >= 0, "Cannot add a negative number of instances (provided: %d)", n); + int instance_count = (int)amount_.size() - 2; + for (int i = instance_count; i < n + instance_count; i++) { + amount_["instance_" + std::to_string(i)] = amount_.at("instance_0"); + queued_firings_["instance_" + std::to_string(i)] = 0; + running_instances_["instance_" + std::to_string(i)] = 0; + count_["instance_" + std::to_string(i)] = 0; + parallelism_degree_["instance_" + std::to_string(i)] = parallelism_degree_.at("instance_0"); + current_activities_["instance_" + std::to_string(i)] = {}; + internal_bytes_to_send_["instance_" + std::to_string(i)] = internal_bytes_to_send_.at("instance_0"); + ; + } +} + +/** + * @brief TODO + */ +void Task::remove_instances(int n) +{ + int instance_count = (int)amount_.size() - 2; + xbt_assert(n >= 0, "Cannot remove a negative number of instances (provided: %d)", n); + xbt_assert(instance_count - n > 0, "The number of instances must be above 0 (instances: %d, provided: %d)", + instance_count, n); + for (int i = instance_count - 1; i >= instance_count - n; i--) + xbt_assert(running_instances_.at("instance_" + std::to_string(i)) == 0, + "Cannot remove a running instance (instances: %d)", i); + for (int i = instance_count - 1; i >= instance_count - n; i--) { + amount_.erase("instance_" + std::to_string(i)); + queued_firings_.erase("instance_" + std::to_string(i)); + running_instances_.erase("instance_" + std::to_string(i)); + count_.erase("instance_" + std::to_string(i)); + parallelism_degree_.erase("instance_" + std::to_string(i)); + current_activities_.erase("instance_" + std::to_string(i)); + } +} + /** * @brief Default constructor. */ -ExecTask::ExecTask(const std::string& name) : Task(name) {} +ExecTask::ExecTask(const std::string& name) : Task(name) +{ + set_load_balancing_function([]() { return "instance_0"; }); +} -/** @ingroup plugin_task +/** * @brief Smart Constructor. */ ExecTaskPtr ExecTask::init(const std::string& name) @@ -183,7 +285,7 @@ ExecTaskPtr ExecTask::init(const std::string& name) return ExecTaskPtr(new ExecTask(name)); } -/** @ingroup plugin_task +/** * @brief Smart Constructor. */ ExecTaskPtr ExecTask::init(const std::string& name, double flops, Host* host) @@ -193,43 +295,104 @@ ExecTaskPtr ExecTask::init(const std::string& name, double flops, Host* host) /** * @brief Do one execution of the Task. - * @note Call the on_this_start() func. Set its working status as true. + * @note Call the on_this_start() func. * Init and start the underlying Activity. */ -void ExecTask::fire() +void ExecTask::fire(std::string instance) { - Task::fire(); - auto exec = Exec::init()->set_name(get_name())->set_flops_amount(get_amount())->set_host(host_); - exec->start(); - exec->on_this_completion_cb([this](Exec const&) { this->complete(); }); - set_current_activity(exec); + Task::fire(instance); + if (instance == "dispatcher" or instance == "collector") { + auto exec = Exec::init() + ->set_name(get_name() + "_" + instance) + ->set_flops_amount(get_amount(instance)) + ->set_host(host_[instance]); + exec->start(); + exec->on_this_completion_cb([this, instance](Exec const&) { complete(instance); }); + store_activity(exec, instance); + } else { + auto exec = Exec::init()->set_name(get_name())->set_flops_amount(get_amount())->set_host(host_[instance]); + if (host_["dispatcher"] == host_[instance]) { + exec->start(); + store_activity(exec, instance); + } else { + auto comm = Comm::sendto_init(host_["dispatcher"], host_[instance]) + ->set_name(get_name() + "_dispatcher_to_" + instance) + ->set_payload_size(get_internal_bytes("dispatcher")); + comm->add_successor(exec); + comm->start(); + store_activity(comm, instance); + } + if (host_[instance] == host_["collector"]) { + exec->on_this_completion_cb([this, instance](Exec const&) { complete(instance); }); + if (host_["dispatcher"] != host_[instance]) + store_activity(exec, instance); + } else { + auto comm = Comm::sendto_init(host_[instance], host_["collector"]) + ->set_name(get_name() + instance + "_to_collector") + ->set_payload_size(get_internal_bytes(instance)); + exec->add_successor(comm); + comm->on_this_completion_cb([this, instance](Comm const&) { complete(instance); }); + comm.detach(); + } + } } -/** @ingroup plugin_task +/** * @param host The host to set. * @brief Set a new host. */ -ExecTaskPtr ExecTask::set_host(Host* host) +ExecTaskPtr ExecTask::set_host(Host* host, std::string instance) { - kernel::actor::simcall_answered([this, host] { host_ = host; }); + kernel::actor::simcall_answered([this, host, &instance] { + if (instance == "all") + for (auto& [key, value] : host_) + host_[key] = host; + else + host_[instance] = host; + }); return this; } -/** @ingroup plugin_task +/** * @param flops The amount of flops to set. */ -ExecTaskPtr ExecTask::set_flops(double flops) +ExecTaskPtr ExecTask::set_flops(double flops, std::string instance) { - kernel::actor::simcall_answered([this, flops] { set_amount(flops); }); + kernel::actor::simcall_answered([this, flops, &instance] { set_amount(flops, instance); }); return this; } +/** + * @brief TODO + */ +void ExecTask::add_instances(int n) +{ + Task::add_instances(n); + int instance_count = (int)host_.size() - 2; + for (int i = instance_count; i < n + instance_count; i++) + host_["instance_" + std::to_string(i)] = host_.at("instance_0"); +} + +/** + * @brief TODO + */ +void ExecTask::remove_instances(int n) +{ + Task::remove_instances(n); + int instance_count = (int)host_.size() - 2; + for (int i = instance_count - 1; i >= instance_count - n; i--) + host_.erase("instance_" + std::to_string(i)); +} + /** * @brief Default constructor. */ -CommTask::CommTask(const std::string& name) : Task(name) {} +CommTask::CommTask(const std::string& name) : Task(name) +{ + set_load_balancing_function([]() { return "instance_0"; }); +} -/** @ingroup plugin_task +/** * @brief Smart constructor. */ CommTaskPtr CommTask::init(const std::string& name) @@ -237,7 +400,7 @@ CommTaskPtr CommTask::init(const std::string& name) return CommTaskPtr(new CommTask(name)); } -/** @ingroup plugin_task +/** * @brief Smart constructor. */ CommTaskPtr CommTask::init(const std::string& name, double bytes, Host* source, Host* destination) @@ -247,19 +410,29 @@ CommTaskPtr CommTask::init(const std::string& name, double bytes, Host* source, /** * @brief Do one execution of the Task. - * @note Call the on_this_start() func. Set its working status as true. + * @note Call the on_this_start() func. * Init and start the underlying Activity. */ -void CommTask::fire() +void CommTask::fire(std::string instance) { - Task::fire(); - auto comm = Comm::sendto_init(source_, destination_)->set_name(get_name())->set_payload_size(get_amount()); - comm->start(); - comm->on_this_completion_cb([this](Comm const&) { this->complete(); }); - set_current_activity(comm); + Task::fire(instance); + if (instance == "dispatcher" or instance == "collector") { + auto exec = Exec::init() + ->set_name(get_name() + "_" + instance) + ->set_flops_amount(get_amount(instance)) + ->set_host(instance == "dispatcher" ? source_ : destination_); + exec->start(); + exec->on_this_completion_cb([this, instance](Exec const&) { complete(instance); }); + store_activity(exec, instance); + } else { + auto comm = Comm::sendto_init(source_, destination_)->set_name(get_name())->set_payload_size(get_amount()); + comm->start(); + comm->on_this_completion_cb([this, instance](Comm const&) { complete(instance); }); + store_activity(comm, instance); + } } -/** @ingroup plugin_task +/** * @param source The host to set. * @brief Set a new source host. */ @@ -269,7 +442,7 @@ CommTaskPtr CommTask::set_source(Host* source) return this; } -/** @ingroup plugin_task +/** * @param destination The host to set. * @brief Set a new destination host. */ @@ -279,7 +452,7 @@ CommTaskPtr CommTask::set_destination(Host* destination) return this; } -/** @ingroup plugin_task +/** * @param bytes The amount of bytes to set. */ CommTaskPtr CommTask::set_bytes(double bytes) @@ -291,9 +464,12 @@ CommTaskPtr CommTask::set_bytes(double bytes) /** * @brief Default constructor. */ -IoTask::IoTask(const std::string& name) : Task(name) {} +IoTask::IoTask(const std::string& name) : Task(name) +{ + set_load_balancing_function([]() { return "instance_0"; }); +} -/** @ingroup plugin_task +/** * @brief Smart Constructor. */ IoTaskPtr IoTask::init(const std::string& name) @@ -301,7 +477,7 @@ IoTaskPtr IoTask::init(const std::string& name) return IoTaskPtr(new IoTask(name)); } -/** @ingroup plugin_task +/** * @brief Smart Constructor. */ IoTaskPtr IoTask::init(const std::string& name, double bytes, Disk* disk, Io::OpType type) @@ -309,7 +485,7 @@ IoTaskPtr IoTask::init(const std::string& name, double bytes, Disk* disk, Io::Op return init(name)->set_bytes(bytes)->set_disk(disk)->set_op_type(type); } -/** @ingroup plugin_task +/** * @param disk The disk to set. * @brief Set a new disk. */ @@ -319,7 +495,7 @@ IoTaskPtr IoTask::set_disk(Disk* disk) return this; } -/** @ingroup plugin_task +/** * @param bytes The amount of bytes to set. */ IoTaskPtr IoTask::set_bytes(double bytes) @@ -328,20 +504,30 @@ IoTaskPtr IoTask::set_bytes(double bytes) return this; } -/** @ingroup plugin_task */ +/** */ IoTaskPtr IoTask::set_op_type(Io::OpType type) { kernel::actor::simcall_answered([this, type] { type_ = type; }); return this; } -void IoTask::fire() +void IoTask::fire(std::string instance) { - Task::fire(); - auto io = Io::init()->set_name(get_name())->set_size(get_amount())->set_disk(disk_)->set_op_type(type_); - io->start(); - io->on_this_completion_cb([this](Io const&) { this->complete(); }); - set_current_activity(io); + Task::fire(instance); + if (instance == "dispatcher" or instance == "collector") { + auto exec = Exec::init() + ->set_name(get_name() + "_" + instance) + ->set_flops_amount(get_amount(instance)) + ->set_host(disk_->get_host()); + exec->start(); + exec->on_this_completion_cb([this, instance](Exec const&) { complete(instance); }); + store_activity(exec, instance); + } else { + auto io = Io::init()->set_name(get_name())->set_size(get_amount())->set_disk(disk_)->set_op_type(type_); + io->start(); + io->on_this_completion_cb([this, instance](Io const&) { complete(instance); }); + store_activity(io, instance); + } } } // namespace simgrid::s4u