History

lavu/threadmessage: switch to new FIFO API

Anton Khirnov committed 4 years ago

Commit 90eef1c1 parent d1bd1891 file · +19 −19Browse files

Unified · Split

libavutil/threadmessage.c+19−19
@@ -24,7 +24,7 @@ · 23 unmodified lines
24 24
25struct AVThreadMessageQueue {25struct AVThreadMessageQueue {
26#if HAVE_THREADS26#if HAVE_THREADS
27 AVFifoBuffer *fifo;27 AVFifo *fifo;
28 pthread_mutex_t lock;28 pthread_mutex_t lock;
29 pthread_cond_t cond_recv;29 pthread_cond_t cond_recv;
30 pthread_cond_t cond_send;30 pthread_cond_t cond_send;
@@ -64,7 +64,7 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq, · 33 unmodified lines
64 av_free(rmq);64 av_free(rmq);
65 return AVERROR(ret);65 return AVERROR(ret);
66 }66 }
67 if (!(rmq->fifo = av_fifo_alloc(elsize * nelem))) {67 if (!(rmq->fifo = av_fifo_alloc2(nelem, elsize, 0))) {
68 pthread_cond_destroy(&rmq->cond_send);68 pthread_cond_destroy(&rmq->cond_send);
69 pthread_cond_destroy(&rmq->cond_recv);69 pthread_cond_destroy(&rmq->cond_recv);
70 pthread_mutex_destroy(&rmq->lock);70 pthread_mutex_destroy(&rmq->lock);
@@ -93,7 +93,7 @@ void av_thread_message_queue_free(AVThreadMessageQueue **mq) · 22 unmodified lines
93#if HAVE_THREADS93#if HAVE_THREADS
94 if (*mq) {94 if (*mq) {
95 av_thread_message_flush(*mq);95 av_thread_message_flush(*mq);
96 av_fifo_freep(&(*mq)->fifo);96 av_fifo_freep2(&(*mq)->fifo);
97 pthread_cond_destroy(&(*mq)->cond_send);97 pthread_cond_destroy(&(*mq)->cond_send);
98 pthread_cond_destroy(&(*mq)->cond_recv);98 pthread_cond_destroy(&(*mq)->cond_recv);
99 pthread_mutex_destroy(&(*mq)->lock);99 pthread_mutex_destroy(&(*mq)->lock);
@@ -107,9 +107,9 @@ int av_thread_message_queue_nb_elems(AVThreadMessageQueue *mq) · 7 unmodified lines
107#if HAVE_THREADS107#if HAVE_THREADS
108 int ret;108 int ret;
109 pthread_mutex_lock(&mq->lock);109 pthread_mutex_lock(&mq->lock);
110 ret = av_fifo_size(mq->fifo);110 ret = av_fifo_can_read(mq->fifo);
111 pthread_mutex_unlock(&mq->lock);111 pthread_mutex_unlock(&mq->lock);
112 return ret / mq->elsize;112 return ret;
113#else113#else
114 return AVERROR(ENOSYS);114 return AVERROR(ENOSYS);
115#endif115#endif
@@ -121,14 +121,14 @@ static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq, · 5 unmodified lines
121 void *msg,121 void *msg,
122 unsigned flags)122 unsigned flags)
123{123{
124 while (!mq->err_send && av_fifo_space(mq->fifo) < mq->elsize) {124 while (!mq->err_send && !av_fifo_can_write(mq->fifo)) {
125 if ((flags & AV_THREAD_MESSAGE_NONBLOCK))125 if ((flags & AV_THREAD_MESSAGE_NONBLOCK))
126 return AVERROR(EAGAIN);126 return AVERROR(EAGAIN);
127 pthread_cond_wait(&mq->cond_send, &mq->lock);127 pthread_cond_wait(&mq->cond_send, &mq->lock);
128 }128 }
129 if (mq->err_send)129 if (mq->err_send)
130 return mq->err_send;130 return mq->err_send;
131 av_fifo_generic_write(mq->fifo, msg, mq->elsize, NULL);131 av_fifo_write(mq->fifo, msg, 1);
132 /* one message is sent, signal one receiver */132 /* one message is sent, signal one receiver */
133 pthread_cond_signal(&mq->cond_recv);133 pthread_cond_signal(&mq->cond_recv);
134 return 0;134 return 0;
@@ -138,14 +138,14 @@ static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq, · 3 unmodified lines
138 void *msg,138 void *msg,
139 unsigned flags)139 unsigned flags)
140{140{
141 while (!mq->err_recv && av_fifo_size(mq->fifo) < mq->elsize) {141 while (!mq->err_recv && !av_fifo_can_read(mq->fifo)) {
142 if ((flags & AV_THREAD_MESSAGE_NONBLOCK))142 if ((flags & AV_THREAD_MESSAGE_NONBLOCK))
143 return AVERROR(EAGAIN);143 return AVERROR(EAGAIN);
144 pthread_cond_wait(&mq->cond_recv, &mq->lock);144 pthread_cond_wait(&mq->cond_recv, &mq->lock);
145 }145 }
146 if (av_fifo_size(mq->fifo) < mq->elsize)146 if (!av_fifo_can_read(mq->fifo))
147 return mq->err_recv;147 return mq->err_recv;
148 av_fifo_generic_read(mq->fifo, msg, mq->elsize, NULL);148 av_fifo_read(mq->fifo, msg, 1);
149 /* one message space appeared, signal one sender */149 /* one message space appeared, signal one sender */
150 pthread_cond_signal(&mq->cond_send);150 pthread_cond_signal(&mq->cond_send);
151 return 0;151 return 0;
@@ -208,25 +208,25 @@ void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq, · 56 unmodified lines
208}208}
209 209
210#if HAVE_THREADS210#if HAVE_THREADS
211static void free_func_wrap(void *arg, void *msg, int size)211static int free_func_wrap(void *arg, void *buf, size_t *nb_elems)
212{212{
213 AVThreadMessageQueue *mq = arg;213 AVThreadMessageQueue *mq = arg;
214 mq->free_func(msg);214 uint8_t *msg = buf;
215 for (size_t i = 0; i < *nb_elems; i++)
216 mq->free_func(msg + i * mq->elsize);
217 return 0;
215}218}
216#endif219#endif
217 220
218void av_thread_message_flush(AVThreadMessageQueue *mq)221void av_thread_message_flush(AVThreadMessageQueue *mq)
219{222{
220#if HAVE_THREADS223#if HAVE_THREADS
221 int used, off;224 size_t used;
222 void *free_func = mq->free_func;
223 225
224 pthread_mutex_lock(&mq->lock);226 pthread_mutex_lock(&mq->lock);
225 used = av_fifo_size(mq->fifo);227 used = av_fifo_can_read(mq->fifo);
226 if (free_func)228 if (mq->free_func)
227 for (off = 0; off < used; off += mq->elsize)229 av_fifo_read_to_cb(mq->fifo, free_func_wrap, mq, &used);
228 av_fifo_generic_peek_at(mq->fifo, mq, off, mq->elsize, free_func_wrap);
229 av_fifo_drain(mq->fifo, used);
230 /* only the senders need to be notified since the queue is empty and there230 /* only the senders need to be notified since the queue is empty and there
231 * is nothing to read */231 * is nothing to read */
232 pthread_cond_broadcast(&mq->cond_send);232 pthread_cond_broadcast(&mq->cond_send);

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t