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