X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/5834b3e37665fac531ccb6118397895604d5d569..3318d4f4f09b51bc1158737c5cb0391df264d018:/src/xbt/xbt_queue.c diff --git a/src/xbt/xbt_queue.c b/src/xbt/xbt_queue.c index a31ac73fe3..6a48254a67 100644 --- a/src/xbt/xbt_queue.c +++ b/src/xbt/xbt_queue.c @@ -57,11 +57,11 @@ void xbt_queue_free(xbt_queue_t *queue) { } /** @brief Get the queue size */ -unsigned long xbt_queue_length(xbt_queue_t queue) { +unsigned long xbt_queue_length(const xbt_queue_t queue) { unsigned long res; - xbt_mutex_lock(queue->mutex); + xbt_mutex_acquire(queue->mutex); res=xbt_dynar_length(queue->data); - xbt_mutex_unlock(queue->mutex); + xbt_mutex_release(queue->mutex); return res; } @@ -72,14 +72,14 @@ unsigned long xbt_queue_length(xbt_queue_t queue) { * @see #xbt_dynar_push */ void xbt_queue_push(xbt_queue_t queue, const void *src) { - xbt_mutex_lock(queue->mutex); + xbt_mutex_acquire(queue->mutex); while (queue->capacity != 0 && queue->capacity == xbt_dynar_length(queue->data)) { DEBUG2("Capacity of %p exceded (=%d). Waiting",queue,queue->capacity); xbt_cond_wait(queue->not_full,queue->mutex); } xbt_dynar_push(queue->data,src); xbt_cond_signal(queue->not_empty); - xbt_mutex_unlock(queue->mutex); + xbt_mutex_release(queue->mutex); } @@ -91,14 +91,14 @@ void xbt_queue_push(xbt_queue_t queue, const void *src) { * */ void xbt_queue_pop(xbt_queue_t queue, void* const dst) { - xbt_mutex_lock(queue->mutex); + xbt_mutex_acquire(queue->mutex); while (xbt_dynar_length(queue->data) == 0) { DEBUG1("Queue %p empty. Waiting",queue); xbt_cond_wait(queue->not_empty,queue->mutex); } xbt_dynar_pop(queue->data,dst); xbt_cond_signal(queue->not_full); - xbt_mutex_unlock(queue->mutex); + xbt_mutex_release(queue->mutex); } /** @brief Unshift something to the message exchange queue. @@ -108,14 +108,14 @@ void xbt_queue_pop(xbt_queue_t queue, void* const dst) { * @see #xbt_dynar_unshift */ void xbt_queue_unshift(xbt_queue_t queue, const void *src) { - xbt_mutex_lock(queue->mutex); + xbt_mutex_acquire(queue->mutex); while (queue->capacity != 0 && queue->capacity == xbt_dynar_length(queue->data)) { DEBUG2("Capacity of %p exceded (=%d). Waiting",queue,queue->capacity); xbt_cond_wait(queue->not_full,queue->mutex); } xbt_dynar_unshift(queue->data,src); xbt_cond_signal(queue->not_empty); - xbt_mutex_unlock(queue->mutex); + xbt_mutex_release(queue->mutex); } @@ -127,14 +127,14 @@ void xbt_queue_unshift(xbt_queue_t queue, const void *src) { * */ void xbt_queue_shift(xbt_queue_t queue, void* const dst) { - xbt_mutex_lock(queue->mutex); + xbt_mutex_acquire(queue->mutex); while (xbt_dynar_length(queue->data) == 0) { DEBUG1("Queue %p empty. Waiting",queue); xbt_cond_wait(queue->not_empty,queue->mutex); } xbt_dynar_shift(queue->data,dst); xbt_cond_signal(queue->not_full); - xbt_mutex_unlock(queue->mutex); + xbt_mutex_release(queue->mutex); } @@ -145,15 +145,39 @@ void xbt_queue_shift(xbt_queue_t queue, void* const dst) { * @see #xbt_queue_push */ void xbt_queue_push_timed(xbt_queue_t queue, const void *src,double delay) { - double timeout = xbt_time() + delay; - xbt_mutex_lock(queue->mutex); - while ( (queue->capacity != 0) && (queue->capacity == xbt_dynar_length(queue->data)) && (xbt_time() < timeout) ) { - DEBUG2("Capacity of %p exceded (=%d). Waiting",queue,queue->capacity); - xbt_cond_timedwait(queue->not_full,queue->mutex, timeout - xbt_time()); - } - xbt_dynar_push(queue->data,src); - xbt_cond_signal(queue->not_empty); - xbt_mutex_unlock(queue->mutex); + double timeout = xbt_time() + delay; + xbt_ex_t e; + + xbt_mutex_acquire(queue->mutex); + + if (delay == 0) { + if (queue->capacity != 0 && + queue->capacity == xbt_dynar_length(queue->data)) { + + xbt_mutex_release(queue->mutex); + THROW2(timeout_error,0,"Capacity of %p exceded (=%d), and delay = 0", + queue,queue->capacity); + } + } else { + while (queue->capacity != 0 && + queue->capacity == xbt_dynar_length(queue->data) && + (delay<0 || xbt_time() < timeout) ) { + + DEBUG2("Capacity of %p exceded (=%d). Waiting", + queue,queue->capacity); + TRY { + xbt_cond_timedwait(queue->not_full,queue->mutex, + delay < 0 ? -1 : timeout - xbt_time()); + } CATCH(e) { + xbt_mutex_release(queue->mutex); + RETHROW; + } + } + } + + xbt_dynar_push(queue->data,src); + xbt_cond_signal(queue->not_empty); + xbt_mutex_release(queue->mutex); } @@ -163,15 +187,33 @@ void xbt_queue_push_timed(xbt_queue_t queue, const void *src,double delay) { * */ void xbt_queue_pop_timed(xbt_queue_t queue, void* const dst,double delay) { - double timeout = xbt_time() + delay; - xbt_mutex_lock(queue->mutex); - while ( (xbt_dynar_length(queue->data) == 0) && (xbt_time() < timeout) ) { + double timeout = xbt_time() + delay; + xbt_ex_t e; + + xbt_mutex_acquire(queue->mutex); + + if (delay == 0) { + if (xbt_dynar_length(queue->data) == 0) { + xbt_mutex_release(queue->mutex); + THROW0(timeout_error,0,"Delay = 0, and queue is empty"); + } + } else { + while ( (xbt_dynar_length(queue->data) == 0) && + (delay<0 || xbt_time() < timeout) ) { DEBUG1("Queue %p empty. Waiting",queue); - xbt_cond_timedwait(queue->not_empty,queue->mutex, timeout - xbt_time()); - } - xbt_dynar_pop(queue->data,dst); - xbt_cond_signal(queue->not_full); - xbt_mutex_unlock(queue->mutex); + TRY { + xbt_cond_timedwait(queue->not_empty,queue->mutex, + delay<0 ? -1 : timeout - xbt_time()); + } CATCH(e) { + xbt_mutex_release(queue->mutex); + RETHROW; + } + } + } + + xbt_dynar_pop(queue->data,dst); + xbt_cond_signal(queue->not_full); + xbt_mutex_release(queue->mutex); } /** @brief Unshift something to the message exchange queue, with a timeout. @@ -179,15 +221,39 @@ void xbt_queue_pop_timed(xbt_queue_t queue, void* const dst,double delay) { * @see #xbt_queue_unshift */ void xbt_queue_unshift_timed(xbt_queue_t queue, const void *src,double delay) { - double timeout = xbt_time() + delay; - xbt_mutex_lock(queue->mutex); - while ( (queue->capacity != 0) && (queue->capacity == xbt_dynar_length(queue->data)) && (xbt_time() < timeout) ) { - DEBUG2("Capacity of %p exceded (=%d). Waiting",queue,queue->capacity); - xbt_cond_timedwait(queue->not_full,queue->mutex, timeout - xbt_time()); - } - xbt_dynar_unshift(queue->data,src); - xbt_cond_signal(queue->not_empty); - xbt_mutex_unlock(queue->mutex); + double timeout = xbt_time() + delay; + xbt_ex_t e; + + xbt_mutex_acquire(queue->mutex); + + if (delay==0) { + if (queue->capacity != 0 && + queue->capacity == xbt_dynar_length(queue->data)) { + + xbt_mutex_release(queue->mutex); + THROW2(timeout_error,0,"Capacity of %p exceded (=%d), and delay = 0", + queue,queue->capacity); + } + } else { + while (queue->capacity != 0 && + queue->capacity == xbt_dynar_length(queue->data) && + (delay<0 || xbt_time() < timeout) ) { + + DEBUG2("Capacity of %p exceded (=%d). Waiting", + queue,queue->capacity); + TRY { + xbt_cond_timedwait(queue->not_full,queue->mutex, + delay < 0 ? -1 : timeout - xbt_time()); + } CATCH(e) { + xbt_mutex_release(queue->mutex); + RETHROW; + } + } + } + + xbt_dynar_unshift(queue->data,src); + xbt_cond_signal(queue->not_empty); + xbt_mutex_release(queue->mutex); } @@ -197,13 +263,31 @@ void xbt_queue_unshift_timed(xbt_queue_t queue, const void *src,double delay) { * */ void xbt_queue_shift_timed(xbt_queue_t queue, void* const dst,double delay) { - double timeout = xbt_time() + delay; - xbt_mutex_lock(queue->mutex); - while ( (xbt_dynar_length(queue->data) == 0) && (xbt_time() < timeout) ) { + double timeout = xbt_time() + delay; + xbt_ex_t e; + + xbt_mutex_acquire(queue->mutex); + + if (delay == 0) { + if (xbt_dynar_length(queue->data) == 0) { + xbt_mutex_release(queue->mutex); + THROW0(timeout_error,0,"Delay = 0, and queue is empty"); + } + } else { + while ( (xbt_dynar_length(queue->data) == 0) && + (delay<0 || xbt_time() < timeout) ) { DEBUG1("Queue %p empty. Waiting",queue); - xbt_cond_timedwait(queue->not_empty,queue->mutex, timeout - xbt_time()); - } - xbt_dynar_shift(queue->data,dst); - xbt_cond_signal(queue->not_full); - xbt_mutex_unlock(queue->mutex); + TRY { + xbt_cond_timedwait(queue->not_empty,queue->mutex, + delay<0 ? -1 : timeout - xbt_time()); + } CATCH(e) { + xbt_mutex_release(queue->mutex); + RETHROW; + } + } + } + + xbt_dynar_shift(queue->data,dst); + xbt_cond_signal(queue->not_full); + xbt_mutex_release(queue->mutex); }