From: navarrop <navarrop@48e7efb5-ca39-0410-a469-dd3cf9ba447f>
Date: Fri, 30 Jul 2010 13:29:30 +0000 (+0000)
Subject: Add an example for waitany fct.
X-Git-Tag: v3_5~742
X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/commitdiff_plain/c6eefaa5a3198ebff98651ac72501367e4195ded?hp=d4256cb7fe003cad9b262af0143313a70f4f0ff7

Add an example for waitany fct.

git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@8079 48e7efb5-ca39-0410-a469-dd3cf9ba447f
---

diff --git a/examples/msg/icomms/CMakeLists.txt b/examples/msg/icomms/CMakeLists.txt
index 2ed45e89e0..870bfb40ac 100755
--- a/examples/msg/icomms/CMakeLists.txt
+++ b/examples/msg/icomms/CMakeLists.txt
@@ -6,4 +6,7 @@ add_executable(peer peer.c)
 target_link_libraries(peer simgrid)
 
 add_executable(peer2 peer2.c)
-target_link_libraries(peer2 simgrid)
\ No newline at end of file
+target_link_libraries(peer2 simgrid)
+
+add_executable(peer3 peer3.c)
+target_link_libraries(peer3 simgrid)
\ No newline at end of file
diff --git a/examples/msg/icomms/deployment_peer.xml b/examples/msg/icomms/deployment_peer.xml
index b4415187d9..2e5004439c 100755
--- a/examples/msg/icomms/deployment_peer.xml
+++ b/examples/msg/icomms/deployment_peer.xml
@@ -11,5 +11,6 @@
   <!-- The receiver processes -->
   <process host="Ruby" function="receiver">
     <argument value="0"/>
+    <argument value="3"/>       <!-- Number of tasks -->
   </process>
 </platform>
