Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
remove this damn fflush that spoils the performance (even though it was useful when...
[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/error.h"
11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(global, msg,
12                                 "Logging specific to MSG (global)");
13
14 int __stop_at_time = -1.0 ;
15
16 MSG_Global_t msg_global = NULL;
17
18 /* static void MarkAsFailed(m_task_t t, TBX_HashTable_t failedProcessList); */
19 /* static xbt_fifo_t MSG_buildFailedHostList(double a, double b); */
20
21 /** \defgroup msg_simulation   MSG simulation Functions
22  *  \brief This section describes the functions you need to know to
23  *  set up a simulation. You should have a look at \ref MSG_examples 
24  *  to have an overview of their usage.
25  */
26
27 /********************************* MSG **************************************/
28
29 /** \ingroup msg_simulation
30  * \brief Initialize some MSG internal data.
31  */
32 void MSG_global_init(void)
33 {
34   int argc=0;
35   char **argv=NULL;
36
37   CRITICAL0("Function MSG_global_init() is deprecated by MSG_global_init_args().");
38   MSG_global_init_args(&argc,argv);
39 }
40
41 /** \ingroup msg_simulation
42  * \brief Initialize some MSG internal data.
43  */
44 void MSG_global_init_args(int *argc, char **argv)
45 {
46   if (!msg_global) {
47     surf_init(argc, argv);      /* Initialize some common structures. Warning, it sets msg_global=NULL */
48      
49     msg_global = xbt_new0(s_MSG_Global_t,1);
50
51     xbt_context_init();
52     msg_global->host = xbt_fifo_new();
53     msg_global->process_to_run = xbt_fifo_new();
54     msg_global->process_list = xbt_fifo_new();
55     msg_global->max_channel = 0;
56     msg_global->current_process = NULL;
57     msg_global->registered_functions = xbt_dict_new();
58     msg_global->PID = 1;
59   }
60 }
61
62 /** \ingroup msg_easier_life
63  * \brief Traces MSG events in the Paje format.
64  */
65 void MSG_paje_output(const char *filename)
66 {
67   int i;
68   const char *paje_preembule="%EventDef SetLimits       0\n"
69     "%  StartTime       date\n"
70     "%  EndTime date\n"
71     "%EndEventDef\n"
72     "%EventDef  PajeDefineContainerType 1\n"
73     "%  NewType string\n"
74     "%  ContainerType   string\n"
75     "%  NewName string\n"
76     "%EndEventDef\n"
77     "%EventDef  PajeDefineEventType     2\n"
78     "%  NewType string\n"
79     "%  ContainerType   string\n"
80     "%  NewName string\n"
81     "%EndEventDef\n"
82     "%EventDef  PajeDefineStateType     3\n"
83     "%  NewType string\n"
84     "%  ContainerType   string\n"
85     "%  NewName string\n"
86     "%EndEventDef\n"
87     "%EventDef  PajeDefineVariableType  4\n"
88     "%  NewType string\n"
89     "%  ContainerType   string\n"
90     "%  NewName string\n"
91     "%EndEventDef\n"
92     "%EventDef  PajeDefineLinkType      5\n"
93     "%  NewType string\n"
94     "%  ContainerType   string\n"
95     "%  SourceContainerType     string\n"
96     "%  DestContainerType       string\n"
97     "%  NewName string\n"
98     "%EndEventDef\n"
99     "%EventDef  PajeDefineEntityValue   6\n"
100     "%  NewValue        string\n"
101     "%  EntityType      string\n"
102     "%  NewName string\n"
103     "%EndEventDef\n"
104     "%EventDef  PajeCreateContainer     7\n"
105     "%  Time    date\n"
106     "%  NewContainer    string\n"
107     "%  NewContainerType        string\n"
108     "%  Container       string\n"
109     "%  NewName string\n"
110     "%EndEventDef\n"
111     "%EventDef  PajeDestroyContainer    8\n"
112     "%  Time    date\n"
113     "%  Name    string\n"
114     "%  Type    string\n"
115     "%EndEventDef\n"
116     "%EventDef  PajeNewEvent    9\n"
117     "%  Time    date\n"
118     "%  EntityType      string\n"
119     "%  Container       string\n"
120     "%  Value   string\n"
121     "%EndEventDef\n"
122     "%EventDef  PajeSetState    10\n"
123     "%  Time    date\n"
124     "%  EntityType      string\n"
125     "%  Container       string\n"
126     "%  Value   string\n"
127     "%EndEventDef\n"
128     "%EventDef  PajeSetState    101\n"
129     "%  Time    date\n"
130     "%  EntityType      string\n"
131     "%  Container       string\n"
132     "%  Value   string\n"
133     "%  FileName        string\n"
134     "%  LineNumber      int\n"
135     "%EndEventDef\n"
136     "%EventDef  PajePushState   111\n"
137     "%  Time    date\n"
138     "%  EntityType      string\n"
139     "%  Container       string\n"
140     "%  Value   string\n"
141     "%  FileName        string\n"
142     "%  LineNumber      int\n"
143     "%EndEventDef\n"
144     "%EventDef  PajePushState   11\n"
145     "%  Time    date\n"
146     "%  EntityType      string\n"
147     "%  Container       string\n"
148     "%  Value   string\n"
149     "%EndEventDef\n"
150     "%EventDef  PajePopState    12\n"
151     "%  Time    date\n"
152     "%  EntityType      string\n"
153     "%  Container       string\n"
154     "%EndEventDef\n"
155     "%EventDef  PajeSetVariable 13\n"
156     "%  Time    date\n"
157     "%  EntityType      string\n"
158     "%  Container       string\n"
159     "%  Value   double\n"
160     "%EndEventDef\n"
161     "%EventDef  PajeAddVariable 14\n"
162     "%  Time    date\n"
163     "%  EntityType      string\n"
164     "%  Container       string\n"
165     "%  Value   double\n"
166     "%EndEventDef\n"
167     "%EventDef  PajeSubVariable 15\n"
168     "%  Time    date\n"
169     "%  EntityType      string\n"
170     "%  Container       string\n"
171     "%  Value   double\n"
172     "%EndEventDef\n"
173     "%EventDef  PajeStartLink   16\n"
174     "%  Time    date\n"
175     "%  EntityType      string\n"
176     "%  Container       string\n"
177     "%  Value   string\n"
178     "%  SourceContainer string\n"
179     "%  Key     string\n"
180     "%EndEventDef\n"
181     "%EventDef  PajeEndLink     17\n"
182     "%  Time    date\n"
183     "%  EntityType      string\n"
184     "%  Container       string\n"
185     "%  Value   string\n"
186     "%  DestContainer   string\n"
187     "%  Key     string\n"
188     "%EndEventDef\n";
189
190   const char *type_definitions = "1     Sim_t   0       Simulation_t\n"
191     "1  H_t     Sim_t   m_host_t\n"
192     "1  P_t     H_t     m_process_t\n"
193     "3  S_t     P_t     \"Process State\"\n"
194     "6  E       S_t     Executing\n"
195     "6  B       S_t     Blocked\n"
196     "6  C       S_t     Communicating\n"
197     "5  Comm    Sim_t   P_t     P_t     Communication_t\n";
198
199   const char *ext = ".trace";
200   int ext_len = strlen(ext);
201   int len;
202   m_host_t host;
203   m_process_t process;
204   xbt_fifo_item_t item = NULL;
205
206   xbt_assert0(msg_global, "Initialize MSG first\n");
207   xbt_assert0(!msg_global->paje_output, "Paje output already defined\n");
208   xbt_assert0(filename, "Need a real file name\n");
209
210   len = strlen(filename);
211   if((len<ext_len) || (strncmp(filename+len-ext_len,ext,ext_len))) {
212     CRITICAL2("%s does not end by \"%s\". It may cause troubles when using Paje\n",
213               filename,ext);
214   }
215
216   msg_global->paje_output=fopen(filename,"w");
217   xbt_assert1(msg_global->paje_output, "Failed to open %s \n",filename);
218
219   fprintf(msg_global->paje_output,"%s",paje_preembule);
220   fprintf(msg_global->paje_output,"%s",type_definitions);
221
222   /*    Channels    */
223   for(i=0; i<msg_global->max_channel; i++) {
224     fprintf(msg_global->paje_output, "6 COMM_%d Comm    \"Channel %d\"\n" ,i,i);
225   }
226   fprintf(msg_global->paje_output,
227           "7    0.0     CUR     Sim_t   0       \"MSG simulation\"\n");
228
229   /*    Hosts       */
230   xbt_fifo_foreach(msg_global->host,item,host,m_host_t) {
231     PAJE_HOST_NEW(host);
232   }
233
234   /*    Process       */
235   xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
236     PAJE_PROCESS_NEW(process);
237   }
238 }
239
240 /** \ingroup msg_simulation
241  * \brief Defines the verbosity of the simulation.
242  */
243 void MSG_set_verbosity(MSG_outputmode_t mode)
244 {
245   CRITICAL0("MSG_set_verbosity : Deprecated function. Use the XBT logging interface.");
246 }
247
248 /** \defgroup m_channel_management    Understanding channels
249  *  \brief This section briefly describes the channel notion of MSG
250  *  (#m_channel_t).
251  *
252  *  For convenience, the simulator provides the notion of channel
253  *  that is close to the tag notion in MPI. A channel is not a
254  *  socket. It doesn't need to be opened neither closed. It rather
255  *  corresponds to the ports opened on the different machines.
256  */
257
258
259 /** \ingroup m_channel_management
260  * \brief Set the number of channel in the simulation.
261  *
262  * This function has to be called to fix the number of channel in the
263    simulation before creating any host. Indeed, each channel is
264    represented by a different mailbox on each #m_host_t. This
265    function can then be called only once. This function takes only one
266    parameter.
267  * \param number the number of channel in the simulation. It has to be >0
268  */
269 MSG_error_t MSG_set_channel_number(int number)
270 {
271   xbt_assert0((msg_global) && (msg_global->max_channel == 0), "Channel number already set!");
272
273   msg_global->max_channel = number;
274
275   return MSG_OK;
276 }
277
278 /** \ingroup m_simulation
279  * \brief Set the sharing policy used for the links
280  *
281  * This function can be called to change the sharing policy used for the links 
282    (see \ref paper_tcp). By default the store and forward mechanism is used
283    with a parameter equal to 0.1. This function has to be called before creating 
284    any link. 
285  * \param mode the sharing policy used for the links: can be either 
286    MSG_STORE_AND_FORWARD or MSG_TCP.
287  * \param param a parameter for the sharing policy. It has to be >0. It is 
288    currently used only for the MSG_STORE_AND_FORWARD flavor and represents the
289    granularity of the communications (i.e. the packet size).
290  */
291 MSG_error_t MSG_set_sharing_policy(MSG_sharing_t mode, double param)
292 {
293   CRITICAL0("MSG_set_sharing_policy: this function is now deprecated and useless. Store and forward does not exist anymore. Please stop using it.");
294   
295   return MSG_OK;
296 }
297
298 /** \ingroup m_channel_management
299  * \brief Return the number of channel in the simulation.
300  *
301  * This function has to be called once the number of channel is fixed. I can't 
302    figure out a reason why anyone would like to call this function but nevermind.
303  * \return the number of channel in the simulation.
304  */
305 int MSG_get_channel_number(void)
306 {
307   xbt_assert0((msg_global)&&(msg_global->max_channel != 0), "Channel number not set yet!");
308
309   return msg_global->max_channel;
310 }
311
312 /** \ingroup msg_simulation
313  * \brief Launch the MSG simulation
314  */
315 MSG_error_t MSG_main(void)
316 {
317   m_process_t process = NULL;
318   int nbprocess,i;
319   double elapsed_time = 0.0;
320
321   /* Clean IO before the run */
322   fflush(stdout);
323   fflush(stderr);
324
325   surf_solve(); /* Takes traces into account. Returns 0.0 */
326 /* xbt_fifo_size(msg_global->process_to_run) */
327   while (1) {
328     xbt_context_empty_trash();
329     if(xbt_fifo_size(msg_global->process_to_run) && (elapsed_time>0)) {
330       DEBUG0("**************************************************");
331     }
332     if((__stop_at_time>0) && (MSG_getClock() >= __stop_at_time)) {
333       DEBUG0("Let's stop here!");
334     }
335
336     while ((process = xbt_fifo_pop(msg_global->process_to_run))) {
337       DEBUG3("Scheduling  %s(%d) on %s",             
338              process->name,process->simdata->PID,
339              process->simdata->host->name);
340       msg_global->current_process = process;
341 /*       fflush(NULL); */
342       xbt_context_schedule(process->simdata->context);
343       msg_global->current_process = NULL;
344     }
345     DEBUG1("%lg : Calling surf_solve",MSG_getClock());
346     elapsed_time = surf_solve();
347     DEBUG1("Elapsed_time %lg",elapsed_time);
348
349 /*     fprintf(stderr, "====== %lg =====\n",Now); */
350 /*     if (elapsed_time==0.0) { */
351 /*       fprintf(stderr, "No change in time\n"); */
352 /*     } */
353     if (elapsed_time<0.0) {
354 /*       fprintf(stderr, "We're done %lg\n",elapsed_time); */
355       break;
356     }
357
358     {
359       surf_action_t action = NULL;
360       surf_resource_t resource = NULL;
361       m_task_t task = NULL;
362
363       void *fun = NULL;
364       void *arg = NULL;
365       while (surf_timer_resource->extension_public->get(&fun,(void*)&arg)) {
366         DEBUG2("got %p %p", fun, arg);
367         if(fun==MSG_process_create_with_arguments) {
368           process_arg_t args = arg;
369           DEBUG2("Launching %s on %s", args->name, args->host->name);
370           process = MSG_process_create_with_arguments(args->name, args->code, 
371                                                       args->data, args->host,
372                                                       args->argc,args->argv);
373           if(args->kill_time > MSG_getClock()) {
374             surf_timer_resource->extension_public->set(args->kill_time, 
375                                                        (void*) &MSG_process_kill,
376                                                        (void*) process);
377           }
378           xbt_free(args);
379         }
380         if(fun==MSG_process_kill) {
381           process = arg;
382           DEBUG3("Killing %s(%d) on %s", process->name, process->simdata->PID, 
383                  process->simdata->host->name);
384           MSG_process_kill(process);
385         }
386       }
387       
388       xbt_dynar_foreach(resource_list, i, resource) {
389         while ((action =
390                xbt_swag_extract(resource->common_public->states.
391                                 failed_action_set))) {
392           task = action->data;
393           if(task) {
394             int _cursor;
395             DEBUG1("** %s failed **",task->name);
396             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
397               DEBUG3("\t preparing to wake up %s(%d) on %s",         
398                      process->name,process->simdata->PID,
399                      process->simdata->host->name);
400               xbt_fifo_unshift(msg_global->process_to_run, process);
401             }
402             process=NULL;
403           }
404         }
405         while ((action =
406                xbt_swag_extract(resource->common_public->states.
407                                 done_action_set))) {
408           task = action->data;
409           if(task) {
410             int _cursor;
411             DEBUG1("** %s done **",task->name);
412             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
413               DEBUG3("\t preparing to wake up %s(%d) on %s",         
414                      process->name,process->simdata->PID,
415                      process->simdata->host->name);
416               xbt_fifo_unshift(msg_global->process_to_run, process);
417             }
418             process=NULL;
419           }
420         }
421       }
422     }
423   }
424
425   if ((nbprocess=xbt_fifo_size(msg_global->process_list)) == 0) {
426     fprintf(stderr,
427             "MSG: Congratulations ! Simulation terminated : all process are over\n");
428     return MSG_OK;
429   } else {
430     xbt_fifo_item_t item = NULL;
431     fprintf(stderr,"MSG: Oops ! Deadlock or code not perfectly clean.\n");
432     fprintf(stderr,"MSG: %d processes are still running, waiting for something.\n",
433             nbprocess);
434     /*  List the process and their state */
435     fprintf(stderr,"MSG: <process>(<pid>) on <host>: <status>.\n");
436     xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
437       simdata_process_t p_simdata = (simdata_process_t) process->simdata;
438       simdata_host_t h_simdata=(simdata_host_t)p_simdata->host->simdata;
439       
440
441       fprintf(stderr,"MSG:  %s(%d) on %s: ",
442              process->name,p_simdata->PID,
443              p_simdata->host->name);
444
445       if (process->simdata->blocked)      
446         fprintf(stderr,"[blocked] ");
447       if (process->simdata->suspended)    
448         fprintf(stderr,"[suspended] ");
449
450       for (i=0; i<msg_global->max_channel; i++) {
451         if (h_simdata->sleeping[i] == process) {
452           fprintf(stderr,"Listening on channel %d.\n",i);
453           break;
454         }
455       }
456       if (i==msg_global->max_channel) {
457         if(p_simdata->waiting_task) {
458           if(p_simdata->waiting_task->simdata->compute) {
459             if(p_simdata->put_host) 
460               fprintf(stderr,"Trying to send a task on Host %s, channel %d.\n",
461                       p_simdata->put_host->name, p_simdata->put_channel);
462             else 
463               fprintf(stderr,"Waiting for %s to finish.\n",p_simdata->waiting_task->name);
464           } else if (p_simdata->waiting_task->simdata->comm)
465             fprintf(stderr,"Waiting for %s to be finished transfered.\n",
466                     p_simdata->waiting_task->name);
467           else
468             fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
469         }
470         else { /* Must be trying to put a task somewhere */
471           fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
472         }
473       } 
474     }
475     if(XBT_LOG_ISENABLED(msg, xbt_log_priority_debug) ||
476        XBT_LOG_ISENABLED(global, xbt_log_priority_debug)) {
477       DEBUG0("Aborting!");
478       xbt_abort();
479     }
480
481     return MSG_WARNING;
482   }
483 }
484
485 /* static void MarkAsFailed(m_task_t t, TBX_HashTable_t failedProcessList)  */
486 /* { */
487 /*   simdata_task_t simdata = NULL; */
488 /*   xbt_fifo_item_t i = NULL; */
489 /*   m_process_t p = NULL; */
490   
491 /*   xbt_assert0((t!=NULL),"Invalid task"); */
492 /*   simdata = t->simdata; */
493
494 /* #define KILL(task) if(task) SG_failTask(task) */
495 /*   KILL(simdata->compute); */
496 /*   KILL(simdata->TCP_comm); */
497 /*   KILL(simdata->s[0]); */
498 /*   KILL(simdata->s[1]); */
499 /*   KILL(simdata->s[2]); */
500 /*   KILL(simdata->s[3]); */
501 /*   KILL(simdata->sleep); */
502 /* #undef KILL   */
503 /* /\*   if(simdata->comm) SG_failEndToEndTransfer(simdata->comm); *\/ */
504   
505 /*   xbt_fifo_foreach(simdata->sleeping,i,p,m_process_t) { */
506 /*     if(!TBX_HashTable_isInList(failedProcessList,p,TBX_basicHash))  */
507 /*       TBX_HashTable_insert(failedProcessList,p,TBX_basicHash); */
508 /*   } */
509   
510 /* } */
511
512 /* static xbt_fifo_t MSG_buildFailedHostList(double begin, double end) */
513 /* { */
514 /*   xbt_fifo_t failedHostList = xbt_fifo_new(); */
515 /*   m_host_t host = NULL; */
516 /*   xbt_fifo_item_t i; */
517
518 /*   xbt_fifo_foreach(msg_global->host,i,host,m_host_t) { */
519 /*     SG_Resource r= ((simdata_host_t) (host->simdata))->host; */
520
521 /*     if(SG_evaluateFailureTrace(r->failure_trace,begin,end)!=-1.0) */
522 /*       xbt_fifo_insert(failedHostList,host); */
523 /*   } */
524
525 /*   return failedHostList; */
526 /* } */
527
528 /** \ingroup msg_simulation
529  * \brief Kill all running process
530
531  * \param reset_PIDs should we reset the PID numbers. A negative
532  *   number means no reset and a positive number will be used to set the PID
533  *   of the next newly created process.
534  */
535 int MSG_process_killall(int reset_PIDs)
536 {
537   xbt_fifo_item_t i = NULL;
538   m_process_t p = NULL;
539   m_process_t self = MSG_process_self();
540
541   while((p=xbt_fifo_shift(msg_global->process_list))) {
542     if(p!=self) MSG_process_kill(p);
543   }
544
545   if(reset_PIDs>0) {
546     msg_global->PID = reset_PIDs; 
547     msg_global->session++;
548  }
549
550   xbt_context_empty_trash();
551
552   if(self) {
553     xbt_context_yield();
554   }
555
556   return msg_global->PID;
557 }
558
559 /** \ingroup msg_simulation
560  * \brief Clean the MSG simulation
561  */
562 MSG_error_t MSG_clean(void)
563 {
564   xbt_fifo_item_t i = NULL;
565   m_host_t h = NULL;
566   m_process_t p = NULL;
567
568
569   while((p=xbt_fifo_shift(msg_global->process_list))) {
570     MSG_process_kill(p);
571   }
572   xbt_context_exit();
573
574   xbt_fifo_foreach(msg_global->host,i,h,m_host_t) {
575     __MSG_host_destroy(h);
576   }
577   xbt_fifo_free(msg_global->host);
578   xbt_fifo_free(msg_global->process_to_run);
579   xbt_fifo_free(msg_global->process_list);
580   xbt_dict_free(&(msg_global->registered_functions));
581
582   if(msg_global->paje_output) {
583     fclose(msg_global->paje_output);
584     msg_global->paje_output = NULL;
585   }
586   free(msg_global);
587   surf_finalize();
588
589   return MSG_OK;
590 }
591
592
593 /** \ingroup msg_easier_life
594  * \brief A clock (in second).
595  * \deprecated Use MSG_get_clock
596  */
597 double MSG_getClock(void) {
598   return surf_get_clock();
599 }
600
601 /** \ingroup msg_easier_life
602  * \brief A clock (in second).
603  */
604 double MSG_get_clock(void) {
605   return surf_get_clock();
606 }
607