include teshsuite/s4u/listen_async/listen_async.cpp
include teshsuite/s4u/listen_async/listen_async.tesh
include teshsuite/s4u/monkey-masterworkers/monkey-masterworkers.cpp
+include teshsuite/s4u/monkey-masterworkers/monkey-masterworkers.tesh
include teshsuite/s4u/ns3-from-src-to-itself/ns3-from-src-to-itself.cpp
include teshsuite/s4u/ns3-from-src-to-itself/ns3-from-src-to-itself.tesh
include teshsuite/s4u/ns3-simultaneous-send-rcv/ns3-simultaneous-send-rcv.cpp
auto c12 = sg4::Comm::sendto_async(h1, h2, 1.5e7); // Creates and start a direct communication
auto c34 = sg4::Comm::sendto_init(h3, h4); // Creates but do not start another direct communication
- c34->set_remaining(1e7); // Specify the amount of bytes to exchange in this comm
+ c34->set_payload_size(1e7); // Specify the amount of bytes to exchange in this comm
// You can also detach() communications that you never plan to test() or wait().
// Here we create a communication that only slows down the other ones
auto noise = sg4::Comm::sendto_init(h1, h2);
- noise->set_remaining(10000);
+ noise->set_payload_size(10000);
noise->detach();
XBT_INFO("After creation, c12 is %s (remaining: %.2e bytes); c34 is %s (remaining: %.2e bytes)",
/* As usual, you don't have to explicitly start communications that were just init()ed.
The wait() will start it automatically. */
auto c14 = sg4::Comm::sendto_init(h1, h4);
- c14->set_remaining(100)->wait(); // Chaining 2 operations on this new communication
+ c14->set_payload_size(100)->wait(); // Chaining 2 operations on this new communication
}
int main(int argc, char* argv[])
$ ${bindir:=.}/s4u-comm-host2host ${platfdir}/small_platform.xml "--log=root.fmt:[%10.6r]%e(%i:%a@%h)%e%m%n"
> [ 0.000000] (1:sender@Boivin) Send c12 with sendto_async(Tremblay -> Jupiter), and c34 with sendto_init(Fafard -> Ginette)
-> [ 0.000000] (1:sender@Boivin) After creation, c12 is STARTED (remaining: 1.50e+07 bytes); c34 is INITED (remaining: 1.00e+07 bytes)
-> [ 1.000000] (1:sender@Boivin) One sec later, c12 is STARTED (remaining: 8.48e+06 bytes); c34 is INITED (remaining: 1.00e+07 bytes)
+> [ 0.000000] (1:sender@Boivin) After creation, c12 is STARTED (remaining: 1.50e+07 bytes); c34 is STARTING (remaining: 1.00e+07 bytes)
+> [ 1.000000] (1:sender@Boivin) One sec later, c12 is STARTED (remaining: 8.48e+06 bytes); c34 is STARTING (remaining: 1.00e+07 bytes)
> [ 1.000000] (1:sender@Boivin) After c34->start,c12 is STARTED (remaining: 8.48e+06 bytes); c34 is STARTED (remaining: 1.00e+07 bytes)
> [ 2.272621] (1:sender@Boivin) After c12->wait, c12 is FINISHED (remaining: 0.00e+00 bytes); c34 is STARTED (remaining: 5.33e+05 bytes)
> [ 2.343278] (1:sender@Boivin) After c34->wait, c12 is FINISHED (remaining: 0.00e+00 bytes); c34 is FINISHED (remaining: 0.00e+00 bytes)
}
AnyActivity* set_name(const std::string& name)
{
- xbt_assert(get_state() == State::INITED, "Cannot change the name of an activity after its start");
name_ = name;
return static_cast<AnyActivity*>(this);
}
* Represents all asynchronous communications, that you can test or wait onto.
*/
class XBT_PUBLIC Comm : public Activity_T<Comm> {
+ friend Mailbox; // Factory of comms
+ /* specified for normal mailbox-based communications*/
Mailbox* mailbox_ = nullptr;
- kernel::actor::ActorImpl* sender_ = nullptr; /* specified for normal mailbox-based communications*/
+ kernel::actor::ActorImpl* sender_ = nullptr;
kernel::actor::ActorImpl* receiver_ = nullptr;
- Host* from_ = nullptr; /* specified only for direct host-to-host communications */
- Host* to_ = nullptr;
double rate_ = -1;
void* dst_buff_ = nullptr;
size_t dst_buff_size_ = 0;
void* src_buff_ = nullptr;
size_t src_buff_size_ = sizeof(void*);
+
/* FIXME: expose these elements in the API */
bool detached_ = false;
bool (*match_fun_)(void*, void*, kernel::activity::CommImpl*) = nullptr;
Comm() = default;
public:
-#ifndef DOXYGEN
- friend Mailbox; // Factory of comms
-#endif
-
- ~Comm() override;
-
- /*! Creates a communication that bypasses the mailbox mechanism. */
- static CommPtr sendto_init();
- /*! Creates a communication beween the two given hosts, bypassing the mailbox mechanism. */
- static CommPtr sendto_init(Host* from, Host* to);
- /** Do an asynchronous communication between two arbitrary hosts.
- *
- * This initializes a communication that completely bypass the mailbox and actors mechanism.
- * There is really no limit on the hosts involved. In particular, the actor does not have to be on one of the involved
- * hosts.
- */
- static CommPtr sendto_async(Host* from, Host* to, uint64_t simulated_size_in_bytes);
- /** Do a blocking communication between two arbitrary hosts.
- *
- * This starts a blocking communication right away, bypassing the mailbox and actors mechanism.
- * The calling actor is blocked until the end of the communication; there is really no limit on the hosts involved.
- * In particular, the actor does not have to be on one of the involved hosts. Enjoy the comfort of the simulator :)
- */
- static void sendto(Host* from, Host* to, uint64_t simulated_size_in_bytes);
-
- static void on_send_cb(const std::function<void(Comm const&)>& cb) { on_send.connect(cb); }
- static void on_recv_cb(const std::function<void(Comm const&)>& cb) { on_recv.connect(cb); }
- static void on_start_cb(const std::function<void(Comm const&)>& cb) { on_start.connect(cb); }
- static void on_completion_cb(const std::function<void(Activity const&)>& cb) { on_completion.connect(cb); }
+ /* signals and related callbacks */
#ifndef DOXYGEN
/* FIXME signals should be private */
static xbt::signal<void(Comm const&)> on_send;
static xbt::signal<void(Comm const&)> on_completion;
#endif
- /*! take a vector s4u::CommPtr and return when one of them is finished.
- * The return value is the rank of the first finished CommPtr. */
- static ssize_t wait_any(const std::vector<CommPtr>& comms) { return wait_any_for(comms, -1); }
- /*! Same as wait_any, but with a timeout. Return -1 if the timeout occurs.*/
- static ssize_t wait_any_for(const std::vector<CommPtr>& comms, double timeout);
-
- /*! take a vector s4u::CommPtr and return when all of them is finished. */
- static void wait_all(const std::vector<CommPtr>& comms);
- /*! Same as wait_all, but with a timeout. Return the number of terminated comm (less than comms.size() if the timeout
- * occurs). */
- static size_t wait_all_for(const std::vector<CommPtr>& comms, double timeout);
- /*! take a vector s4u::CommPtr and return the rank of the first finished one (or -1 if none is done). */
- static ssize_t test_any(const std::vector<CommPtr>& comms);
-
-#ifndef DOXYGEN
- XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter")
- static int wait_any(const std::vector<CommPtr>* comms) { return static_cast<int>(wait_any_for(*comms, -1)); }
- XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for first parameter")
- static int wait_any_for(const std::vector<CommPtr>* comms, double timeout) { return static_cast<int>(wait_any_for(*comms, timeout)); }
- XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter")
- static void wait_all(const std::vector<CommPtr>* comms) { wait_all(*comms); }
- XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter")
- static int test_any(const std::vector<CommPtr>* comms) { return static_cast<int>(test_any(*comms)); }
-#endif
+ static void on_send_cb(const std::function<void(Comm const&)>& cb) { on_send.connect(cb); }
+ static void on_recv_cb(const std::function<void(Comm const&)>& cb) { on_recv.connect(cb); }
+ static void on_start_cb(const std::function<void(Comm const&)>& cb) { on_start.connect(cb); }
+ static void on_completion_cb(const std::function<void(Activity const&)>& cb) { on_completion.connect(cb); }
+ /* More callbacks */
+ CommPtr set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t));
+ static void copy_buffer_callback(kernel::activity::CommImpl*, void*, size_t);
+ static void copy_pointer_callback(kernel::activity::CommImpl*, void*, size_t);
- Comm* start() override;
- Comm* wait_for(double timeout) override;
+ ~Comm() override;
- /** Start the comm, and ignore its result. It can be completely forgotten after that. */
- Comm* detach();
- /** Start the comm, and ignore its result. It can be completely forgotten after that. */
- Comm* detach(void (*clean_function)(void*))
- {
- clean_fun_ = clean_function;
- return detach();
- }
+ /* "One-sided" communications. This way of communicating bypasses the mailbox and actors mechanism. It creates a
+ * communication (vetoabled, asynchronous, or synchronous) directly between two hosts. There is really no limit on
+ * the hosts involved. In particular, the actor creating such a communication does not have to be on one of the
+ * involved hosts! Enjoy the comfort of the simulator :)
+ */
+ static CommPtr sendto_init(); /* Source and Destination hosts have to be set before the communication can start */
+ static CommPtr sendto_init(Host* from, Host* to);
+ static CommPtr sendto_async(Host* from, Host* to, uint64_t simulated_size_in_bytes);
+ static void sendto(Host* from, Host* to, uint64_t simulated_size_in_bytes);
- /** Set the source and destination of communications that bypass the mailbox mechanism */
CommPtr set_source(Host* from);
- Host* get_source() const { return from_; }
+ Host* get_source() const;
CommPtr set_destination(Host* to);
- Host* get_destination() const { return to_; }
+ Host* get_destination() const;
- /** Sets the maximal communication rate (in byte/sec). Must be done before start */
- CommPtr set_rate(double rate);
+ /* Mailbox-based communications */
+ CommPtr set_mailbox(Mailbox* mailbox);
+ /** Retrieve the mailbox on which this comm acts */
+ Mailbox* get_mailbox() const { return mailbox_; }
/** Specify the data to send.
*
* @endrst
*/
CommPtr set_src_data_size(size_t size);
-
- /** Specify the amount of bytes which exchange should be simulated (not to be mixed with set_src_data_size())
- *
- * @beginrst
- * That's the size of the simulated data, that's completely related from the actual data size (given by
- * :cpp:func:`simgrid::s4u::Comm::set_src_data_size`).
- * @endrst
- */
- CommPtr set_payload_size(uint64_t bytes);
-
/** Specify the data to send and its size (not to be mixed with set_payload_size())
*
* @beginrst
* That's a buffer where the sent data will be copied */
CommPtr set_dst_data(void** buff, size_t size);
/** Retrieve where the data will be copied on the receiver side */
- void* get_dst_data();
-
- /** Retrieve the mailbox on which this comm acts */
- Mailbox* get_mailbox() const;
+ void* get_dst_data() { return dst_buff_; }
/** Retrieve the size of the received data. Not to be mixed with @ref Activity::set_remaining() */
- size_t get_dst_data_size() const;
+ size_t get_dst_data_size() const { return dst_buff_size_; }
+
+ /* Common functions */
+ /** Specify the amount of bytes which exchange should be simulated (not to be mixed with set_src_data_size())
+ *
+ * @beginrst
+ * That's the size of the simulated data, that's completely unrelated from the actual data size (given by
+ * :cpp:func:`simgrid::s4u::Comm::set_src_data_size`).
+ * @endrst
+ */
+ CommPtr set_payload_size(uint64_t bytes);
+ /** Sets the maximal communication rate (in byte/sec). Must be done before start */
+ CommPtr set_rate(double rate);
+
+ bool is_assigned() const override;
Actor* get_sender() const;
- bool is_assigned() const override { return (to_ != nullptr && from_ != nullptr) || (mailbox_ != nullptr); }
+ /* Comm life cycle */
+ Comm* start() override;
+ /** Start the comm, and ignore its result. It can be completely forgotten after that. */
+ Comm* detach();
+ /** Start the comm, and ignore its result. It can be completely forgotten after that. */
+ Comm* detach(void (*clean_function)(void*))
+ {
+ clean_fun_ = clean_function;
+ return detach();
+ }
- CommPtr set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t));
- static void copy_buffer_callback(kernel::activity::CommImpl*, void*, size_t);
- static void copy_pointer_callback(kernel::activity::CommImpl*, void*, size_t);
+ Comm* wait_for(double timeout) override;
+
+ /*! take a vector s4u::CommPtr and return the rank of the first finished one (or -1 if none is done). */
+ static ssize_t test_any(const std::vector<CommPtr>& comms);
+
+ /*! take a vector s4u::CommPtr and return when one of them is finished.
+ * The return value is the rank of the first finished CommPtr. */
+ static ssize_t wait_any(const std::vector<CommPtr>& comms) { return wait_any_for(comms, -1); }
+ /*! Same as wait_any, but with a timeout. Return -1 if the timeout occurs.*/
+ static ssize_t wait_any_for(const std::vector<CommPtr>& comms, double timeout);
+
+ /*! take a vector s4u::CommPtr and return when all of them is finished. */
+ static void wait_all(const std::vector<CommPtr>& comms);
+ /*! Same as wait_all, but with a timeout. Return the number of terminated comm (less than comms.size() if the timeout
+ * occurs). */
+ static size_t wait_all_for(const std::vector<CommPtr>& comms, double timeout);
+
+#ifndef DOXYGEN
+ XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter") static int wait_any(
+ const std::vector<CommPtr>* comms)
+ {
+ return static_cast<int>(wait_any_for(*comms, -1));
+ }
+ XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for first parameter") static int wait_any_for(
+ const std::vector<CommPtr>* comms, double timeout)
+ {
+ return static_cast<int>(wait_any_for(*comms, timeout));
+ }
+ XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter") static void wait_all(
+ const std::vector<CommPtr>* comms)
+ {
+ wait_all(*comms);
+ }
+ XBT_ATTRIB_DEPRECATED_v332("Please use a plain vector for parameter") static int test_any(
+ const std::vector<CommPtr>* comms)
+ {
+ return static_cast<int>(test_any(*comms));
+ }
+#endif
};
} // namespace s4u
} // namespace simgrid
SG_BEGIN_DECL
#ifdef COVERAGE
+
+#if defined(__GNUC__) && __GNUC__ >= 11
+#include "gcov.h"
+#define coverage_checkpoint() \
+ do { \
+ __gcov_dump(); \
+ __gcov_reset(); \
+ } while (0)
+#else
extern void __gcov_flush();
#define coverage_checkpoint() __gcov_flush()
+#endif
+
#else
#define coverage_checkpoint() (void)0
#endif
copy_data_callback_ = callback;
}
+CommImpl& CommImpl::set_type(CommImplType type)
+{
+ type_ = type;
+ return *this;
+}
+
+CommImpl& CommImpl::set_source(s4u::Host* from)
+{
+ from_ = from;
+ return *this;
+}
+
+CommImpl& CommImpl::set_destination(s4u::Host* to)
+{
+ to_ = to;
+ return *this;
+}
+
CommImpl& CommImpl::set_size(double size)
{
size_ = size;
return *this;
}
-CommImpl::CommImpl(s4u::Host* from, s4u::Host* to, double bytes) : size_(bytes), detached_(true), from_(from), to_(to)
-{
- set_state(State::READY);
-}
-
CommImpl::~CommImpl()
{
XBT_DEBUG("Really free communication %p in state %s (detached = %d)", this, get_state_str(), detached_);
XBT_DEBUG("send from mailbox %p", mbox);
/* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
- CommImplPtr this_comm(new CommImpl(CommImpl::Type::SEND));
+ CommImplPtr this_comm(new CommImpl());
+ this_comm->set_type(CommImplType::SEND);
/* Look for communication synchro matching our needs. We also provide a description of
* ourself so that the other side also gets a chance of choosing if it wants to match with us.
*
* If it is not found then push our communication into the rendez-vous point */
CommImplPtr other_comm =
- mbox->find_matching_comm(CommImpl::Type::RECEIVE, observer->get_match_fun(), observer->get_payload(), this_comm,
+ mbox->find_matching_comm(CommImplType::RECEIVE, observer->get_match_fun(), observer->get_payload(), this_comm,
/*done*/ false, /*remove_matching*/ true);
if (not other_comm) {
ActivityImplPtr CommImpl::irecv(actor::CommIrecvSimcall* observer)
{
- CommImplPtr this_synchro(new CommImpl(CommImpl::Type::RECEIVE));
+ CommImplPtr this_synchro(new CommImpl());
+ this_synchro->set_type(CommImplType::RECEIVE);
+
auto* mbox = observer->get_mailbox();
XBT_DEBUG("recv from mbox %p. this_synchro=%p", mbox, this_synchro.get());
if (mbox->is_permanent() && mbox->has_some_done_comm()) {
XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
// find a match in the list of already received comms
- other_comm = mbox->find_matching_comm(CommImpl::Type::SEND, observer->get_match_fun(), observer->get_payload(),
+ other_comm = mbox->find_matching_comm(CommImplType::SEND, observer->get_match_fun(), observer->get_payload(),
this_synchro, /*done*/ true, /*remove_matching*/ true);
// if not found, assume the receiver came first, register it to the mailbox in the classical way
if (not other_comm) {
* ourself so that the other side also gets a chance of choosing if it wants to match with us.
*
* If it is not found then push our communication into the rendez-vous point */
- other_comm = mbox->find_matching_comm(CommImpl::Type::SEND, observer->get_match_fun(), observer->get_payload(),
+ other_comm = mbox->find_matching_comm(CommImplType::SEND, observer->get_match_fun(), observer->get_payload(),
this_synchro, /*done*/ false, /*remove_matching*/ true);
if (other_comm == nullptr) {
namespace kernel {
namespace activity {
+enum class CommImplType { SEND, RECEIVE };
+
class XBT_PUBLIC CommImpl : public ActivityImpl_T<CommImpl> {
~CommImpl() override;
void cleanup_surf();
long mbox_id_ = -1; /* ID of the rendez-vous where the comm was first queued (for MC) */
s4u::Host* from_ = nullptr; /* Pre-determined only for direct host-to-host communications */
s4u::Host* to_ = nullptr; /* Otherwise, computed at start() time from the actors */
+ CommImplType type_ = CommImplType::SEND; /* Type of the communication (SEND or RECEIVE) */
public:
- enum class Type { SEND, RECEIVE };
+ CommImpl() = default;
static void set_copy_data_callback(void (*callback)(CommImpl*, void*, size_t));
- explicit CommImpl(Type type) : type_(type) {}
- CommImpl(s4u::Host* from, s4u::Host* to, double bytes);
-
+ CommImpl& set_type(CommImplType type);
+ CommImplType get_type() const { return type_; }
+ CommImpl& set_source(s4u::Host* from);
+ s4u::Host* get_source() const { return from_; }
+ CommImpl& set_destination(s4u::Host* to);
+ s4u::Host* get_destination() const { return to_; }
CommImpl& set_size(double size);
CommImpl& set_src_buff(unsigned char* buff, size_t size);
CommImpl& set_dst_buff(unsigned char* buff, size_t* size);
MailboxImpl* get_mailbox() const { return mbox_; }
long get_mailbox_id() const { return mbox_id_; }
bool detached() const { return detached_; }
+ bool is_assigned() { return (to_ != nullptr && from_ != nullptr); }
std::vector<s4u::Link*> get_traversed_links() const;
void copy_data();
void set_exception(actor::ActorImpl* issuer) override;
void finish() override;
- const Type type_ = Type::SEND; /* Type of the communication (SEND or RECEIVE) */
#if SIMGRID_HAVE_MC
MailboxImpl* mbox_cpy = nullptr; /* Copy of the rendez-vous where the comm is queued, MC needs it for DPOR
{
XBT_DEBUG("iprobe from %p %p", this, &comm_queue_);
- CommImplPtr this_comm;
- CommImpl::Type smx_type;
+ CommImplPtr this_comm(new CommImpl);
+ CommImplType other_type;
if (type == 1) {
- this_comm = CommImplPtr(new CommImpl(CommImpl::Type::SEND));
- smx_type = CommImpl::Type::RECEIVE;
+ this_comm->set_type(CommImplType::SEND);
+ other_type = CommImplType::RECEIVE;
} else {
- this_comm = CommImplPtr(new CommImpl(CommImpl::Type::RECEIVE));
- smx_type = CommImpl::Type::SEND;
+ this_comm->set_type(CommImplType::RECEIVE);
+ other_type = CommImplType::SEND;
}
CommImplPtr other_comm = nullptr;
if (permanent_receiver_ != nullptr && not done_comm_queue_.empty()) {
XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something");
- other_comm = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false);
+ other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false);
}
if (not other_comm) {
XBT_DEBUG("check if we have more luck in the normal mailbox");
- other_comm = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false);
+ other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false);
}
return other_comm;
* @param remove_matching whether or not to clean the found object from the queue
* @return The communication activity if found, nullptr otherwise
*/
-CommImplPtr MailboxImpl::find_matching_comm(CommImpl::Type type, bool (*match_fun)(void*, void*, CommImpl*),
+CommImplPtr MailboxImpl::find_matching_comm(CommImplType type, bool (*match_fun)(void*, void*, CommImpl*),
void* this_user_data, const CommImplPtr& my_synchro, bool done,
bool remove_matching)
{
auto iter = std::find_if(
comm_queue.begin(), comm_queue.end(), [&type, &match_fun, &this_user_data, &my_synchro](const CommImplPtr& comm) {
- void* other_user_data = (comm->type_ == CommImpl::Type::SEND ? comm->src_data_ : comm->dst_data_);
- return (comm->type_ == type && (not match_fun || match_fun(this_user_data, other_user_data, comm.get())) &&
+ void* other_user_data = (comm->get_type() == CommImplType::SEND ? comm->src_data_ : comm->dst_data_);
+ return (comm->get_type() == type && (not match_fun || match_fun(this_user_data, other_user_data, comm.get())) &&
(not comm->match_fun || comm->match_fun(other_user_data, this_user_data, my_synchro.get())));
});
if (iter == comm_queue.end()) {
void remove(const CommImplPtr& comm);
void clear();
CommImplPtr iprobe(int type, bool (*match_fun)(void*, void*, CommImpl*), void* data);
- CommImplPtr find_matching_comm(CommImpl::Type type, bool (*match_fun)(void*, void*, CommImpl*), void* this_user_data,
+ CommImplPtr find_matching_comm(CommImplType type, bool (*match_fun)(void*, void*, CommImpl*), void* this_user_data,
const CommImplPtr& my_synchro, bool done, bool remove_matching);
bool is_permanent() const { return permanent_receiver_ != nullptr; }
actor::ActorImplPtr get_permanent_receiver() const { return permanent_receiver_; }
namespace simgrid {
namespace mc {
-/** Statically "upcast" a s_smx_actor_t into an ActorInformation
- *
- * This gets 'actorInfo' from '&actorInfo->copy'. It upcasts in the
- * sense that we could achieve the same thing by having ActorInformation
- * inherit from s_smx_actor_t but we don't really want to do that.
- */
-static simgrid::mc::ActorInformation* actor_info_cast(smx_actor_t actor)
-{
- simgrid::mc::ActorInformation temp;
- std::size_t offset = (char*)temp.copy.get_buffer() - (char*)&temp;
-
- auto* process_info = reinterpret_cast<simgrid::mc::ActorInformation*>((char*)actor - offset);
- return process_info;
-}
-
simgrid::mc::Exploration* Api::initialize(char** argv, simgrid::mc::CheckerAlgorithm algo)
{
session_ = std::make_unique<simgrid::mc::Session>([argv] {
simgrid::mc::Snapshot* take_snapshot(long num_state) const;
// SESSION APIs
- simgrid::mc::Session const& get_session() { return *session_; }
+ simgrid::mc::Session const& get_session() const { return *session_; }
void s_close();
// AUTOMATION APIs
member.byte_size = MC_dwarf_attr_integrate_uint(&child, DW_AT_byte_size, 0);
member.type_id = MC_dwarf_at_type(&child);
- xbt_assert(not dwarf_hasattr(&child, DW_AT_data_bit_offset), "Can't groke DW_AT_data_bit_offset.");
+ if (dwarf_hasattr(&child, DW_AT_data_bit_offset)) {
+ XBT_WARN("Can't groke DW_AT_data_bit_offset for %s", name);
+ continue;
+ }
MC_dwarf_fill_member_location(type, &member, &child);
xbt::signal<void(Comm const&)> Comm::on_recv;
xbt::signal<void(Comm const&)> Comm::on_completion;
+CommPtr Comm::set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t))
+{
+ copy_data_function_ = callback;
+ return this;
+}
+
+void Comm::copy_buffer_callback(kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
+{
+ XBT_DEBUG("Copy the data over");
+ memcpy(comm->dst_buff_, buff, buff_size);
+ if (comm->detached()) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the
+ // original buffer available to the application ASAP
+ xbt_free(buff);
+ comm->src_buff_ = nullptr;
+ }
+}
+
+void Comm::copy_pointer_callback(kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
+{
+ xbt_assert((buff_size == sizeof(void*)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
+ *(void**)(comm->dst_buff_) = buff;
+}
+
Comm::~Comm()
{
if (state_ == State::STARTED && not detached_ &&
}
}
-ssize_t Comm::wait_any_for(const std::vector<CommPtr>& comms, double timeout)
+CommPtr Comm::sendto_init()
{
- std::vector<ActivityPtr> activities;
- for (const auto& comm : comms)
- activities.push_back(boost::dynamic_pointer_cast<Activity>(comm));
- ssize_t changed_pos;
- try {
- changed_pos = Activity::wait_any_for(activities, timeout);
- } catch (const NetworkFailureException& e) {
- changed_pos = -1;
- for (auto c : comms) {
- if (c->pimpl_->get_state() == kernel::activity::State::FAILED) {
- c->complete(State::FAILED);
- }
- }
- e.rethrow_nested(XBT_THROW_POINT, boost::core::demangle(typeid(e).name()) + " raised in kernel mode.");
- }
- return changed_pos;
+ CommPtr res(new Comm());
+ res->pimpl_ = kernel::activity::CommImplPtr(new kernel::activity::CommImpl());
+ boost::static_pointer_cast<kernel::activity::CommImpl>(res->pimpl_)->detach();
+ res->sender_ = kernel::actor::ActorImpl::self();
+ return res;
}
-void Comm::wait_all(const std::vector<CommPtr>& comms)
+CommPtr Comm::sendto_init(Host* from, Host* to)
{
- // TODO: this should be a simcall or something
- for (auto& comm : comms)
- comm->wait();
+ auto res = Comm::sendto_init()->set_source(from)->set_destination(to);
+ res->set_state(State::STARTING);
+ return res;
}
-size_t Comm::wait_all_for(const std::vector<CommPtr>& comms, double timeout)
+CommPtr Comm::sendto_async(Host* from, Host* to, uint64_t simulated_size_in_bytes)
{
- if (timeout < 0.0) {
- wait_all(comms);
- return comms.size();
- }
+ return Comm::sendto_init()->set_payload_size(simulated_size_in_bytes)->set_source(from)->set_destination(to);
+}
- double deadline = Engine::get_clock() + timeout;
- std::vector<CommPtr> waited_comm(1, nullptr);
- for (size_t i = 0; i < comms.size(); i++) {
- double wait_timeout = std::max(0.0, deadline - Engine::get_clock());
- waited_comm[0] = comms[i];
- // Using wait_any_for() here (and not wait_for) because we don't want comms to be invalidated on timeout
- if (wait_any_for(waited_comm, wait_timeout) == -1) {
- XBT_DEBUG("Timeout (%g): i = %zu", wait_timeout, i);
- return i;
- }
- }
- return comms.size();
+void Comm::sendto(Host* from, Host* to, uint64_t simulated_size_in_bytes)
+{
+ sendto_async(from, to, simulated_size_in_bytes)->wait();
}
CommPtr Comm::set_source(Host* from)
{
xbt_assert(state_ == State::INITED || state_ == State::STARTING,
"Cannot change the source of a Comm once it's started (state: %s)", to_c_str(state_));
- from_ = from;
- // Setting 'from_' may allow to start the activity, let's try
- vetoable_start();
+ boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->set_source(from);
+ // Setting 'source' may allow to start the activity, let's try
+ if (state_ == State::STARTING && remains_ <= 0)
+ XBT_DEBUG("This communication has a payload size of 0 byte. It cannot start yet");
+ else
+ vetoable_start();
return this;
}
+Host* Comm::get_source() const
+{
+ return pimpl_ ? boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->get_source() : nullptr;
+}
CommPtr Comm::set_destination(Host* to)
{
xbt_assert(state_ == State::INITED || state_ == State::STARTING,
"Cannot change the destination of a Comm once it's started (state: %s)", to_c_str(state_));
- to_ = to;
- // Setting 'to_' may allow to start the activity, let's try
- vetoable_start();
+ boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->set_destination(to);
+ // Setting 'destination' may allow to start the activity, let's try
+ if (state_ == State::STARTING && remains_ <= 0)
+ XBT_DEBUG("This communication has a payload size of 0 byte. It cannot start yet");
+ else
+ vetoable_start();
return this;
}
+Host* Comm::get_destination() const
+{
+ return pimpl_ ? boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->get_destination() : nullptr;
+}
+
CommPtr Comm::set_rate(double rate)
{
xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
return this;
}
+CommPtr Comm::set_mailbox(Mailbox* mailbox)
+{
+ xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
+ __FUNCTION__);
+ mailbox_ = mailbox;
+ return this;
+}
+
CommPtr Comm::set_src_data(void* buff)
{
xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
dst_buff_ = buff;
return this;
}
-void* Comm::get_dst_data()
-{
- return dst_buff_;
-}
-size_t Comm::get_dst_data_size() const
-{
- return dst_buff_size_;
-}
CommPtr Comm::set_dst_data(void** buff, size_t size)
{
xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
dst_buff_size_ = size;
return this;
}
+
CommPtr Comm::set_payload_size(uint64_t bytes)
{
Activity::set_remaining(bytes);
+ if (pimpl_) {
+ boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->set_size(bytes);
+ }
return this;
}
-CommPtr Comm::sendto_init()
-{
- CommPtr res(new Comm());
- res->sender_ = kernel::actor::ActorImpl::self();
- return res;
-}
-
-CommPtr Comm::sendto_init(Host* from, Host* to)
-{
- auto res = Comm::sendto_init();
- res->from_ = from;
- res->to_ = to;
-
- return res;
-}
-
-CommPtr Comm::sendto_async(Host* from, Host* to, uint64_t simulated_size_in_bytes)
+Actor* Comm::get_sender() const
{
- auto res = Comm::sendto_init(from, to)->set_payload_size(simulated_size_in_bytes);
- res->vetoable_start();
- return res;
+ kernel::actor::ActorImplPtr sender = nullptr;
+ if (pimpl_)
+ sender = boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->src_actor_;
+ return sender ? sender->get_ciface() : nullptr;
}
-void Comm::sendto(Host* from, Host* to, uint64_t simulated_size_in_bytes)
+bool Comm::is_assigned() const
{
- sendto_async(from, to, simulated_size_in_bytes)->wait();
+ return (pimpl_ && boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->is_assigned()) ||
+ mailbox_ != nullptr;
}
Comm* Comm::start()
{
xbt_assert(get_state() == State::INITED || get_state() == State::STARTING,
"You cannot use %s() once your communication started (not implemented)", __FUNCTION__);
- if (from_ != nullptr || to_ != nullptr) {
- xbt_assert(from_ != nullptr && to_ != nullptr, "When either from_ or to_ is specified, both must be.");
+ if (get_source() != nullptr || get_destination() != nullptr) {
+ xbt_assert(is_assigned(), "When either from_ or to_ is specified, both must be.");
xbt_assert(src_buff_ == nullptr && dst_buff_ == nullptr,
"Direct host-to-host communications cannot carry any data.");
- pimpl_ = kernel::actor::simcall_answered([this] {
- kernel::activity::CommImplPtr res(new kernel::activity::CommImpl(this->from_, this->to_, this->get_remaining()));
- res->start();
- return res;
+ XBT_DEBUG("host-to-host Comm. Pimpl already created and set, just start it.");
+ kernel::actor::simcall_answered([this] {
+ pimpl_->set_state(kernel::activity::State::READY);
+ boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->start();
});
-
} else if (src_buff_ != nullptr) { // Sender side
on_send(*this);
kernel::actor::CommIsendSimcall observer{sender_,
return this;
}
+Comm* Comm::detach()
+{
+ xbt_assert(state_ == State::INITED || state_ == State::STARTING,
+ "You cannot use %s() once your communication is %s (not implemented)", __FUNCTION__, get_state_str());
+ xbt_assert(dst_buff_ == nullptr && dst_buff_size_ == 0, "You can only detach sends, not recvs");
+ detached_ = true;
+ vetoable_start();
+ return this;
+}
+
+ssize_t Comm::test_any(const std::vector<CommPtr>& comms)
+{
+ std::vector<ActivityPtr> activities;
+ for (const auto& comm : comms)
+ activities.push_back(boost::dynamic_pointer_cast<Activity>(comm));
+ return Activity::test_any(activities);
+}
+
/** @brief Block the calling actor until the communication is finished, or until timeout
*
* On timeout, an exception is thrown and the communication is invalidated.
throw NetworkFailureException(XBT_THROW_POINT, "Cannot wait for a failed communication");
case State::INITED:
case State::STARTING: // It's not started yet. Do it in one simcall if it's a regular communication
- if (from_ != nullptr || to_ != nullptr) {
+ if (get_source() != nullptr || get_destination() != nullptr) {
return vetoable_start()->wait_for(timeout); // In the case of host2host comm, do it in two simcalls
} else if (src_buff_ != nullptr) {
on_send(*this);
return this;
}
-ssize_t Comm::test_any(const std::vector<CommPtr>& comms)
+ssize_t Comm::wait_any_for(const std::vector<CommPtr>& comms, double timeout)
{
std::vector<ActivityPtr> activities;
for (const auto& comm : comms)
activities.push_back(boost::dynamic_pointer_cast<Activity>(comm));
- return Activity::test_any(activities);
-}
-
-Comm* Comm::detach()
-{
- xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication is %s (not implemented)",
- __FUNCTION__, get_state_str());
- xbt_assert(dst_buff_ == nullptr && dst_buff_size_ == 0, "You can only detach sends, not recvs");
- detached_ = true;
- vetoable_start();
- return this;
-}
-
-Mailbox* Comm::get_mailbox() const
-{
- return mailbox_;
+ ssize_t changed_pos;
+ try {
+ changed_pos = Activity::wait_any_for(activities, timeout);
+ } catch (const NetworkFailureException& e) {
+ changed_pos = -1;
+ for (auto c : comms) {
+ if (c->pimpl_->get_state() == kernel::activity::State::FAILED) {
+ c->complete(State::FAILED);
+ }
+ }
+ e.rethrow_nested(XBT_THROW_POINT, boost::core::demangle(typeid(e).name()) + " raised in kernel mode.");
+ }
+ return changed_pos;
}
-Actor* Comm::get_sender() const
+void Comm::wait_all(const std::vector<CommPtr>& comms)
{
- kernel::actor::ActorImplPtr sender = nullptr;
- if (pimpl_)
- sender = boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->src_actor_;
- return sender ? sender->get_ciface() : nullptr;
+ // TODO: this should be a simcall or something
+ for (auto& comm : comms)
+ comm->wait();
}
-CommPtr Comm::set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t))
-{
- copy_data_function_ = callback;
- return this;
-}
-void Comm::copy_buffer_callback(kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
+size_t Comm::wait_all_for(const std::vector<CommPtr>& comms, double timeout)
{
- XBT_DEBUG("Copy the data over");
- memcpy(comm->dst_buff_, buff, buff_size);
- if (comm->detached()) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the
- // original buffer available to the application ASAP
- xbt_free(buff);
- comm->src_buff_ = nullptr;
+ if (timeout < 0.0) {
+ wait_all(comms);
+ return comms.size();
}
-}
-void Comm::copy_pointer_callback(kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
-{
- xbt_assert((buff_size == sizeof(void*)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
- *(void**)(comm->dst_buff_) = buff;
+ double deadline = Engine::get_clock() + timeout;
+ std::vector<CommPtr> waited_comm(1, nullptr);
+ for (size_t i = 0; i < comms.size(); i++) {
+ double wait_timeout = std::max(0.0, deadline - Engine::get_clock());
+ waited_comm[0] = comms[i];
+ // Using wait_any_for() here (and not wait_for) because we don't want comms to be invalidated on timeout
+ if (wait_any_for(waited_comm, wait_timeout) == -1) {
+ XBT_DEBUG("Timeout (%g): i = %zu", wait_timeout, i);
+ return i;
+ }
+ }
+ return comms.size();
}
-
} // namespace s4u
} // namespace simgrid
/* **************************** Public C interface *************************** */
{
CommPtr res(new Comm());
res->sender_ = kernel::actor::ActorImpl::self();
- res->mailbox_ = this;
+ res->set_mailbox(this);
return res;
}
CommPtr Mailbox::get_init()
{
- CommPtr res(new Comm());
+ auto res = CommPtr(new Comm())->set_mailbox(this);
res->receiver_ = kernel::actor::ActorImpl::self();
- res->mailbox_ = this;
return res;
}
cd "$WORKSPACE"
#convert all gcov reports to xml cobertura reports
- gcovr -r . --xml-pretty -e teshsuite -e examples/smpi/NAS -e examples/smpi/mc -u -o "$BUILDFOLDER"/xml_coverage.xml
+ gcovr -g -r . --xml-pretty -e teshsuite -e examples/smpi/NAS -e examples/smpi/mc -u -o "$BUILDFOLDER"/xml_coverage.xml
xsltproc "$WORKSPACE"/tools/jenkins/ctest2junit.xsl build/Testing/"$( head -n 1 < build/Testing/TAG )"/Test.xml > CTestResults_memcheck.xml
#generate sloccount report