diff --git a/examples/msg/icomms/deployment_peer05.xml b/examples/msg/icomms/deployment_peer05.xml
new file mode 100755
index 0000000000..0b89a0dab8
--- /dev/null
+++ b/examples/msg/icomms/deployment_peer05.xml
@@ -0,0 +1,20 @@
+<?xml version='1.0'?>
+<!DOCTYPE platform SYSTEM "simgrid.dtd">
+<platform version="2">
+  <!-- The master process (with some arguments) -->
+  <process 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 -->
+  </process>
+  <!-- The receiver processes -->
+  <process host="Ruby" function="receiver">
+    <argument value="0"/>
+    <argument value="3"/>       <!-- Number of tasks -->
+  </process>
+    <process host="Perl" function="receiver">
+    <argument value="1"/>
+    <argument value="3"/>       <!-- Number of tasks -->
+  </process>
+</platform>
diff --git a/examples/msg/icomms/peer3.c b/examples/msg/icomms/peer3.c
new file mode 100755
index 0000000000..8ca347b6d7
--- /dev/null
+++ b/examples/msg/icomms/peer3.c
@@ -0,0 +1,166 @@
+/* Copyright (c) 2010. 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 <stdio.h>
+#include "msg/msg.h" /* Yeah! If you want to use msg, you need to include msg/msg.h */
+#include "xbt/sysdep.h" /* calloc, printf */
+#include <math.h>
+/* Create a log channel to have nice outputs. */
+#include "xbt/log.h"
+#include "xbt/asserts.h"
+XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,"Messages specific for this msg example");
+
+int sender(int argc, char *argv[]);
+int receiver(int argc, char *argv[]);
+
+MSG_error_t test_all(const char *platform_file, const char *application_file);
+
+/** Sender function  */
+int sender(int argc, char *argv[])
+{
+  long number_of_tasks = atol(argv[1]);
+  double task_comp_size = atof(argv[2]);
+  double task_comm_size = atof(argv[3]);
+  long receivers_count = atol(argv[4]);
+  xbt_dynar_t d = NULL;
+  int i;
+  m_task_t task = NULL;
+  char mailbox[256];
+  char sprintf_buffer[256];
+  d = xbt_dynar_new(sizeof(msg_comm_t), NULL);
+  msg_comm_t *comm = malloc((number_of_tasks+receivers_count)*sizeof(msg_comm_t));
+  msg_comm_t res_irecv = NULL;
+  for (i = 0; i < (number_of_tasks); i++)
+  {
+    sprintf(mailbox,"receiver-%ld",(i % receivers_count));
+    sprintf(sprintf_buffer, "Task_%d", i);
+    task = MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size/pow(10,i), NULL);
+    comm[i] = MSG_task_isend(task, mailbox);
+    MSG_task_refcount_dec(task);
+    xbt_dynar_push_as(d, msg_comm_t, comm[i]);
+    INFO3("Send to receiver-%ld %s comm_size %f",i % receivers_count,sprintf_buffer,task_comm_size/pow(10,i));
+  }
+  /* Here we are waiting for the completion of all communications*/
+
+  while(d->used)
+  {
+    xbt_dynar_remove_at(d, MSG_comm_waitany(d), &res_irecv);
+  }
+  xbt_dynar_free(&d);
+
+
+  /* Here we are waiting for the completion of all tasks*/
+  sprintf(mailbox,"finalize");
+
+  for(i=0; i<receivers_count; i++)
+  {
+	  task = NULL;
+	  res_irecv = NULL;
+	  res_irecv = MSG_task_irecv(&(task), mailbox);
+	  xbt_assert0(MSG_comm_wait(res_irecv,-1) == MSG_OK, "MSG_task_get failed");
+	  MSG_task_destroy(task);
+	  MSG_comm_destroy(res_irecv);
+  }
+
+  INFO0("Goodbye now!");
+  return 0;
+} /* end_of_sender */
+
+/** Receiver function  */
+int receiver(int argc, char *argv[])
+{
+  int id = -1;
+  int i;
+  char mailbox[80];
+  xbt_dynar_t comms = NULL;
+  int tasks = atof(argv[2]);
+  comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
+
+  xbt_assert1(sscanf(argv[1],"%d", &id),
+	 "Invalid argument %s\n",argv[1]);
+  sprintf(mailbox,"receiver-%d",id);
+
+  msg_comm_t res_irecv = NULL;
+  m_task_t task = NULL;
+  m_task_t task_com = NULL;
+  for(i=0; i<tasks ;i++)
+  {
+	INFO1("Wait to receive task %d",i);
+	res_irecv = MSG_task_irecv(&task, mailbox);
+	xbt_dynar_push_as(comms, msg_comm_t , res_irecv);
+	task = NULL;
+	res_irecv = NULL;
+  }
+
+  /* Here we are waiting for the receiving of all communications*/
+  while(comms->used)
+  {
+	task_com = NULL;
+	res_irecv = NULL;
+	int num = MSG_comm_waitany(comms);
+	xbt_dynar_remove_at(comms, num, &res_irecv);
+	task_com = MSG_comm_get_task(res_irecv);
+	INFO1("Processing \"%s\"", MSG_task_get_name(task_com) );
+	MSG_task_execute(task_com);
+	INFO1("\"%s\" done", MSG_task_get_name(task_com));
+	MSG_task_destroy(task_com);
+  }
+  xbt_dynar_free(&comms);
+
+  /* Here we tell to sender that all tasks are done*/
+  sprintf(mailbox,"finalize");
+  res_irecv = MSG_task_isend(MSG_task_create("end", 0, 0, NULL), mailbox);
+
+  INFO0("I'm done. See you!");
+  return 0;
+} /* end_of_receiver */
+
+
+
+
+/** Test function */
+MSG_error_t test_all(const char *platform_file,
+			    const char *application_file)
+{
+  MSG_error_t res = MSG_OK;
+
+  /* MSG_config("workstation/model","KCCFLN05"); */
+  {				/*  Simulation setting */
+    MSG_set_channel_number(0);
+    MSG_create_environment(platform_file);
+  }
+  {                            /*   Application deployment */
+    MSG_function_register("sender", sender);
+    MSG_function_register("receiver", receiver);
+    MSG_launch_application(application_file);
+  }
+  res = MSG_main();
+  
+  INFO1("Simulation time %g",MSG_get_clock());
+  return res;
+} /* end_of_test_all */
+
+
+/** Main function */
+int main(int argc, char *argv[])
+{
+  MSG_error_t res = MSG_OK;
+
+  MSG_global_init(&argc,argv);
+  if (argc < 3) {
+     printf ("Usage: %s platform_file deployment_file\n",argv[0]);
+     printf ("example: %s msg_platform.xml msg_deployment.xml\n",argv[0]);
+     exit(1);
+  }
+  res = test_all(argv[1],argv[2]);
+   SIMIX_message_sizes_output("toto.txt");
+  MSG_clean();
+
+  if(res==MSG_OK)
+    return 0;
+  else
+    return 1;
+} /* end_of_main */
diff --git a/include/msg/msg.h b/include/msg/msg.h
index a82b4224f7..11faa2bd3f 100644
--- a/include/msg/msg.h
+++ b/include/msg/msg.h
@@ -115,6 +115,7 @@ XBT_PUBLIC(m_host_t) MSG_task_get_source(m_task_t task);
 XBT_PUBLIC(const char *) MSG_task_get_name(m_task_t task);
 XBT_PUBLIC(MSG_error_t) MSG_task_cancel(m_task_t task);
 XBT_PUBLIC(MSG_error_t) MSG_task_destroy(m_task_t task);
