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).
  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);
  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 {
  85   simdata_process_t simdata = xbt_new0(s_simdata_process_t,1);
  86   m_process_t process = xbt_new0(s_m_process_t,1);
  87   m_process_t self = NULL;
  88
  89   xbt_assert0(((code != NULL) && (host != NULL)), "Invalid parameters");
  90   /* Simulator Data */
  91
  92   simdata->PID = msg_global->PID++;
  93   simdata->host = host;
  94   simdata->waiting_task = NULL;
  95   simdata->argc = argc;
  96   simdata->argv = argv;
  97   simdata->context = xbt_context_new(code, NULL, NULL,
  98                                      MSG_process_cleanup, process,
  99                                      simdata->argc, simdata->argv);
 100
 101   if((self=msg_global->current_process)) {
 102     simdata->PPID = MSG_process_get_PID(self);
 103   } else {
 104     simdata->PPID = -1;
 105   }
 106   simdata->last_errno=MSG_OK;
 107
 108
 109   /* Process structure */
 110   process->name = xbt_strdup(name);
 111   process->simdata = simdata;
 112   process->data = data;
 113
 114   xbt_fifo_push(host->simdata->process_list, process);
 115
 116   /* /////////////// FIX du current_process !!! ////////////// */
 117   self = msg_global->current_process;
 118   xbt_context_start(process->simdata->context);
 119   msg_global->current_process = self;
 120
 121   xbt_fifo_push(msg_global->process_list, process);
 122   xbt_fifo_push(msg_global->process_to_run, process);
 123
 124   PAJE_PROCESS_NEW(process);
 125
 126   return process;
 127 }
 128
 129 /** \ingroup m_process_management
 130  * \param process poor victim
 131  *
 132  * This function simply kills a \a process... scarry isn't it ? :)
 133  */
 134 void MSG_process_kill(m_process_t process)
 135 {
 136   int i;
 137   simdata_process_t p_simdata = process->simdata;
 138   simdata_host_t h_simdata= p_simdata->host->simdata;
 139
 140 /*   fprintf(stderr,"Killing %s(%d) on %s.\n",process->name, */
 141 /*        p_simdata->PID,p_simdata->host->name); */
 142
 143   for (i=0; i<msg_global->max_channel; i++) {
 144     if (h_simdata->sleeping[i] == process) {
 145       h_simdata->sleeping[i] = NULL;
 146       break;
 147     }
 148   }
 149   if (i==msg_global->max_channel) {
 150     if(p_simdata->waiting_task) {
 151       if(p_simdata->waiting_task->simdata->compute)
 152         surf_workstation_resource->common_public->
 153           action_free(p_simdata->waiting_task->simdata->compute);
 154       else if (p_simdata->waiting_task->simdata->comm)
 155         surf_workstation_resource->common_public->
 156           action_change_state(p_simdata->waiting_task->simdata->comm,SURF_ACTION_FAILED);
 157       else
 158         fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
 159     } else { /* Must be trying to put a task somewhere */
 160       fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
 161     }
 162   }
 163
 164   xbt_fifo_remove(msg_global->process_list,process);
 165   xbt_context_free(process->simdata->context);
 166 }
 167
 168 /** \ingroup m_process_management
 169  * \brief Migrates an agent to another location.
 170  *
 171  * This functions checks whether \a process and \a host are valid pointers
 172    and change the value of the #m_host_t on which \a process is running.
 173  */
 174 MSG_error_t MSG_process_change_host(m_process_t process, m_host_t host)
 175 {
 176   simdata_process_t simdata = NULL;
 177
 178   /* Sanity check */
 179
 180   xbt_assert0(((process) && (process->simdata)
 181           && (host)), "Invalid parameters");
 182   simdata = process->simdata;
 183
 184   xbt_fifo_remove(simdata->host->simdata->process_list,process);
 185   simdata->host = host;
 186   xbt_fifo_push(host->simdata->process_list,process);
 187
 188   return MSG_OK;
 189 }
 190
 191 /** \ingroup m_process_management
 192  * \brief Return the user data of a #m_process_t.
 193  *
 194  * This functions checks whether \a process is a valid pointer or not
 195    and return the user data associated to \a process if it is possible.
 196  */
 197 void *MSG_process_get_data(m_process_t process)
 198 {
 199   xbt_assert0((process != NULL), "Invalid parameters");
 200
 201   return (process->data);
 202 }
 203
 204 /** \ingroup m_process_management
 205  * \brief Set the user data of a #m_process_t.
 206  *
 207  * This functions checks whether \a process is a valid pointer or not
 208    and set the user data associated to \a process if it is possible.
 209  */
 210 MSG_error_t MSG_process_set_data(m_process_t process,void *data)
 211 {
 212   xbt_assert0((process != NULL), "Invalid parameters");
 213   xbt_assert0((process->data == NULL), "Data already set");
 214
 215   process->data = data;
 216
 217   return MSG_OK;
 218 }
 219
 220 /** \ingroup m_process_management
 221  * \brief Return the location on which an agent is running.
 222  *
 223  * This functions checks whether \a process is a valid pointer or not
 224    and return the m_host_t corresponding to the location on which \a
 225    process is running.
 226  */
 227 m_host_t MSG_process_get_host(m_process_t process)
 228 {
 229   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 230
 231   return (((simdata_process_t) process->simdata)->host);
 232 }
 233
 234 /** \ingroup m_process_management
 235  *
 236  * \brief Return a #m_process_t given its PID.
 237  *
 238  * This functions search in the list of all the created m_process_t for a m_process_t
 239    whose PID is equal to \a PID. If no host is found, \c NULL is returned.
 240    Note that the PID are uniq in the whole simulation, not only on a given host.
 241  */
 242 m_process_t MSG_process_from_PID(int PID)
 243 {
 244   xbt_fifo_item_t i = NULL;
 245   m_process_t process = NULL;
 246
 247   xbt_fifo_foreach(msg_global->process_list,i,process,m_process_t) {
 248     if(MSG_process_get_PID(process) == PID) return process;
 249   }
 250   return NULL;
 251 }
 252
 253 /** \ingroup m_process_management
 254  * \brief Returns the process ID of \a process.
 255  *
 256  * This functions checks whether \a process is a valid pointer or not
 257    and return its PID.
 258  */
 259 int MSG_process_get_PID(m_process_t process)
 260 {
 261   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 262
 263   return (((simdata_process_t) process->simdata)->PID);
 264 }
 265
 266 /** \ingroup m_process_management
 267  * \brief Returns the process ID of the parent of \a process.
 268  *
 269  * This functions checks whether \a process is a valid pointer or not
 270    and return its PID. Returns -1 if the agent has not been created by
 271    another agent.
 272  */
 273 int MSG_process_get_PPID(m_process_t process)
 274 {
 275   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 276
 277   return (((simdata_process_t) process->simdata)->PPID);
 278 }
 279
 280 /** \ingroup m_process_management
 281  * \brief Return the name of an agent.
 282  *
 283  * This functions checks whether \a process is a valid pointer or not
 284    and return its name.
 285  */
 286 const char *MSG_process_get_name(m_process_t process)
 287 {
 288   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 289
 290   return (process->name);
 291 }
 292
 293 /** \ingroup m_process_management
 294  * \brief Return the PID of the current agent.
 295  *
 296  * This functions returns the PID of the currently running #m_process_t.
 297  */
 298 int MSG_process_self_PID(void)
 299 {
 300   return (MSG_process_get_PID(MSG_process_self()));
 301 }
 302
 303 /** \ingroup m_process_management
 304  * \brief Return the PPID of the current agent.
 305  *
 306  * This functions returns the PID of the parent of the currently
 307  * running #m_process_t.
 308  */
 309 int MSG_process_self_PPID(void)
 310 {
 311   return (MSG_process_get_PPID(MSG_process_self()));
 312 }
 313
 314 /** \ingroup m_process_management
 315  * \brief Return the current agent.
 316  *
 317  * This functions returns the currently running #m_process_t.
 318  */
 319 m_process_t MSG_process_self(void)
 320 {
 321   return msg_global ? msg_global->current_process : NULL;
 322 }
 323
 324 /** \ingroup m_process_management
 325  * \brief Suspend the process.
 326  *
 327  * This functions suspend the process by suspending the task on which
 328  * it was waiting for the completion.
 329  */
 330 MSG_error_t MSG_process_suspend(m_process_t process)
 331 {
 332   simdata_process_t simdata = NULL;
 333   simdata_task_t simdata_task = NULL;
 334   int i;
 335
 336   xbt_assert0(((process) && (process->simdata)), "Invalid parameters");
 337
 338   PAJE_PROCESS_PUSH_STATE(process,"S");
 339
 340   if(process!=MSG_process_self()) {
 341     simdata = process->simdata;
 342
 343     xbt_assert0(simdata->waiting_task,"Process not waiting for anything else. Weird !");
 344
 345     simdata_task = simdata->waiting_task->simdata;
 346
 347     simdata->suspended = 1;
 348     if(simdata->blocked) return MSG_OK;
 349
 350     xbt_assert0(((simdata_task->compute)||(simdata_task->comm))&&
 351                 !((simdata_task->compute)&&(simdata_task->comm)),
 352                 "Got a problem in deciding which action to choose !");
 353     simdata->suspended = 1;
 354     if(simdata_task->compute)
 355       surf_workstation_resource->common_public->suspend(simdata_task->compute);
 356     else
 357       surf_workstation_resource->common_public->suspend(simdata_task->comm);
 358   } else {
 359     m_task_t dummy = MSG_TASK_UNINITIALIZED;
 360     dummy = MSG_task_create("suspended", 0.0, 0, NULL);
 361
 362     simdata = process->simdata;
 363     simdata->suspended = 1;
 364     __MSG_task_execute(process,dummy);
 365     surf_workstation_resource->common_public->suspend(dummy->simdata->compute);
 366     __MSG_wait_for_computation(process,dummy);
 367     simdata->suspended = 0;
 368
 369     MSG_task_destroy(dummy);
 370   }
 371   return MSG_OK;
 372 }
 373
 374 /** \ingroup m_process_management
 375  * \brief Resume a suspended process.
 376  *
 377  * This functions resume a suspended process by resuming the task on
 378  * which it was waiting for the completion.
 379  */
 380 MSG_error_t MSG_process_resume(m_process_t process)
 381 {
 382   simdata_process_t simdata = NULL;
 383   simdata_task_t simdata_task = NULL;
 384   int i;
 385
 386   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 387   CHECK_HOST();
 388
 389   simdata = process->simdata;
 390
 391
 392   if(simdata->blocked) {
 393     PAJE_PROCESS_POP_STATE(process);
 394
 395     simdata->suspended = 0; /* He'll wake up by itself */
 396     MSG_RETURN(MSG_OK);
 397   }
 398
 399   if(!(simdata->waiting_task)) {
 400     xbt_assert0(0,"Process not waiting for anything else. Weird !");
 401     return MSG_WARNING;
 402   }
 403   simdata_task = simdata->waiting_task->simdata;
 404
 405
 406   if(simdata_task->compute) {
 407     surf_workstation_resource->common_public->resume(simdata_task->compute);
 408     PAJE_PROCESS_POP_STATE(process);
 409   }
 410   else {
 411     PAJE_PROCESS_POP_STATE(process);
 412     surf_workstation_resource->common_public->resume(simdata_task->comm);
 413   }
 414
 415   MSG_RETURN(MSG_OK);
 416 }
 417
 418 /** \ingroup m_process_management
 419  * \brief Returns true if the process is suspended .
 420  *
 421  * This checks whether a process is suspended or not by inspecting the
 422  * task on which it was waiting for the completion.
 423  */
 424 int MSG_process_isSuspended(m_process_t process)
 425 {
 426   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 427
 428   return (process->simdata->suspended);
 429 }
 430
 431 MSG_error_t __MSG_process_block(void)
 432 {
 433   m_process_t process = MSG_process_self();
 434
 435   m_task_t dummy = MSG_TASK_UNINITIALIZED;
 436   dummy = MSG_task_create("blocked", 0.0, 0, NULL);
 437
 438   PAJE_PROCESS_PUSH_STATE(process,"B");
 439
 440   process->simdata->blocked=1;
 441   __MSG_task_execute(process,dummy);
 442   surf_workstation_resource->common_public->suspend(dummy->simdata->compute);
 443   __MSG_wait_for_computation(process,dummy);
 444   process->simdata->blocked=0;
 445
 446   if(process->simdata->suspended)
 447     MSG_process_suspend(process);
 448
 449   MSG_task_destroy(dummy);
 450
 451   return MSG_OK;
 452 }
 453
 454 MSG_error_t __MSG_process_unblock(m_process_t process)
 455 {
 456   simdata_process_t simdata = NULL;
 457   simdata_task_t simdata_task = NULL;
 458   int i;
 459
 460   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 461   CHECK_HOST();
 462
 463   simdata = process->simdata;
 464   if(!(simdata->waiting_task)) {
 465     xbt_assert0(0,"Process not waiting for anything else. Weird !");
 466     return MSG_WARNING;
 467   }
 468   simdata_task = simdata->waiting_task->simdata;
 469
 470   xbt_assert0(simdata->blocked,"Process not blocked");
 471
 472   surf_workstation_resource->common_public->resume(simdata_task->compute);
 473
 474   PAJE_PROCESS_POP_STATE(process);
 475
 476   MSG_RETURN(MSG_OK);
 477 }
 478
 479 int __MSG_process_isBlocked(m_process_t process)
 480 {
 481   xbt_assert0(((process != NULL) && (process->simdata)), "Invalid parameters");
 482
 483   return (process->simdata->blocked);
 484 }
 485
 486
 487 const char* xbt_procname(void) {
 488   char *res = NULL;
 489   m_process_t process = MSG_process_self();
 490   if ((process != NULL) && (process->simdata))
 491     res = (char *)MSG_process_get_name(process);
 492   if (res)
 493     return (const char*)res;
 494   else
 495     return "";
 496 }