Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
ba982f87d898ddc6486bd9e0e76bf5dcd59cb69f
[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 parameter.
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   m_process_t process = MSG_process_self();
47   m_task_t t = NULL;
48   m_host_t h = NULL;
49   simdata_task_t t_simdata = NULL;
50   simdata_host_t h_simdata = NULL;
51   int warning = 0;
52   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
53   
54   CHECK_HOST();
55   /* Sanity check */
56   xbt_assert0(task,"Null pointer for the task\n");
57
58   if (*task) 
59     CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
60
61   /* Get the task */
62   h = MSG_host_self();
63   h_simdata = h->simdata;
64   while ((t = xbt_fifo_pop(h_simdata->mbox[channel])) == NULL) {
65     xbt_assert2(!(h_simdata->sleeping[channel]),
66                 "A process (%s(%d)) is already blocked on this channel",
67                 h_simdata->sleeping[channel]->name,
68                 h_simdata->sleeping[channel]->simdata->PID);
69     h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
70     __MSG_process_block();
71     if(surf_workstation_resource->extension_public->get_state(h_simdata->host) 
72        == SURF_CPU_OFF)
73       MSG_RETURN(MSG_HOST_FAILURE);
74     h_simdata->sleeping[channel] = NULL;
75     /* OK, we should both be ready now. Are you there ? */
76   }
77
78   t_simdata = t->simdata;
79   /*   *task = __MSG_task_copy(t); */
80   *task=t;
81
82   /* Transfer */
83   t_simdata->using++;
84
85   t_simdata->comm = surf_workstation_resource->extension_public->
86     communicate(MSG_process_get_host(t_simdata->sender)->simdata->host,
87                 h->simdata->host, t_simdata->message_size,t_simdata->rate);
88   
89   surf_workstation_resource->common_public->action_set_data(t_simdata->comm,t);
90
91   if(__MSG_process_isBlocked(t_simdata->sender)) 
92     __MSG_process_unblock(t_simdata->sender);
93
94   PAJE_PROCESS_PUSH_STATE(process,"C");  
95
96   do {
97     __MSG_task_wait_event(process, t);
98     state=surf_workstation_resource->common_public->action_get_state(t_simdata->comm);
99   } while (state==SURF_ACTION_RUNNING);
100
101   if(t->simdata->using>1) {
102     xbt_fifo_unshift(msg_global->process_to_run,process);
103     xbt_context_yield();
104   }
105
106   PAJE_PROCESS_POP_STATE(process);  
107   PAJE_COMM_STOP(process,t,channel);
108
109   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
110   else if(surf_workstation_resource->extension_public->get_state(h_simdata->host) 
111           == SURF_CPU_OFF)
112     MSG_RETURN(MSG_HOST_FAILURE);
113   else MSG_RETURN(MSG_TRANSFER_FAILURE);
114 }
115
116 /** \ingroup msg_gos_functions
117  * \brief Test whether there is a pending communication on a channel.
118  *
119  * It takes one parameter.
120  * \param channel the channel on which the agent should be
121    listening. This value has to be >=0 and < than the maximal
122    number of channels fixed with MSG_set_channel_number().
123  * \return 1 if there is a pending communication and 0 otherwise
124  */
125 int MSG_task_Iprobe(m_channel_t channel)
126 {
127   m_host_t h = NULL;
128   simdata_host_t h_simdata = NULL;
129
130   CHECK_HOST();
131   h = MSG_host_self();
132   h_simdata = h->simdata;
133   return(xbt_fifo_getFirstItem(h_simdata->mbox[channel])!=NULL);
134 }
135
136 /** \ingroup msg_gos_functions
137  * \brief Test whether there is a pending communication on a channel, and who sent it.
138  *
139  * It takes one parameter.
140  * \param channel the channel on which the agent should be
141    listening. This value has to be >=0 and < than the maximal
142    number of channels fixed with MSG_set_channel_number().
143  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
144  */
145 int MSG_task_probe_from(m_channel_t channel)
146 {
147   m_host_t h = NULL;
148   simdata_host_t h_simdata = NULL;
149   xbt_fifo_item_t item;
150   m_task_t t;
151
152   CHECK_HOST();
153   h = MSG_host_self();
154   h_simdata = h->simdata;
155    
156   item = xbt_fifo_getFirstItem(((simdata_host_t)h->simdata)->mbox[channel]);
157   if (!item || !(t = xbt_fifo_get_item_content(item)) || (simdata_task_t)t->simdata)
158     return -1;
159    
160   return MSG_process_get_PID(((simdata_task_t)t->simdata)->sender);
161 }
162
163 /** \ingroup msg_gos_functions
164  * \brief Put a task on a channel of an host and waits for the end of the
165  * transmission.
166  *
167  * This function is used for describing the behavior of an agent. It
168  * takes three parameter.
169  * \param task a #m_task_t to send on another location. This task
170    will not be usable anymore when the function will return. There is
171    no automatic task duplication and you have to save your parameters
172    before calling this function. Tasks are unique and once it has been
173    sent to another location, you should not access it anymore. You do
174    not need to call MSG_task_destroy() but to avoid using, as an
175    effect of inattention, this task anymore, you definitely should
176    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
177    can be transfered iff it has been correctly created with
178    MSG_task_create().
179  * \param dest the destination of the message
180  * \param channel the channel on which the agent should put this
181    task. This value has to be >=0 and < than the maximal number of
182    channels fixed with MSG_set_channel_number().
183  * \return #MSG_FATAL if \a task is not properly initialized and
184  * #MSG_OK otherwise.
185  */
186 MSG_error_t MSG_task_put(m_task_t task,
187                          m_host_t dest, m_channel_t channel)
188 {
189   m_process_t process = MSG_process_self();
190   simdata_task_t task_simdata = NULL;
191   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
192   m_host_t local_host = NULL;
193   m_host_t remote_host = NULL;
194
195   CHECK_HOST();
196
197   task_simdata = task->simdata;
198   task_simdata->sender = process;
199   xbt_assert0(task_simdata->using==1,"Gargl!");
200   task_simdata->comm = NULL;
201   
202   local_host = ((simdata_process_t) process->simdata)->host;
203   remote_host = dest;
204
205   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
206                 mbox[channel], task);
207
208   PAJE_COMM_START(process,task,channel);
209     
210   if(remote_host->simdata->sleeping[channel]) 
211     __MSG_process_unblock(remote_host->simdata->sleeping[channel]);
212
213   process->simdata->put_host = dest;
214   process->simdata->put_channel = channel;
215   while(!(task_simdata->comm)) 
216     __MSG_process_block();
217   process->simdata->put_host = NULL;
218   process->simdata->put_channel = -1;
219
220
221   PAJE_PROCESS_PUSH_STATE(process,"C");  
222
223   state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
224   while (state==SURF_ACTION_RUNNING) {
225     __MSG_task_wait_event(process, task);
226     state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
227   }
228     
229   MSG_task_destroy(task);
230
231   PAJE_PROCESS_POP_STATE(process);  
232
233   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
234   else if(surf_workstation_resource->extension_public->get_state(local_host->simdata->host) 
235           == SURF_CPU_OFF)
236     MSG_RETURN(MSG_HOST_FAILURE);
237   else MSG_RETURN(MSG_TRANSFER_FAILURE);
238 }
239
240 /** \ingroup msg_gos_functions
241  * \brief Does exactly the same as MSG_task_put but with a bounded transmition 
242  * rate.
243  *
244  * \sa MSG_task_put
245  */
246 MSG_error_t MSG_task_put_bounded(m_task_t task,
247                                  m_host_t dest, m_channel_t channel,
248                                  long double max_rate)
249 {
250   task->simdata->rate=max_rate;
251   return(MSG_task_put(task, dest, channel));
252   task->simdata->rate=-1.0;
253 }
254
255 /** \ingroup msg_gos_functions
256  * \brief Executes a task and waits for its termination.
257  *
258  * This function is used for describing the behavior of an agent. It
259  * takes only one parameter.
260  * \param task a #m_task_t to execute on the location on which the
261    agent is running.
262  * \return #MSG_FATAL if \a task is not properly initialized and
263  * #MSG_OK otherwise.
264  */
265 MSG_error_t MSG_task_execute(m_task_t task)
266 {
267   m_process_t process = MSG_process_self();
268   MSG_error_t res;
269   __MSG_task_execute(process, task);
270
271   PAJE_PROCESS_PUSH_STATE(process,"E");  
272   res = __MSG_wait_for_computation(process,task);
273   PAJE_PROCESS_POP_STATE(process);
274   return res;
275 }
276
277 void __MSG_task_execute(m_process_t process, m_task_t task)
278 {
279   simdata_task_t simdata = NULL;
280
281   CHECK_HOST();
282
283   simdata = task->simdata;
284
285   simdata->compute = surf_workstation_resource->extension_public->
286     execute(MSG_process_get_host(process)->simdata->host,
287             simdata->computation_amount);
288   surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
289 }
290
291 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
292 {
293   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
294   simdata_task_t simdata = task->simdata;
295
296   simdata->using++;
297   do {
298     __MSG_task_wait_event(process, task);
299     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
300   } while (state==SURF_ACTION_RUNNING);
301   simdata->using--;
302     
303
304   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
305   else if(surf_workstation_resource->extension_public->
306           get_state(MSG_process_get_host(process)->simdata->host) 
307           == SURF_CPU_OFF)
308     MSG_RETURN(MSG_HOST_FAILURE);
309   else MSG_RETURN(MSG_TRANSFER_FAILURE);
310 }
311
312 /** \ingroup msg_gos_functions
313  * \brief Sleep for the specified number of seconds
314  *
315  * Makes the current process sleep until \a time seconds have elapsed.
316  *
317  * \param nb_sec a number of second
318  */
319 MSG_error_t MSG_process_sleep(long double nb_sec)
320 {
321   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
322   m_process_t process = MSG_process_self();
323   m_task_t dummy = NULL;
324   simdata_task_t simdata = NULL;
325
326   CHECK_HOST();
327   dummy = MSG_task_create("MSG_sleep", nb_sec, 0.0, NULL);
328   simdata = dummy->simdata;
329
330   simdata->compute = surf_workstation_resource->extension_public->
331     sleep(MSG_process_get_host(process)->simdata->host,
332             simdata->computation_amount);
333   surf_workstation_resource->common_public->action_set_data(simdata->compute,dummy);
334
335   
336   simdata->using++;
337   do {
338     __MSG_task_wait_event(process, dummy);
339     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
340   } while (state==SURF_ACTION_RUNNING);
341   simdata->using--;
342     
343   if(state == SURF_ACTION_DONE) {
344     if(surf_workstation_resource->extension_public->
345        get_state(MSG_process_get_host(process)->simdata->host) 
346        == SURF_CPU_OFF)
347       MSG_RETURN(MSG_HOST_FAILURE);
348
349     if(__MSG_process_isBlocked(process)) {
350       __MSG_process_unblock(MSG_process_self());
351     }
352     if(surf_workstation_resource->extension_public->
353        get_state(MSG_process_get_host(process)->simdata->host) 
354        == SURF_CPU_OFF)
355       MSG_RETURN(MSG_HOST_FAILURE);
356     MSG_task_destroy(dummy);
357     MSG_RETURN(MSG_OK);
358   } else MSG_RETURN(MSG_HOST_FAILURE);
359 }
360
361 /** \ingroup msg_gos_functions
362  * \brief Return the number of MSG tasks currently running on a
363  * the host of the current running process.
364  */
365 int MSG_get_msgload(void) 
366 {
367   CHECK_HOST();
368   xbt_assert0(0,"Not implemented yet!");
369   
370   return 1;
371 }
372
373 /** \ingroup msg_gos_functions
374  *
375  * \brief Return the the last value returned by a MSG function (except
376  * MSG_get_errno...).
377  */
378 MSG_error_t MSG_get_errno(void)
379 {
380   return PROCESS_GET_ERRNO();
381 }