Index: extensions/app_radgw/rgw_worker.c
===================================================================
--- extensions/app_radgw/rgw_worker.c	(revision 741)
+++ extensions/app_radgw/rgw_worker.c	(revision 767)
@@ -327,5 +327,5 @@
 	memset(workers, 0, sizeof(workers));
 	
-	CHECK_FCT( fd_fifo_new ( &work_stack ) );
+	CHECK_FCT( fd_fifo_new ( &work_stack, 30 ) );
 	
 	/* Create the worker thread(s) */
Index: extensions/dbg_interactive/queues.i
===================================================================
--- extensions/dbg_interactive/queues.i	(revision 741)
+++ extensions/dbg_interactive/queues.i	(revision 767)
@@ -42,7 +42,7 @@
 
 %extend fifo {
-	fifo() {
+	fifo(int max = 0) {
 		struct fifo * q = NULL;
-		int ret = fd_fifo_new(&q);
+		int ret = fd_fifo_new(&q, max);
 		if (ret != 0) {
 			DI_ERROR(ret, NULL, NULL);
@@ -147,5 +147,5 @@
 		ret = fd_fifo_tryget($self, &obj);
 		if (ret == EWOULDBLOCK) {
-			Py_XINCREF(Py_None);
+			Py_INCREF(Py_None);
 			return Py_None;
 		}
@@ -182,5 +182,5 @@
 		ret = fd_fifo_timedget($self, &obj, &ts);
 		if (ret == ETIMEDOUT) {
-			Py_XINCREF(Py_None);
+			Py_INCREF(Py_None);
 			return Py_None;
 		}
Index: include/freeDiameter/libfdproto.h
===================================================================
--- include/freeDiameter/libfdproto.h	(revision 764)
+++ include/freeDiameter/libfdproto.h	(revision 767)
@@ -2759,4 +2759,6 @@
  * PARAMETERS:
  *  queue	: Upon success, a pointer to the new queue is saved here.
+ *  max		: max number of items in the queue. Above this number, adding a new item becomes a
+ *		  blocking operation. Use 0 to disable this maximum.
  *
  * DESCRIPTION: 
@@ -2768,5 +2770,5 @@
  *  ENOMEM	: Not enough memory to complete the creation.  
  */
-int fd_fifo_new ( struct fifo ** queue );
+int fd_fifo_new ( struct fifo ** queue, int max );
 
 /*
Index: libfdcore/cnxctx.c
===================================================================
--- libfdcore/cnxctx.c	(revision 740)
+++ libfdcore/cnxctx.c	(revision 767)
@@ -90,5 +90,5 @@
 
 	if (full) {
-		CHECK_FCT_DO( fd_fifo_new ( &conn->cc_incoming ), return NULL );
+		CHECK_FCT_DO( fd_fifo_new ( &conn->cc_incoming, 5 ), return NULL );
 	}
 
Index: libfdcore/config.c
===================================================================
--- libfdcore/config.c	(revision 740)
+++ libfdcore/config.c	(revision 767)
@@ -69,5 +69,5 @@
 	
 	CHECK_FCT( fd_dict_init(&fd_g_config->cnf_dict) );
-	CHECK_FCT( fd_fifo_new(&fd_g_config->cnf_main_ev) );
+	CHECK_FCT( fd_fifo_new(&fd_g_config->cnf_main_ev, 0) );
 	
 	/* TLS parameters */
Index: libfdcore/p_psm.c
===================================================================
--- libfdcore/p_psm.c	(revision 740)
+++ libfdcore/p_psm.c	(revision 767)
@@ -856,5 +856,5 @@
 	
 	/* Create the FIFO for events */
-	CHECK_FCT( fd_fifo_new(&peer->p_events) );
+	CHECK_FCT( fd_fifo_new(&peer->p_events, 0) );
 	
 	/* Create the PSM controler thread */
Index: libfdcore/peers.c
===================================================================
--- libfdcore/peers.c	(revision 749)
+++ libfdcore/peers.c	(revision 767)
@@ -77,5 +77,5 @@
 	fd_list_init(&p->p_actives, p);
 	fd_list_init(&p->p_expiry, p);
-	CHECK_FCT( fd_fifo_new(&p->p_tosend) );
+	CHECK_FCT( fd_fifo_new(&p->p_tosend, 5) );
 	p->p_hbh = lrand48();
 	
Index: libfdcore/queues.c
===================================================================
--- libfdcore/queues.c	(revision 740)
+++ libfdcore/queues.c	(revision 767)
@@ -45,7 +45,7 @@
 {
 	TRACE_ENTRY();
-	CHECK_FCT( fd_fifo_new ( &fd_g_incoming ) );
-	CHECK_FCT( fd_fifo_new ( &fd_g_outgoing ) );
-	CHECK_FCT( fd_fifo_new ( &fd_g_local ) );
+	CHECK_FCT( fd_fifo_new ( &fd_g_incoming, 20 ) );
+	CHECK_FCT( fd_fifo_new ( &fd_g_outgoing, 30 ) );
+	CHECK_FCT( fd_fifo_new ( &fd_g_local, 25 ) );
 	return 0;
 }
Index: libfdcore/sctps.c
===================================================================
--- libfdcore/sctps.c	(revision 740)
+++ libfdcore/sctps.c	(revision 767)
@@ -497,5 +497,5 @@
 		conn->cc_sctps_data.array[i].parent = conn;
 		conn->cc_sctps_data.array[i].strid  = i;
-		CHECK_FCT( fd_fifo_new(&conn->cc_sctps_data.array[i].raw_recv) );
+		CHECK_FCT( fd_fifo_new(&conn->cc_sctps_data.array[i].raw_recv, 10) );
 	}
 	
Index: libfdproto/fifo.c
===================================================================
--- libfdproto/fifo.c	(revision 740)
+++ libfdproto/fifo.c	(revision 767)
@@ -53,9 +53,13 @@
 	
 	pthread_mutex_t	mtx;	/* Mutex protecting this queue */
-	pthread_cond_t	cond;	/* condition variable of the list */
+	pthread_cond_t	cond_pull;	/* condition variable for pulling threads */
+	pthread_cond_t	cond_push;	/* condition variable for pushing threads */
 	
 	struct fd_list	list;	/* sentinel for the list of elements */
 	int		count;	/* number of objects in the list */
 	int		thrs;	/* number of threads waiting for a new element (when count is 0) */
+	
+	int 		max;	/* maximum number of items to accept if not 0 */
+	int		thrs_push; /* number of threads waitnig to push an item */
 	
 	uint16_t	high;	/* High level threshold (see libfreeDiameter.h for details) */
@@ -75,6 +79,6 @@
 
 
-/* Create a new queue */
-int fd_fifo_new ( struct fifo ** queue )
+/* Create a new queue, with max number of items -- use 0 for no max */
+int fd_fifo_new ( struct fifo ** queue, int max )
 {
 	struct fifo * new;
@@ -92,5 +96,7 @@
 	new->eyec = FIFO_EYEC;
 	CHECK_POSIX( pthread_mutex_init(&new->mtx, NULL) );
-	CHECK_POSIX( pthread_cond_init(&new->cond, NULL) );
+	CHECK_POSIX( pthread_cond_init(&new->cond_pull, NULL) );
+	CHECK_POSIX( pthread_cond_init(&new->cond_push, NULL) );
+	new->max = max;
 	
 	fd_list_init(&new->list, NULL);
@@ -119,4 +125,5 @@
 	CHECK_POSIX_DO(  pthread_mutex_lock( &queue->mtx ), /* continue */  );
 	fd_log_debug("   %d elements in queue / %d threads waiting\n", queue->count, queue->thrs);
+	fd_log_debug("   %d elements max / %d threads waiting to push\n", queue->max, queue->thrs_push);
 	fd_log_debug("   thresholds: %d / %d (h:%d), cb: %p,%p (%p), highest: %d\n",
 			queue->high, queue->low, queue->highest, 
@@ -162,9 +169,6 @@
 	while (q->thrs) {
 		CHECK_POSIX(  pthread_mutex_unlock( &q->mtx ));
-		CHECK_POSIX(  pthread_cond_signal(&q->cond)  );
-		sched_yield();
-		if (loops >= 10)
-			/* sleep for a few milliseconds */
-			usleep(50000);
+		CHECK_POSIX(  pthread_cond_signal(&q->cond_pull)  );
+		usleep(1000);
 		
 		CHECK_POSIX(  pthread_mutex_lock( &q->mtx )  );
@@ -178,5 +182,7 @@
 	CHECK_POSIX(  pthread_mutex_unlock( &q->mtx )  );
 	
-	CHECK_POSIX_DO(  pthread_cond_destroy( &q->cond ),  );
+	CHECK_POSIX_DO(  pthread_cond_destroy( &q->cond_pull ),  );
+	
+	CHECK_POSIX_DO(  pthread_cond_destroy( &q->cond_push ),  );
 	
 	CHECK_POSIX_DO(  pthread_mutex_destroy( &q->mtx ),  );
@@ -207,4 +213,10 @@
 	/* Lock the queues */
 	CHECK_POSIX(  pthread_mutex_lock( &old->mtx )  );
+	
+	CHECK_PARAMS_DO( (! old->thrs_push), {
+			pthread_mutex_unlock( &old->mtx );
+			return EINVAL;
+		} );
+	
 	CHECK_POSIX(  pthread_mutex_lock( &new->mtx )  );
 	
@@ -213,12 +225,9 @@
 	while (old->thrs) {
 		CHECK_POSIX(  pthread_mutex_unlock( &old->mtx ));
-		CHECK_POSIX(  pthread_cond_signal(&old->cond)  );
-		sched_yield();
-		if (loops >= 10)
-			/* sleep for a few milliseconds */
-			usleep(50000);
+		CHECK_POSIX(  pthread_cond_signal( &old->cond_pull )  );
+		usleep(1000);
 		
 		CHECK_POSIX(  pthread_mutex_lock( &old->mtx )  );
-		ASSERT( ++loops < 20 ); /* detect infinite loops */
+		ASSERT( loops < 20 ); /* detect infinite loops */
 	}
 	
@@ -226,5 +235,5 @@
 	fd_list_move_end( &new->list, &old->list );
 	if (old->count && (!new->count)) {
-		CHECK_POSIX(  pthread_cond_signal(&new->cond)  );
+		CHECK_POSIX(  pthread_cond_signal(&new->cond_pull)  );
 	}
 	new->count += old->count;
@@ -296,4 +305,22 @@
 }
 
+
+/* This handler is called when a thread is blocked on a queue, and cancelled */
+static void fifo_cleanup_push(void * queue)
+{
+	struct fifo * q = (struct fifo *)queue;
+	TRACE_ENTRY( "%p", queue );
+	
+	/* The thread has been cancelled, therefore it does not wait on the queue anymore */
+	q->thrs_push--;
+	
+	/* Now unlock the queue, and we're done */
+	CHECK_POSIX_DO(  pthread_mutex_unlock( &q->mtx ),  /* nothing */  );
+	
+	/* End of cleanup handler */
+	return;
+}
+
+
 /* Post a new item in the queue */
 int fd_fifo_post_int ( struct fifo * queue, void ** item )
@@ -307,12 +334,29 @@
 	CHECK_PARAMS( CHECK_FIFO( queue ) && item && *item );
 	
+	/* lock the queue */
+	CHECK_POSIX(  pthread_mutex_lock( &queue->mtx )  );
+	
+	if (queue->max) {
+		while (queue->count >= queue->max) {
+			int ret = 0;
+			
+			/* We have to wait for an item to be pulled */
+			queue->thrs_push++ ;
+			pthread_cleanup_push( fifo_cleanup_push, queue);
+			ret = pthread_cond_wait( &queue->cond_push, &queue->mtx );
+			pthread_cleanup_pop(0);
+			queue->thrs_push-- ;
+			
+			ASSERT( ret == 0 );
+		}
+	}
+	
 	/* Create a new list item */
-	CHECK_MALLOC(  new = malloc (sizeof (struct fd_list))  );
+	CHECK_MALLOC_DO(  new = malloc (sizeof (struct fd_list)) , {
+			pthread_mutex_unlock( &queue->mtx );
+		} );
 	
 	fd_list_init(new, *item);
 	*item = NULL;
-	
-	/* lock the queue */
-	CHECK_POSIX(  pthread_mutex_lock( &queue->mtx )  );
 	
 	/* Add the new item at the end */
@@ -328,5 +372,9 @@
 	/* Signal if threads are asleep */
 	if (queue->thrs > 0) {
-		CHECK_POSIX(  pthread_cond_signal(&queue->cond)  );
+		CHECK_POSIX(  pthread_cond_signal(&queue->cond_pull)  );
+	}
+	if (queue->thrs_push > 0) {
+		/* cascade */
+		CHECK_POSIX(  pthread_cond_signal(&queue->cond_push)  );
 	}
 	
@@ -355,4 +403,8 @@
 	free(li);
 	
+	if (queue->thrs_push) {
+		CHECK_POSIX_DO( pthread_cond_signal( &queue->cond_push ), );
+	}
+	
 	return ret;
 }
@@ -388,8 +440,19 @@
 	/* Check queue status */
 	if (queue->count > 0) {
+got_item:
 		/* There are elements in the queue, so pick the first one */
 		*item = mq_pop(queue);
 		call_cb = test_l_cb(queue);
 	} else {
+		if (queue->thrs_push > 0) {
+			/* A thread is trying to push something, let's give it a chance */
+			CHECK_POSIX(  pthread_mutex_unlock( &queue->mtx )  );
+			CHECK_POSIX(  pthread_cond_signal( &queue->cond_push )  );
+			usleep(1000);
+			CHECK_POSIX(  pthread_mutex_lock( &queue->mtx )  );
+			if (queue->count > 0)
+				goto got_item;
+		}
+		
 		wouldblock = 1;
 		*item = NULL;
@@ -457,7 +520,7 @@
 		pthread_cleanup_push( fifo_cleanup, queue);
 		if (istimed) {
-			ret = pthread_cond_timedwait( &queue->cond, &queue->mtx, abstime );
+			ret = pthread_cond_timedwait( &queue->cond_pull, &queue->mtx, abstime );
 		} else {
-			ret = pthread_cond_wait( &queue->cond, &queue->mtx );
+			ret = pthread_cond_wait( &queue->cond_pull, &queue->mtx );
 		}
 		pthread_cleanup_pop(0);
Index: tests/testcnx.c
===================================================================
--- tests/testcnx.c	(revision 740)
+++ tests/testcnx.c	(revision 767)
@@ -1499,5 +1499,5 @@
 		/* fd_cnx_recv_setaltfifo */
 		CHECK( 0, fd_cnx_send(client_side, cer_buf, cer_sz, 0));
-		CHECK( 0, fd_fifo_new(&myfifo) );
+		CHECK( 0, fd_fifo_new(&myfifo, 0) );
 		CHECK( 0, fd_cnx_recv_setaltfifo(server_side, myfifo) );
 		CHECK( 0, clock_gettime(CLOCK_REALTIME, &now) );
@@ -1591,5 +1591,5 @@
 		/* fd_cnx_recv_setaltfifo */
 		CHECK( 0, fd_cnx_send(client_side, cer_buf, cer_sz, 0));
-		CHECK( 0, fd_fifo_new(&myfifo) );
+		CHECK( 0, fd_fifo_new(&myfifo, 0) );
 		CHECK( 0, fd_cnx_recv_setaltfifo(server_side, myfifo) );
 		CHECK( 0, clock_gettime(CLOCK_REALTIME, &now) );
Index: tests/testfifo.c
===================================================================
--- tests/testfifo.c	(revision 751)
+++ tests/testfifo.c	(revision 767)
@@ -111,4 +111,23 @@
 }
 
+/* The test function, to be threaded */
+static int iter  = 0;
+static void * test_fct2(void * data)
+{
+	int i;
+	int * item;
+	struct test_data * td = (struct test_data *) data;
+	
+	for (i=0; i< td->nbr; i++) {
+		item = malloc(sizeof(int));
+		CHECK( 1, item ? 1 : 0 );
+		*item = i;
+		CHECK( 0, fd_fifo_post(td->queue, &item) );
+		iter++;
+	}
+	
+	return NULL;
+}
+
 
 /* Main test routine */
@@ -145,5 +164,5 @@
 		
 		/* Create the queue */
-		CHECK( 0, fd_fifo_new(&queue) );
+		CHECK( 0, fd_fifo_new(&queue, 0) );
 		
 		/* Check the count is 0 */
@@ -233,5 +252,5 @@
 		
 		/* Create the queue */
-		CHECK( 0, fd_fifo_new(&queue) );
+		CHECK( 0, fd_fifo_new(&queue, 0) );
 		
 		/* Create the barrier */
@@ -303,5 +322,5 @@
 		
 		/* Create the queue */
-		CHECK( 0, fd_fifo_new(&queue) );
+		CHECK( 0, fd_fifo_new(&queue, 0) );
 		
 		/* Create the barrier */
@@ -370,5 +389,5 @@
 		
 		/* Create the queue */
-		CHECK( 0, fd_fifo_new(&queue) );
+		CHECK( 0, fd_fifo_new(&queue, 0) );
 		
 		/* Prepare the test data */
@@ -442,4 +461,55 @@
 	}
 	
+	/* Test max queue limit */
+	{
+		struct fifo      	*queue = NULL;
+		struct test_data	 td;
+		pthread_t		 th;
+		int *			item, i;
+		
+		/* Create the queue */
+		CHECK( 0, fd_fifo_new(&queue, 10) );
+		
+		/* Initialize the test data structures */
+		td.queue = queue;
+		td.nbr = 15;
+		
+		CHECK( 0, pthread_create( &th, NULL, test_fct2, &td ) );
+		
+		usleep(1000); /* 1 millisec */
+		
+		CHECK( 10, iter );
+		
+		CHECK( 0, fd_fifo_tryget(queue, &item) );
+		CHECK( 0, *item);
+		free(item);
+		
+		usleep(1000); /* 1 millisec */
+		
+		CHECK( 11, iter );
+		
+		for (i=1; i<4; i++) {
+			CHECK( 0, fd_fifo_get(queue, &item) );
+			CHECK( i, *item);
+			free(item);
+		}
+		
+		usleep(1000); /* 1 millisec */
+		
+		CHECK( 14, iter );
+		
+		/* fd_fifo_dump(0, "test", queue, NULL); */
+		
+		for (; i < td.nbr; i++) {
+			CHECK( 0, fd_fifo_tryget(queue, &item) );
+			CHECK( i, *item);
+			free(item);
+		}
+		
+		CHECK( 0, pthread_join( th, NULL ) );
+		CHECK( 15, iter );
+		
+	}
+	
 	/* Delete the messages */
 	CHECK( 0, fd_msg_free( msg1 ) );
