src/msg/m_process.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 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_process, msg,
  12                                 "Logging specific to MSG (process)");
  13
  14 /** \defgroup m_process_management Management Functions of Agents
  15  *  \brief This section describes the agent structure of MSG
  16  */
  17 /** \addtogroup m_process_management
  18  *  (#m_process_t) and the functions for managing it.
  19  *    \htmlonly <!-- DOXYGEN_NAVBAR_LABEL="Agents" --> \endhtmlonly
  20  *
  21  *  We need to simulate many independent scheduling decisions, so
  22  *  the concept of <em>process</em> is at the heart of the
  23  *  simulator. A process may be defined as a <em>code</em>, with
  24  *  some <em>private data</em>, executing in a <em>location</em>.
  25  *  \see m_process_t
  26  */
  27
  28 /******************************** Process ************************************/
  29 /** \ingroup m_process_management
  30  * \brief Creates and runs a new #m_process_t.
  31  *
  32  * Does exactly the same as #MSG_process_create_with_arguments but without
  33    providing standard arguments (\a argc, \a argv, \a start_time, \a kill_time).
  34  * \sa MSG_process_create_with_arguments
  35  */
  36 m_process_t MSG_process_create(const char *name,
  37                                m_process_code_t code, void *data,
  38                                m_host_t host)
  39 {
  40   return MSG_process_create_with_arguments(name, code, data, host, -1, NULL);
  41 }
  42
  43 static void MSG_process_cleanup(void *arg)
  44 {
  45
  46   while(((m_process_t)arg)->simdata->paje_state) {
  47     PAJE_PROCESS_POP_STATE((m_process_t)arg);
  48   }
  49
  50   PAJE_PROCESS_FREE(arg);
  51
  52   xbt_fifo_remove(msg_global->process_list, arg);
  53   xbt_fifo_remove(msg_global->process_to_run, arg);
  54   xbt_fifo_remove(((m_process_t) arg)->simdata->host->simdata->process_list, arg);
  55   free(((m_process_t) arg)->name);
  56   ((m_process_t) arg)->name = NULL;
  57   free(((m_process_t) arg)->simdata);
  58   ((m_process_t) arg)->simdata = NULL;
  59   free(arg);
  60 }
  61
  62 /** \ingroup m_process_management
  63  * \brief Creates and runs a new #m_process_t.
  64
  65  * A constructor for #m_process_t taking four arguments and returning the
  66  * corresponding object. The structure (and the corresponding thread) is
  67  * created, and put in the list of ready process.
  68  * \param name a name for the object. It is for user-level information
  69    and can be NULL.
  70  * \param code is a function describing the behavior of the agent. It
  71    should then only use functions described in \ref
  72    m_process_management (to create a new #m_process_t for example),
  73    in \ref m_host_management (only the read-only functions i.e. whose
  74    name contains the word get), in \ref m_task_management (to create
  75    or destroy some #m_task_t for example) and in \ref
  76    msg_gos_functions (to handle file transfers and task processing).
  77  * \param data a pointer to any data one may want to attach to the new
  78    object.  It is for user-level information and can be NULL. It can
  79    be retrieved with the function \ref MSG_process_get_data.
  80  * \param host the location where the new agent is executed.
  81  * \param argc first argument passed to \a code
  82  * \param argv second argument passed to \a code
  83  * \see m_process_t
  84  * \return The new corresponding object.
  85  */
  86 m_process_t MSG_process_create_with_arguments(const char *name,
  87                                               m_process_code_t code, void *data,
  88                                               m_host_t host, int argc, char **argv)
  89 {
  90   simdata_process_t simdata = xbt_new0(s_simdata_process_t,1);
  91   m_process_t process = xbt_new0(s_m_process_t,1);
  92   m_process_t self = NULL;
  93
  94   xbt_assert0(((code != NULL) && (host != NULL)), "Invalid parameters");
  95   /* Simulator Data */
  96
  97   simdata->PID = msg_global->PID++;
  98   simdata->host = host;
  99   simdata->waiting_task = NULL;
 100   simdata->argc = argc;
 101   simdata->argv = argv;
 102   simdata->context = xbt_context_new(code, NULL, NULL,
 103                                      MSG_process_cleanup, process,
 104                                      simdata->argc, simdata->argv);
 105
 106   if((self=msg_global->current_process)) {
 107     simdata->PPID = MSG_process_get_PID(self);
 108   } else {
 109     simdata->PPID = -1;
 110   }
 111   simdata->last_errno=MSG_OK;
 112
 113
 114   /* Process structure */
 115   process->name = xbt_strdup(name);
 116   process->simdata = simdata;
 117   process->data = data;
 118
 119   xbt_fifo_unshift(host->simdata->process_list, process);
 120
 121   /* *************** FIX du current_process !!! *************** */
 122   self = msg_global->current_process;
 123   xbt_context_start(process->simdata->context);
 124   msg_global->current_process = self;
 125
 126   xbt_fifo_unshift(msg_global->process_list, process);
 127   DEBUG2("Inserting %s(%s) in the to_run list",process->name,
 128          host->name);
 129   xbt_fifo_unshift(msg_global->process_to_run, process);
 130
 131   PAJE_PROCESS_NEW(process);
 132
 133   return process;
 134 }
 135
 136 /** \ingroup m_process_management
 137  * \param process poor victim
 138  *
 139  * This function simply kills a \a process... scarry isn't it ? :)
 140  */
 141 void MSG_process_kill(m_process_t process)
 142 {
 143   int i;
 144   simdata_process_t p_simdata = process->simdata;
 145   simdata_host_t h_simdata= p_simdata->host->simdata;
 146   int _cursor;
 147   m_process_t proc = NULL;
 148
 149   DEBUG3("Killing %s(%d) on %s",process->name, p_simdata->PID,
 150          p_simdata->host->name);
 151
 152   for (i=0; i<msg_global->max_channel; i++) {
 153     if (h_simdata->sleeping[i] == process) {
 154       h_simdata->sleeping[i] = NULL;
 155       break;
 156     }
 157   }
 158
 159   if(p_simdata->waiting_task) {
 160     xbt_dynar_foreach(p_simdata->waiting_task->simdata->sleeping,_cursor,proc) {
 161       if(proc==process)
 162         xbt_dynar_remove_at(p_simdata->waiting_task->simdata->sleeping,_cursor,&proc);
 163     }
 164     if(p_simdata->waiting_task->simdata->compute)
 165       surf_workstation_resource->common_public->
 166         action_free(p_simdata->waiting_task->simdata->compute);
 167     else if (p_simdata->waiting_task->simdata->comm) {
 168       surf_workstation_resource->common_public->
 169         action_change_state(p_simdata->waiting_task->simdata->comm,SURF_ACTION_FAILED);
 170       surf_workstation_resource->common_public->
 171         action_free(p_simdata->waiting_task->simdata->comm);
 172     } else {
 173       xbt_die("UNKNOWN STATUS. Please report this bug.");
 174     }
 175   }
 176
 177   if ((i==msg_global->max_channel) && (process!=MSG_process_self()) &&
 178       (!p_simdata->waiting_task)) {
 179     xbt_die("UNKNOWN STATUS. Please report this bug.");
 180   }
 181
 182   xbt_fifo_remove(msg_global->process_to_run,process);
 183   xbt_fifo_remove(msg_global->process_list,process);
 184   xbt_context_kill(process->simdata->context);
 185
 186   if(process==MSG_process_self()) {
 187     /* I just killed myself */
 188     xbt_context_yield();
 189   }
 190 }
 191
 192 /** \ingroup m_process_management
 193  * \brief Migrates an agent to another location.
 194  *
 195  * This functions checks whether \a process and \a host are valid pointers
 196    and change the value of the #m_host_t on which \a process is running.
 197  */
 198 MSG_error_t MSG_process_change_host(m_process_t process, m_host_t host)
 199 {
 200   simdata_process_t simdata = NULL;
 201
 202   /* Sanity check */
 203
 204   xbt_assert0(((process) && (process->simdata)
 205           && (host)), "Invalid parameters");
 206   simdata = process->simdata;
 207
 208   xbt_fifo_remove(simdata->host->simdata->process_list,process);
 209   simdata->host = host;
 210   xbt_fifo_unshift(host->simdata->process_list,process);
 211
 212   return MSG_OK;
 213 }
 214
 215 /** \ingroup m_process_management
 216  * \brief Return the user data of a #m_process_t.
 217  *
 218  * This functions checks whether \a process is a valid pointer or not
 219    and return the user data associated to \a process if it is possible.
 220  */
 221 void *MSG_process_get_data(m_process_t process)
 222 {
 223   xbt_assert0((process != NULL), "Invalid parameters");
 224
 225   return (process->data);
 226 }
 227
 228 /** \ingroup m_process_management
 229  * \brief Set the user data of a #m_process_t.
 230  *
 231  * This functions checks whether \a process is a valid pointer or not
 232    and set the user data associated to \a process if it is possible.
 233  */
 234 MSG_error_t MSG_process_set_data(m_process_t process,void *data)
 235 {
 236   xbt_assert0((process != NULL), "Invalid parameters");
 237   xbt_assert0((process->data == NULL), "Data already set");
 238
 239   process->data = data;
 240
 241   return MSG_OK;
 242 }
 243
 244 /** \ingroup m_process_management
 245  * \brief Return the location on which an agent is running.
 246  *
 247  * This functions checks whether \a process is a valid pointer or not
 248    and return the m_host_t corresponding to the location on which \a
 249    process is running.
 250  */
 251 m_host_t MSG_process_get_host(m_process_t process)
 252 {
 253   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 254
 255   return (((simdata_process_t) process->simdata)->host);
 256 }
 257
 258 /** \ingroup m_process_management
 259  *
 260  * \brief Return a #m_process_t given its PID.
 261  *
 262  * This functions search in the list of all the created m_process_t for a m_process_t
 263    whose PID is equal to \a PID. If no host is found, \c NULL is returned.
 264    Note that the PID are uniq in the whole simulation, not only on a given host.
 265  */
 266 m_process_t MSG_process_from_PID(int PID)
 267 {
 268   xbt_fifo_item_t i = NULL;
 269   m_process_t process = NULL;
 270
 271   xbt_fifo_foreach(msg_global->process_list,i,process,m_process_t) {
 272     if(MSG_process_get_PID(process) == PID) return process;
 273   }
 274   return NULL;
 275 }
 276
 277 /** \ingroup m_process_management
 278  * \brief Returns the process ID of \a process.
 279  *
 280  * This functions checks whether \a process is a valid pointer or not
 281    and return its PID.
 282  */
 283 int MSG_process_get_PID(m_process_t process)
 284 {
 285   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 286
 287   return (((simdata_process_t) process->simdata)->PID);
 288 }
 289
 290 /** \ingroup m_process_management
 291  * \brief Returns the process ID of the parent of \a process.
 292  *
 293  * This functions checks whether \a process is a valid pointer or not
 294    and return its PID. Returns -1 if the agent has not been created by
 295    another agent.
 296  */
 297 int MSG_process_get_PPID(m_process_t process)
 298 {
 299   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 300
 301   return (((simdata_process_t) process->simdata)->PPID);
 302 }
 303
 304 /** \ingroup m_process_management
 305  * \brief Return the name of an agent.
 306  *
 307  * This functions checks whether \a process is a valid pointer or not
 308    and return its name.
 309  */
 310 const char *MSG_process_get_name(m_process_t process)
 311 {
 312   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 313
 314   return (process->name);
 315 }
 316
 317 /** \ingroup m_process_management
 318  * \brief Return the PID of the current agent.
 319  *
 320  * This functions returns the PID of the currently running #m_process_t.
 321  */
 322 int MSG_process_self_PID(void)
 323 {
 324   return (MSG_process_get_PID(MSG_process_self()));
 325 }
 326
 327 /** \ingroup m_process_management
 328  * \brief Return the PPID of the current agent.
 329  *
 330  * This functions returns the PID of the parent of the currently
 331  * running #m_process_t.
 332  */
 333 int MSG_process_self_PPID(void)
 334 {
 335   return (MSG_process_get_PPID(MSG_process_self()));
 336 }
 337
 338 /** \ingroup m_process_management
 339  * \brief Return the current agent.
 340  *
 341  * This functions returns the currently running #m_process_t.
 342  */
 343 m_process_t MSG_process_self(void)
 344 {
 345   return msg_global ? msg_global->current_process : NULL;
 346 }
 347
 348 /** \ingroup m_process_management
 349  * \brief Suspend the process.
 350  *
 351  * This functions suspend the process by suspending the task on which
 352  * it was waiting for the completion.
 353  */
 354 MSG_error_t MSG_process_suspend(m_process_t process)
 355 {
 356   simdata_process_t simdata = NULL;
 357   simdata_task_t simdata_task = NULL;
 358
 359   XBT_IN2("(%p(%s))", process, process->name);
 360
 361   xbt_assert0(((process) && (process->simdata)), "Invalid parameters");
 362
 363   PAJE_PROCESS_PUSH_STATE(process,"S",NULL);
 364
 365   if(process!=MSG_process_self()) {
 366     simdata = process->simdata;
 367
 368     xbt_assert0(simdata->waiting_task,"Process not waiting for anything else. Weird !");
 369
 370     simdata_task = simdata->waiting_task->simdata;
 371
 372     simdata->suspended = 1;
 373     if(simdata->blocked) {
 374       XBT_OUT;
 375       return MSG_OK;
 376     }
 377
 378     xbt_assert0(((simdata_task->compute)||(simdata_task->comm))&&
 379                 !((simdata_task->compute)&&(simdata_task->comm)),
 380                 "Got a problem in deciding which action to choose !");
 381     simdata->suspended = 1;
 382     if(simdata_task->compute)
 383       surf_workstation_resource->common_public->suspend(simdata_task->compute);
 384     else
 385       surf_workstation_resource->common_public->suspend(simdata_task->comm);
 386   } else {
 387     m_task_t dummy = MSG_TASK_UNINITIALIZED;
 388     dummy = MSG_task_create("suspended", 0.0, 0, NULL);
 389
 390     simdata = process->simdata;
 391     simdata->suspended = 1;
 392     __MSG_task_execute(process,dummy);
 393     surf_workstation_resource->common_public->suspend(dummy->simdata->compute);
 394     __MSG_wait_for_computation(process,dummy);
 395     simdata->suspended = 0;
 396
 397     MSG_task_destroy(dummy);
 398   }
 399   XBT_OUT;
 400   return MSG_OK;
 401 }
 402
 403 /** \ingroup m_process_management
 404  * \brief Resume a suspended process.
 405  *
 406  * This functions resume a suspended process by resuming the task on
 407  * which it was waiting for the completion.
 408  */
 409 MSG_error_t MSG_process_resume(m_process_t process)
 410 {
 411   simdata_process_t simdata = NULL;
 412   simdata_task_t simdata_task = NULL;
 413
 414   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 415   CHECK_HOST();
 416
 417   XBT_IN2("(%p(%s))", process, process->name);
 418
 419   if(process == MSG_process_self()) {
 420     XBT_OUT;
 421     MSG_RETURN(MSG_OK);
 422   }
 423
 424   simdata = process->simdata;
 425
 426   if(simdata->blocked) {
 427     PAJE_PROCESS_POP_STATE(process);
 428
 429     simdata->suspended = 0; /* He'll wake up by itself */
 430     XBT_OUT;
 431     MSG_RETURN(MSG_OK);
 432   }
 433
 434   if(!(simdata->waiting_task)) {
 435     xbt_assert0(0,"Process not waiting for anything else. Weird !");
 436     XBT_OUT;
 437     return MSG_WARNING;
 438   }
 439   simdata_task = simdata->waiting_task->simdata;
 440
 441
 442   if(simdata_task->compute) {
 443     surf_workstation_resource->common_public->resume(simdata_task->compute);
 444     PAJE_PROCESS_POP_STATE(process);
 445   }
 446   else {
 447     PAJE_PROCESS_POP_STATE(process);
 448     surf_workstation_resource->common_public->resume(simdata_task->comm);
 449   }
 450
 451   XBT_OUT;
 452   MSG_RETURN(MSG_OK);
 453 }
 454
 455 /** \ingroup m_process_management
 456  * \brief Returns true if the process is suspended .
 457  *
 458  * This checks whether a process is suspended or not by inspecting the
 459  * task on which it was waiting for the completion.
 460  */
 461 int MSG_process_is_suspended(m_process_t process)
 462 {
 463   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 464
 465   return (process->simdata->suspended);
 466 }
 467
 468 int __MSG_process_block(double max_duration, const char *info)
 469 {
 470   m_process_t process = MSG_process_self();
 471   m_task_t dummy = MSG_TASK_UNINITIALIZED;
 472   char blocked_name[512];
 473   snprintf(blocked_name,512,"blocked [%s] (%s:%s)",
 474           info, process->name, process->simdata->host->name);
 475
 476   XBT_IN1(": max_duration=%g",max_duration);
 477
 478   dummy = MSG_task_create(blocked_name, 0.0, 0, NULL);
 479
 480   PAJE_PROCESS_PUSH_STATE(process,"B",NULL);
 481
 482   process->simdata->blocked=1;
 483   __MSG_task_execute(process,dummy);
 484   surf_workstation_resource->common_public->suspend(dummy->simdata->compute);
 485   if(max_duration>=0)
 486     surf_workstation_resource->common_public->set_max_duration(dummy->simdata->compute,
 487                                                                max_duration);
 488   __MSG_wait_for_computation(process,dummy);
 489   MSG_task_destroy(dummy);
 490   process->simdata->blocked=0;
 491
 492   if(process->simdata->suspended) {
 493     DEBUG0("I've been suspended in the meantime");
 494     MSG_process_suspend(process);
 495     DEBUG0("I've been resumed, let's keep going");
 496   }
 497
 498   PAJE_PROCESS_POP_STATE(process);
 499
 500   XBT_OUT;
 501   return 1;
 502 }
 503
 504 MSG_error_t __MSG_process_unblock(m_process_t process)
 505 {
 506   simdata_process_t simdata = NULL;
 507   simdata_task_t simdata_task = NULL;
 508
 509   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 510   CHECK_HOST();
 511
 512   XBT_IN2(": %s unblocking %s", MSG_process_self()->name,process->name);
 513
 514   simdata = process->simdata;
 515   if(!(simdata->waiting_task)) {
 516     xbt_assert0(0,"Process not waiting for anything else. Weird !");
 517     XBT_OUT;
 518     return MSG_WARNING;
 519   }
 520   simdata_task = simdata->waiting_task->simdata;
 521
 522   xbt_assert0(simdata->blocked,"Process not blocked");
 523
 524   surf_workstation_resource->common_public->resume(simdata_task->compute);
 525
 526   XBT_OUT;
 527
 528   MSG_RETURN(MSG_OK);
 529 }
 530
 531 int __MSG_process_isBlocked(m_process_t process)
 532 {
 533   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 534
 535   return (process->simdata->blocked);
 536 }