#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"
Operations are designed to represent workflows, i.e, graphs of Operations.
Operations can only be instancied using either
-:cpp:func:`simgrid::plugins::ExecOp::create` or :cpp:func:`simgrid::plugins::CommOp::create`
+: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>`.
xbt::signal<void(Operation*)> Operation::on_start;
xbt::signal<void(Operation*)> Operation::on_end;
-Operation::Operation(const std::string& name, double amount) : name_(name), amount_(amount) {}
-
-std::string Operation::get_name()
-{
- return name_;
-}
+Operation::Operation(const std::string& name) : name_(name) {}
/**
* @param predecessor The Operation to add.
*/
bool Operation::ready_to_run() const
{
- if (working_ or queued_execs_ <= 0)
- return false;
- else
- return true;
+ return not working_ && queued_execs_ > 0;
}
/**
working_ = false;
count_++;
});
- end_func_(this);
+ for (auto end_func : end_func_handlers_)
+ end_func(this);
Operation::on_end(this);
for (auto const& op : successors_)
op->receive(this);
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::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();
});
}
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);
+ }
+ });
+}
+
/** @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(std::function<void(Operation*)> func)
+void Operation::on_this_start(const std::function<void(Operation*)>& func)
{
- simgrid::kernel::actor::simcall_answered([this, func] { start_func_ = func; });
+ simgrid::kernel::actor::simcall_answered([this, &func] { start_func_handlers_.push_back(func); });
}
/** @ingroup plugin_operation
* @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(std::function<void(Operation*)> func)
+void Operation::on_this_end(const std::function<void(Operation*)>& func)
{
- simgrid::kernel::actor::simcall_answered([this, func] { end_func_ = func; });
+ simgrid::kernel::actor::simcall_answered([this, &func] { end_func_handlers_.push_back(func); });
}
/** @ingroup plugin_operation
* @brief Return the number of completed executions.
*/
-int Operation::get_count()
+int Operation::get_count() const
{
return count_;
}
/**
* @brief Default constructor.
*/
-ExecOp::ExecOp(const std::string& name, double flops, simgrid::s4u::Host* host) : Operation(name, flops), host_(host) {}
+ExecOp::ExecOp(const std::string& name) : Operation(name) {}
/** @ingroup plugin_operation
* @brief Smart Constructor.
*/
-ExecOpPtr ExecOp::create(const std::string& name, double flops, simgrid::s4u::Host* host)
+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.
- * Create and start the underlying Activity.
+ * Init and start the underlying Activity.
*/
void ExecOp::execute()
{
- start_func_(this);
+ for (auto start_func : start_func_handlers_)
+ start_func(this);
Operation::on_start(this);
- simgrid::kernel::actor::simcall_answered([this] {
+ kernel::actor::simcall_answered([this] {
working_ = true;
queued_execs_ = std::max(queued_execs_ - 1, 0);
});
- simgrid::s4u::ExecPtr exec = simgrid::s4u::Exec::init();
+ 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;
- simgrid::kernel::actor::simcall_answered([this, exec] { 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.
*/
-void ExecOp::set_host(simgrid::s4u::Host* host)
+ExecOpPtr ExecOp::set_host(s4u::Host* host)
{
- simgrid::kernel::actor::simcall_answered([this, host] { 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;
}
/**
* @brief Default constructor.
*/
-CommOp::CommOp(const std::string& name, double bytes, simgrid::s4u::Host* source, simgrid::s4u::Host* destination)
- : Operation(name, bytes), source_(source), destination_(destination)
+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));
}
/** @ingroup plugin_operation
* @brief Smart constructor.
*/
-CommOpPtr CommOp::create(const std::string& name, double bytes, simgrid::s4u::Host* source,
- simgrid::s4u::Host* destination)
+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.
- * Create and start the underlying Activity.
+ * Init and start the underlying Activity.
*/
void CommOp::execute()
{
- start_func_(this);
+ for (auto start_func : start_func_handlers_)
+ start_func(this);
Operation::on_start(this);
- simgrid::kernel::actor::simcall_answered([this] {
+ kernel::actor::simcall_answered([this] {
working_ = true;
queued_execs_ = std::max(queued_execs_ - 1, 0);
});
- simgrid::s4u::CommPtr comm = simgrid::s4u::Comm::sendto_init(source_, destination_);
+ 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;
- simgrid::kernel::actor::simcall_answered([this, comm] { 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.
*/
-void CommOp::set_source(simgrid::s4u::Host* source)
+CommOpPtr CommOp::set_source(s4u::Host* source)
{
- simgrid::kernel::actor::simcall_answered([this, source] { source_ = 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.
*/
-void CommOp::set_destination(simgrid::s4u::Host* destination)
+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)
{
- simgrid::kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
+ 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>
