Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
9d4e7ec3fde6310248b6febceeb313bda1b7417f
[simgrid.git] / src / msg / global.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/log.h"
11 #include "xbt/ex.h" /* ex_backtrace_display */
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_kernel, msg,
13                                 "Logging specific to MSG (kernel)");
14
15 int __stop_at_time = -1.0 ;
16
17 MSG_Global_t msg_global = NULL;
18
19 /* static void MarkAsFailed(m_task_t t, TBX_HashTable_t failedProcessList); */
20 /* static xbt_fifo_t MSG_buildFailedHostList(double a, double b); */
21
22 /** \defgroup msg_simulation   MSG simulation Functions
23  *  \brief This section describes the functions you need to know to
24  *  set up a simulation. You should have a look at \ref MSG_examples 
25  *  to have an overview of their usage.
26  *    \htmlonly <!-- DOXYGEN_NAVBAR_LABEL="Simulation functions" --> \endhtmlonly
27  */
28
29 /********************************* MSG **************************************/
30
31 /** \ingroup msg_simulation
32  * \brief Initialize some MSG internal data.
33  */
34 void MSG_global_init_args(int *argc, char **argv)
35 {
36   MSG_global_init(argc,argv);
37 }
38
39 /** \ingroup msg_simulation
40  * \brief Initialize some MSG internal data.
41  */
42 void MSG_global_init(int *argc, char **argv)
43 {
44   if (!msg_global) {
45     surf_init(argc, argv);      /* Initialize some common structures. Warning, it sets msg_global=NULL */
46      
47     msg_global = xbt_new0(s_MSG_Global_t,1);
48
49     xbt_context_init();
50     msg_global->host = xbt_fifo_new();
51     msg_global->process_to_run = xbt_fifo_new();
52     msg_global->process_list = xbt_fifo_new();
53     msg_global->max_channel = 0;
54     msg_global->current_process = NULL;
55     msg_global->registered_functions = xbt_dict_new();
56     msg_global->PID = 1;
57     msg_global->task_mallocator = xbt_mallocator_new(256,
58                                                      (pvoid_f_void_t*) task_mallocator_new_f,
59                                                      (void_f_pvoid_t*) task_mallocator_free_f,
60                                                      (void_f_pvoid_t*) task_mallocator_reset_f);
61   }
62 }
63
64 /** \ingroup msg_easier_life
65  * \brief Traces MSG events in the Paje format.
66  */
67 void MSG_paje_output(const char *filename)
68 {
69   int i;
70   const char *paje_preembule=
71     "%EventDef  PajeDefineContainerType 1\n"
72     "%  NewType string\n"
73     "%  ContainerType   string\n"
74     "%  NewName string\n"
75     "%EndEventDef\n"
76     "%EventDef  PajeDefineStateType     3\n"
77     "%  NewType string\n"
78     "%  ContainerType   string\n"
79     "%  NewName string\n"
80     "%EndEventDef\n"
81     "%EventDef  PajeDefineLinkType      5\n"
82     "%  NewType string\n"
83     "%  ContainerType   string\n"
84     "%  SourceContainerType     string\n"
85     "%  DestContainerType       string\n"
86     "%  NewName string\n"
87     "%EndEventDef\n"
88     "%EventDef  PajeDefineEntityValue   6\n"
89     "%  NewValue        string\n"
90     "%  EntityType      string\n"
91     "%  NewName string\n"
92     "%EndEventDef\n"
93     "%EventDef  PajeCreateContainer     7\n"
94     "%  Time    date\n"
95     "%  NewContainer    string\n"
96     "%  NewContainerType        string\n"
97     "%  Container       string\n"
98     "%  NewName string\n"
99     "%EndEventDef\n"
100     "%EventDef  PajeDestroyContainer    8\n"
101     "%  Time    date\n"
102     "%  Name    string\n"
103     "%  Type    string\n"
104     "%EndEventDef\n"
105     "%EventDef  PajeSetState    10\n"
106     "%  Time    date\n"
107     "%  EntityType      string\n"
108     "%  Container       string\n"
109     "%  Value   string\n"
110     "%EndEventDef\n"
111     "%EventDef  PajePushState   11\n"
112     "%  Time    date\n"
113     "%  EntityType      string\n"
114     "%  Container       string\n"
115     "%  Value   string\n"
116     "%  TaskName        string\n"
117     "%EndEventDef\n"
118     "%EventDef  PajePopState    12\n"
119     "%  Time    date\n"
120     "%  EntityType      string\n"
121     "%  Container       string\n"
122     "%EndEventDef\n"
123     "%EventDef  PajeStartLink   16\n"
124     "%  Time    date\n"
125     "%  EntityType      string\n"
126     "%  Container       string\n"
127     "%  Value   string\n"
128     "%  SourceContainer string\n"
129     "%  Key     string\n"
130     "%EndEventDef\n"
131     "%EventDef  PajeEndLink     17\n"
132     "%  Time    date\n"
133     "%  EntityType      string\n"
134     "%  Container       string\n"
135     "%  Value   string\n"
136     "%  DestContainer   string\n"
137     "%  Key     string\n"
138     "%EndEventDef\n";
139
140   const char *type_definitions = "1     Sim_t   0       Simulation_t\n"
141     "1  H_t     Sim_t   m_host_t\n"
142     "1  P_t     H_t     m_process_t\n"
143     "3  S_t     P_t     \"Process State\"\n"
144     "6  E       S_t     Executing\n"
145     "6  B       S_t     Blocked\n"
146     "6  C       S_t     Communicating\n"
147     "5  Comm    Sim_t   P_t     P_t     Communication_t\n";
148
149   const char *ext = ".trace";
150   int ext_len = strlen(ext);
151   int len;
152   m_host_t host;
153   m_process_t process;
154   xbt_fifo_item_t item = NULL;
155
156   xbt_assert0(msg_global, "Initialize MSG first\n");
157   xbt_assert0(!msg_global->paje_output, "Paje output already defined\n");
158   xbt_assert0(filename, "Need a real file name\n");
159
160   len = strlen(filename);
161   if((len<ext_len) || (strncmp(filename+len-ext_len,ext,ext_len))) {
162     CRITICAL2("The name of the Paje trace file \"%s\" does not end by \"%s\". Paje will cause difficulties to read it.\n",
163               filename,ext);
164   }
165
166   msg_global->paje_output=fopen(filename,"w");
167   xbt_assert1(msg_global->paje_output, "Failed to open %s \n",filename);
168
169   fprintf(msg_global->paje_output,"%s",paje_preembule);
170   fprintf(msg_global->paje_output,"%s",type_definitions);
171
172   /*    Channels    */
173   for(i=0; i<msg_global->max_channel; i++) {
174     fprintf(msg_global->paje_output, "6 COMM_%d Comm    \"Channel %d\"\n" ,i,i);
175   }
176   fprintf(msg_global->paje_output,
177           "7    0.0     CUR     Sim_t   0       \"MSG simulation\"\n");
178
179   /*    Hosts       */
180   xbt_fifo_foreach(msg_global->host,item,host,m_host_t) {
181     PAJE_HOST_NEW(host);
182   }
183
184   /*    Process       */
185   xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
186     PAJE_PROCESS_NEW(process);
187   }
188 }
189
190 /** \defgroup m_channel_management    Understanding channels
191  *  \brief This section briefly describes the channel notion of MSG
192  *  (#m_channel_t).
193  *    \htmlonly <!-- DOXYGEN_NAVBAR_LABEL="Channels" --> \endhtmlonly
194  * 
195  *
196  *  For convenience, the simulator provides the notion of channel
197  *  that is close to the tag notion in MPI. A channel is not a
198  *  socket. It doesn't need to be opened neither closed. It rather
199  *  corresponds to the ports opened on the different machines.
200  */
201
202
203 /** \ingroup m_channel_management
204  * \brief Set the number of channel in the simulation.
205  *
206  * This function has to be called to fix the number of channel in the
207    simulation before creating any host. Indeed, each channel is
208    represented by a different mailbox on each #m_host_t. This
209    function can then be called only once. This function takes only one
210    parameter.
211  * \param number the number of channel in the simulation. It has to be >0
212  */
213 MSG_error_t MSG_set_channel_number(int number)
214 {
215   xbt_assert0((msg_global) && (msg_global->max_channel == 0), "Channel number already set!");
216
217   msg_global->max_channel = number;
218
219   return MSG_OK;
220 }
221
222 /** \ingroup m_channel_management
223  * \brief Return the number of channel in the simulation.
224  *
225  * This function has to be called once the number of channel is fixed. I can't 
226    figure out a reason why anyone would like to call this function but nevermind.
227  * \return the number of channel in the simulation.
228  */
229 int MSG_get_channel_number(void)
230 {
231   xbt_assert0((msg_global)&&(msg_global->max_channel != 0), "Channel number not set yet!");
232
233   return msg_global->max_channel;
234 }
235
236 void __MSG_display_process_status(void)
237 {
238    m_process_t process = NULL;
239    xbt_fifo_item_t item = NULL;
240    int i;
241    int nbprocess=xbt_fifo_size(msg_global->process_list);
242    
243    INFO1("MSG: %d processes are still running, waiting for something.",
244          nbprocess);
245    /*  List the process and their state */
246    INFO0("MSG: <process>(<pid>) on <host>: <status>.");
247    xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
248       simdata_process_t p_simdata = (simdata_process_t) process->simdata;
249       simdata_host_t h_simdata=(simdata_host_t)p_simdata->host->simdata;
250       char *who;
251         
252       asprintf(&who,"MSG:  %s(%d) on %s: %s",
253                process->name,p_simdata->PID,
254                p_simdata->host->name,
255                (process->simdata->blocked)?"[blocked] "
256                :((process->simdata->suspended)?"[suspended] ":""));
257       
258       for (i=0; i<msg_global->max_channel; i++) {
259          if (h_simdata->sleeping[i] == process) {
260             INFO2("%s\tListening on channel %d",who,i);
261             break;
262          }
263       }
264       if (i==msg_global->max_channel) {
265          if(p_simdata->waiting_task) {
266             if(p_simdata->waiting_task->simdata->compute) {
267                if(p_simdata->put_host) {
268                   INFO4("%s\tTrying to send the task '%s' to Host %s, channel %d.",
269                         who, p_simdata->waiting_task->name,p_simdata->put_host->name, p_simdata->put_channel);
270                } else {
271                   INFO2("%s\tWaiting for %s to finish.",who,p_simdata->waiting_task->name);
272                }
273             } else if (p_simdata->waiting_task->simdata->comm) {
274                INFO2("%s\tWaiting for %s to be finished transfered.",
275                      who,p_simdata->waiting_task->name);
276             } else {
277               INFO1("%s\tUNKNOWN STATUS. Please report this bug.",who);
278             }
279 /*          The following would display the trace of where the maestro thread is, 
280             since this is the thread calling this. I'd like to get the other threads to 
281             run this to see where they were blocked, but I'm not sure of how to do this */
282 /*          xbt_backtrace_display(); */
283          } else { /* Must be trying to put a task somewhere */
284             INFO1("%s\tUNKNOWN STATUS. Please report this bug.",who);
285          }
286       }
287       free(who);
288    }
289 }
290
291 /* FIXME: Yeah, I'll do it in a portable maner one day [Mt] */
292 #include <signal.h>
293
294 static void _XBT_CALL inthandler(int ignored)
295 {
296    INFO0("CTRL-C pressed. Displaying status and bailing out");
297    __MSG_display_process_status();
298    exit(1);
299 }
300
301 /** \ingroup msg_simulation
302  * \brief Launch the MSG simulation
303  */
304 MSG_error_t MSG_main(void)
305 {
306   m_process_t process = NULL;
307   int i;
308   double elapsed_time = 0.0;
309   int state_modifications = 1;
310    
311   /* Prepare to display some more info when dying on Ctrl-C pressing */
312   signal(SIGINT,inthandler);
313    
314   /* Clean IO before the run */
315   fflush(stdout);
316   fflush(stderr);
317
318   surf_solve(); /* Takes traces into account. Returns 0.0 */
319 /* xbt_fifo_size(msg_global->process_to_run) */
320   while (1) {
321     xbt_context_empty_trash();
322     if(xbt_fifo_size(msg_global->process_to_run) && (elapsed_time>0)) {
323       DEBUG0("**************************************************");
324     }
325     if((__stop_at_time>0) && (MSG_get_clock() >= __stop_at_time)) {
326       DEBUG0("Let's stop here!");
327     }
328
329     while ((process = xbt_fifo_pop(msg_global->process_to_run))) {
330       DEBUG3("Scheduling %s(%d) on %s",      
331              process->name,process->simdata->PID,
332              process->simdata->host->name);
333       msg_global->current_process = process;
334 /*       fflush(NULL); */
335       xbt_context_schedule(process->simdata->context);
336       msg_global->current_process = NULL;
337     }
338     
339     {
340       surf_action_t action = NULL;
341       surf_resource_t resource = NULL;
342       m_task_t task = NULL;
343
344       void *fun = NULL;
345       void *arg = NULL;
346
347       xbt_dynar_foreach(resource_list, i, resource) {
348         if(xbt_swag_size(resource->common_public->states.failed_action_set) ||
349            xbt_swag_size(resource->common_public->states.done_action_set))
350           state_modifications = 1;
351       }
352       
353       if(!state_modifications) {
354         DEBUG1("%f : Calling surf_solve",MSG_get_clock());
355         elapsed_time = surf_solve();
356         DEBUG1("Elapsed_time %f",elapsed_time);
357         
358         if (elapsed_time<0.0) {
359           /*       fprintf(stderr, "We're done %g\n",elapsed_time); */
360           break;
361         }
362       }
363
364       while (surf_timer_resource->extension_public->get(&fun,(void*)&arg)) {
365         DEBUG2("got %p %p", fun, arg);
366         if(fun==MSG_process_create_with_arguments) {
367           process_arg_t args = arg;
368           DEBUG2("Launching %s on %s", args->name, args->host->name);
369           process = MSG_process_create_with_arguments(args->name, args->code, 
370                                                       args->data, args->host,
371                                                       args->argc,args->argv);
372           if(args->kill_time > MSG_get_clock()) {
373             surf_timer_resource->extension_public->set(args->kill_time, 
374                                                        (void*) &MSG_process_kill,
375                                                        (void*) process);
376           }
377           xbt_free(args);
378         }
379         if(fun==MSG_process_kill) {
380           process = arg;
381           DEBUG3("Killing %s(%d) on %s", process->name, process->simdata->PID, 
382                  process->simdata->host->name);
383           MSG_process_kill(process);
384         }
385       }
386       
387       xbt_dynar_foreach(resource_list, i, resource) {
388         while ((action =
389                 xbt_swag_extract(resource->common_public->states.
390                                  failed_action_set))) {
391           task = action->data;
392           if(task) {
393             int _cursor;
394             DEBUG1("** %s failed **",task->name);
395             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
396               DEBUG3("\t preparing to wake up %s(%d) on %s",         
397                      process->name,process->simdata->PID,
398                      process->simdata->host->name);
399               xbt_fifo_unshift(msg_global->process_to_run, process);
400             }
401             process=NULL;
402           }
403         }
404         while ((action =
405                 xbt_swag_extract(resource->common_public->states.
406                                  done_action_set))) {
407           task = action->data;
408           if(task) {
409             int _cursor;
410             DEBUG1("** %s done **",task->name);
411             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
412               DEBUG3("\t preparing to wake up %s(%d) on %s",         
413                      process->name,process->simdata->PID,
414                      process->simdata->host->name);
415               xbt_fifo_unshift(msg_global->process_to_run, process);
416             }
417             process=NULL;
418           }
419         }
420       }
421     }
422     state_modifications = 0;
423   }
424
425   if (xbt_fifo_size(msg_global->process_list) == 0) {
426     INFO0("Congratulations ! Simulation terminated : all processes are over");
427     return MSG_OK;
428   } else {
429     INFO0("Oops ! Deadlock or code not perfectly clean.");
430     __MSG_display_process_status();
431     if(XBT_LOG_ISENABLED(msg, xbt_log_priority_debug) ||
432        XBT_LOG_ISENABLED(msg_kernel, xbt_log_priority_debug)) {
433       DEBUG0("Aborting!");
434       xbt_abort();
435     }
436
437     INFO0("Return a Warning.");
438     return MSG_WARNING;
439   }
440 }
441
442 /** \ingroup msg_simulation
443  * \brief Kill all running process
444
445  * \param reset_PIDs should we reset the PID numbers. A negative
446  *   number means no reset and a positive number will be used to set the PID
447  *   of the next newly created process.
448  */
449 int MSG_process_killall(int reset_PIDs)
450 {
451   m_process_t p = NULL;
452   m_process_t self = MSG_process_self();
453
454   while((p=xbt_fifo_pop(msg_global->process_list))) {
455     if(p!=self) MSG_process_kill(p);
456   }
457
458   if(reset_PIDs>0) {
459     msg_global->PID = reset_PIDs; 
460     msg_global->session++;
461  }
462
463   xbt_context_empty_trash();
464
465   if(self) {
466     xbt_context_yield();
467   }
468
469   return msg_global->PID;
470 }
471
472 /** \ingroup msg_simulation
473  * \brief Clean the MSG simulation
474  */
475 MSG_error_t MSG_clean(void)
476 {
477   xbt_fifo_item_t i = NULL;
478   m_host_t h = NULL;
479   m_process_t p = NULL;
480
481
482   while((p=xbt_fifo_pop(msg_global->process_list))) {
483     MSG_process_kill(p);
484   }
485   xbt_context_exit();
486
487   xbt_fifo_foreach(msg_global->host,i,h,m_host_t) {
488     __MSG_host_destroy(h);
489   }
490   xbt_fifo_free(msg_global->host);
491   xbt_fifo_free(msg_global->process_to_run);
492   xbt_fifo_free(msg_global->process_list);
493   xbt_dict_free(&(msg_global->registered_functions));
494   xbt_mallocator_free(msg_global->task_mallocator);
495
496   if(msg_global->paje_output) {
497     fclose(msg_global->paje_output);
498     msg_global->paje_output = NULL;
499   }
500   msg_config_finalize();
501   free(msg_global);
502   surf_exit();
503
504   return MSG_OK;
505 }
506
507
508 /** \ingroup msg_easier_life
509  * \brief A clock (in second).
510  */
511 double MSG_get_clock(void) {
512   return surf_get_clock();
513 }
514