+XBT_PUBLIC(void) MSG_task_refcount_dec(m_task_t task);
 
 XBT_PUBLIC(MSG_error_t) MSG_task_get(m_task_t * task, m_channel_t channel);
 XBT_PUBLIC(MSG_error_t) MSG_task_get_with_timeout(m_task_t * task,
@@ -174,7 +175,8 @@ XBT_PUBLIC(int)               MSG_comm_test(msg_comm_t comm);
 XBT_PUBLIC(void)              MSG_comm_destroy(msg_comm_t comm);
 XBT_PUBLIC(MSG_error_t) MSG_comm_wait(msg_comm_t comm,double timeout);
 XBT_PUBLIC(void) MSG_comm_waitall(msg_comm_t *comm,int nb_elem, double timeout);
-
+XBT_PUBLIC(int) MSG_comm_waitany(xbt_dynar_t comms);
+XBT_PUBLIC(m_task_t) MSG_comm_get_task(msg_comm_t comm);
 
 XBT_PUBLIC(int) MSG_task_listen(const char *alias);
 
diff --git a/src/msg/gos.c b/src/msg/gos.c
index b7a99af3f9..a1266eb648 100644
--- a/src/msg/gos.c
+++ b/src/msg/gos.c
@@ -466,6 +466,7 @@ msg_comm_t MSG_task_irecv(m_task_t * task, const char *alias) {
 	/* Try to receive it by calling SIMIX network layer */
 	return SIMIX_network_irecv(rdv, task, &size);
 }
+
 int MSG_comm_test(msg_comm_t comm) {
 	return SIMIX_network_test(comm);
 }
@@ -531,6 +532,15 @@ void MSG_comm_waitall(msg_comm_t *comm,int nb_elem, double timeout) {
 	}
 }
 
+int MSG_comm_waitany(xbt_dynar_t comms) {
+	return SIMIX_network_waitany(comms);
+}
+
+m_task_t MSG_comm_get_task(msg_comm_t comm) {
+	xbt_assert0(comm, "Invalid parameters");
+	return (m_task_t) SIMIX_communication_get_src_buf(comm);
+}
+
 /** \ingroup msg_gos_functions
  * \brief Put a task on a channel of an host and waits for the end of the
  * transmission.
diff --git a/src/msg/task.c b/src/msg/task.c
index 89f3cdb7b7..3479e4703b 100644
--- a/src/msg/task.c
+++ b/src/msg/task.c
@@ -136,6 +136,10 @@ const char *MSG_task_get_name(m_task_t task)
   return task->name;
 }
 
+void MSG_task_refcount_dec(m_task_t task)
+{
+  task->simdata->refcount--;
+}
 
 /** \ingroup m_task_management
  * \brief Destroy a #m_task_t.