foreach (example actions-comm actions-storage actor-create actor-daemon actor-kill actor-migration actor-suspend
- app-masterworker app-pingpong app-token-ring plugin-hostload io mutex )
+ app-masterworker app-pingpong app-token-ring plugin-hostload io mutex async-waitany async-waitall)
add_executable (s4u_${example} ${example}/s4u_${example}.cpp)
target_link_libraries(s4u_${example} simgrid)
set_target_properties(s4u_${example} PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/${example})
set(examples_src ${examples_src} PARENT_SCOPE)
set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/s4u_app-bittorrent.tesh
- ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.tesh PARENT_SCOPE)
+ ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.tesh
+ ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.tesh
+ ${CMAKE_CURRENT_SOURCE_DIR}/async-waitany/s4u_async-waitany.tesh
+ ${CMAKE_CURRENT_SOURCE_DIR}/async-waitall/s4u_async-waitall.tesh PARENT_SCOPE)
set(xml_files ${xml_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_split_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-storage/s4u_actions-storage_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actor-create/s4u_actor-create_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/s4u_app-bittorrent_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/app-masterworker/s4u_app-masterworker_d.xml
+ ${CMAKE_CURRENT_SOURCE_DIR}/async-waitany/s4u_async-waitany_d.xml
+ ${CMAKE_CURRENT_SOURCE_DIR}/async-waitany/s4u_async-waitall_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord_d.xml PARENT_SCOPE)
set(txt_files ${txt_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_split_p0.txt
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_split_p1.txt
${CMAKE_CURRENT_SOURCE_DIR}/README.doc PARENT_SCOPE)
foreach(example actions-comm actions-storage actor-create actor-daemon actor-kill actor-migration actor-suspend
- app-bittorrent app-masterworker app-pingpong app-token-ring dht-chord plugin-hostload io mutex )
+ app-bittorrent app-masterworker app-pingpong app-token-ring dht-chord plugin-hostload io mutex async-waitall async-waitany )
ADD_TESH_FACTORIES(s4u-${example} "thread;ucontext;raw;boost" --setenv bindir=${CMAKE_CURRENT_BINARY_DIR}/${example} --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/examples/s4u/${example} s4u_${example}.tesh)
endforeach()
@brief Find the S4U example fitting your needs in the archive.
- @ref s4u_ex_basics
+ - @ref s4u_ex_async
- @ref s4u_ex_actors
- @ref s4u_ex_synchro
- @ref s4u_ex_actions
- <b>Master Workers:</b> @ref examples/s4u/app-masterworker/s4u_app-masterworker.cpp \n
Another good old example, where one Master process has a bunch of task to dispatch to a set of several Worker
processes.
+
+@section msg_ex_async Asynchronous communications
+
+ - <b>Waiting for all communications in a set</b>.
+ @ref examples/s4u/async-waitall/s4u_async-waitall.cpp\n
+ The @ref S4U_comm_waitall function is useful when you want to block
+ until all activities in a given set have completed.
+
+ - <b>Waiting for the first completed communication in a set</b>.
+ @ref examples/s4u/async-waitany/s4u_async-waitany.cpp\n
+ The @ref S4U_comm_waitany function is useful when you want to block
+ until one activity of the set completes, no matter which terminates
+ first.
@section s4u_ex_actors Acting on Actors
--- /dev/null
+/* Copyright (c) 2010-2016. The SimGrid Team. All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+ #include "simgrid/s4u.hpp"
+ #include "xbt/str.h"
+ #include <cstdlib>
+ #include <iostream>
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(msg_async_waitall, "Messages specific for this msg example");
+
+class sender {
+ long number_of_tasks = 0; /* - Number of tasks */
+ long receivers_count = 0; /* - Number of workers */
+
+public:
+ explicit sender(std::vector<std::string> args)
+{
+ xbt_assert(args.size()== 4, "This function expects 5 parameters from the XML deployment file");
+ number_of_tasks = std::stol(args[0]);
+ double task_comp_size = std::stod(args[1]);
+ double task_comm_size = std::stod(args[2]);
+ receivers_count = std::stol(args[3]);
+
+}
+void operator()()
+{
+ simgrid::s4u::MailboxPtr mbox = simgrid::s4u::Mailbox::byName("sender_mailbox");
+ simgrid::s4u::CommPtr* comms = new simgrid::s4u::CommPtr[number_of_tasks + receivers_count] ;
+
+ for (int i = 0; i < number_of_tasks; i++) {
+ char mailbox[80];
+ char taskname[80];
+ snprintf(mailbox,79, "receiver-%ld", i % receivers_count);
+ snprintf(taskname,79, "Task_%d", i);
+ comms[i] = mbox->put_async((void*)taskname, 42);
+ XBT_INFO("Send to receiver-%ld Task_%d", i % receivers_count, i);
+ }
+ for (int i = 0; i < receivers_count; i++) {
+ char mailbox[80];
+ snprintf(mailbox,79, "receiver-%ld", i % receivers_count);
+ comms[i + number_of_tasks] = mbox->put_async((void*)"finalize", 42);
+ XBT_INFO("Send to receiver-%ld finalize", i % receivers_count);
+ }
+
+ /* Here we are waiting for the completion of all communications */
+ for (int i = 0; i < number_of_tasks + receivers_count; i++)
+ comms[i]->wait();
+
+ delete [] comms;
+ XBT_INFO("Goodbye now!");
+}
+};
+
+class receiver {
+public:
+ explicit receiver(std::vector<std::string> args)
+{
+ xbt_assert(args.size() == 1,"This function expects 1 parameter from the XML deployment file");
+ int id = xbt_str_parse_int(args[0].c_str(), "Any process of this example must have a numerical name, not %s");
+ char mailbox[80];
+ snprintf(mailbox,79, "receiver-%d", id);
+
+ simgrid::s4u::this_actor::sleep_for(10.0);
+
+ XBT_INFO("I'm done. See you!");
+}
+void operator()()
+{
+ simgrid::s4u::MailboxPtr mbox = simgrid::s4u::Mailbox::byName("receiver_mailbox");
+ while (1) {
+ XBT_INFO("Wait to receive a task");
+ void *received = NULL;
+ simgrid::s4u::CommPtr comm = mbox->get_async(&received);
+ comm->wait();
+ std::string* receivedStr = static_cast<std::string*>(received);
+ if (receivedStr->compare("finalize") == 0) {
+ delete receivedStr;
+ break;
+ }
+ double *comp_size = static_cast<double*>(received);
+ /* - Otherwise, process the task */
+ simgrid::s4u::this_actor::execute(*comp_size);
+ }
+ XBT_INFO("I'm done. See you!");
+}
+};
+
+int main(int argc, char *argv[])
+{
+ simgrid::s4u::Engine *e = new simgrid::s4u::Engine(&argc,argv);
+
+ xbt_assert(argc > 2, "Usage: %s platform_file deployment_file\n"
+ "\tExample: %s msg_platform.xml msg_deployment.xml\n", argv[0], argv[0]);
+
+ e->registerFunction<sender>("sender");
+ e->registerFunction<receiver>("receiver");
+
+ e->loadDeployment(argv[2]);
+ e->run();
+
+ return 0;
+}
--- /dev/null
+#! ./tesh
+p Test1 MSG_comm_waitall() for sender
+
+! output sort 19
+$ $SG_TEST_EXENV ${bindir:=.}/async-waitall ${srcdir:=.}/small_platform_fatpipe.xml ${srcdir:=.}/../msg/async-waitall/async-waitall_d.xml "--log=root.fmt:[%10.6r]%e(%i:%P@%h)%e%m%n"
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_0
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_1
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_2
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 finalize
+> [ 10.000000] (2:receiver@Ruby) Wait to receive a task
+> [ 10.004022] (2:receiver@Ruby) Received "Task_0"
+> [ 10.004022] (2:receiver@Ruby) Processing "Task_0"
+> [ 10.513732] (2:receiver@Ruby) "Task_0" done
+> [ 10.513732] (2:receiver@Ruby) Wait to receive a task
+> [ 10.517753] (2:receiver@Ruby) Received "Task_1"
+> [ 10.517753] (2:receiver@Ruby) Processing "Task_1"
+> [ 11.027463] (2:receiver@Ruby) "Task_1" done
+> [ 11.027463] (2:receiver@Ruby) Wait to receive a task
+> [ 11.031485] (2:receiver@Ruby) Received "Task_2"
+> [ 11.031485] (2:receiver@Ruby) Processing "Task_2"
+> [ 11.541195] (2:receiver@Ruby) "Task_2" done
+> [ 11.541195] (2:receiver@Ruby) Wait to receive a task
+> [ 11.543146] (0:maestro@) Simulation time 11.5431
+> [ 11.543146] (1:sender@Tremblay) Goodbye now!
+> [ 11.543146] (2:receiver@Ruby) Received "finalize"
+> [ 11.543146] (2:receiver@Ruby) I'm done. See you!
--- /dev/null
+<?xml version='1.0'?>
+<!DOCTYPE platform SYSTEM "http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd">
+<platform version="4.1">
+ <!-- The master actor (with some arguments) -->
+ <actor host="Tremblay" function="sender">
+ <argument value="3"/> <!-- Number of tasks -->
+ <argument value="50000000"/> <!-- Computation size of tasks -->
+ <argument value="1000000"/> <!-- Communication size of tasks -->
+ <argument value="1"/> <!-- Number of receivers -->
+ </actor>
+ <!-- The receiver processes -->
+ <actor host="Ruby" function="receiver">
+ <argument value="0"/>
+ </actor>
+</platform>
--- /dev/null
+/* Copyright (c) 2010-2016. The SimGrid Team. All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include "simgrid/s4u.hpp"
+#include "xbt/str.h"
+#include <cstdlib>
+#include <iostream>
+#include <vector>
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(msg_async_waitany, "Messages specific for this msg example");
+
+class sender {
+ long number_of_tasks = 0; /* - Number of tasks */
+ long receivers_count = 0; /* - Number of workers */
+ int diff_com = 0;
+
+public:
+ explicit sender(std::vector<std::string> args)
+{
+ xbt_assert(args.size()== 5, "This function expects 5 parameters from the XML deployment file");
+ number_of_tasks = std::stol(args[0]);
+ double task_comp_size = std::stod(args[1]);
+ double task_comm_size = std::stod(args[2]);
+ receivers_count = std::stol(args[3]);
+ diff_com = std::stoi(args[4]);
+
+}
+void operator()()
+{
+ std::vector<simgrid::s4u::CommPtr> comms;
+ simgrid::s4u::CommPtr comm;
+ simgrid::s4u::MailboxPtr mbox = simgrid::s4u::Mailbox::byName("receiver_mailbox");
+ /* First pack the communications in the dynar */
+ for (int i = 0; i < number_of_tasks; i++) {
+ double coef = (diff_com == 0) ? 1 : (i + 1);
+ char mailbox[80];
+ char taskname[80];
+ snprintf(mailbox,79, "receiver-%ld", (i % receivers_count));
+ snprintf(taskname,79, "Task_%d", i);
+ comm = mbox->put_async((void*)taskname, 42.0);
+ comms.push_back(comm);
+ }
+
+ /* Here we are waiting for the completion of all communications */
+ while (!comms.empty()) {
+ comm=comms.back();
+ comms.pop_back();
+ comm->wait();
+ }
+ comms.clear();
+ comm = nullptr;
+
+ /* Here we are waiting for the completion of all tasks */
+ for (int i = 0; i < receivers_count; i++) {
+ void* received = nullptr;
+ simgrid::s4u::CommPtr comm = mbox->get_async(&received);
+ comm->wait();
+ comm = nullptr;
+ }
+ XBT_INFO("Goodbye now!");
+}
+};
+
+class receiver {
+ int id = 0;
+ int task_amount = 0;
+public:
+ explicit receiver(std::vector<std::string> args)
+{
+ xbt_assert(args.size() == 2, "This function expects 2 parameters from the XML deployment file");
+ id = std::stoi(args[0]);
+ task_amount = std::stoi(args[1]);
+}
+void operator()()
+{
+ void *received;
+ std::vector<simgrid::s4u::CommPtr> comms;
+ simgrid::s4u::CommPtr comm;
+ simgrid::s4u::MailboxPtr mbox;
+
+ simgrid::s4u::this_actor::sleep_for(10.0);
+ for (int i = 0; i < task_amount; i++) {
+ XBT_INFO("Wait to receive task %d", i);
+ received = NULL;
+ comm = mbox->get_async(&received);
+ comms.push_back(comm);
+ }
+
+ /* Here we are waiting for the receiving of all communications */
+ while (!comms.empty()) {
+ // returns the rank of the comm that just ended. Remove it.
+ comm=comms.back();
+ comms.pop_back();
+ comm->wait();
+ }
+ comms.clear();
+ comm = nullptr;
+ /* Here we tell to sender that all tasks are done */
+ simgrid::s4u::Mailbox::byName("finalize")->put(nullptr, 1);
+ XBT_INFO("I'm done. See you!");
+}
+};
+
+int main(int argc, char *argv[])
+{
+ simgrid::s4u::Engine *e = new simgrid::s4u::Engine(&argc,argv);
+
+ e->registerFunction<sender>("sender");
+ e->registerFunction<receiver>("receiver");
+
+ e->loadDeployment(argv[2]);
+ e->run();
+
+ delete e;
+ return 0;
+}
--- /dev/null
+#! ./tesh
+
+p Testing the MSG_comm_waitany function
+
+! output sort 19
+$ $SG_TEST_EXENV ${bindir:=.}/async-waitany ${srcdir:=.}/small_platform.xml ${srcdir:=.}/../msg/async-waitany/async-waitany_d.xml "--log=root.fmt:[%10.6r]%e(%i:%P@%h)%e%m%n"
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_0 comm_size 1000000.000000
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-1 Task_1 comm_size 1000000.000000
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_2 comm_size 1000000.000000
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-1 Task_3 comm_size 1000000.000000
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-0 Task_4 comm_size 1000000.000000
+> [ 0.000000] (1:sender@Tremblay) Send to receiver-1 Task_5 comm_size 1000000.000000
+> [ 10.000000] (2:receiver@Fafard) Wait to receive task 0
+> [ 10.000000] (2:receiver@Fafard) Wait to receive task 1
+> [ 10.000000] (2:receiver@Fafard) Wait to receive task 2
+> [ 10.000000] (3:receiver@Jupiter) Wait to receive task 0
+> [ 10.000000] (3:receiver@Jupiter) Wait to receive task 1
+> [ 10.000000] (3:receiver@Jupiter) Wait to receive task 2
+> [ 10.423774] (2:receiver@Fafard) Processing "Task_4"
+> [ 10.469435] (3:receiver@Jupiter) Processing "Task_3"
+> [ 11.079116] (2:receiver@Fafard) "Task_4" done
+> [ 11.079116] (2:receiver@Fafard) Processing "Task_0"
+> [ 11.124778] (3:receiver@Jupiter) "Task_3" done
+> [ 11.124778] (3:receiver@Jupiter) Processing "Task_1"
+> [ 11.734459] (2:receiver@Fafard) "Task_0" done
+> [ 11.734459] (2:receiver@Fafard) Processing "Task_2"
+> [ 11.780120] (3:receiver@Jupiter) "Task_1" done
+> [ 11.780120] (3:receiver@Jupiter) Processing "Task_5"
+> [ 12.389801] (2:receiver@Fafard) "Task_2" done
+> [ 12.415509] (2:receiver@Fafard) I'm done. See you!
+> [ 12.435462] (3:receiver@Jupiter) "Task_5" done
+> [ 12.454477] (0:maestro@) Simulation time 12.4545
+> [ 12.454477] (1:sender@Tremblay) Goodbye now!
+> [ 12.454477] (3:receiver@Jupiter) I'm done. See you!
--- /dev/null
+<?xml version='1.0'?>
+<!DOCTYPE platform SYSTEM "http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd">
+<platform version="4.1">
+ <!-- The master actor (with some arguments) -->
+ <actor host="Tremblay" function="sender">
+ <argument value="6"/> <!-- Number of tasks -->
+ <argument value="50000000"/> <!-- Computation size of tasks -->
+ <argument value="1000000"/> <!-- Communication size of tasks -->
+ <argument value="2"/> <!-- Number of receivers -->
+ <argument value="0"/> <!-- Use diff comm size -->
+ </actor>
+ <!-- The receiver processes -->
+ <actor host="Fafard" function="receiver">
+ <argument value="0"/>
+ <argument value="3"/> <!-- Number of tasks -->
+ </actor>
+ <actor host="Jupiter" function="receiver">
+ <argument value="1"/>
+ <argument value="3"/> <!-- Number of tasks -->
+ </actor>
+</platform>
