fifo.c @ 14:14cf6daf716d

Last change on this file since 14:14cf6daf716d was 14:14cf6daf716d, checked in by Sebastien Decugis <sdecugis@nict.go.jp>, 12 years ago
Some progress on peers module
File size: 12.4 KB

Line
1	/*********************************************************************************************************
2	* Software License Agreement (BSD License) *
3	* Author: Sebastien Decugis <sdecugis@nict.go.jp> *
4	* *
5	* Copyright (c) 2008, WIDE Project and NICT *
6	* All rights reserved. *
7	* *
8	* Redistribution and use of this software in source and binary forms, with or without modification, are *
9	* permitted provided that the following conditions are met: *
10	* *
11	* * Redistributions of source code must retain the above *
12	* copyright notice, this list of conditions and the *
13	* following disclaimer. *
14	* *
15	* * Redistributions in binary form must reproduce the above *
16	* copyright notice, this list of conditions and the *
17	* following disclaimer in the documentation and/or other *
18	* materials provided with the distribution. *
19	* *
20	* * Neither the name of the WIDE Project or NICT nor the *
21	* names of its contributors may be used to endorse or *
22	* promote products derived from this software without *
23	* specific prior written permission of WIDE Project and *
24	* NICT. *
25	* *
26	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED *
27	* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A *
28	* PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR *
29	* ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
30	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS *
31	* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR *
32	* TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF *
33	* ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
34	*********************************************************************************************************/
35
36	/* FIFO queues module.
37	*
38	* The threads that call these functions must be in the cancellation state PTHREAD_CANCEL_ENABLE and type PTHREAD_CANCEL_DEFERRED.
39	* This is the default state and type on thread creation.
40	*
41	* In order to destroy properly a queue, the application must:
42	* -> shutdown any process that can add into the queue first.
43	* -> pthread_cancel any thread that could be waiting on the queue.
44	* -> consume any element that is in the queue, using fd_qu_tryget_int.
45	* -> then destroy the queue using fd_mq_del.
46	*/
47
48	#include "libfD.h"
49
50	/* Definition of a FIFO queue object */
51	struct fifo {
52	int eyec; /* An eye catcher, also used to check a queue is valid. FIFO_EYEC */
53
54	pthread_mutex_t mtx; /* Mutex protecting this queue */
55	pthread_cond_t cond; /* condition variable of the list */
56
57	struct fd_list list; /* sentinel for the list of elements */
58	int count; /* number of objects in the list */
59	int thrs; /* number of threads waiting for a new element (when count is 0) */
60
61	uint16_t high; /* High level threshold (see libfreeDiameter.h for details) */
62	uint16_t low; /* Low level threshhold */
63	void data; / Opaque pointer for threshold callbacks */
64	void (h_cb)(struct fifo , void *); / The callbacks */
65	void (l_cb)(struct fifo , void **);
66	int highest;/* The highest count value for which h_cb has been called */
67	int highest_ever; /* The max count value this queue has reached (for tweaking) */
68	};
69
70	/* The eye catcher value */
71	#define FIFO_EYEC 0xe7ec1130
72
73	/* Macro to check a pointer */
74	#define CHECK_FIFO( _queue ) (( (_queue) != NULL) && ( (_queue)->eyec == FIFO_EYEC) )
75
76
77	/* Create a new queue */
78	int fd_fifo_new ( struct fifo ** queue )
79	{
80	struct fifo * new;
81
82	TRACE_ENTRY( "%p", queue );
83
84	CHECK_PARAMS( queue );
85
86	/* Create a new object */
87	CHECK_MALLOC( new = malloc (sizeof (struct fifo) ) );
88
89	/* Initialize the content */
90	memset(new, 0, sizeof(struct fifo));
91
92	new->eyec = FIFO_EYEC;
93	CHECK_POSIX( pthread_mutex_init(&new->mtx, NULL) );
94	CHECK_POSIX( pthread_cond_init(&new->cond, NULL) );
95
96	fd_list_init(&new->list, NULL);
97
98	/* We're done */
99	*queue = new;
100	return 0;
101	}
102
103	/* Dump the content of a queue */
104	void fd_fifo_dump(int level, char * name, struct fifo * queue, void (dump_item)(int level, void item))
105	{
106	TRACE_ENTRY("%i %p %p %p", level, name, queue, dump_item);
107
108	if (!TRACE_BOOL(level))
109	return;
110
111	fd_log_debug("Dumping queue '%s' (%p):\n", name ?: "?", queue);
112	if (!CHECK_FIFO( queue )) {
113	fd_log_debug(" Queue invalid!\n");
114	if (queue)
115	fd_log_debug(" (%x != %x)\n", queue->eyec, FIFO_EYEC);
116	return;
117	}
118
119	CHECK_POSIX_DO( pthread_mutex_lock( &queue->mtx ), /* continue */ );
120	fd_log_debug(" %d elements in queue / %d threads waiting\n", queue->count, queue->thrs);
121	fd_log_debug(" thresholds: %d / %d (h:%d), cb: %p,%p (%p), highest: %d\n",
122	queue->high, queue->low, queue->highest,
123	queue->h_cb, queue->l_cb, queue->data,
124	queue->highest_ever);
125
126	if (dump_item) {
127	struct fd_list * li;
128	int i = 0;
129	for (li = queue->list.next; li != &queue->list; li = li->next) {
130	fd_log_debug(" [%i] item %p in fifo %p:\n", i++, li->o, queue);
131	(*dump_item)(level, li->o);
132	}
133	}
134	CHECK_POSIX_DO( pthread_mutex_unlock( &queue->mtx ), /* continue */ );
135
136	}
137
138	/* Delete a queue. It must be unused. */
139	int fd_fifo_del ( struct fifo ** queue )
140	{
141	struct fifo * q;
142
143	TRACE_ENTRY( "%p", queue );
144
145	CHECK_PARAMS( queue && CHECK_FIFO( *queue ) );
146
147	q = *queue;
148
149	CHECK_POSIX( pthread_mutex_lock( &q->mtx ) );
150
151	if ((q->count != 0) \|\| (q->thrs != 0) \|\| (q->data != NULL)) {
152	TRACE_DEBUG(INFO, "The queue cannot be destroyed (%d, %d, %p)", q->count, q->thrs, q->data);
153	CHECK_POSIX_DO( pthread_mutex_unlock( &q->mtx ), /* no fallback */ );
154	return EINVAL;
155	}
156
157	/* sanity check */
158	ASSERT(FD_IS_LIST_EMPTY(&q->list));
159
160	/* Ok, now invalidate the queue */
161	q->eyec = 0xdead;
162
163	/* And destroy it */
164	CHECK_POSIX( pthread_mutex_unlock( &q->mtx ) );
165
166	CHECK_POSIX( pthread_cond_destroy( &q->cond ) );
167
168	CHECK_POSIX( pthread_mutex_destroy( &q->mtx ) );
169
170	free(q);
171	*queue = NULL;
172
173	return 0;
174	}
175
176	/* Get the length of the queue */
177	int fd_fifo_length ( struct fifo * queue, int * length )
178	{
179	TRACE_ENTRY( "%p %p", queue, length );
180
181	/* Check the parameters */
182	CHECK_PARAMS( CHECK_FIFO( queue ) && length );
183
184	/* lock the queue */
185	CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) );
186
187	/* Retrieve the count */
188	*length = queue->count;
189
190	/* Unlock */
191	CHECK_POSIX( pthread_mutex_unlock( &queue->mtx ) );
192
193	/* Done */
194	return 0;
195	}
196
197	/* alternate version with no error checking */
198	int fd_fifo_length_noerr ( struct fifo * queue )
199	{
200	if ( !CHECK_FIFO( queue ) )
201	return 0;
202
203	return queue->count; /* Let's hope it's read atomically, since we are not locking... */
204	}
205
206	/* Set the thresholds of the queue */
207	int fd_fifo_setthrhd ( struct fifo * queue, void * data, uint16_t high, void (h_cb)(struct fifo , void *), uint16_t low, void (l_cb)(struct fifo , void *) )
208	{
209	TRACE_ENTRY( "%p %p %hu %p %hu %p", queue, data, high, h_cb, low, l_cb );
210
211	/* Check the parameters */
212	CHECK_PARAMS( CHECK_FIFO( queue ) && (high > low) && (queue->data == NULL) );
213
214	/* lock the queue */
215	CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) );
216
217	/* Save the values */
218	queue->high = high;
219	queue->low = low;
220	queue->data = data;
221	queue->h_cb = h_cb;
222	queue->l_cb = l_cb;
223
224	/* Unlock */
225	CHECK_POSIX( pthread_mutex_unlock( &queue->mtx ) );
226
227	/* Done */
228	return 0;
229	}
230
231	/* Post a new item in the queue */
232	int fd_fifo_post_int ( struct fifo * queue, void ** item )
233	{
234	struct fd_list * new;
235	int call_cb = 0;
236
237	TRACE_ENTRY( "%p %p", queue, item );
238
239	/* Check the parameters */
240	CHECK_PARAMS( CHECK_FIFO( queue ) && item && *item );
241
242	/* Create a new list item */
243	CHECK_MALLOC( new = malloc (sizeof (struct fd_list)) );
244
245	fd_list_init(new, *item);
246	*item = NULL;
247
248	/* lock the queue */
249	CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) );
250
251	/* Add the new item at the end */
252	fd_list_insert_before( &queue->list, new);
253	queue->count++;
254	if (queue->highest_ever < queue->count)
255	queue->highest_ever = queue->count;
256	if (queue->high && ((queue->count % queue->high) == 0)) {
257	call_cb = 1;
258	queue->highest = queue->count;
259	}
260
261	/* Signal if threads are asleep */
262	if (queue->thrs > 0) {
263	CHECK_POSIX( pthread_cond_signal(&queue->cond) );
264	}
265
266	/* Unlock */
267	CHECK_POSIX( pthread_mutex_unlock( &queue->mtx ) );
268
269	/* Call high-watermark cb as needed */
270	if (call_cb && queue->h_cb)
271	(*queue->h_cb)(queue, &queue->data);
272
273	/* Done */
274	return 0;
275	}
276
277	/* Pop the first item from the queue */
278	static void * mq_pop(struct fifo * queue)
279	{
280	void * ret = NULL;
281	struct fd_list * li;
282
283	ASSERT( ! FD_IS_LIST_EMPTY(&queue->list) );
284
285	fd_list_unlink(li = queue->list.next);
286	queue->count--;
287	ret = li->o;
288	free(li);
289
290	return ret;
291	}
292
293	/* Check if the low watermark callback must be called. */
294	static __inline__ int test_l_cb(struct fifo * queue)
295	{
296	if ((queue->high == 0) \|\| (queue->low == 0) \|\| (queue->l_cb == 0))
297	return 0;
298
299	if (((queue->count % queue->high) == queue->low) && (queue->highest > queue->count)) {
300	queue->highest -= queue->high;
301	return 1;
302	}
303
304	return 0;
305	}
306
307	/* Try poping an item */
308	int fd_fifo_tryget_int ( struct fifo * queue, void ** item )
309	{
310	int wouldblock = 0;
311	int call_cb = 0;
312
313	TRACE_ENTRY( "%p %p", queue, item );
314
315	/* Check the parameters */
316	CHECK_PARAMS( CHECK_FIFO( queue ) && item );
317
318	/* lock the queue */
319	CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) );
320
321	/* Check queue status */
322	if (queue->count > 0) {
323	/* There are elements in the queue, so pick the first one */
324	*item = mq_pop(queue);
325	call_cb = test_l_cb(queue);
326	} else {
327	wouldblock = 1;
328	*item = NULL;
329	}
330
331	/* Unlock */
332	CHECK_POSIX( pthread_mutex_unlock( &queue->mtx ) );
333
334	/* Call low watermark callback as needed */
335	if (call_cb)
336	(*queue->l_cb)(queue, &queue->data);
337
338	/* Done */
339	return wouldblock ? EWOULDBLOCK : 0;
340	}
341
342	/* This handler is called when a thread is blocked on a queue, and cancelled */
343	static void fifo_cleanup(void * queue)
344	{
345	struct fifo * q = (struct fifo *)queue;
346	TRACE_ENTRY( "%p", queue );
347
348	/* Check the parameter */
349	if ( ! CHECK_FIFO( q )) {
350	TRACE_DEBUG(INFO, "Invalid queue, skipping handler");
351	return;
352	}
353
354	/* The thread has been cancelled, therefore it does not wait on the queue anymore */
355	q->thrs--;
356
357	/* Now unlock the queue, and we're done */
358	CHECK_POSIX_DO( pthread_mutex_unlock( &q->mtx ), /* nothing */ );
359
360	/* End of cleanup handler */
361	return;
362	}
363
364	/* The internal function for fd_fifo_timedget and fd_fifo_get */
365	static int fifo_tget ( struct fifo * queue, void ** item, int istimed, const struct timespec *abstime)
366	{
367	int timedout = 0;
368	int call_cb = 0;
369
370	/* Check the parameters */
371	CHECK_PARAMS( CHECK_FIFO( queue ) && item && (abstime \|\| !istimed) );
372
373	/* Initialize the return value */
374	*item = NULL;
375
376	/* lock the queue */
377	CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) );
378
379	awaken:
380	/* Check queue status */
381	if (queue->count > 0) {
382	/* There are items in the queue, so pick the first one */
383	*item = mq_pop(queue);
384	call_cb = test_l_cb(queue);
385	} else {
386	int ret = 0;
387	/* We have to wait for a new item */
388	queue->thrs++ ;
389	pthread_cleanup_push( fifo_cleanup, queue);
390	if (istimed) {
391	ret = pthread_cond_timedwait( &queue->cond, &queue->mtx, abstime );
392	} else {
393	ret = pthread_cond_wait( &queue->cond, &queue->mtx );
394	}
395	pthread_cleanup_pop(0);
396	queue->thrs-- ;
397	if (ret == 0)
398	goto awaken; /* test for spurious wake-ups */
399
400	if (istimed && (ret == ETIMEDOUT)) {
401	timedout = 1;
402	} else {
403	/* Unexpected error condition (means we need to debug) */
404	ASSERT( ret == 0 /* never true */ );
405	}
406	}
407
408	/* Unlock */
409	CHECK_POSIX( pthread_mutex_unlock( &queue->mtx ) );
410
411	/* Call low watermark callback as needed */
412	if (call_cb)
413	(*queue->l_cb)(queue, &queue->data);
414
415	/* Done */
416	return timedout ? ETIMEDOUT : 0;
417	}
418
419	/* Get the next available item, block until there is one */
420	int fd_fifo_get_int ( struct fifo * queue, void ** item )
421	{
422	TRACE_ENTRY( "%p %p", queue, item );
423	return fifo_tget(queue, item, 0, NULL);
424	}
425
426	/* Get the next available item, block until there is one, or the timeout expires */
427	int fd_fifo_timedget_int ( struct fifo * queue, void ** item, const struct timespec *abstime )
428	{
429	TRACE_ENTRY( "%p %p %p", queue, item, abstime );
430	return fifo_tget(queue, item, 1, abstime);
431	}
432

Note: See TracBrowser for help on using the repository browser.

Navigation

source: freeDiameter/libfreeDiameter/fifo.c @ 14:14cf6daf716d

Download in other formats: