Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Add an example for waitany fct.
[simgrid.git] / examples / msg / icomms / peer3.c
1 /* Copyright (c) 2010. The SimGrid Team.
2  * All rights reserved.                                                     */
3
4 /* This program is free software; you can redistribute it and/or modify it
5  * under the terms of the license (GNU LGPL) which comes with this package. */
6
7 #include <stdio.h>
8 #include "msg/msg.h" /* Yeah! If you want to use msg, you need to include msg/msg.h */
9 #include "xbt/sysdep.h" /* calloc, printf */
10 #include <math.h>
11 /* Create a log channel to have nice outputs. */
12 #include "xbt/log.h"
13 #include "xbt/asserts.h"
14 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,"Messages specific for this msg example");
15
16 int sender(int argc, char *argv[]);
17 int receiver(int argc, char *argv[]);
18
19 MSG_error_t test_all(const char *platform_file, const char *application_file);
20
21 /** Sender function  */
22 int sender(int argc, char *argv[])
23 {
24   long number_of_tasks = atol(argv[1]);
25   double task_comp_size = atof(argv[2]);
26   double task_comm_size = atof(argv[3]);
27   long receivers_count = atol(argv[4]);
28   xbt_dynar_t d = NULL;
29   int i;
30   m_task_t task = NULL;
31   char mailbox[256];
32   char sprintf_buffer[256];
33   d = xbt_dynar_new(sizeof(msg_comm_t), NULL);
34   msg_comm_t *comm = malloc((number_of_tasks+receivers_count)*sizeof(msg_comm_t));
35   msg_comm_t res_irecv = NULL;
36   for (i = 0; i < (number_of_tasks); i++)
37   {
38     sprintf(mailbox,"receiver-%ld",(i % receivers_count));
39     sprintf(sprintf_buffer, "Task_%d", i);
40     task = MSG_task_create(sprintf_buffer, task_comp_size, task_comm_size/pow(10,i), NULL);
41     comm[i] = MSG_task_isend(task, mailbox);
42     MSG_task_refcount_dec(task);
43     xbt_dynar_push_as(d, msg_comm_t, comm[i]);
44     INFO3("Send to receiver-%ld %s comm_size %f",i % receivers_count,sprintf_buffer,task_comm_size/pow(10,i));
45   }
46   /* Here we are waiting for the completion of all communications*/
47
48   while(d->used)
49   {
50     xbt_dynar_remove_at(d, MSG_comm_waitany(d), &res_irecv);
51   }
52   xbt_dynar_free(&d);
53
54
55   /* Here we are waiting for the completion of all tasks*/
56   sprintf(mailbox,"finalize");
57
58   for(i=0; i<receivers_count; i++)
59   {
60           task = NULL;
61           res_irecv = NULL;
62           res_irecv = MSG_task_irecv(&(task), mailbox);
63           xbt_assert0(MSG_comm_wait(res_irecv,-1) == MSG_OK, "MSG_task_get failed");
64           MSG_task_destroy(task);
65           MSG_comm_destroy(res_irecv);
66   }
67
68   INFO0("Goodbye now!");
69   return 0;
70 } /* end_of_sender */
71
72 /** Receiver function  */
73 int receiver(int argc, char *argv[])
74 {
75   int id = -1;
76   int i;
77   char mailbox[80];
78   xbt_dynar_t comms = NULL;
79   int tasks = atof(argv[2]);
80   comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
81
82   xbt_assert1(sscanf(argv[1],"%d", &id),
83          "Invalid argument %s\n",argv[1]);
84   sprintf(mailbox,"receiver-%d",id);
85
86   msg_comm_t res_irecv = NULL;
87   m_task_t task = NULL;
88   m_task_t task_com = NULL;
89   for(i=0; i<tasks ;i++)
90   {
91         INFO1("Wait to receive task %d",i);
92         res_irecv = MSG_task_irecv(&task, mailbox);
93         xbt_dynar_push_as(comms, msg_comm_t , res_irecv);
94         task = NULL;
95         res_irecv = NULL;
96   }
97
98   /* Here we are waiting for the receiving of all communications*/
99   while(comms->used)
100   {
101         task_com = NULL;
102         res_irecv = NULL;
103         int num = MSG_comm_waitany(comms);
104         xbt_dynar_remove_at(comms, num, &res_irecv);
105         task_com = MSG_comm_get_task(res_irecv);
106         INFO1("Processing \"%s\"", MSG_task_get_name(task_com) );
107         MSG_task_execute(task_com);
108         INFO1("\"%s\" done", MSG_task_get_name(task_com));
109         MSG_task_destroy(task_com);
110   }
111   xbt_dynar_free(&comms);
112
113   /* Here we tell to sender that all tasks are done*/
114   sprintf(mailbox,"finalize");
115   res_irecv = MSG_task_isend(MSG_task_create("end", 0, 0, NULL), mailbox);
116
117   INFO0("I'm done. See you!");
118   return 0;
119 } /* end_of_receiver */
120
121
122
123
124 /** Test function */
125 MSG_error_t test_all(const char *platform_file,
126                             const char *application_file)
127 {
128   MSG_error_t res = MSG_OK;
129
130   /* MSG_config("workstation/model","KCCFLN05"); */
131   {                             /*  Simulation setting */
132     MSG_set_channel_number(0);
133     MSG_create_environment(platform_file);
134   }
135   {                            /*   Application deployment */
136     MSG_function_register("sender", sender);
137     MSG_function_register("receiver", receiver);
138     MSG_launch_application(application_file);
139   }
140   res = MSG_main();
141   
142   INFO1("Simulation time %g",MSG_get_clock());
143   return res;
144 } /* end_of_test_all */
145
146
147 /** Main function */
148 int main(int argc, char *argv[])
149 {
150   MSG_error_t res = MSG_OK;
151
152   MSG_global_init(&argc,argv);
153   if (argc < 3) {
154      printf ("Usage: %s platform_file deployment_file\n",argv[0]);
155      printf ("example: %s msg_platform.xml msg_deployment.xml\n",argv[0]);
156      exit(1);
157   }
158   res = test_all(argv[1],argv[2]);
159    SIMIX_message_sizes_output("toto.txt");
160   MSG_clean();
161
162   if(res==MSG_OK)
163     return 0;
164   else
165     return 1;
166 } /* end_of_main */