Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
"new ruby host method"
[simgrid.git] / src / xbt / xbt_queue.c
1 /* $Id$ */
2
3 /* A (synchronized) message queue.                                          */
4 /* Popping an empty queue is blocking, as well as pushing a full one        */
5
6 /* Copyright (c) 2007 Martin Quinson. All rights reserved.                  */
7
8 /* This program is free software; you can redistribute it and/or modify it
9  * under the terms of the license (GNU LGPL) which comes with this package. */
10
11 #include "xbt/misc.h"
12 #include "xbt/sysdep.h"
13 #include "xbt/log.h"
14 #include "xbt/dynar.h"
15
16 #include "xbt/synchro.h"
17 #include "xbt/queue.h"          /* this module */
18 #include "gras/virtu.h"
19 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_queue, xbt, "Message exchanging queue");
20
21 typedef struct s_xbt_queue_ {
22   int capacity;
23   xbt_dynar_t data;
24   xbt_mutex_t mutex;
25   xbt_cond_t not_full, not_empty;
26 } s_xbt_queue_t;
27
28 /** @brief Create a new message exchange queue.
29  *
30  * @param capacity the capacity of the queue. If non-nul, any attempt to push an item which would let the size of the queue over this number will be blocking until someone else pop some data
31  * @param elm_size size of each element stored in it (see #xbt_dynar_new)
32  */
33 xbt_queue_t xbt_queue_new(int capacity, unsigned long elm_size)
34 {
35   xbt_queue_t res = xbt_new0(s_xbt_queue_t, 1);
36   xbt_assert0(capacity >= 0, "Capacity cannot be negative");
37
38   res->capacity = capacity;
39   res->data = xbt_dynar_new(elm_size, NULL);
40   res->mutex = xbt_mutex_init();
41   res->not_full = xbt_cond_init();
42   res->not_empty = xbt_cond_init();
43   return res;
44 }
45
46 /** @brief Destroy a message exchange queue.
47  *
48  * Any remaining content is leaked.
49  */
50 void xbt_queue_free(xbt_queue_t * queue)
51 {
52
53   xbt_dynar_free(&((*queue)->data));
54   xbt_mutex_destroy((*queue)->mutex);
55   xbt_cond_destroy((*queue)->not_full);
56   xbt_cond_destroy((*queue)->not_empty);
57   free((*queue));
58   *queue = NULL;
59 }
60
61 /** @brief Get the queue size */
62 unsigned long xbt_queue_length(const xbt_queue_t queue)
63 {
64   unsigned long res;
65   xbt_mutex_acquire(queue->mutex);
66   res = xbt_dynar_length(queue->data);
67   xbt_mutex_release(queue->mutex);
68   return res;
69 }
70
71 /** @brief Push something to the message exchange queue.
72  *
73  * This is blocking if the declared capacity is non-nul, and if this amount is reached.
74  *
75  * @see #xbt_dynar_push
76  */
77 void xbt_queue_push(xbt_queue_t queue, const void *src)
78 {
79   xbt_mutex_acquire(queue->mutex);
80   while (queue->capacity != 0
81          && queue->capacity == xbt_dynar_length(queue->data)) {
82     DEBUG2("Capacity of %p exceded (=%d). Waiting", queue, queue->capacity);
83     xbt_cond_wait(queue->not_full, queue->mutex);
84   }
85   xbt_dynar_push(queue->data, src);
86   xbt_cond_signal(queue->not_empty);
87   xbt_mutex_release(queue->mutex);
88 }
89
90
91 /** @brief Pop something from the message exchange queue.
92  *
93  * This is blocking if the queue is empty.
94  *
95  * @see #xbt_dynar_pop
96  *
97  */
98 void xbt_queue_pop(xbt_queue_t queue, void *const dst)
99 {
100   xbt_mutex_acquire(queue->mutex);
101   while (xbt_dynar_length(queue->data) == 0) {
102     DEBUG1("Queue %p empty. Waiting", queue);
103     xbt_cond_wait(queue->not_empty, queue->mutex);
104   }
105   xbt_dynar_pop(queue->data, dst);
106   xbt_cond_signal(queue->not_full);
107   xbt_mutex_release(queue->mutex);
108 }
109
110 /** @brief Unshift something to the message exchange queue.
111  *
112  * This is blocking if the declared capacity is non-nul, and if this amount is reached.
113  *
114  * @see #xbt_dynar_unshift
115  */
116 void xbt_queue_unshift(xbt_queue_t queue, const void *src)
117 {
118   xbt_mutex_acquire(queue->mutex);
119   while (queue->capacity != 0
120          && queue->capacity == xbt_dynar_length(queue->data)) {
121     DEBUG2("Capacity of %p exceded (=%d). Waiting", queue, queue->capacity);
122     xbt_cond_wait(queue->not_full, queue->mutex);
123   }
124   xbt_dynar_unshift(queue->data, src);
125   xbt_cond_signal(queue->not_empty);
126   xbt_mutex_release(queue->mutex);
127 }
128
129
130 /** @brief Shift something from the message exchange queue.
131  *
132  * This is blocking if the queue is empty.
133  *
134  * @see #xbt_dynar_shift
135  *
136  */
137 void xbt_queue_shift(xbt_queue_t queue, void *const dst)
138 {
139   xbt_mutex_acquire(queue->mutex);
140   while (xbt_dynar_length(queue->data) == 0) {
141     DEBUG1("Queue %p empty. Waiting", queue);
142     xbt_cond_wait(queue->not_empty, queue->mutex);
143   }
144   xbt_dynar_shift(queue->data, dst);
145   xbt_cond_signal(queue->not_full);
146   xbt_mutex_release(queue->mutex);
147 }
148
149
150
151
152 /** @brief Push something to the message exchange queue, with a timeout.
153  *
154  * @see #xbt_queue_push
155  */
156 void xbt_queue_push_timed(xbt_queue_t queue, const void *src, double delay)
157 {
158   double begin = xbt_time();
159   xbt_ex_t e;
160
161   xbt_mutex_acquire(queue->mutex);
162
163   if (delay == 0) {
164     if (queue->capacity != 0 &&
165         queue->capacity == xbt_dynar_length(queue->data)) {
166
167       xbt_mutex_release(queue->mutex);
168       THROW2(timeout_error, 0, "Capacity of %p exceded (=%d), and delay = 0",
169              queue, queue->capacity);
170     }
171   } else {
172     while (queue->capacity != 0 &&
173            queue->capacity == xbt_dynar_length(queue->data) &&
174            (delay < 0 || (xbt_time() - begin) <= delay)) {
175
176       DEBUG2("Capacity of %p exceded (=%d). Waiting", queue, queue->capacity);
177       TRY {
178         xbt_cond_timedwait(queue->not_full, queue->mutex,
179                            delay < 0 ? -1 : delay - (xbt_time() - begin));
180       }
181       CATCH(e) {
182         xbt_mutex_release(queue->mutex);
183         RETHROW;
184       }
185     }
186   }
187
188   xbt_dynar_push(queue->data, src);
189   xbt_cond_signal(queue->not_empty);
190   xbt_mutex_release(queue->mutex);
191 }
192
193
194 /** @brief Pop something from the message exchange queue, with a timeout.
195  *
196  * @see #xbt_queue_pop
197  *
198  */
199 void xbt_queue_pop_timed(xbt_queue_t queue, void *const dst, double delay)
200 {
201   double begin = xbt_time();
202   xbt_ex_t e;
203
204   xbt_mutex_acquire(queue->mutex);
205
206   if (delay == 0) {
207     if (xbt_dynar_length(queue->data) == 0) {
208       xbt_mutex_release(queue->mutex);
209       THROW0(timeout_error, 0, "Delay = 0, and queue is empty");
210     }
211   } else {
212     while ((xbt_dynar_length(queue->data) == 0) &&
213            (delay < 0 || (xbt_time() - begin) <= delay)) {
214       DEBUG1("Queue %p empty. Waiting", queue);
215       TRY {
216         xbt_cond_timedwait(queue->not_empty, queue->mutex,
217                            delay < 0 ? -1 : delay - (xbt_time() - begin));
218       }
219       CATCH(e) {
220         xbt_mutex_release(queue->mutex);
221         RETHROW;
222       }
223     }
224   }
225
226   xbt_dynar_pop(queue->data, dst);
227   xbt_cond_signal(queue->not_full);
228   xbt_mutex_release(queue->mutex);
229 }
230
231 /** @brief Unshift something to the message exchange queue, with a timeout.
232  *
233  * @see #xbt_queue_unshift
234  */
235 void xbt_queue_unshift_timed(xbt_queue_t queue, const void *src, double delay)
236 {
237   double begin = xbt_time();
238   xbt_ex_t e;
239
240   xbt_mutex_acquire(queue->mutex);
241
242   if (delay == 0) {
243     if (queue->capacity != 0 &&
244         queue->capacity == xbt_dynar_length(queue->data)) {
245
246       xbt_mutex_release(queue->mutex);
247       THROW2(timeout_error, 0, "Capacity of %p exceded (=%d), and delay = 0",
248              queue, queue->capacity);
249     }
250   } else {
251     while (queue->capacity != 0 &&
252            queue->capacity == xbt_dynar_length(queue->data) &&
253            (delay < 0 || (xbt_time() - begin) <= delay)) {
254
255       DEBUG2("Capacity of %p exceded (=%d). Waiting", queue, queue->capacity);
256       TRY {
257         xbt_cond_timedwait(queue->not_full, queue->mutex,
258                            delay < 0 ? -1 : delay - (xbt_time() - begin));
259       }
260       CATCH(e) {
261         xbt_mutex_release(queue->mutex);
262         RETHROW;
263       }
264     }
265   }
266
267   xbt_dynar_unshift(queue->data, src);
268   xbt_cond_signal(queue->not_empty);
269   xbt_mutex_release(queue->mutex);
270 }
271
272
273 /** @brief Shift something from the message exchange queue, with a timeout.
274  *
275  * @see #xbt_queue_shift
276  *
277  */
278 void xbt_queue_shift_timed(xbt_queue_t queue, void *const dst, double delay)
279 {
280   double begin = xbt_time();
281   xbt_ex_t e;
282
283   xbt_mutex_acquire(queue->mutex);
284
285   if (delay == 0) {
286     if (xbt_dynar_length(queue->data) == 0) {
287       xbt_mutex_release(queue->mutex);
288       THROW0(timeout_error, 0, "Delay = 0, and queue is empty");
289     }
290   } else {
291     while ((xbt_dynar_length(queue->data) == 0) &&
292            (delay < 0 || (xbt_time() - begin) <= delay)) {
293       DEBUG1("Queue %p empty. Waiting", queue);
294       TRY {
295         xbt_cond_timedwait(queue->not_empty, queue->mutex,
296                            delay < 0 ? -1 : delay - (xbt_time() - begin));
297       }
298       CATCH(e) {
299         xbt_mutex_release(queue->mutex);
300         RETHROW;
301       }
302     }
303   }
304
305   if (xbt_dynar_length(queue->data) == 0) {
306     xbt_mutex_release(queue->mutex);
307     THROW1(timeout_error, 0, "Timeout (%f) elapsed, but queue still empty",
308            delay);
309   }
310
311   xbt_dynar_shift(queue->data, dst);
312   xbt_cond_signal(queue->not_full);
313   xbt_mutex_release(queue->mutex);
314 }