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