Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
37d82bbbdf45d530b5f5a91698a8e15d7654fa51
[simgrid.git] / src / msg / gos.c
1 /*      $Id$     */
2
3 /* Copyright (c) 2002,2003,2004 Arnaud Legrand. All rights reserved.        */
4
5 /* This program is free software; you can redistribute it and/or modify it
6  * under the terms of the license (GNU LGPL) which comes with this package. */
7
8 #include"private.h"
9 #include"xbt/sysdep.h"
10 #include "xbt/error.h"
11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gos, msg,
12                                 "Logging specific to MSG (gos)");
13
14 /** \defgroup msg_gos_functions MSG Operating System Functions
15  *  \brief This section describes the functions that can be used
16  *  by an agent for handling some task.
17  */
18
19 /** \ingroup msg_gos_functions
20  * \brief This function is now deprecated and useless. Please stop using it.
21  */
22 MSG_error_t MSG_process_start(m_process_t process)
23 {
24   xbt_assert0(0,"This function is now deprecated and useless. Please stop using it.");
25   
26   return MSG_OK;
27 }
28
29 /** \ingroup msg_gos_functions
30  * \brief Listen on a channel and wait for receiving a task.
31  *
32  * It takes two parameters.
33  * \param task a memory location for storing a #m_task_t. It will
34    hold a task when this function will return. Thus \a task should not
35    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
36    those two condition does not hold, there will be a warning message.
37  * \param channel the channel on which the agent should be
38    listening. This value has to be >=0 and < than the maximal
39    number of channels fixed with MSG_set_channel_number().
40  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
41  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
42  */
43 MSG_error_t MSG_task_get(m_task_t * task,
44                          m_channel_t channel)
45 {
46   return MSG_task_get_with_time_out(task, channel, -1);
47 }
48
49 /** \ingroup msg_gos_functions
50  * \brief Listen on a channel and wait for receiving a task with a timeout.
51  *
52  * It takes three parameters.
53  * \param task a memory location for storing a #m_task_t. It will
54    hold a task when this function will return. Thus \a task should not
55    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
56    those two condition does not hold, there will be a warning message.
57  * \param channel the channel on which the agent should be
58    listening. This value has to be >=0 and < than the maximal
59    number of channels fixed with MSG_set_channel_number().
60  * \param timeout the maximum time to wait for a task before giving
61     up. In such a case, \a task will not be modified and will still be
62     equal to \c NULL when returning.
63  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
64    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
65  */
66
67 MSG_error_t MSG_task_get_with_time_out(m_task_t * task,
68                                        m_channel_t channel,
69                                        double max_duration)
70 {
71   m_process_t process = MSG_process_self();
72   m_task_t t = NULL;
73   m_host_t h = NULL;
74   simdata_task_t t_simdata = NULL;
75   simdata_host_t h_simdata = NULL;
76   int warning = 0;
77   int first_time = 1;
78   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
79   
80   CHECK_HOST();
81   /* Sanity check */
82   xbt_assert0(task,"Null pointer for the task\n");
83
84   if (*task) 
85     CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
86
87   /* Get the task */
88   h = MSG_host_self();
89   h_simdata = h->simdata;
90
91   DEBUG2("Waiting for a task on channel %d (%s)", channel,h->name);
92
93   while ((t = xbt_fifo_shift(h_simdata->mbox[channel])) == NULL) {
94     if(max_duration>0) {
95       if(!first_time) {
96         MSG_RETURN(MSG_OK);
97       }
98     }
99     xbt_assert2(!(h_simdata->sleeping[channel]),
100                 "A process (%s(%d)) is already blocked on this channel",
101                 h_simdata->sleeping[channel]->name,
102                 h_simdata->sleeping[channel]->simdata->PID);
103     h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
104     if(max_duration>0) {
105       __MSG_process_block(max_duration);
106     } else {
107       __MSG_process_block(-1);
108     }
109     if(surf_workstation_resource->extension_public->get_state(h_simdata->host) 
110        == SURF_CPU_OFF)
111       MSG_RETURN(MSG_HOST_FAILURE);
112     h_simdata->sleeping[channel] = NULL;
113     first_time = 0;
114     /* OK, we should both be ready now. Are you there ? */
115   }
116
117   t_simdata = t->simdata;
118   /*   *task = __MSG_task_copy(t); */
119   *task=t;
120
121   /* Transfer */
122   t_simdata->using++;
123
124   t_simdata->comm = surf_workstation_resource->extension_public->
125     communicate(MSG_process_get_host(t_simdata->sender)->simdata->host,
126                 h->simdata->host, t_simdata->message_size,t_simdata->rate);
127   
128   surf_workstation_resource->common_public->action_set_data(t_simdata->comm,t);
129
130   if(__MSG_process_isBlocked(t_simdata->sender)) 
131     __MSG_process_unblock(t_simdata->sender);
132
133   PAJE_PROCESS_PUSH_STATE(process,"C");  
134
135   do {
136     __MSG_task_wait_event(process, t);
137     state=surf_workstation_resource->common_public->action_get_state(t_simdata->comm);
138   } while (state==SURF_ACTION_RUNNING);
139
140   if(t->simdata->using>1) {
141     xbt_fifo_unshift(msg_global->process_to_run,process);
142     xbt_context_yield();
143   }
144
145   PAJE_PROCESS_POP_STATE(process);
146   PAJE_COMM_STOP(process,t,channel);
147
148   if(state == SURF_ACTION_DONE) {
149     if(surf_workstation_resource->common_public->action_free(t_simdata->comm)) 
150       t_simdata->comm = NULL;
151     MSG_RETURN(MSG_OK);
152   } else if(surf_workstation_resource->extension_public->get_state(h_simdata->host) 
153           == SURF_CPU_OFF) {
154     if(surf_workstation_resource->common_public->action_free(t_simdata->comm)) 
155       t_simdata->comm = NULL;
156     MSG_RETURN(MSG_HOST_FAILURE);
157   } else {
158     if(surf_workstation_resource->common_public->action_free(t_simdata->comm)) 
159       t_simdata->comm = NULL;
160     MSG_RETURN(MSG_TRANSFER_FAILURE);
161   }
162 }
163
164 /** \ingroup msg_gos_functions
165  * \brief Test whether there is a pending communication on a channel.
166  *
167  * It takes one parameter.
168  * \param channel the channel on which the agent should be
169    listening. This value has to be >=0 and < than the maximal
170    number of channels fixed with MSG_set_channel_number().
171  * \return 1 if there is a pending communication and 0 otherwise
172  */
173 int MSG_task_Iprobe(m_channel_t channel)
174 {
175   m_host_t h = NULL;
176   simdata_host_t h_simdata = NULL;
177
178   DEBUG2("Probing on channel %d (%s)", channel,h->name);
179   CHECK_HOST();
180   h = MSG_host_self();
181   h_simdata = h->simdata;
182   return(xbt_fifo_getFirstItem(h_simdata->mbox[channel])!=NULL);
183 }
184
185 /** \ingroup msg_gos_functions
186  * \brief Test whether there is a pending communication on a channel, and who sent it.
187  *
188  * It takes one parameter.
189  * \param channel the channel on which the agent should be
190    listening. This value has to be >=0 and < than the maximal
191    number of channels fixed with MSG_set_channel_number().
192  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
193  */
194 int MSG_task_probe_from(m_channel_t channel)
195 {
196   m_host_t h = NULL;
197   simdata_host_t h_simdata = NULL;
198   xbt_fifo_item_t item;
199   m_task_t t;
200
201   CHECK_HOST();
202   h = MSG_host_self();
203   h_simdata = h->simdata;
204
205   DEBUG2("Probing on channel %d (%s)", channel,h->name);
206    
207   item = xbt_fifo_getFirstItem(h->simdata->mbox[channel]);
208   if (!item || !(t = xbt_fifo_get_item_content(item)))
209     return -1;
210    
211   return MSG_process_get_PID(t->simdata->sender);
212 }
213
214 MSG_error_t MSG_channel_select_from(m_channel_t channel, double max_duration,
215                                     int *PID)
216 {
217   m_host_t h = NULL;
218   simdata_host_t h_simdata = NULL;
219   xbt_fifo_item_t item;
220   m_task_t t;
221   int first_time = 1;
222   m_process_t process = MSG_process_self();
223
224   if(PID) {
225     *PID = -1;
226   }
227
228   if(max_duration==0.0) {
229     return MSG_task_probe_from(channel);
230   } else {
231     CHECK_HOST();
232     h = MSG_host_self();
233     h_simdata = h->simdata;
234     
235     DEBUG2("Probing on channel %d (%s)", channel,h->name);
236     while((item = xbt_fifo_getFirstItem(h->simdata->mbox[channel]))) {
237       if(max_duration>0) {
238         if(!first_time) {
239           MSG_RETURN(MSG_OK);
240         }
241       }
242       xbt_assert2(!(h_simdata->sleeping[channel]),
243                   "A process (%s(%d)) is already blocked on this channel",
244                   h_simdata->sleeping[channel]->name,
245                   h_simdata->sleeping[channel]->simdata->PID);
246       h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
247       if(max_duration>0) {
248         __MSG_process_block(max_duration);
249       } else {
250         __MSG_process_block(-1);
251       }
252       if(surf_workstation_resource->extension_public->get_state(h_simdata->host) 
253          == SURF_CPU_OFF) {
254         MSG_RETURN(MSG_HOST_FAILURE);
255       }
256       h_simdata->sleeping[channel] = NULL;
257       first_time = 0;
258     }
259     if (!item || !(t = xbt_fifo_get_item_content(item))) {
260       MSG_RETURN(MSG_OK);
261     }
262     if(PID) {
263       *PID = MSG_process_get_PID(t->simdata->sender);
264     }
265     MSG_RETURN(MSG_OK);
266   }
267 }
268 /** \ingroup msg_gos_functions
269  * \brief Put a task on a channel of an host and waits for the end of the
270  * transmission.
271  *
272  * This function is used for describing the behavior of an agent. It
273  * takes three parameter.
274  * \param task a #m_task_t to send on another location. This task
275    will not be usable anymore when the function will return. There is
276    no automatic task duplication and you have to save your parameters
277    before calling this function. Tasks are unique and once it has been
278    sent to another location, you should not access it anymore. You do
279    not need to call MSG_task_destroy() but to avoid using, as an
280    effect of inattention, this task anymore, you definitely should
281    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
282    can be transfered iff it has been correctly created with
283    MSG_task_create().
284  * \param dest the destination of the message
285  * \param channel the channel on which the agent should put this
286    task. This value has to be >=0 and < than the maximal number of
287    channels fixed with MSG_set_channel_number().
288  * \return #MSG_FATAL if \a task is not properly initialized and
289  * #MSG_OK otherwise.
290  */
291 MSG_error_t MSG_task_put(m_task_t task,
292                          m_host_t dest, m_channel_t channel)
293 {
294   m_process_t process = MSG_process_self();
295   simdata_task_t task_simdata = NULL;
296   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
297   m_host_t local_host = NULL;
298   m_host_t remote_host = NULL;
299
300   CHECK_HOST();
301
302   task_simdata = task->simdata;
303   task_simdata->sender = process;
304   xbt_assert0(task_simdata->using==1,"Gargl!");
305   task_simdata->comm = NULL;
306   
307   local_host = ((simdata_process_t) process->simdata)->host;
308   remote_host = dest;
309
310   DEBUG4("Trying to send a task (%lg Mb) from %s to %s on channel %d", 
311          task->simdata->message_size,local_host->name, remote_host->name, channel);
312
313   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
314                 mbox[channel], task);
315
316   PAJE_COMM_START(process,task,channel);
317     
318   if(remote_host->simdata->sleeping[channel]) 
319     __MSG_process_unblock(remote_host->simdata->sleeping[channel]);
320
321   process->simdata->put_host = dest;
322   process->simdata->put_channel = channel;
323   while(!(task_simdata->comm)) 
324     __MSG_process_block(-1);
325   surf_workstation_resource->common_public->action_use(task_simdata->comm);
326   process->simdata->put_host = NULL;
327   process->simdata->put_channel = -1;
328
329
330   PAJE_PROCESS_PUSH_STATE(process,"C");  
331
332   state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
333   while (state==SURF_ACTION_RUNNING) {
334     __MSG_task_wait_event(process, task);
335     state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
336   }
337     
338   MSG_task_destroy(task);
339
340   PAJE_PROCESS_POP_STATE(process);  
341
342   if(state == SURF_ACTION_DONE) {
343     if(surf_workstation_resource->common_public->action_free(task_simdata->comm)) 
344       task_simdata->comm = NULL;
345     MSG_RETURN(MSG_OK);
346   } else if(surf_workstation_resource->extension_public->get_state(local_host->simdata->host) 
347             == SURF_CPU_OFF) {
348     if(surf_workstation_resource->common_public->action_free(task_simdata->comm)) 
349       task_simdata->comm = NULL;
350     MSG_RETURN(MSG_HOST_FAILURE);
351   } else { 
352     if(surf_workstation_resource->common_public->action_free(task_simdata->comm)) 
353       task_simdata->comm = NULL;
354     MSG_RETURN(MSG_TRANSFER_FAILURE);
355   }
356 }
357
358 /** \ingroup msg_gos_functions
359  * \brief Does exactly the same as MSG_task_put but with a bounded transmition 
360  * rate.
361  *
362  * \sa MSG_task_put
363  */
364 MSG_error_t MSG_task_put_bounded(m_task_t task,
365                                  m_host_t dest, m_channel_t channel,
366                                  double max_rate)
367 {
368   task->simdata->rate=max_rate;
369   return(MSG_task_put(task, dest, channel));
370   task->simdata->rate=-1.0;
371 }
372
373 /** \ingroup msg_gos_functions
374  * \brief Executes a task and waits for its termination.
375  *
376  * This function is used for describing the behavior of an agent. It
377  * takes only one parameter.
378  * \param task a #m_task_t to execute on the location on which the
379    agent is running.
380  * \return #MSG_FATAL if \a task is not properly initialized and
381  * #MSG_OK otherwise.
382  */
383 MSG_error_t MSG_task_execute(m_task_t task)
384 {
385   m_process_t process = MSG_process_self();
386   MSG_error_t res;
387
388   DEBUG1("Computing on %s", process->simdata->host->name);
389
390   __MSG_task_execute(process, task);
391
392   PAJE_PROCESS_PUSH_STATE(process,"E");  
393   res = __MSG_wait_for_computation(process,task);
394   PAJE_PROCESS_POP_STATE(process);
395   return res;
396 }
397
398 void __MSG_task_execute(m_process_t process, m_task_t task)
399 {
400   simdata_task_t simdata = NULL;
401
402   CHECK_HOST();
403
404   simdata = task->simdata;
405
406   simdata->compute = surf_workstation_resource->extension_public->
407     execute(MSG_process_get_host(process)->simdata->host,
408             simdata->computation_amount);
409   surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
410 }
411
412 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
413 {
414   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
415   simdata_task_t simdata = task->simdata;
416
417   simdata->using++;
418   do {
419     __MSG_task_wait_event(process, task);
420     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
421   } while (state==SURF_ACTION_RUNNING);
422   simdata->using--;
423     
424
425   if(state == SURF_ACTION_DONE) {
426     if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
427       simdata->compute = NULL;
428     MSG_RETURN(MSG_OK);
429   } else if(surf_workstation_resource->extension_public->
430             get_state(MSG_process_get_host(process)->simdata->host) 
431             == SURF_CPU_OFF) {
432     if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
433       simdata->compute = NULL;
434     MSG_RETURN(MSG_HOST_FAILURE);
435   } else {
436     if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
437       simdata->compute = NULL;
438     MSG_RETURN(MSG_TRANSFER_FAILURE);
439   }
440 }
441
442 /** \ingroup msg_gos_functions
443  * \brief Sleep for the specified number of seconds
444  *
445  * Makes the current process sleep until \a time seconds have elapsed.
446  *
447  * \param nb_sec a number of second
448  */
449 MSG_error_t MSG_process_sleep(double nb_sec)
450 {
451   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
452   m_process_t process = MSG_process_self();
453   m_task_t dummy = NULL;
454   simdata_task_t simdata = NULL;
455
456   CHECK_HOST();
457   dummy = MSG_task_create("MSG_sleep", nb_sec, 0.0, NULL);
458   simdata = dummy->simdata;
459
460   simdata->compute = surf_workstation_resource->extension_public->
461     sleep(MSG_process_get_host(process)->simdata->host,
462             simdata->computation_amount);
463   surf_workstation_resource->common_public->action_set_data(simdata->compute,dummy);
464
465   
466   simdata->using++;
467   do {
468     __MSG_task_wait_event(process, dummy);
469     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
470   } while (state==SURF_ACTION_RUNNING);
471   simdata->using--;
472     
473   if(state == SURF_ACTION_DONE) {
474     if(surf_workstation_resource->extension_public->
475        get_state(MSG_process_get_host(process)->simdata->host) 
476        == SURF_CPU_OFF) {
477       if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
478         simdata->compute = NULL;
479       MSG_RETURN(MSG_HOST_FAILURE);
480     }
481     if(__MSG_process_isBlocked(process)) {
482       __MSG_process_unblock(MSG_process_self());
483     }
484     if(surf_workstation_resource->extension_public->
485        get_state(MSG_process_get_host(process)->simdata->host) 
486        == SURF_CPU_OFF) {
487       if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
488         simdata->compute = NULL;
489       MSG_RETURN(MSG_HOST_FAILURE);
490     }
491     if(surf_workstation_resource->common_public->action_free(simdata->compute)) 
492       simdata->compute = NULL;
493     MSG_task_destroy(dummy);
494     MSG_RETURN(MSG_OK);
495   } else MSG_RETURN(MSG_HOST_FAILURE);
496 }
497
498 /** \ingroup msg_gos_functions
499  * \brief Return the number of MSG tasks currently running on a
500  * the host of the current running process.
501  */
502 static int MSG_get_msgload(void) 
503 {
504   m_process_t process;
505    
506   CHECK_HOST();
507   
508   xbt_assert0(0, "This function is still to be specified correctly (what do you mean by 'load', exactly?). In the meantime, please don't use it");
509   process = MSG_process_self();
510   return xbt_fifo_size(process->simdata->host->simdata->process_list);
511 }
512
513 /** \ingroup msg_gos_functions
514  *
515  * \brief Return the the last value returned by a MSG function (except
516  * MSG_get_errno...).
517  */
518 MSG_error_t MSG_get_errno(void)
519 {
520   return PROCESS_GET_ERRNO();
521 }