block_FifoPace: proper waiting for overflowing queue

include/vlc_block.h

@@ -300,6 +300,8 @@ static inline block_t *block_ChainGather( block_t *p_list )
 
 VLC_EXPORT( block_fifo_t *, block_FifoNew,      ( void ) LIBVLC_USED );
 VLC_EXPORT( void,           block_FifoRelease,  ( block_fifo_t * ) );
+/* TODO: do we need to export this? */
+void block_FifoPace (block_fifo_t *fifo, size_t max_depth, size_t max_size);
 VLC_EXPORT( void,           block_FifoEmpty,    ( block_fifo_t * ) );
 VLC_EXPORT( size_t,         block_FifoPut,      ( block_fifo_t *, block_t * ) );
 VLC_EXPORT( void,           block_FifoWake,     ( block_fifo_t * ) );

src/misc/block.c

@@ -32,10 +32,13 @@
 #include <sys/stat.h>
 #include "vlc_block.h"
 
-/*****************************************************************************
- * Block functions.
- *****************************************************************************/
-/* private */
+/**
+ * @section Block handling functions.
+ */
+
+/**
+ * Internal state for heap block.
+  */
 struct block_sys_t
 {
     block_t     self;
@@ -342,13 +345,18 @@ block_t *block_File (int fd)
     return block;
 }
 
-/*****************************************************************************
- * block_fifo_t management
- *****************************************************************************/
+/**
+ * @section Thread-safe block queue functions
+ */
+
+/**
+ * Internal state for block queues
+ */
 struct block_fifo_t
 {
     vlc_mutex_t         lock;                         /* fifo data lock */
-    vlc_cond_t          wait;         /* fifo data conditional variable */
+    vlc_cond_t          wait;      /**< Wait for data */
+    vlc_cond_t          wait_room; /**< Wait for queue depth to shrink */
 
     block_t             *p_first;
     block_t             **pp_last;
@@ -365,6 +373,7 @@ block_fifo_t *block_FifoNew( void )
 
     vlc_mutex_init( &p_fifo->lock );
     vlc_cond_init( &p_fifo->wait );
+    vlc_cond_init( &p_fifo->wait_room );
     p_fifo->p_first = NULL;
     p_fifo->pp_last = &p_fifo->p_first;
     p_fifo->i_depth = p_fifo->i_size = 0;
@@ -376,6 +385,7 @@ block_fifo_t *block_FifoNew( void )
 void block_FifoRelease( block_fifo_t *p_fifo )
 {
     block_FifoEmpty( p_fifo );
+    vlc_cond_destroy( &p_fifo->wait_room );
     vlc_cond_destroy( &p_fifo->wait );
     vlc_mutex_destroy( &p_fifo->lock );
     free( p_fifo );
@@ -398,9 +408,39 @@ void block_FifoEmpty( block_fifo_t *p_fifo )
     p_fifo->i_depth = p_fifo->i_size = 0;
     p_fifo->p_first = NULL;
     p_fifo->pp_last = &p_fifo->p_first;
+    vlc_cond_broadcast( &p_fifo->wait_room );
     vlc_mutex_unlock( &p_fifo->lock );
 }
 
+/**
+ * Wait until the FIFO gets below a certain size (if needed).
+ *
+ * Note that if more than one thread writes to the FIFO, you cannot assume that
+ * the FIFO is actually below the requested size upon return (since another
+ * thread could have refilled it already). This is typically not an issue, as
+ * this function is meant for (relaxed) congestion control.
+ *
+ * This function may be a cancellation point and it is cancel-safe.
+ *
+ * @param fifo queue to wait on
+ * @param max_depth wait until the queue has no more than this many blocks
+ *                  (use SIZE_MAX to ignore this constraint)
+ * @param max_size wait until the queue has no more than this many bytes
+ *                  (use SIZE_MAX to ignore this constraint)
+ * @return nothing.
+ */
+void block_FifoPace (block_fifo_t *fifo, size_t max_depth, size_t max_size)
+{
+    vlc_mutex_lock (&fifo->lock);
+    while ((fifo->i_depth > max_depth) || (fifo->i_size > max_size))
+    {
+         mutex_cleanup_push (&fifo->lock);
+         vlc_cond_wait (&fifo->wait_room, &fifo->lock);
+         vlc_cleanup_pop ();
+    }
+    vlc_mutex_unlock (&fifo->lock);
+}
+
 /**
  * Immediately queue one block at the end of a FIFO.
  * @param fifo queue
@@ -471,6 +511,8 @@ block_t *block_FifoGet( block_fifo_t *p_fifo )
         p_fifo->pp_last = &p_fifo->p_first;
     }
 
+    /* We don't know how many threads can queue new packets now. */
+    vlc_cond_broadcast( &p_fifo->wait_room );
     vlc_mutex_unlock( &p_fifo->lock );
 
     b->p_next = NULL;