From 39997124732a66cfd24f1b50cf816bfb7ed43b3b Mon Sep 17 00:00:00 2001 From: mquinson Date: Tue, 20 Oct 2009 13:09:36 +0000 Subject: [PATCH] Enough is enough. Add semaphores as SIMIX synchronization mechanism. Check the SIMIX_sem_release_forever() function, which allows something between conditions and pure semaphores. git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@6806 48e7efb5-ca39-0410-a469-dd3cf9ba447f --- ChangeLog | 3 +- src/include/simix/datatypes.h | 1 + src/include/simix/simix.h | 25 +++++- src/simix/private.h | 11 +++ src/simix/smx_action.c | 57 +++++++++--- src/simix/smx_synchro.c | 164 ++++++++++++++++++++++++++++++++-- 6 files changed, 241 insertions(+), 20 deletions(-) diff --git a/ChangeLog b/ChangeLog index 65499d7e71..603559d356 100644 --- a/ChangeLog +++ b/ChangeLog @@ -51,7 +51,8 @@ SimGrid (3.3.4) unstable; urgency=low SIMIX: * New SIMIX network module It provides a rendez-vous point (aka mailbox) based send/recv API to perform - synchronous communication. + synchronous communication. + * Add semaphores as SIMIX synchronization mechanism. SimDag: * new function SD_daxload(char*) to load a DAX file diff --git a/src/include/simix/datatypes.h b/src/include/simix/datatypes.h index dd998ac525..863070d1c0 100644 --- a/src/include/simix/datatypes.h +++ b/src/include/simix/datatypes.h @@ -35,6 +35,7 @@ SG_BEGIN_DECL() /* ******************************** Syncro ************************************ */ typedef struct s_smx_mutex *smx_mutex_t; typedef struct s_smx_cond *smx_cond_t; + typedef struct s_smx_sem *smx_sem_t; /********************************** Action *************************************/ typedef struct s_smx_action *smx_action_t; diff --git a/src/include/simix/simix.h b/src/include/simix/simix.h index c0a0fa8eb3..1fe9aa004e 100644 --- a/src/include/simix/simix.h +++ b/src/include/simix/simix.h @@ -138,6 +138,18 @@ XBT_PUBLIC(void) SIMIX_cond_destroy(smx_cond_t cond); XBT_PUBLIC(xbt_fifo_t) SIMIX_cond_get_actions(smx_cond_t cond); XBT_PUBLIC(void) SIMIX_cond_display_info(smx_cond_t cond); +/*****Semaphores*******/ + + +XBT_PUBLIC(smx_sem_t) SIMIX_sem_init(int capacity); +XBT_PUBLIC(void) SIMIX_sem_destroy(smx_sem_t sem); +XBT_PUBLIC(void) SIMIX_sem_release(smx_sem_t sem); +XBT_PUBLIC(void) SIMIX_sem_release_forever(smx_sem_t sem); +XBT_PUBLIC(int) SIMIX_sem_would_block(smx_sem_t sem); +XBT_PUBLIC(void) SIMIX_sem_acquire(smx_sem_t sem); +XBT_PUBLIC(void) SIMIX_sem_acquire_timeout(smx_sem_t sem, double max_duration); + + /************************** Action handling ************************************/ XBT_PUBLIC(smx_action_t) SIMIX_action_communicate(smx_host_t sender, smx_host_t receiver, @@ -153,10 +165,15 @@ XBT_PUBLIC(void) SIMIX_action_set_priority(smx_action_t action, XBT_PUBLIC(int) SIMIX_action_destroy(smx_action_t action); XBT_PUBLIC(void) SIMIX_action_use(smx_action_t action); XBT_PUBLIC(void) SIMIX_action_release(smx_action_t action); + XBT_PUBLIC(void) SIMIX_register_action_to_condition(smx_action_t action, smx_cond_t cond); XBT_PUBLIC(void) SIMIX_unregister_action_to_condition(smx_action_t action, smx_cond_t cond); +XBT_PUBLIC(void) SIMIX_register_action_to_semaphore(smx_action_t action, smx_sem_t sem); +XBT_PUBLIC(void) SIMIX_unregister_action_to_semaphore(smx_action_t action, smx_sem_t sem); + + XBT_PUBLIC(double) SIMIX_action_get_remains(smx_action_t action); XBT_PUBLIC(e_surf_action_state_t) SIMIX_action_get_state(smx_action_t action); @@ -197,8 +214,12 @@ XBT_PUBLIC(void) SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate size_t src_buff_size, smx_comm_t *comm, void *data); XBT_PUBLIC(void) SIMIX_network_recv(smx_rdv_t rdv, double timeout, void *dst_buff, size_t *dst_buff_size, smx_comm_t *comm); -XBT_PUBLIC(void) SIMIX_network_wait(smx_action_t comm, double timeout); -XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm); +XBT_PUBLIC(smx_comm_t) SIMIX_network_isend(smx_rdv_t rdv, double task_size, double rate, + void *src_buff, size_t src_buff_size, void *data); +XBT_PUBLIC(smx_comm_t) SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_size); + +XBT_PUBLIC(void) SIMIX_network_wait(smx_comm_t comm, double timeout); +XBT_PUBLIC(int) SIMIX_network_test(smx_comm_t comm); SG_END_DECL() #endif /* _SIMIX_SIMIX_H */ diff --git a/src/simix/private.h b/src/simix/private.h index 6560c3ecb5..e797a2193f 100644 --- a/src/simix/private.h +++ b/src/simix/private.h @@ -72,6 +72,7 @@ extern SIMIX_Global_t simix_global; int iwannadie : 1; smx_mutex_t mutex; /* mutex on which the process is blocked */ smx_cond_t cond; /* cond on which the process is blocked */ + smx_sem_t sem; /* semaphore on which the process is blocked */ smx_action_t waiting_action; xbt_dict_t properties; void *data; /* kept for compatibility, it should be replaced with moddata */ @@ -117,6 +118,15 @@ typedef struct s_smx_cond { } s_smx_cond_t; +typedef struct s_smx_sem { + /* KEEP IT IN SYNC WITH src/xbt_sg_thread.c::struct s_xbt_sem */ + xbt_swag_t sleeping; /* list of sleeping process */ + int capacity; + xbt_fifo_t actions; /* list of actions */ + /* KEEP IT IN SYNC WITH src/xbt_sg_thread.c::struct s_xbt_sem */ + +} s_smx_sem_t; + /******************************* Networking ***********************************/ /** @brief Rendez-vous point datatype */ @@ -168,6 +178,7 @@ typedef enum {ready, ongoing, done, failed} smx_action_state_t; typedef struct s_smx_action { char *name; /**< @brief action name if any */ xbt_fifo_t cond_list; /*< conditional variables that must be signaled when the action finish. */ + xbt_fifo_t sem_list; /*< semaphores that must be signaled when the action finish. */ void *data; /**< @brief user data */ int refcount; /**< @brief reference counter */ surf_action_t surf_action; /* SURF modeling of computation */ diff --git a/src/simix/smx_action.c b/src/simix/smx_action.c index a61c9230eb..4003661164 100644 --- a/src/simix/smx_action.c +++ b/src/simix/smx_action.c @@ -45,6 +45,7 @@ smx_action_t SIMIX_action_communicate(smx_host_t sender, /* alloc structures */ act = xbt_new0(s_smx_action_t, 1); act->cond_list = xbt_fifo_new(); + act->sem_list = xbt_fifo_new(); /* initialize them */ act->name = xbt_strdup(name); @@ -84,6 +85,7 @@ smx_action_t SIMIX_action_execute(smx_host_t host, const char *name, /* alloc structures */ act = xbt_new0(s_smx_action_t, 1); act->cond_list = xbt_fifo_new(); + act->sem_list = xbt_fifo_new(); /* initialize them */ act->source = host; @@ -124,6 +126,7 @@ smx_action_t SIMIX_action_sleep(smx_host_t host, double duration) /* alloc structures */ act = xbt_new0(s_smx_action_t, 1); act->cond_list = xbt_fifo_new(); + act->sem_list = xbt_fifo_new(); /* initialize them */ act->source = host; @@ -189,11 +192,16 @@ int SIMIX_action_destroy(smx_action_t action) "Conditional list not empty %d. There is a problem. Cannot destroy it now!", xbt_fifo_size(action->cond_list)); + xbt_assert1((xbt_fifo_size(action->sem_list) == 0), + "Semaphore list not empty %d. There is a problem. Cannot destroy it now!", + xbt_fifo_size(action->sem_list)); + DEBUG1("Destroy action %p", action); if (action->name) xbt_free(action->name); xbt_fifo_free(action->cond_list); + xbt_fifo_free(action->sem_list); if (action->surf_action) action->surf_action->model_type->action_unref(action->surf_action); @@ -232,11 +240,11 @@ void SIMIX_action_release(smx_action_t action) } /** - * \brief Set an action to a condition + * \brief Set an action to a condition * - * Creates the "link" between an action and a condition. You have to call this function when you create an action and want to wait its ending. - * \param action SIMIX action - * \param cond SIMIX cond + * Creates the "link" between an action and a condition. You have to call this function when you create an action and want to wait its ending. + * \param action SIMIX action + * \param cond SIMIX cond */ void SIMIX_register_action_to_condition(smx_action_t action, smx_cond_t cond) { @@ -263,11 +271,11 @@ void SIMIX_register_action_to_condition(smx_action_t action, smx_cond_t cond) } /** - * \brief Unset an action to a condition. + * \brief Unset an action to a condition. * - * Destroys the "links" from the condition to this action. - * \param action SIMIX action - * \param cond SIMIX cond + * Destroys the "links" from the condition to this action. + * \param action SIMIX action + * \param cond SIMIX cond */ void SIMIX_unregister_action_to_condition(smx_action_t action, smx_cond_t cond) @@ -284,12 +292,34 @@ void SIMIX_unregister_action_to_condition(smx_action_t action, if(XBT_LOG_ISENABLED(simix_action, xbt_log_priority_debug)) __SIMIX_action_display_conditions(action); - + xbt_fifo_remove_all(action->cond_list, cond); if(XBT_LOG_ISENABLED(simix_action, xbt_log_priority_debug)) __SIMIX_action_display_conditions(action); } +/** + * \brief Link an action to a semaphore + * + * When the action terminates, the semaphore gets signaled automatically. + */ +void SIMIX_register_action_to_semaphore(smx_action_t action, smx_sem_t sem) { + + DEBUG2("Register action %p to semaphore %p (and otherwise)", action, sem); + xbt_fifo_push(sem->actions, action); + xbt_fifo_push(action->sem_list, sem); +} +/** + * \brief Unset an action to a semaphore. + * + * Destroys the "links" from the semaphore to this action. + */ +void SIMIX_unregister_action_to_semaphore(smx_action_t action, + smx_sem_t sem) +{ + xbt_fifo_remove_all(sem->actions, action); + xbt_fifo_remove_all(action->sem_list, sem); +} /** * \brief Return how much remais to be done in the action. @@ -316,6 +346,7 @@ smx_action_t SIMIX_action_parallel_execute(char *name, int host_nb, /* alloc structures */ act = xbt_new0(s_smx_action_t, 1); act->cond_list = xbt_fifo_new(); + act->sem_list = xbt_fifo_new(); /* initialize them */ act->name = xbt_strdup(name); @@ -373,12 +404,14 @@ void SIMIX_action_set_name(smx_action_t action,char *name) action->name = name; } -void SIMIX_action_signal_all(smx_action_t action) -{ +/** @brief broadcast any condition and release any semaphore including this action */ +void SIMIX_action_signal_all(smx_action_t action){ smx_cond_t cond; + smx_sem_t sem; while ((cond = xbt_fifo_pop(action->cond_list))) SIMIX_cond_broadcast(cond); - return; + while ((sem = xbt_fifo_pop(action->sem_list))) + SIMIX_sem_release(sem); } diff --git a/src/simix/smx_synchro.c b/src/simix/smx_synchro.c index 8dd790cc53..8272cce96c 100644 --- a/src/simix/smx_synchro.c +++ b/src/simix/smx_synchro.c @@ -131,7 +131,8 @@ void SIMIX_mutex_destroy(smx_mutex_t mutex) /** * \brief Initialize a condition. * - * Allocs and creates the data for the condition. It have to be called before the utilisation of the condition. + * Allocates and creates the data for the condition. + * It have to be called before the use of the condition. * \return A condition */ smx_cond_t SIMIX_cond_init() @@ -267,8 +268,6 @@ void SIMIX_cond_wait_timeout(smx_cond_t cond, smx_mutex_t mutex, /* get the mutex again if necessary */ if(mutex != NULL) SIMIX_mutex_lock(cond->mutex); - - return; } /** @@ -289,8 +288,6 @@ void SIMIX_cond_broadcast(smx_cond_t cond) xbt_swag_remove(proc, cond->sleeping); xbt_swag_insert(proc, simix_global->process_to_run); } - - return; } /** @@ -339,3 +336,160 @@ void SIMIX_cond_display_info(smx_cond_t cond) } } } + +/* ************************** Semaphores ************************************** */ +#define SMX_SEM_NOLIMIT 99999 +/** @brief Initialize a semaphore */ +smx_sem_t SIMIX_sem_init(int capacity) { + smx_sem_t sem = xbt_new0(s_smx_sem_t, 1); + s_smx_process_t p; + + sem->sleeping = xbt_swag_new(xbt_swag_offset(p, synchro_hookup)); + sem->actions = xbt_fifo_new(); + sem->capacity = capacity; + return sem; +} +/** @brief Destroys a semaphore */ +void SIMIX_sem_destroy(smx_sem_t sem) { + DEBUG1("Destroy semaphore %p", sem); + if (sem == NULL) + return; + + xbt_fifo_item_t item = NULL; + smx_action_t action = NULL; + + xbt_assert0(xbt_swag_size(sem->sleeping) == 0, + "Cannot destroy semaphore since someone is still using it"); + xbt_swag_free(sem->sleeping); + + DEBUG1("%d actions registered", xbt_fifo_size(sem->actions)); + xbt_fifo_foreach(sem->actions, item, action, smx_action_t) { + SIMIX_unregister_action_to_semaphore(action, sem); + } + + xbt_fifo_free(sem->actions); + xbt_free(sem); +} + +/** @brief release the semaphore + * + * The first locked process on this semaphore is unlocked. + * If no one was blocked, the semaphore capacity is increased by 1. + * */ +void SIMIX_sem_release(smx_sem_t sem) { + if (xbt_swag_size(sem->sleeping) >= 1) { + smx_process_t proc = xbt_swag_extract(sem->sleeping); + xbt_swag_insert(proc, simix_global->process_to_run); + } else if (sem->capacity != SMX_SEM_NOLIMIT) { + sem->capacity++; + } +} +/** @brief make sure the semaphore will never be blocking again + * + * This function is not really in the semaphore spirit. It makes + * sure that the semaphore will never be blocking anymore. + * + * Releasing and acquiring the semaphore after calling this + * function is a noop. Such "broken" semaphores are useful to + * implement something between condition variables (with broadcast) + * and semaphore (with memory). It's like a semaphore signaled for ever. + * + * There is no way to reset the semaphore to a more regular state afterward. + * */ +void SIMIX_sem_release_forever(smx_sem_t sem) { + smx_process_t proc = NULL; + smx_process_t proc_next = NULL; + + DEBUG1("Broadcast semaphore %p", sem); + xbt_swag_foreach_safe(proc, proc_next, sem->sleeping) { + xbt_swag_remove(proc, sem->sleeping); + xbt_swag_insert(proc, simix_global->process_to_run); + } +} + +static inline void SIMIX_sem_block_onto(smx_sem_t sem) { + smx_process_t self = SIMIX_process_self(); + + /* process status */ + self->sem = sem; + xbt_swag_insert(self, sem->sleeping); + SIMIX_process_yield(); + self->sem = NULL; + while (self->suspended) + SIMIX_process_yield(); +} + +/** @brief Returns true if acquiring this semaphore would block */ +int SIMIX_sem_would_block(smx_sem_t sem) { + return (sem->capacity>0); +} + +/** + * \brief Waits on a semaphore + * + * If the capacity>0, decrease the capacity. + * + * If capacity==0, locks the current process + * until someone call SIMIX_sem_release() on this semaphore + */ +void SIMIX_sem_acquire(smx_sem_t sem) { + smx_action_t act_sleep; + + DEBUG1("Wait semaphore %p", sem); + + if (sem->capacity == SMX_SEM_NOLIMIT) + return; /* don't even decrease it if wide open */ + + /* If capacity sufficient, decrease it */ + if (sem->capacity>0) { + sem->capacity--; + return; + } + + /* Always create an action null in case there is a host failure */ + act_sleep = SIMIX_action_sleep(SIMIX_host_self(), -1); + SIMIX_action_set_name(act_sleep,bprintf("Locked in semaphore %p", sem)); + SIMIX_process_self()->waiting_action = act_sleep; + SIMIX_register_action_to_semaphore(act_sleep, sem); + SIMIX_sem_block_onto(sem); + SIMIX_process_self()->waiting_action = NULL; + SIMIX_unregister_action_to_semaphore(act_sleep, sem); + SIMIX_action_destroy(act_sleep); +} +/** + * \brief Tries to acquire a semaphore before a timeout + * + * Same behavior of #SIMIX_sem_acquire, but waits a maximum time and throws an timeout_error if it happens. + */ +void SIMIX_sem_acquire_timeout(smx_sem_t sem, double max_duration) { + smx_action_t act_sleep; + + DEBUG2("Timed wait semaphore %p (timeout:%f)", sem,max_duration); + + if (sem->capacity == SMX_SEM_NOLIMIT) + return; /* don't even decrease it if wide open */ + + /* If capacity sufficient, decrease it */ + if (sem->capacity>0) { + sem->capacity--; + return; + } + + if (max_duration >= 0) { + act_sleep = SIMIX_action_sleep(SIMIX_host_self(), max_duration); + SIMIX_action_set_name(act_sleep,bprintf("Timed wait semaphore %p (max_duration:%f)", sem,max_duration)); + SIMIX_register_action_to_semaphore(act_sleep, sem); + SIMIX_process_self()->waiting_action = act_sleep; + SIMIX_sem_block_onto(sem); + SIMIX_process_self()->waiting_action = NULL; + SIMIX_unregister_action_to_semaphore(act_sleep, sem); + if (SIMIX_action_get_state(act_sleep) == SURF_ACTION_DONE) { + SIMIX_action_destroy(act_sleep); + THROW1(timeout_error, 0, "Semaphore acquire timeouted after %f",max_duration); + } else { + SIMIX_action_destroy(act_sleep); + } + + } else + SIMIX_sem_block_onto(sem); +} -- 2.20.1