#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"
Operation::Operation(const std::string& name) : name_(name) {}
-std::string Operation::get_name()
-{
- return name_;
-}
-
/**
* @param predecessor The Operation to add.
* @brief Add a predecessor to this Operation.
working_ = false;
count_++;
});
- if (end_func_)
- 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_;
}
*/
ExecOpPtr ExecOp::init(const std::string& name)
{
- auto op = ExecOpPtr(new ExecOp(name));
- return op;
+ return ExecOpPtr(new ExecOp(name));
}
/** @ingroup plugin_operation
* @brief Smart Constructor.
*/
-ExecOpPtr ExecOp::init(const std::string& name, double flops, simgrid::s4u::Host* host)
+ExecOpPtr ExecOp::init(const std::string& name, double flops, s4u::Host* host)
{
- auto op = ExecOpPtr(new ExecOp(name));
- op->set_flops(flops);
- op->set_host(host);
- return op;
+ return init(name)->set_flops(flops)->set_host(host);
}
/**
*/
void ExecOp::execute()
{
- if (start_func_)
- 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.
*/
-void ExecOp::set_flops(double flops)
+ExecOpPtr ExecOp::set_flops(double flops)
{
- simgrid::kernel::actor::simcall_answered([this, flops] { amount_ = flops; });
+ kernel::actor::simcall_answered([this, flops] { amount_ = flops; });
+ return this;
}
/**
*/
CommOpPtr CommOp::init(const std::string& name)
{
- auto op = CommOpPtr(new CommOp(name));
- return op;
+ return CommOpPtr(new CommOp(name));
}
/** @ingroup plugin_operation
* @brief Smart constructor.
*/
-CommOpPtr CommOp::init(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));
- op->set_bytes(bytes);
- op->set_source(source);
- op->set_destination(destination);
- return op;
+ return init(name)->set_bytes(bytes)->set_source(source)->set_destination(destination);
}
/**
*/
void CommOp::execute()
{
- if (start_func_)
- 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)
{
- simgrid::kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
+ kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
+ return this;
}
/** @ingroup plugin_operation
* @param bytes The amount of bytes to set.
*/
-void CommOp::set_bytes(double bytes)
+CommOpPtr CommOp::set_bytes(double bytes)
{
- simgrid::kernel::actor::simcall_answered([this, bytes] { amount_ = 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 */
+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>