#include <simgrid/plugins/operation.hpp>
#include <simgrid/s4u/Comm.hpp>
#include <simgrid/s4u/Exec.hpp>
+#include <simgrid/s4u/Io.hpp>
+#include <simgrid/simix.hpp>
#include "src/simgrid/module.hpp"
SIMGRID_REGISTER_PLUGIN(operation, "Battery management", nullptr)
+/** @defgroup plugin_operation plugin_operation Plugin Operation
+
+ @beginrst
+
+This is the operation plugin, enabling management of Operations.
+To activate this plugin, first call :cpp:func:`Operation::init`.
+
+Operations are designed to represent workflows, i.e, graphs of Operations.
+Operations can only be instancied using either
+:cpp:func:`simgrid::plugins::ExecOp::init` or :cpp:func:`simgrid::plugins::CommOp::init`
+An ExecOp is an Execution Operation. Its underlying Activity is an :ref:`Exec <API_s4u_Exec>`.
+A CommOp is a Communication Operation. Its underlying Activity is a :ref:`Comm <API_s4u_Comm>`.
+
+ @endrst
+ */
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(Operation, kernel, "Logging specific to the operation plugin");
namespace simgrid::plugins {
-Operation::Operation(const std::string& name, double amount) : name_(name), amount_(amount) {}
-std::string Operation::get_name()
+xbt::signal<void(Operation*)> Operation::on_start;
+xbt::signal<void(Operation*)> Operation::on_end;
+
+Operation::Operation(const std::string& name) : name_(name) {}
+
+/**
+ * @param predecessor The Operation to add.
+ * @brief Add a predecessor to this Operation.
+ */
+void Operation::add_predecessor(Operation* predecessor)
{
- return this->name_;
+ if (predecessors_.find(predecessor) == predecessors_.end())
+ simgrid::kernel::actor::simcall_answered([this, predecessor] { predecessors_[predecessor] = 0; });
}
-void Operation::add_predecessor(Operation* predecessor)
+/**
+ * @param predecessor The Operation to remove.
+ * @brief Remove a predecessor from this Operation.
+ */
+void Operation::remove_predecessor(Operation* predecessor)
{
- this->predecessors_[predecessor] = 0;
+ simgrid::kernel::actor::simcall_answered([this, predecessor] { predecessors_.erase(predecessor); });
}
+/**
+ * @brief Return True if the Operation can start a new Activity.
+ * @note The Operation is ready if not already doing something and there is at least one execution waiting in queue.
+ */
bool Operation::ready_to_run() const
{
- if (this->working_ or (this->iteration_count_ != -1 and this->iteration_count_ >= this->iteration_limit_))
- return false;
- for (auto const& [key, val] : this->predecessors_)
- if (val < 1)
- return false;
- return true;
+ return not working_ && queued_execs_ > 0;
}
+/**
+ * @param source The sender.
+ * @brief Receive a token from another Operation.
+ * @note Check upon reception if the Operation has received a token from each of its predecessors,
+ * and in this case consumes those tokens and enqueue an execution.
+ */
void Operation::receive(Operation* source)
{
- auto it = this->predecessors_.find(source);
- it->second++;
- if (this->ready_to_run())
- this->execute();
+ XBT_DEBUG("Operation %s received a token from %s", name_.c_str(), source->name_.c_str());
+ auto it = predecessors_.find(source);
+ simgrid::kernel::actor::simcall_answered([this, it] {
+ it->second++;
+ bool enough_tokens = true;
+ for (auto const& [key, val] : predecessors_)
+ if (val < 1) {
+ enough_tokens = false;
+ break;
+ }
+ if (enough_tokens) {
+ for (auto& [key, val] : predecessors_)
+ val--;
+ enqueue_execs(1);
+ }
+ });
}
+/**
+ * @brief Operation routine when finishing an execution.
+ * @note Set its working status as false.
+ * Add 1 to its count of finished executions.
+ * Call the on_this_end func.
+ * Fire on_end callback.
+ * Send a token to each of its successors.
+ * Start a new execution if possible.
+ */
void Operation::complete()
{
- working_ = false;
- call_end();
- for (auto const& op : this->successors_)
+ simgrid::kernel::actor::simcall_answered([this] {
+ working_ = false;
+ count_++;
+ });
+ for (auto end_func : end_func_handlers_)
+ end_func(this);
+ Operation::on_end(this);
+ for (auto const& op : successors_)
op->receive(this);
if (ready_to_run())
execute();
}
-void Operation::consume()
+/** @ingroup plugin_operation
+ * @brief Init the Operation plugin.
+ * @note Add a completion callback to all Activities to call Operation::complete().
+ */
+void Operation::init()
+{
+ if (Operation::inited_)
+ return;
+ Operation::inited_ = true;
+ ExtendedAttributeActivity::EXTENSION_ID = simgrid::s4u::Activity::extension_create<ExtendedAttributeActivity>();
+ simgrid::s4u::Exec::on_completion_cb([](simgrid::s4u::Exec const& exec) {
+ exec.extension<ExtendedAttributeActivity>()->operation_->complete();
+ });
+ simgrid::s4u::Comm::on_completion_cb([](simgrid::s4u::Comm const& comm) {
+ comm.extension<ExtendedAttributeActivity>()->operation_->complete();
+ });
+ simgrid::s4u::Io::on_completion_cb([](simgrid::s4u::Io const& io) {
+ io.extension<ExtendedAttributeActivity>()->operation_->complete();
+ });
+}
+
+/** @ingroup plugin_operation
+ * @param n The number of executions to enqueue.
+ * @brief Enqueue executions.
+ * @note Immediatly starts an execution if possible.
+ */
+void Operation::enqueue_execs(int n)
{
- for (auto const& [key, val] : predecessors_)
- predecessors_[key] = predecessors_[key] > 0 ? predecessors_[key] - 1 : 0;
+ simgrid::kernel::actor::simcall_answered([this, n] {
+ queued_execs_ += n;
+ if (ready_to_run())
+ execute();
+ });
}
-void Operation::call_end()
+/** @ingroup plugin_operation
+ * @param amount The amount to set.
+ * @brief Set the amout of work to do.
+ * @note Amount in flop for ExecOp and in bytes for CommOp.
+ */
+void Operation::set_amount(double amount)
{
- end_func(this);
+ simgrid::kernel::actor::simcall_answered([this, amount] { amount_ = amount; });
}
-void Operation::init()
+/** @ingroup plugin_operation
+ * @param successor The Operation to add.
+ * @brief Add a successor to this Operation.
+ * @note It also adds this as a predecessor of successor.
+ */
+void Operation::add_successor(OperationPtr successor)
{
- if (Operation::inited_)
- return;
- Operation::inited_ = true;
- ExtendedAttributeActivity::EXTENSION_ID = simgrid::s4u::Activity::extension_create<ExtendedAttributeActivity>();
- simgrid::s4u::Activity::on_completion_cb([&](simgrid::s4u::Activity const& activity) {
- activity.extension<ExtendedAttributeActivity>()->operation_->complete();
+ simgrid::kernel::actor::simcall_answered([this, successor] { successors_.insert(successor.get()); });
+ successor->add_predecessor(this);
+}
+
+/** @ingroup plugin_operation
+ * @param successor The Operation to remove.
+ * @brief Remove a successor from this Operation.
+ * @note It also remove this from the predecessors of successor.
+ */
+void Operation::remove_successor(OperationPtr successor)
+{
+ simgrid::kernel::actor::simcall_answered([this, successor] { successors_.erase(successor.get()); });
+ successor->remove_predecessor(this);
+}
+
+void Operation::remove_all_successors()
+{
+ simgrid::kernel::actor::simcall_answered([this] {
+ while (not successors_.empty()) {
+ auto* successor = *(successors_.begin());
+ successor->predecessors_.erase(this);
+ successors_.erase(successor);
+ }
});
}
-void Operation::set_iteration_limit(unsigned int n)
+/** @ingroup plugin_operation
+ * @param func The function to set.
+ * @brief Set a function to be called before each execution.
+ * @note The function is called before the underlying Activity starts.
+ */
+void Operation::on_this_start(const std::function<void(Operation*)>& func)
{
- iteration_limit_ = n;
+ simgrid::kernel::actor::simcall_answered([this, &func] { start_func_handlers_.push_back(func); });
}
-void Operation::add_successor(OperationPtr op)
+/** @ingroup plugin_operation
+ * @param func The function to set.
+ * @brief Set a function to be called after each execution.
+ * @note The function is called after the underlying Activity ends, but before sending tokens to successors.
+ */
+void Operation::on_this_end(const std::function<void(Operation*)>& func)
{
- successors_.insert(op.get());
- op->add_predecessor(this);
+ simgrid::kernel::actor::simcall_answered([this, &func] { end_func_handlers_.push_back(func); });
}
-void Operation::on_end(std::function<void(Operation*)> func)
+/** @ingroup plugin_operation
+ * @brief Return the number of completed executions.
+ */
+int Operation::get_count() const
{
- end_func = func;
+ return count_;
}
-ExecOp::ExecOp(const std::string& name, double flops, simgrid::s4u::Host* host) : Operation(name, flops), host_(host) {}
+/**
+ * @brief Default constructor.
+ */
+ExecOp::ExecOp(const std::string& name) : Operation(name) {}
-ExecOpPtr ExecOp::create(const std::string& name, double flops, simgrid::s4u::Host* host)
+/** @ingroup plugin_operation
+ * @brief Smart Constructor.
+ */
+ExecOpPtr ExecOp::init(const std::string& name)
{
- auto op = ExecOpPtr(new ExecOp(name, flops, host));
- return op;
+ return ExecOpPtr(new ExecOp(name));
}
+/** @ingroup plugin_operation
+ * @brief Smart Constructor.
+ */
+ExecOpPtr ExecOp::init(const std::string& name, double flops, s4u::Host* host)
+{
+ return init(name)->set_flops(flops)->set_host(host);
+}
+
+/**
+ * @brief Do one execution of the Operation.
+ * @note Call the on_this_start() func. Set its working status as true.
+ * Init and start the underlying Activity.
+ */
void ExecOp::execute()
{
- iteration_count_++;
- working_ = true;
- consume();
- simgrid::s4u::ExecPtr exec = simgrid::s4u::Exec::init();
+ for (auto start_func : start_func_handlers_)
+ start_func(this);
+ Operation::on_start(this);
+ kernel::actor::simcall_answered([this] {
+ working_ = true;
+ queued_execs_ = std::max(queued_execs_ - 1, 0);
+ });
+ s4u::ExecPtr exec = s4u::Exec::init();
exec->set_name(name_);
exec->set_flops_amount(amount_);
exec->set_host(host_);
exec->start();
exec->extension_set(new ExtendedAttributeActivity());
exec->extension<ExtendedAttributeActivity>()->operation_ = this;
- current_activity_ = exec;
+ kernel::actor::simcall_answered([this, exec] { current_activity_ = exec; });
+}
+
+/** @ingroup plugin_operation
+ * @param host The host to set.
+ * @brief Set a new host.
+ */
+ExecOpPtr ExecOp::set_host(s4u::Host* host)
+{
+ kernel::actor::simcall_answered([this, host] { host_ = host; });
+ return this;
+}
+
+/** @ingroup plugin_operation
+ * @param flops The amount of flops to set.
+ */
+ExecOpPtr ExecOp::set_flops(double flops)
+{
+ kernel::actor::simcall_answered([this, flops] { amount_ = flops; });
+ return this;
}
-CommOp::CommOp(const std::string& name, double bytes, simgrid::s4u::Host* source, simgrid::s4u::Host* destination)
- : Operation(name, bytes), source_(source), destination_(destination)
+/**
+ * @brief Default constructor.
+ */
+CommOp::CommOp(const std::string& name) : Operation(name) {}
+
+/** @ingroup plugin_operation
+ * @brief Smart constructor.
+ */
+CommOpPtr CommOp::init(const std::string& name)
{
+ return CommOpPtr(new CommOp(name));
}
-CommOpPtr CommOp::create(const std::string& name, double bytes, simgrid::s4u::Host* source,
- simgrid::s4u::Host* destination)
+/** @ingroup plugin_operation
+ * @brief Smart constructor.
+ */
+CommOpPtr CommOp::init(const std::string& name, double bytes, s4u::Host* source,
+ s4u::Host* destination)
{
- auto op = CommOpPtr(new CommOp(name, bytes, source, destination));
- return op;
+ return init(name)->set_bytes(bytes)->set_source(source)->set_destination(destination);
}
+/**
+ * @brief Do one execution of the Operation.
+ * @note Call the on_this_start() func. Set its working status as true.
+ * Init and start the underlying Activity.
+ */
void CommOp::execute()
{
- iteration_count_++;
- working_ = true;
- consume();
- simgrid::s4u::CommPtr comm = simgrid::s4u::Comm::sendto_init(source_, destination_);
+ for (auto start_func : start_func_handlers_)
+ start_func(this);
+ Operation::on_start(this);
+ kernel::actor::simcall_answered([this] {
+ working_ = true;
+ queued_execs_ = std::max(queued_execs_ - 1, 0);
+ });
+ s4u::CommPtr comm = s4u::Comm::sendto_init(source_, destination_);
comm->set_name(name_);
comm->set_payload_size(amount_);
comm->start();
comm->extension_set(new ExtendedAttributeActivity());
comm->extension<ExtendedAttributeActivity>()->operation_ = this;
- current_activity_ = comm;
+ kernel::actor::simcall_answered([this, comm] { current_activity_ = comm; });
+}
+
+/** @ingroup plugin_operation
+ * @param source The host to set.
+ * @brief Set a new source host.
+ */
+CommOpPtr CommOp::set_source(s4u::Host* source)
+{
+ kernel::actor::simcall_answered([this, source] { source_ = source; });
+ return this;
+}
+
+/** @ingroup plugin_operation
+ * @param destination The host to set.
+ * @brief Set a new destination host.
+ */
+CommOpPtr CommOp::set_destination(s4u::Host* destination)
+{
+ kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
+ return this;
+}
+
+/** @ingroup plugin_operation
+ * @param bytes The amount of bytes to set.
+ */
+CommOpPtr CommOp::set_bytes(double bytes)
+{
+ kernel::actor::simcall_answered([this, bytes] { amount_ = bytes; });
+ return this;
+}
+
+/**
+ * @brief Default constructor.
+ */
+IoOp::IoOp(const std::string& name) : Operation(name) {}
+
+/** @ingroup plugin_operation
+ * @brief Smart Constructor.
+ */
+IoOpPtr IoOp::init(const std::string& name)
+{
+ return IoOpPtr(new IoOp(name));
+}
+
+/** @ingroup plugin_operation
+ * @brief Smart Constructor.
+ */
+IoOpPtr IoOp::init(const std::string& name, double bytes, s4u::Disk* disk, s4u::Io::OpType type)
+{
+ return init(name)->set_bytes(bytes)->set_disk(disk)->set_op_type(type);
+}
+
+/** @ingroup plugin_operation
+ * @param disk The disk to set.
+ * @brief Set a new disk.
+ */
+IoOpPtr IoOp::set_disk(s4u::Disk* disk)
+{
+ kernel::actor::simcall_answered([this, disk] { disk_ = disk; });
+ return this;
+}
+
+/** @ingroup plugin_operation
+ * @param bytes The amount of bytes to set.
+ */
+IoOpPtr IoOp::set_bytes(double bytes)
+{
+ kernel::actor::simcall_answered([this, bytes] { amount_ = bytes; });
+ return this;
}
+
+/** @ingroup plugin_operation
+ * @param bytes The amount of bytes to set.
+ */
+IoOpPtr IoOp::set_op_type(s4u::Io::OpType type)
+{
+ kernel::actor::simcall_answered([this, type] { type_ = type; });
+ return this;
+}
+
+void IoOp::execute()
+{
+ for (auto start_func : start_func_handlers_)
+ start_func(this);
+ Operation::on_start(this);
+ kernel::actor::simcall_answered([this] {
+ working_ = true;
+ queued_execs_ = std::max(queued_execs_ - 1, 0);
+ });
+ s4u::IoPtr io = s4u::Io::init();
+ io->set_name(name_);
+ io->set_size(amount_);
+ io->set_disk(disk_);
+ io->set_op_type(type_);
+ io->start();
+ io->extension_set(new ExtendedAttributeActivity());
+ io->extension<ExtendedAttributeActivity>()->operation_ = this;
+ kernel::actor::simcall_answered([this, io] { current_activity_ = io; });
+}
+
+
} // namespace simgrid::plugins
simgrid::xbt::Extension<simgrid::s4u::Activity, simgrid::plugins::ExtendedAttributeActivity>