| Message ID | 20220111204610.14262-31-anton@khirnov.net |
|---|---|
| State | New |
| Headers | show |
| Series | [FFmpeg-devel,01/35] lavu/fifo: disallow overly large fifo sizes | expand |
| Context | Check | Description |
|---|---|---|
| andriy/configurex86 | warning | Failed to apply patch |
| andriy/configureppc | warning | Failed to apply patch |
| andriy/configureaarch64_jetson | warning | Failed to apply patch |
Anton Khirnov: > --- > libavutil/threadmessage.c | 34 +++++++++++++++++----------------- > 1 file changed, 17 insertions(+), 17 deletions(-) > > diff --git a/libavutil/threadmessage.c b/libavutil/threadmessage.c > index 764b7fb813..39d3525a78 100644 > --- a/libavutil/threadmessage.c > +++ b/libavutil/threadmessage.c > @@ -64,7 +64,7 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq, > av_free(rmq); > return AVERROR(ret); > } > - if (!(rmq->fifo = av_fifo_alloc(elsize * nelem))) { > + if (!(rmq->fifo = av_fifo_alloc2(nelem, elsize, 0))) { > pthread_cond_destroy(&rmq->cond_send); > pthread_cond_destroy(&rmq->cond_recv); > pthread_mutex_destroy(&rmq->lock); > @@ -107,9 +107,9 @@ int av_thread_message_queue_nb_elems(AVThreadMessageQueue *mq) > #if HAVE_THREADS > int ret; > pthread_mutex_lock(&mq->lock); > - ret = av_fifo_size(mq->fifo); > + ret = av_fifo_can_read(mq->fifo); > pthread_mutex_unlock(&mq->lock); > - return ret / mq->elsize; > + return ret; > #else > return AVERROR(ENOSYS); > #endif > @@ -121,14 +121,14 @@ static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq, > void *msg, > unsigned flags) > { > - while (!mq->err_send && av_fifo_space(mq->fifo) < mq->elsize) { > + while (!mq->err_send && !av_fifo_can_write(mq->fifo)) { > if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) > return AVERROR(EAGAIN); > pthread_cond_wait(&mq->cond_send, &mq->lock); > } > if (mq->err_send) > return mq->err_send; > - av_fifo_generic_write(mq->fifo, msg, mq->elsize, NULL); > + av_fifo_write(mq->fifo, msg, 1); > /* one message is sent, signal one receiver */ > pthread_cond_signal(&mq->cond_recv); > return 0; > @@ -138,14 +138,14 @@ static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq, > void *msg, > unsigned flags) > { > - while (!mq->err_recv && av_fifo_size(mq->fifo) < mq->elsize) { > + while (!mq->err_recv && !av_fifo_can_read(mq->fifo)) { > if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) > return AVERROR(EAGAIN); > pthread_cond_wait(&mq->cond_recv, &mq->lock); > } > - if (av_fifo_size(mq->fifo) < mq->elsize) > + if (!av_fifo_can_read(mq->fifo)) > return mq->err_recv; > - av_fifo_generic_read(mq->fifo, msg, mq->elsize, NULL); > + av_fifo_read(mq->fifo, msg, 1); > /* one message space appeared, signal one sender */ > pthread_cond_signal(&mq->cond_send); > return 0; > @@ -208,25 +208,25 @@ void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq, > } > > #if HAVE_THREADS > -static void free_func_wrap(void *arg, void *msg, int size) > +static int free_func_wrap(void *arg, void *buf, size_t *nb_elems) > { > AVThreadMessageQueue *mq = arg; > - mq->free_func(msg); > + uint8_t *msg = buf; > + for (size_t i = 0; i < *nb_elems; i++) > + mq->free_func(msg + i * mq->elsize); > + return 0; This function (like most nontrivial callbacks) relies on buf being suitably aligned for whatever type msg is. Therefore one should document that the src/dst in the callbacks is always suitable aligned for all non-over-aligned types. > } > #endif > > void av_thread_message_flush(AVThreadMessageQueue *mq) > { > #if HAVE_THREADS > - int used, off; > - void *free_func = mq->free_func; > + size_t used; > > 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); > + used = av_fifo_can_read(mq->fifo); > + if (mq->free_func) > + av_fifo_read_to_cb(mq->fifo, free_func_wrap, mq, &used); > /* only the senders need to be notified since the queue is empty and there > * is nothing to read */ > pthread_cond_broadcast(&mq->cond_send); >
diff --git a/libavutil/threadmessage.c b/libavutil/threadmessage.c index 764b7fb813..39d3525a78 100644 --- a/libavutil/threadmessage.c +++ b/libavutil/threadmessage.c @@ -64,7 +64,7 @@ int av_thread_message_queue_alloc(AVThreadMessageQueue **mq, av_free(rmq); return AVERROR(ret); } - if (!(rmq->fifo = av_fifo_alloc(elsize * nelem))) { + if (!(rmq->fifo = av_fifo_alloc2(nelem, elsize, 0))) { pthread_cond_destroy(&rmq->cond_send); pthread_cond_destroy(&rmq->cond_recv); pthread_mutex_destroy(&rmq->lock); @@ -107,9 +107,9 @@ int av_thread_message_queue_nb_elems(AVThreadMessageQueue *mq) #if HAVE_THREADS int ret; pthread_mutex_lock(&mq->lock); - ret = av_fifo_size(mq->fifo); + ret = av_fifo_can_read(mq->fifo); pthread_mutex_unlock(&mq->lock); - return ret / mq->elsize; + return ret; #else return AVERROR(ENOSYS); #endif @@ -121,14 +121,14 @@ static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq, void *msg, unsigned flags) { - while (!mq->err_send && av_fifo_space(mq->fifo) < mq->elsize) { + while (!mq->err_send && !av_fifo_can_write(mq->fifo)) { if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) return AVERROR(EAGAIN); pthread_cond_wait(&mq->cond_send, &mq->lock); } if (mq->err_send) return mq->err_send; - av_fifo_generic_write(mq->fifo, msg, mq->elsize, NULL); + av_fifo_write(mq->fifo, msg, 1); /* one message is sent, signal one receiver */ pthread_cond_signal(&mq->cond_recv); return 0; @@ -138,14 +138,14 @@ static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq, void *msg, unsigned flags) { - while (!mq->err_recv && av_fifo_size(mq->fifo) < mq->elsize) { + while (!mq->err_recv && !av_fifo_can_read(mq->fifo)) { if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) return AVERROR(EAGAIN); pthread_cond_wait(&mq->cond_recv, &mq->lock); } - if (av_fifo_size(mq->fifo) < mq->elsize) + if (!av_fifo_can_read(mq->fifo)) return mq->err_recv; - av_fifo_generic_read(mq->fifo, msg, mq->elsize, NULL); + av_fifo_read(mq->fifo, msg, 1); /* one message space appeared, signal one sender */ pthread_cond_signal(&mq->cond_send); return 0; @@ -208,25 +208,25 @@ void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq, } #if HAVE_THREADS -static void free_func_wrap(void *arg, void *msg, int size) +static int free_func_wrap(void *arg, void *buf, size_t *nb_elems) { AVThreadMessageQueue *mq = arg; - mq->free_func(msg); + uint8_t *msg = buf; + for (size_t i = 0; i < *nb_elems; i++) + mq->free_func(msg + i * mq->elsize); + return 0; } #endif void av_thread_message_flush(AVThreadMessageQueue *mq) { #if HAVE_THREADS - int used, off; - void *free_func = mq->free_func; + size_t used; 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); + used = av_fifo_can_read(mq->fifo); + if (mq->free_func) + av_fifo_read_to_cb(mq->fifo, free_func_wrap, mq, &used); /* only the senders need to be notified since the queue is empty and there * is nothing to read */ pthread_cond_broadcast(&mq->cond_send);