avutil/threadmessage: add av_thread_message_flush()

2 files changed, 47 insertions, 0 deletions

diff --git a/libavutil/threadmessage.c b/libavutil/threadmessage.c
index b7fcbe28c0..b7d7dadb5a 100644
--- a/libavutil/threadmessage.c
+++ b/libavutil/threadmessage.c
@@ -40,6 +40,7 @@ struct AVThreadMessageQueue {
     int err_send;
     int err_recv;
     unsigned elsize;
+    void (*free_func)(void *msg);
 #else
     int dummy;
 #endif
@@ -81,10 +82,17 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
 #endif /* HAVE_THREADS */
 }
 
+void av_thread_message_queue_set_free_func(AVThreadMessageQueue *mq,
+                                           void (*free_func)(void *msg))
+{
+    mq->free_func = free_func;
+}
+
 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 +190,26 @@ void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq,
     pthread_mutex_unlock(&mq->lock);
 #endif /* HAVE_THREADS */
 }
+
+static void free_func_wrap(void *arg, void *msg, int size)
+{
+    AVThreadMessageQueue *mq = arg;
+    mq->free_func(msg);
+}
+
+void av_thread_message_flush(AVThreadMessageQueue *mq)
+{
+#if HAVE_THREADS
+    int used, off;
+    void *free_func = mq->free_func;
+
+    pthread_mutex_lock(&mq->lock);
+    used = av_fifo_size(mq->fifo);
+    if (free_func)
+        for (off = 0; off < used; off += mq->elsize)
+            av_fifo_generic_peek_at(mq->fifo, mq, off, mq->elsize, free_func_wrap);
+    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 a8481d8ec3..e256cae9db 100644
--- a/libavutil/threadmessage.h
+++ b/libavutil/threadmessage.h
@@ -88,4 +88,20 @@ void av_thread_message_queue_set_err_send(AVThreadMessageQueue *mq,
 void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq,
                                           int err);
 
+/**
+ * Set the optional free message callback function which will be called if an
+ * operation is removing messages from the queue.
+ */
+void av_thread_message_queue_set_free_func(AVThreadMessageQueue *mq,
+                                           void (*free_func)(void *msg));
+
+/**
+ * Flush the message queue
+ *
+ * This function is mostly equivalent to reading and free-ing every message
+ * except that it will be done in a single operation (no lock/unlock between
+ * reads).
+ */
+void av_thread_message_flush(AVThreadMessageQueue *mq);
+
 #endif /* AVUTIL_THREADMESSAGE_H */