[FFmpeg-devel] [PATCH 1/2] avutil/threadmessage: add av_thread_message_flush()

[Clément Bœsch]

From: Clément Bœsch <clement at stupeflix.com>

---
 libavutil/threadmessage.c | 37 ++++++++++++++++++++++++++++++++++---
 libavutil/threadmessage.h | 21 ++++++++++++++++++---
 2 files changed, 52 insertions(+), 6 deletions(-)

diff --git a/libavutil/threadmessage.c b/libavutil/threadmessage.c
index b7fcbe2..87ce8dc 100644
--- a/libavutil/threadmessage.c
+++ b/libavutil/threadmessage.c
@@ -40,14 +40,16 @@ struct AVThreadMessageQueue {
     int err_send;
     int err_recv;
     unsigned elsize;
+    void (*free_func)(void *msg);
 #else
     int dummy;
 #endif
 };
 
-int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
-                                  unsigned nelem,
-                                  unsigned elsize)
+int av_thread_message_queue_alloc2(AVThreadMessageQueue **mq,
+                                   unsigned nelem,
+                                   unsigned elsize,
+                                   void (*free_func)(void *msg))
 {
 #if HAVE_THREADS
     AVThreadMessageQueue *rmq;
@@ -73,6 +75,7 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
         return AVERROR(ret);
     }
     rmq->elsize = elsize;
+    rmq->free_func = free_func;
     *mq = rmq;
     return 0;
 #else
@@ -81,10 +84,18 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
 #endif /* HAVE_THREADS */
 }
 
+int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
+                                  unsigned nelem,
+                                  unsigned elsize)
+{
+    return av_thread_message_queue_alloc2(mq, nelem, elsize, NULL);
+}
+
 void av_thread_message_queue_free(AVThreadMessageQueue **mq)
 {
 #if HAVE_THREADS
     if (*mq) {
+        av_thread_message_flush(*mq);
         av_fifo_freep(&(*mq)->fifo);
         pthread_cond_destroy(&(*mq)->cond);
         pthread_mutex_destroy(&(*mq)->lock);
@@ -182,3 +193,23 @@ void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq,
     pthread_mutex_unlock(&mq->lock);
 #endif /* HAVE_THREADS */
 }
+
+void av_thread_message_flush(AVThreadMessageQueue *mq)
+{
+#if HAVE_THREADS
+    int used, off;
+
+    pthread_mutex_lock(&mq->lock);
+    used = av_fifo_size(mq->fifo);
+    if (mq->free_func) {
+        for (off = 0; off < used; off += mq->elsize) {
+            void *msg;
+            av_fifo_generic_peek_at(mq->fifo, &msg, off, mq->elsize, NULL);
+            mq->free_func(msg);
+        }
+    }
+    av_fifo_drain(mq->fifo, used);
+    pthread_cond_broadcast(&mq->cond);
+    pthread_mutex_unlock(&mq->lock);
+#endif /* HAVE_THREADS */
+}
diff --git a/libavutil/threadmessage.h b/libavutil/threadmessage.h
index a8481d8..f9004a8 100644
--- a/libavutil/threadmessage.h
+++ b/libavutil/threadmessage.h
@@ -33,17 +33,27 @@ typedef enum AVThreadMessageFlags {
 } AVThreadMessageFlags;
 
 /**
+ * @deprecated use av_thread_message_queue_alloc2 instead
+ */
+attribute_deprecated
+int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
+                                  unsigned nelem,
+                                  unsigned elsize);
+
+/**
  * Allocate a new message queue.
  *
  * @param mq      pointer to the message queue
  * @param nelem   maximum number of elements in the queue
  * @param elsize  size of each element in the queue
+ * @param free_func free message callback function, can be NULL
  * @return  >=0 for success; <0 for error, in particular AVERROR(ENOSYS) if
  *          lavu was built without thread support
  */
-int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
-                                  unsigned nelem,
-                                  unsigned elsize);
+int av_thread_message_queue_alloc2(AVThreadMessageQueue **mq,
+                                   unsigned nelem,
+                                   unsigned elsize,
+                                   void (*free_func)(void *msg));
 
 /**
  * Free a message queue.
@@ -88,4 +98,9 @@ void av_thread_message_queue_set_err_send(AVThreadMessageQueue *mq,
 void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq,
                                           int err);
 
+/**
+ * Flush the message queue
+ */
+void av_thread_message_flush(AVThreadMessageQueue *mq);
+
 #endif /* AVUTIL_THREADMESSAGE_H */
-- 
2.6.2