src/simix/smx_synchro.c

   1 /* Copyright (c) 2007, 2008, 2009, 2010. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "private.h"
   8 #include "xbt/log.h"
   9
  10
  11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_synchro, simix,
  12                                 "Logging specific to SIMIX (synchronization)");
  13
  14 static smx_action_t SIMIX_synchro_wait(smx_host_t smx_host, double timeout);
  15 static void SIMIX_synchro_finish(smx_action_t action);
  16 static void _SIMIX_cond_wait(smx_cond_t cond, smx_mutex_t mutex, double timeout,
  17                              smx_process_t issuer, smx_req_t req);
  18 static void _SIMIX_sem_wait(smx_sem_t sem, double timeout, smx_process_t issuer,
  19                             smx_req_t req);
  20 static void SIMIX_sem_block_onto(smx_sem_t sem);
  21
  22 /***************************** Synchro action *********************************/
  23
  24 static smx_action_t SIMIX_synchro_wait(smx_host_t smx_host, double timeout)
  25 {
  26   XBT_IN("(%p, %f)",smx_host,timeout);
  27   smx_action_t action;
  28   action = xbt_mallocator_get(simix_global->action_mallocator);
  29   action->type = SIMIX_ACTION_SYNCHRO;
  30   action->name = xbt_strdup("synchro");
  31   action->synchro.sleep =
  32     surf_workstation_model->extension.workstation.sleep(smx_host->host, timeout);
  33
  34   surf_workstation_model->action_data_set(action->synchro.sleep, action);
  35   XBT_OUT();
  36   return action;
  37 }
  38
  39 void SIMIX_synchro_stop_waiting(smx_process_t process, smx_req_t req)
  40 {
  41   XBT_IN("(%p, %p)",process,req);
  42   switch (req->call) {
  43
  44     case REQ_MUTEX_LOCK:
  45       xbt_swag_remove(process, req->mutex_lock.mutex->sleeping);
  46       break;
  47
  48     case REQ_COND_WAIT:
  49       xbt_swag_remove(process, req->cond_wait.cond->sleeping);
  50       break;
  51
  52     case REQ_COND_WAIT_TIMEOUT:
  53       xbt_swag_remove(process, req->cond_wait_timeout.cond->sleeping);
  54       break;
  55
  56     case REQ_SEM_ACQUIRE:
  57       xbt_swag_remove(process, req->sem_acquire.sem->sleeping);
  58       break;
  59
  60     case REQ_SEM_ACQUIRE_TIMEOUT:
  61       xbt_swag_remove(process, req->sem_acquire_timeout.sem->sleeping);
  62       break;
  63
  64     default:
  65       THROW_IMPOSSIBLE;
  66   }
  67   XBT_OUT();
  68 }
  69
  70 void SIMIX_synchro_destroy(smx_action_t action)
  71 {
  72   XBT_IN("(%p)",action);
  73   XBT_DEBUG("Destroying synchro %p", action);
  74   action->synchro.sleep->model_type->action_unref(action->synchro.sleep);
  75   xbt_free(action->name);
  76   xbt_mallocator_release(simix_global->action_mallocator, action);
  77   XBT_OUT();
  78 }
  79
  80 void SIMIX_post_synchro(smx_action_t action)
  81 {
  82   XBT_IN("(%p)",action);
  83   if (surf_workstation_model->action_state_get(action->synchro.sleep) == SURF_ACTION_FAILED)
  84     action->state = SIMIX_FAILED;
  85   else if(surf_workstation_model->action_state_get(action->synchro.sleep) == SURF_ACTION_DONE)
  86     action->state = SIMIX_SRC_TIMEOUT;
  87
  88   SIMIX_synchro_finish(action);
  89   XBT_OUT();
  90 }
  91
  92 static void SIMIX_synchro_finish(smx_action_t action)
  93 {
  94   XBT_IN("(%p)",action);
  95   smx_req_t req = xbt_fifo_shift(action->request_list);
  96
  97   switch (action->state) {
  98
  99     case SIMIX_SRC_TIMEOUT:
 100       TRY {
 101         THROWF(timeout_error, 0, "Synchro's wait timeout");
 102       } CATCH(req->issuer->running_ctx->exception) {
 103         req->issuer->doexception = 1;
 104       }
 105       break;
 106
 107     case SIMIX_FAILED:
 108       TRY {
 109         THROWF(host_error, 0, "Host failed");
 110       } CATCH(req->issuer->running_ctx->exception) {
 111         req->issuer->doexception = 1;
 112       }
 113       break;
 114
 115     default:
 116       THROW_IMPOSSIBLE;
 117       break;
 118   }
 119
 120   SIMIX_synchro_stop_waiting(req->issuer, req);
 121   SIMIX_synchro_destroy(action);
 122   SIMIX_request_answer(req);
 123   XBT_OUT();
 124 }
 125 /*********************************** Mutex ************************************/
 126
 127 /**
 128  * \brief Initialize a mutex.
 129  *
 130  * Allocs and creates the data for the mutex.
 131  * \return A mutex
 132  */
 133 smx_mutex_t SIMIX_mutex_init(void)
 134 {
 135   XBT_IN("()");
 136   s_smx_process_t p;            /* useful to initialize sleeping swag */
 137
 138   smx_mutex_t mutex = xbt_new0(s_smx_mutex_t, 1);
 139   mutex->locked = 0;
 140   mutex->sleeping = xbt_swag_new(xbt_swag_offset(p, synchro_hookup));
 141   XBT_OUT();
 142   return mutex;
 143 }
 144
 145 /**
 146  * \brief Handle mutex lock request
 147  * \param req The request
 148  */
 149 void SIMIX_pre_mutex_lock(smx_req_t req)
 150 {
 151   XBT_IN("(%p)",req);
 152   /* FIXME: check where to validate the arguments */
 153   smx_action_t sync_act = NULL;
 154   smx_mutex_t mutex = req->mutex_lock.mutex;
 155   smx_process_t process = req->issuer;
 156
 157   if (mutex->locked) {
 158     /* FIXME: check if the host is active ? */
 159     /* Somebody using the mutex, use a synchro action to get host failures */
 160     sync_act = SIMIX_synchro_wait(process->smx_host, -1);
 161     xbt_fifo_push(sync_act->request_list, req);
 162     req->issuer->waiting_action = sync_act;
 163     xbt_swag_insert(req->issuer, mutex->sleeping);
 164   } else {
 165     /* mutex free */
 166     mutex->locked = 1;
 167     mutex->owner = req->issuer;
 168     SIMIX_request_answer(req);
 169   }
 170   XBT_OUT();
 171 }
 172
 173 /**
 174  * \brief Tries to lock a mutex.
 175  *
 176  * Tries to lock a mutex, return 1 if the mutex is unlocked, else 0.
 177  * This function does not block and wait for the mutex to be unlocked.
 178  * \param mutex The mutex
 179  * \param issuer The process that tries to acquire the mutex
 180  * \return 1 - mutex free, 0 - mutex used
 181  */
 182 int SIMIX_mutex_trylock(smx_mutex_t mutex, smx_process_t issuer)
 183 {
 184   XBT_IN("(%p, %p)",mutex,issuer);
 185   if (mutex->locked){
 186           XBT_OUT();
 187           return 0;
 188   }
 189
 190   mutex->locked = 1;
 191   mutex->owner = issuer;
 192   XBT_OUT();
 193   return 1;
 194 }
 195
 196 /**
 197  * \brief Unlocks a mutex.
 198  *
 199  * Unlocks the mutex and gives it to a process waiting for it.
 200  * If the unlocker is not the owner of the mutex nothing happens.
 201  * If there are no process waiting, it sets the mutex as free.
 202  * \param mutex The mutex
 203  * \param issuer The process trying to unlock the mutex
 204  */
 205 void SIMIX_mutex_unlock(smx_mutex_t mutex, smx_process_t issuer)
 206 {
 207   XBT_IN("(%p, %p)",mutex,issuer);
 208   smx_process_t p;              /*process to wake up */
 209
 210   /* If the mutex is not owned by the issuer do nothing */
 211   if (issuer != mutex->owner){
 212           XBT_OUT();
 213           return;
 214   }
 215
 216   if (xbt_swag_size(mutex->sleeping) > 0) {
 217     p = xbt_swag_extract(mutex->sleeping);
 218     SIMIX_synchro_destroy(p->waiting_action);
 219     p->waiting_action = NULL;
 220     mutex->owner = p;
 221     SIMIX_request_answer(&p->request);
 222   } else {
 223     /* nobody to wake up */
 224     mutex->locked = 0;
 225     mutex->owner = NULL;
 226   }
 227   XBT_OUT();
 228 }
 229
 230 /**
 231  * \brief Destroys a mutex.
 232  *
 233  * Destroys and frees the mutex's memory.
 234  * \param mutex A mutex
 235  */
 236 void SIMIX_mutex_destroy(smx_mutex_t mutex)
 237 {
 238   XBT_IN("(%p)",mutex);
 239   if (mutex){
 240     xbt_swag_free(mutex->sleeping);
 241     xbt_free(mutex);
 242   }
 243   XBT_OUT();
 244 }
 245
 246 /********************************* Condition **********************************/
 247
 248 /**
 249  * \brief Initialize a condition.
 250  *
 251  * Allocates and creates the data for the condition.
 252  * It have to be called before the use of the condition.
 253  * \return A condition
 254  */
 255 smx_cond_t SIMIX_cond_init()
 256 {
 257   XBT_IN("()");
 258   s_smx_process_t p;
 259   smx_cond_t cond = xbt_new0(s_smx_cond_t, 1);
 260   cond->sleeping = xbt_swag_new(xbt_swag_offset(p, synchro_hookup));
 261   cond->mutex = NULL;
 262   XBT_OUT();
 263   return cond;
 264 }
 265
 266 /**
 267  * \brief Handle condition waiting requests without timeouts
 268  * \param The request
 269  */
 270 void SIMIX_pre_cond_wait(smx_req_t req)
 271 {
 272   XBT_IN("(%p)",req);
 273   smx_process_t issuer = req->issuer;
 274   smx_cond_t cond = req->cond_wait.cond;
 275   smx_mutex_t mutex = req->cond_wait.mutex;
 276
 277   _SIMIX_cond_wait(cond, mutex, -1, issuer, req);
 278   XBT_OUT();
 279 }
 280
 281 /**
 282  * \brief Handle condition waiting requests with timeouts
 283  * \param The request
 284  */
 285 void SIMIX_pre_cond_wait_timeout(smx_req_t req)
 286 {
 287   XBT_IN("(%p)",req);
 288   smx_process_t issuer = req->issuer;
 289   smx_cond_t cond = req->cond_wait_timeout.cond;
 290   smx_mutex_t mutex = req->cond_wait_timeout.mutex;
 291   double timeout = req->cond_wait_timeout.timeout;
 292
 293   _SIMIX_cond_wait(cond, mutex, timeout, issuer, req);
 294   XBT_OUT();
 295 }
 296
 297
 298 static void _SIMIX_cond_wait(smx_cond_t cond, smx_mutex_t mutex, double timeout,
 299                              smx_process_t issuer, smx_req_t req)
 300 {
 301   XBT_IN("(%p, %p, %f, %p,%p)",cond,mutex,timeout,issuer,req);
 302   smx_action_t sync_act = NULL;
 303
 304   XBT_DEBUG("Wait condition %p", cond);
 305
 306   /* If there is a mutex unlock it */
 307   /* FIXME: what happens if the issuer is not the owner of the mutex? */
 308   if (mutex != NULL) {
 309     cond->mutex = mutex;
 310     SIMIX_mutex_unlock(mutex, issuer);
 311   }
 312
 313   sync_act = SIMIX_synchro_wait(issuer->smx_host, timeout);
 314   xbt_fifo_unshift(sync_act->request_list, req);
 315   issuer->waiting_action = sync_act;
 316   xbt_swag_insert(req->issuer, cond->sleeping);
 317   XBT_OUT();
 318 }
 319
 320 /**
 321  * \brief Signalizes a condition.
 322  *
 323  * Signalizes a condition and wakes up a sleeping process.
 324  * If there are no process sleeping, no action is done.
 325  * \param cond A condition
 326  */
 327 void SIMIX_cond_signal(smx_cond_t cond)
 328 {
 329   XBT_IN("(%p)",cond);
 330   smx_process_t proc = NULL;
 331   smx_mutex_t mutex = NULL;
 332   smx_req_t req = NULL;
 333
 334   XBT_DEBUG("Signal condition %p", cond);
 335
 336   /* If there are processes waiting for the condition choose one and try
 337      to make it acquire the mutex */
 338   if ((proc = xbt_swag_extract(cond->sleeping))) {
 339
 340     /* Destroy waiter's synchro action */
 341     SIMIX_synchro_destroy(proc->waiting_action);
 342     proc->waiting_action = NULL;
 343
 344     /* Now transform the cond wait request into a mutex lock one */
 345     req = &proc->request;
 346     if(req->call == REQ_COND_WAIT)
 347       mutex = req->cond_wait.mutex;
 348     else
 349       mutex = req->cond_wait_timeout.mutex;
 350
 351     req->call = REQ_MUTEX_LOCK;
 352     req->mutex_lock.mutex = mutex;
 353
 354     SIMIX_pre_mutex_lock(req);
 355   }
 356   XBT_OUT();
 357 }
 358
 359 /**
 360  * \brief Broadcasts a condition.
 361  *
 362  * Signal ALL processes waiting on a condition.
 363  * If there are no process waiting, no action is done.
 364  * \param cond A condition
 365  */
 366 void SIMIX_cond_broadcast(smx_cond_t cond)
 367 {
 368   XBT_IN("(%p)",cond);
 369   XBT_DEBUG("Broadcast condition %p", cond);
 370
 371   /* Signal the condition until nobody is waiting on it */
 372   while (xbt_swag_size(cond->sleeping)) {
 373     SIMIX_cond_signal(cond);
 374   }
 375   XBT_OUT();
 376 }
 377
 378 /**
 379  * \brief Destroys a contidion.
 380  *
 381  * Destroys and frees the condition's memory.
 382  * \param cond A condition
 383  */
 384 void SIMIX_cond_destroy(smx_cond_t cond)
 385 {
 386   XBT_IN("(%p)",cond);
 387   XBT_DEBUG("Destroy condition %p", cond);
 388
 389   if (cond != NULL) {
 390     xbt_assert(xbt_swag_size(cond->sleeping) == 0,
 391                 "Cannot destroy conditional since someone is still using it");
 392
 393     xbt_swag_free(cond->sleeping);
 394     xbt_free(cond);
 395   }
 396   XBT_OUT();
 397 }
 398
 399 /******************************** Semaphores **********************************/
 400 #define SMX_SEM_NOLIMIT 99999
 401 /** @brief Initialize a semaphore */
 402 smx_sem_t SIMIX_sem_init(unsigned int value)
 403 {
 404   XBT_IN("(%d)",value);
 405   s_smx_process_t p;
 406
 407   smx_sem_t sem = xbt_new0(s_smx_sem_t, 1);
 408   sem->sleeping = xbt_swag_new(xbt_swag_offset(p, synchro_hookup));
 409   sem->value = value;
 410   XBT_OUT();
 411   return sem;
 412 }
 413
 414 /** @brief Destroys a semaphore */
 415 void SIMIX_sem_destroy(smx_sem_t sem)
 416 {
 417   XBT_IN("(%p)",sem);
 418   XBT_DEBUG("Destroy semaphore %p", sem);
 419   if (sem != NULL) {
 420     xbt_assert(xbt_swag_size(sem->sleeping) == 0,
 421                 "Cannot destroy semaphore since someone is still using it");
 422     xbt_swag_free(sem->sleeping);
 423     xbt_free(sem);
 424   }
 425   XBT_OUT();
 426 }
 427
 428 /** @brief release the semaphore
 429  *
 430  * Unlock a process waiting on the semaphore.
 431  * If no one was blocked, the semaphore capacity is increased by 1.
 432  */
 433 void SIMIX_sem_release(smx_sem_t sem)
 434 {
 435   XBT_IN("(%p)",sem);
 436   smx_process_t proc;
 437
 438   XBT_DEBUG("Sem release semaphore %p", sem);
 439   if ((proc = xbt_swag_extract(sem->sleeping))) {
 440     proc = xbt_swag_extract(sem->sleeping);
 441     SIMIX_synchro_destroy(proc->waiting_action);
 442     proc->waiting_action = NULL;
 443     SIMIX_request_answer(&proc->request);
 444   } else if (sem->value < SMX_SEM_NOLIMIT) {
 445     sem->value++;
 446   }
 447   XBT_OUT();
 448 }
 449
 450 /** @brief Returns true if acquiring this semaphore would block */
 451 XBT_INLINE int SIMIX_sem_would_block(smx_sem_t sem)
 452 {
 453   XBT_IN("(%p)",sem);
 454   XBT_OUT();
 455   return (sem->value <= 0);
 456 }
 457
 458 /** @brief Returns the current capacity of the semaphore */
 459 int SIMIX_sem_get_capacity(smx_sem_t sem)
 460 {
 461   XBT_IN("(%p)",sem);
 462   XBT_OUT();
 463   return sem->value;
 464 }
 465
 466 static void _SIMIX_sem_wait(smx_sem_t sem, double timeout, smx_process_t issuer,
 467                             smx_req_t req)
 468 {
 469   XBT_IN("(%p, %f, %p, %p)",sem,timeout,issuer,req);
 470   smx_action_t sync_act = NULL;
 471
 472   XBT_DEBUG("Wait semaphore %p (timeout:%f)", sem, timeout);
 473   if (sem->value <= 0) {
 474     sync_act = SIMIX_synchro_wait(issuer->smx_host, timeout);
 475     xbt_fifo_unshift(sync_act->request_list, req);
 476     issuer->waiting_action = sync_act;
 477     xbt_swag_insert(issuer, sem->sleeping);
 478   } else {
 479     sem->value--;
 480     SIMIX_request_answer(req);
 481   }
 482   XBT_OUT();
 483 }
 484
 485 /**
 486  * \brief Handle sem acquire requests without timeouts
 487  */
 488 void SIMIX_pre_sem_acquire(smx_req_t req)
 489 {
 490   XBT_IN("(%p)",req);
 491   _SIMIX_sem_wait(req->sem_acquire.sem, -1, req->issuer, req);
 492   XBT_OUT();
 493 }
 494
 495 /**
 496  * \brief Handle sem acquire requests with timeouts
 497  */
 498 void SIMIX_pre_sem_acquire_timeout(smx_req_t req)
 499 {
 500   XBT_IN("(%p)",req);
 501   _SIMIX_sem_wait(req->sem_acquire_timeout.sem,
 502                   req->sem_acquire_timeout.timeout, req->issuer, req);
 503   XBT_OUT();
 504 }