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
/* ******************************** 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;
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,
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);
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 */
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 */
} 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 */
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 */
/* 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);
/* 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;
/* 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;
"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);
}
/**
- * \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)
{
}
/**
- * \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)
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.
/* 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);
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);
}
/**
* \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()
/* get the mutex again if necessary */
if(mutex != NULL)
SIMIX_mutex_lock(cond->mutex);
-
- return;
}
/**
xbt_swag_remove(proc, cond->sleeping);
xbt_swag_insert(proc, simix_global->process_to_run);
}
-
- return;
}
/**
}
}
}
+
+/* ************************** 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);
+}
