Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
b6aa08e34c30d05497dbbe221b71ad21f5b2350f
[simgrid.git] / src / msg / msg_mailbox.c
1 #include "mailbox.h"
2 #include "msg/private.h"
3
4 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_mailbox, msg,
5                                 "Logging specific to MSG (mailbox)");
6
7 static xbt_dict_t msg_mailboxes = NULL;
8
9 void MSG_mailbox_mod_init(void)
10 {
11   msg_mailboxes = xbt_dict_new();
12 }
13
14 void MSG_mailbox_mod_exit(void)
15 {
16   xbt_dict_free(&msg_mailboxes);
17 }
18
19 msg_mailbox_t MSG_mailbox_create(const char *alias)
20 {
21   msg_mailbox_t mailbox = xbt_new0(s_msg_mailbox_t, 1);
22
23   mailbox->tasks = xbt_fifo_new();
24   mailbox->cond = NULL;
25   mailbox->alias = alias ? xbt_strdup(alias) : NULL;
26   mailbox->hostname = NULL;
27
28   return mailbox;
29 }
30
31 msg_mailbox_t MSG_mailbox_new(const char *alias)
32 {
33   msg_mailbox_t mailbox = MSG_mailbox_create(alias);
34
35   /* add the mbox in the dictionary */
36   xbt_dict_set(msg_mailboxes, alias, mailbox, MSG_mailbox_free);
37
38   return mailbox;
39 }
40
41 void MSG_mailbox_free(void *mailbox)
42 {
43   msg_mailbox_t _mailbox = (msg_mailbox_t) mailbox;
44
45   if (_mailbox->hostname)
46     free(_mailbox->hostname);
47
48   xbt_fifo_free(_mailbox->tasks);
49   free(_mailbox->alias);
50
51   free(_mailbox);
52 }
53
54 smx_cond_t MSG_mailbox_get_cond(msg_mailbox_t mailbox)
55 {
56   return mailbox->cond;
57 }
58
59 void MSG_mailbox_remove(msg_mailbox_t mailbox, m_task_t task)
60 {
61   xbt_fifo_remove(mailbox->tasks, task);
62 }
63
64 int MSG_mailbox_is_empty(msg_mailbox_t mailbox)
65 {
66   return (NULL == xbt_fifo_get_first_item(mailbox->tasks));
67 }
68
69 m_task_t MSG_mailbox_pop_head(msg_mailbox_t mailbox)
70 {
71   return (m_task_t) xbt_fifo_shift(mailbox->tasks);
72 }
73
74 m_task_t MSG_mailbox_get_head(msg_mailbox_t mailbox)
75 {
76   xbt_fifo_item_t item;
77
78   if (!(item = xbt_fifo_get_first_item(mailbox->tasks)))
79     return NULL;
80
81   return (m_task_t) xbt_fifo_get_item_content(item);
82 }
83
84
85 m_task_t MSG_mailbox_get_first_host_task(msg_mailbox_t mailbox, m_host_t host)
86 {
87   m_task_t task = NULL;
88   xbt_fifo_item_t item = NULL;
89
90   xbt_fifo_foreach(mailbox->tasks, item, task, m_task_t)
91     if (task->simdata->source == host) {
92     xbt_fifo_remove_item(mailbox->tasks, item);
93     return task;
94   }
95
96   return NULL;
97 }
98
99 int
100 MSG_mailbox_get_count_host_waiting_tasks(msg_mailbox_t mailbox, m_host_t host)
101 {
102   m_task_t task = NULL;
103   xbt_fifo_item_t item = NULL;
104   int count = 0;
105
106   xbt_fifo_foreach(mailbox->tasks, item, task, m_task_t) {
107     if (task->simdata->source == host)
108       count++;
109   }
110
111   return count;
112 }
113
114 void MSG_mailbox_set_cond(msg_mailbox_t mailbox, smx_cond_t cond)
115 {
116   mailbox->cond = cond;
117 }
118
119 const char *MSG_mailbox_get_alias(msg_mailbox_t mailbox)
120 {
121   return mailbox->alias;
122 }
123
124 const char *MSG_mailbox_get_hostname(msg_mailbox_t mailbox)
125 {
126   return mailbox->hostname;
127 }
128
129 void MSG_mailbox_set_hostname(msg_mailbox_t mailbox, const char *hostname)
130 {
131   mailbox->hostname = xbt_strdup(hostname);
132 }
133
134 msg_mailbox_t MSG_mailbox_get_by_alias(const char *alias)
135 {
136
137   msg_mailbox_t mailbox = xbt_dict_get_or_null(msg_mailboxes, alias);
138
139   if (!mailbox) {
140     mailbox = MSG_mailbox_new(alias);
141     MSG_mailbox_set_hostname(mailbox, MSG_host_self()->name);
142   }
143
144   return mailbox;
145 }
146
147 msg_mailbox_t MSG_mailbox_get_by_channel(m_host_t host, m_channel_t channel)
148 {
149   xbt_assert0((host != NULL), "Invalid host");
150   xbt_assert1((channel >= 0)
151               && (channel < msg_global->max_channel), "Invalid channel %d",
152               channel);
153
154   return host->simdata->mailboxes[(size_t) channel];
155 }
156
157 MSG_error_t
158 MSG_mailbox_get_task_ext(msg_mailbox_t mailbox, m_task_t * task,
159                          m_host_t host, double timeout)
160 {
161   m_process_t process = MSG_process_self();
162   m_task_t t = NULL;
163   m_host_t h = NULL;
164   simdata_task_t t_simdata = NULL;
165   simdata_host_t h_simdata = NULL;
166   double start_time = SIMIX_get_clock();
167
168   smx_cond_t cond = NULL;       //conditional wait if the task isn't on the channel yet
169
170   CHECK_HOST();
171
172   /* Sanity check */
173   xbt_assert0(task, "Null pointer for the task storage");
174
175   if (*task)
176     CRITICAL0
177       ("MSG_task_get() was asked to write in a non empty task struct.");
178
179   /* Get the task */
180   h = MSG_host_self();
181   h_simdata = h->simdata;
182
183   SIMIX_mutex_lock(h_simdata->mutex);   //FIXME: lock the mailbox instead
184
185   if (MSG_mailbox_get_cond(mailbox)) {
186     CRITICAL1
187       ("A process is already blocked on the channel %s (meaning that someone is already doing a get on this)",
188        MSG_mailbox_get_alias(mailbox));
189     SIMIX_cond_display_info(MSG_mailbox_get_cond(mailbox));
190     xbt_die("Go fix your code!");
191   }
192
193   while (1) {
194     /* if the mailbox is not empty (has a task) */
195     if (!MSG_mailbox_is_empty(mailbox)) {
196       if (!host) {
197         /* pop the head of the mailbox */
198         t = MSG_mailbox_pop_head(mailbox);
199         break;
200       } else {
201         /* get the first task of the host */
202         if ((t = MSG_mailbox_get_first_host_task(mailbox, host)))
203           break;
204       }
205     }
206
207     if ((timeout > 0) && (SIMIX_get_clock() - start_time >= timeout)) { // Timeout already elapsed
208       SIMIX_mutex_unlock(h_simdata->mutex);
209       MSG_mailbox_set_cond(mailbox, NULL);
210       SIMIX_cond_destroy(cond);
211       MSG_RETURN(MSG_TRANSFER_FAILURE);
212     }
213
214     if (!cond) {
215       cond = SIMIX_cond_init();
216       MSG_mailbox_set_cond(mailbox, cond);
217     }
218
219     if (timeout > 0)
220       SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout - start_time + SIMIX_get_clock());
221     else
222       SIMIX_cond_wait(cond, h_simdata->mutex);
223
224     if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
225       SIMIX_mutex_unlock(h_simdata->mutex);
226       MSG_mailbox_set_cond(mailbox, NULL);
227       SIMIX_cond_destroy(cond);
228       MSG_RETURN(MSG_HOST_FAILURE);
229     }
230   }
231
232
233   DEBUG1("OK, got a task (%s)", t->name);
234   /* clean conditional */
235   if (cond) {
236     MSG_mailbox_set_cond(mailbox, NULL);
237     SIMIX_cond_destroy(cond);
238   }
239
240   SIMIX_mutex_unlock(h_simdata->mutex);
241
242   t_simdata = t->simdata;
243   t_simdata->receiver = process;
244   *task = t;
245
246   SIMIX_mutex_lock(t_simdata->mutex);
247
248   /* Transfer */
249   /* create SIMIX action to the communication */
250   t_simdata->comm =
251     SIMIX_action_communicate(t_simdata->sender->simdata->m_host->
252                              simdata->smx_host,
253                              process->simdata->m_host->simdata->smx_host,
254                              t->name, t_simdata->message_size,
255                              t_simdata->rate);
256
257   SIMIX_action_use(t_simdata->comm);
258
259   /* if the process is suspend, create the action but stop its execution, it will be restart when the sender process resume */
260   if (MSG_process_is_suspended(t_simdata->sender)) {
261     DEBUG1("Process sender (%s) suspended", t_simdata->sender->name);
262     SIMIX_action_set_priority(t_simdata->comm, 0);
263   }
264   SIMIX_register_action_to_condition(t_simdata->comm, t_simdata->cond);
265   // breaking point if asynchrounous
266   process->simdata->waiting_task = t;
267
268   while (1) {
269     SIMIX_cond_wait(t_simdata->cond, t_simdata->mutex);
270
271     if (SIMIX_action_get_state(t_simdata->comm) != SURF_ACTION_RUNNING)
272       break;
273     if (!SIMIX_host_get_state(h_simdata->smx_host))
274       break;
275     if (!SIMIX_host_get_state(process->simdata->m_host->simdata->smx_host))
276       break;
277   }
278
279   SIMIX_unregister_action_to_condition(t_simdata->comm, t_simdata->cond);
280   process->simdata->waiting_task = NULL;
281
282   /* for this process, don't need to change in get function */
283   SIMIX_mutex_unlock(t_simdata->mutex);
284
285   if (SIMIX_action_get_state(t_simdata->comm) == SURF_ACTION_DONE) {
286     if (SIMIX_action_destroy(t_simdata->comm))
287       t_simdata->comm = NULL;
288     MSG_RETURN(MSG_OK);
289   } else if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
290     if (SIMIX_action_destroy(t_simdata->comm))
291       t_simdata->comm = NULL;
292     MSG_RETURN(MSG_HOST_FAILURE);
293   } else {
294     if (SIMIX_action_destroy(t_simdata->comm))
295       t_simdata->comm = NULL;
296     MSG_RETURN(MSG_TRANSFER_FAILURE);
297   }
298 }
299
300 MSG_error_t
301 MSG_mailbox_put_with_timeout(msg_mailbox_t mailbox, m_task_t task,
302                              double timeout)
303 {
304   m_process_t process = MSG_process_self();
305   const char *hostname;
306   simdata_task_t t_simdata = NULL;
307   m_host_t local_host = NULL;
308   m_host_t remote_host = NULL;
309   smx_cond_t cond = NULL;
310
311   CHECK_HOST();
312
313   t_simdata = task->simdata;
314   t_simdata->sender = process;
315   t_simdata->source = MSG_process_get_host(process);
316
317   xbt_assert0(t_simdata->refcount == 1,
318               "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
319
320   t_simdata->comm = NULL;
321
322   t_simdata->refcount++;
323   local_host = ((simdata_process_t) process->simdata)->m_host;
324   msg_global->sent_msg++;
325
326   /* get the host name containing the mailbox */
327   hostname = MSG_mailbox_get_hostname(mailbox);
328
329   remote_host = MSG_get_host_by_name(hostname);
330
331   if (!remote_host)
332     THROW1(not_found_error, 0, "Host %s not fount", hostname);
333
334
335   DEBUG4("Trying to send a task (%g kB) from %s to %s on the channel %s",
336          t_simdata->message_size / 1000, local_host->name,
337          remote_host->name, MSG_mailbox_get_alias(mailbox));
338
339   SIMIX_mutex_lock(remote_host->simdata->mutex);        /* FIXME: lock the mailbox instead */
340
341   /* put the task in the mailbox */
342   xbt_fifo_push(mailbox->tasks, task);
343
344   if ((cond = MSG_mailbox_get_cond(mailbox))) {
345     DEBUG0("Somebody is listening. Let's wake him up!");
346     SIMIX_cond_signal(cond);
347   }
348
349   SIMIX_mutex_unlock(remote_host->simdata->mutex);
350
351   SIMIX_mutex_lock(t_simdata->mutex);
352
353   process->simdata->waiting_task = task;        // for debugging and status displaying purpose
354
355   if (timeout > 0) {
356     xbt_ex_t e;
357     double time;
358     double time_elapsed;
359     time = SIMIX_get_clock();
360
361     TRY {
362       /*verify if the action that ends is the correct. Call the wait_timeout with the new time. If the timeout occurs, an exception is raised */
363       while (1) {
364         time_elapsed = SIMIX_get_clock() - time;
365         SIMIX_cond_wait_timeout(t_simdata->cond, t_simdata->mutex,
366                                 timeout - time_elapsed);
367
368         if (t_simdata->comm)
369           SIMIX_action_use(t_simdata->comm);
370         if (t_simdata->comm && (SIMIX_action_get_state(t_simdata->comm) !=
371                                 SURF_ACTION_RUNNING))
372           break;
373         if (!SIMIX_host_get_state(local_host->simdata->smx_host))
374           break;
375         if (!SIMIX_host_get_state(remote_host->simdata->smx_host))
376           break;
377       }
378     }
379     CATCH(e) {
380       if (e.category == timeout_error) {
381         xbt_ex_free(e);
382         /* verify if the timeout happened and the communication didn't started yet */
383         if (t_simdata->comm == NULL) {
384           DEBUG1("Action terminated %s (there was a timeout)", task->name);
385           process->simdata->waiting_task = NULL;
386
387           /* remove the task from the mailbox */
388           MSG_mailbox_remove(mailbox, task);
389
390 /*           if (t_simdata->receiver && t_simdata->receiver->simdata) {    /\* receiver still around *\/ */
391 /*             t_simdata->receiver->simdata->waiting_task = NULL; */
392 /*           } */
393
394           SIMIX_mutex_unlock(t_simdata->mutex);
395           MSG_RETURN(MSG_TRANSFER_FAILURE);
396         }
397       } else {
398         RETHROW;
399       }
400     }
401   } else {
402     while (1) {                 //FIXME: factorize with the code right above
403       SIMIX_cond_wait(t_simdata->cond, t_simdata->mutex);
404
405       if (t_simdata->comm)
406         SIMIX_action_use(t_simdata->comm);
407       if (t_simdata->comm
408           && SIMIX_action_get_state(t_simdata->comm) != SURF_ACTION_RUNNING)
409         break;
410       if (!SIMIX_host_get_state(local_host->simdata->smx_host))
411         break;
412       if (!SIMIX_host_get_state(remote_host->simdata->smx_host))
413         break;
414     }
415   }
416
417   DEBUG1("Action terminated %s", task->name);
418   process->simdata->waiting_task = NULL;
419 /*   if (t_simdata->receiver && t_simdata->receiver->simdata) {    /\* receiver still around *\/ */
420 /*     t_simdata->receiver->simdata->waiting_task = NULL; */
421 /*   } */
422
423   SIMIX_mutex_unlock(t_simdata->mutex);
424
425   if (t_simdata->comm
426       && SIMIX_action_get_state(t_simdata->comm) == SURF_ACTION_DONE) {
427     if (SIMIX_action_destroy(t_simdata->comm))
428       t_simdata->comm = NULL;
429     t_simdata->refcount--;
430     MSG_RETURN(MSG_OK);
431   } else if (SIMIX_host_get_state(local_host->simdata->smx_host) == 0) {
432     if (t_simdata->comm && SIMIX_action_destroy(t_simdata->comm))
433       t_simdata->comm = NULL;
434     MSG_RETURN(MSG_HOST_FAILURE);
435   } else {
436     if (t_simdata->comm && SIMIX_action_destroy(t_simdata->comm))
437       t_simdata->comm = NULL;
438     MSG_RETURN(MSG_TRANSFER_FAILURE);
439   }
440 }