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