From patchwork Thu Jun 16 19:55:33 2022 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Anton Khirnov X-Patchwork-Id: 36272 Delivered-To: ffmpegpatchwork2@gmail.com Received: by 2002:a05:6a20:1a22:b0:84:42e0:ad30 with SMTP id cj34csp1128570pzb; Thu, 16 Jun 2022 13:06:12 -0700 (PDT) X-Google-Smtp-Source: AGRyM1vnTNNDyCvPzSuzy0ueGsE1+wv5D9I+BjPEanRTtPdAEoSsf/bNKPkDoDSlrEPGLM9KzDxj X-Received: by 2002:aa7:c7c4:0:b0:431:75d6:6b3 with SMTP id o4-20020aa7c7c4000000b0043175d606b3mr8457413eds.280.1655409972785; Thu, 16 Jun 2022 13:06:12 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1655409972; cv=none; d=google.com; s=arc-20160816; b=xZEU192W0Nb+5fxNlMsN4d2XOsyJa3/VAt9frObe5Ja4N59thiGnrH9IwK4yr0gY/T Y5YkiNap+25R1ZGVw7Vykej/hI3+RQG4RxtVdMaOPxjMT4yip+9ND+/5AjI8wWQr4YI0 hHxQwbF3smjyvJO120Je2MiBeqxIv82Qp/UUSF/TRaL6IQ99vw/7Q4Si2f3Gdbw/5OYU H9S0/YUU5AMgObuyrF0WRqRhaOeb0jMOs1w23e1vGve9eDUWBIHoG2ZR+TOffRDukWIZ 94LNCLZN4m3Y6N1n0VOtkn3RctxjbzeKAa3xJkd7kqXq8sXqSU+6cRw2juZI+LwV3w28 hLvA== ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=google.com; s=arc-20160816; h=sender:errors-to:content-transfer-encoding:reply-to:list-subscribe :list-help:list-post:list-archive:list-unsubscribe:list-id :precedence:subject:mime-version:references:in-reply-to:message-id :date:to:from:delivered-to; bh=MOdu4LTbO9tr+RR5JVL9Yw4ydnk3AElJ8yhsSDMebA8=; b=IxCf1sAuzo2ueGM33gyVrhW/WfQT5kPr5eXxI74T0Rsx1Zx/5bfyKqXxrRN5w2lKLq w92sN88D5m0BsPG/ZBQIwbuXpbdt4Zk6e9L1dqmSuOK3m3I3DhxbGI/8f3OOj3e6vbiI mg9Fd7hKNZHkFxe2h9CWXUDPWp3feJ3Q3c6QLFpEArDh0pLae6JGzbN7CUNyURQPSNwI FlSUm8TXgA3egvAUqYnc3Wp1WZBKV+PHNe1FYHBWIJ4XR4XOi7wvrsVX/T+qXH7Uairu Ox4KacMtrG36q77EWPrNpIedXuDbLm8dyDuvmF34ahO2QaIB+hZ9sUM31+y506GFsa4y fTdA== ARC-Authentication-Results: i=1; mx.google.com; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org Return-Path: Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org. [79.124.17.100]) by mx.google.com with ESMTP id z13-20020a1709063acd00b00707dcb33dc8si725590ejd.402.2022.06.16.13.06.02; Thu, 16 Jun 2022 13:06:12 -0700 (PDT) Received-SPF: pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) client-ip=79.124.17.100; Authentication-Results: mx.google.com; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id D724868B8A0; Thu, 16 Jun 2022 23:03:55 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail0.khirnov.net (red.khirnov.net [176.97.15.12]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id 40EDF68B82F for ; Thu, 16 Jun 2022 23:03:45 +0300 (EEST) Received: from localhost (localhost [IPv6:::1]) by mail0.khirnov.net (Postfix) with ESMTP id E27872404FE for ; Thu, 16 Jun 2022 22:03:44 +0200 (CEST) Received: from mail0.khirnov.net ([IPv6:::1]) by localhost (mail0.khirnov.net [IPv6:::1]) (amavisd-new, port 10024) with ESMTP id HHs-9taLxv89 for ; Thu, 16 Jun 2022 22:03:43 +0200 (CEST) Received: from libav.khirnov.net (libav.khirnov.net [IPv6:2a00:c500:561:201::7]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256 client-signature RSA-PSS (2048 bits) client-digest SHA256) (Client CN "libav.khirnov.net", Issuer "smtp.khirnov.net SMTP CA" (verified OK)) by mail0.khirnov.net (Postfix) with ESMTPS id 784E6240691 for ; Thu, 16 Jun 2022 22:03:34 +0200 (CEST) Received: by libav.khirnov.net (Postfix, from userid 1000) id 1E0133A222E; Thu, 16 Jun 2022 22:03:31 +0200 (CEST) From: Anton Khirnov To: ffmpeg-devel@ffmpeg.org Date: Thu, 16 Jun 2022 21:55:33 +0200 Message-Id: <20220616195534.5278-34-anton@khirnov.net> X-Mailer: git-send-email 2.34.1 In-Reply-To: <20220616195534.5278-1-anton@khirnov.net> References: <20220616195534.5278-1-anton@khirnov.net> MIME-Version: 1.0 Subject: [FFmpeg-devel] [PATCH 34/35] fftools: add a multistream thread-safe queue X-BeenThere: ffmpeg-devel@ffmpeg.org X-Mailman-Version: 2.1.29 Precedence: list List-Id: FFmpeg development discussions and patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: FFmpeg development discussions and patches Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" X-TUID: JrCc6oqppEpZ It is similar to AVThreadMessageQueue, but supports multiple streams, each with its own EOF state. --- fftools/Makefile | 1 + fftools/thread_queue.c | 245 +++++++++++++++++++++++++++++++++++++++++ fftools/thread_queue.h | 81 ++++++++++++++ 3 files changed, 327 insertions(+) create mode 100644 fftools/thread_queue.c create mode 100644 fftools/thread_queue.h diff --git a/fftools/Makefile b/fftools/Makefile index bc57ebe748..6285e6eacb 100644 --- a/fftools/Makefile +++ b/fftools/Makefile @@ -16,6 +16,7 @@ OBJS-ffmpeg += \ fftools/ffmpeg_opt.o \ fftools/objpool.o \ fftools/sync_queue.o \ + fftools/thread_queue.o \ define DOFFTOOL OBJS-$(1) += fftools/cmdutils.o fftools/opt_common.o fftools/$(1).o $(OBJS-$(1)-yes) diff --git a/fftools/thread_queue.c b/fftools/thread_queue.c new file mode 100644 index 0000000000..a1ab4ce92e --- /dev/null +++ b/fftools/thread_queue.c @@ -0,0 +1,245 @@ +/* + * This file is part of FFmpeg. + * + * FFmpeg is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * FFmpeg is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with FFmpeg; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include +#include + +#include "libavutil/avassert.h" +#include "libavutil/error.h" +#include "libavutil/fifo.h" +#include "libavutil/intreadwrite.h" +#include "libavutil/mem.h" +#include "libavutil/thread.h" + +#include "objpool.h" +#include "thread_queue.h" + +enum { + FINISHED_SEND = (1 << 0), + FINISHED_RECV = (1 << 1), +}; + +typedef struct FifoElem { + void *obj; + unsigned int stream_idx; +} FifoElem; + +struct ThreadQueue { + int *finished; + unsigned int nb_streams; + + AVFifo *fifo; + + ObjPool *obj_pool; + void (*obj_move)(void *dst, void *src); + + pthread_mutex_t lock; + pthread_cond_t cond; +}; + +void tq_free(ThreadQueue **ptq) +{ + ThreadQueue *tq = *ptq; + + if (!tq) + return; + + if (tq->fifo) { + FifoElem elem; + while (av_fifo_read(tq->fifo, &elem, 1) >= 0) + objpool_release(tq->obj_pool, &elem.obj); + } + av_fifo_freep2(&tq->fifo); + + objpool_free(&tq->obj_pool); + + av_freep(&tq->finished); + + pthread_cond_destroy(&tq->cond); + pthread_mutex_destroy(&tq->lock); + + av_freep(ptq); +} + +ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, + ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)) +{ + ThreadQueue *tq; + int ret; + + tq = av_mallocz(sizeof(*tq)); + if (!tq) + return NULL; + + ret = pthread_cond_init(&tq->cond, NULL); + if (ret) { + av_freep(&tq); + return NULL; + } + + ret = pthread_mutex_init(&tq->lock, NULL); + if (ret) { + pthread_cond_destroy(&tq->cond); + av_freep(&tq); + return NULL; + } + + tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); + if (!tq->finished) + goto fail; + tq->nb_streams = nb_streams; + + tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0); + if (!tq->fifo) + goto fail; + + tq->obj_pool = obj_pool; + tq->obj_move = obj_move; + + return tq; +fail: + tq_free(&tq); + return NULL; +} + +int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) +{ + int *finished; + int ret; + + av_assert0(stream_idx < tq->nb_streams); + finished = &tq->finished[stream_idx]; + + pthread_mutex_lock(&tq->lock); + + if (*finished & FINISHED_SEND) { + ret = AVERROR(EINVAL); + goto finish; + } + + while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo)) + pthread_cond_wait(&tq->cond, &tq->lock); + + if (*finished & FINISHED_RECV) { + ret = AVERROR_EOF; + *finished |= FINISHED_SEND; + } else { + FifoElem elem = { .stream_idx = stream_idx }; + + ret = objpool_get(tq->obj_pool, &elem.obj); + if (ret < 0) + goto finish; + + tq->obj_move(elem.obj, data); + + ret = av_fifo_write(tq->fifo, &elem, 1); + av_assert0(ret >= 0); + pthread_cond_broadcast(&tq->cond); + } + +finish: + pthread_mutex_unlock(&tq->lock); + + return ret; +} + +static int receive_locked(ThreadQueue *tq, int *stream_idx, + void *data) +{ + FifoElem elem; + unsigned int nb_finished = 0; + + if (av_fifo_read(tq->fifo, &elem, 1) >= 0) { + tq->obj_move(data, elem.obj); + objpool_release(tq->obj_pool, &elem.obj); + *stream_idx = elem.stream_idx; + return 0; + } + + for (unsigned int i = 0; i < tq->nb_streams; i++) { + if (!(tq->finished[i] & FINISHED_SEND)) + continue; + + /* return EOF to the consumer at most once for each stream */ + if (!(tq->finished[i] & FINISHED_RECV)) { + tq->finished[i] |= FINISHED_RECV; + *stream_idx = i; + return AVERROR_EOF; + } + + nb_finished++; + } + + return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN); +} + +int tq_receive(ThreadQueue *tq, int *stream_idx, void *data) +{ + int ret; + + *stream_idx = -1; + + pthread_mutex_lock(&tq->lock); + + while (1) { + ret = receive_locked(tq, stream_idx, data); + if (ret == AVERROR(EAGAIN)) { + pthread_cond_wait(&tq->cond, &tq->lock); + continue; + } + + break; + } + + if (ret == 0) + pthread_cond_broadcast(&tq->cond); + + pthread_mutex_unlock(&tq->lock); + + return ret; +} + +void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) +{ + av_assert0(stream_idx < tq->nb_streams); + + pthread_mutex_lock(&tq->lock); + + /* mark the stream as send-finished; + * next time the consumer thread tries to read this stream it will get + * an EOF and recv-finished flag will be set */ + tq->finished[stream_idx] |= FINISHED_SEND; + pthread_cond_broadcast(&tq->cond); + + pthread_mutex_unlock(&tq->lock); +} + +void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) +{ + av_assert0(stream_idx < tq->nb_streams); + + pthread_mutex_lock(&tq->lock); + + /* mark the stream as recv-finished; + * next time the producer thread tries to send for this stream, it will + * get an EOF and send-finished flag will be set */ + tq->finished[stream_idx] |= FINISHED_RECV; + pthread_cond_broadcast(&tq->cond); + + pthread_mutex_unlock(&tq->lock); +} diff --git a/fftools/thread_queue.h b/fftools/thread_queue.h new file mode 100644 index 0000000000..0cc8c71ebd --- /dev/null +++ b/fftools/thread_queue.h @@ -0,0 +1,81 @@ +/* + * This file is part of FFmpeg. + * + * FFmpeg is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * FFmpeg is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with FFmpeg; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef FFTOOLS_THREAD_QUEUE_H +#define FFTOOLS_THREAD_QUEUE_H + +#include + +#include "objpool.h" + +typedef struct ThreadQueue ThreadQueue; + +/** + * Allocate a queue for sending data between threads. + * + * @param nb_streams number of streams for which a distinct EOF state is + * maintained + * @param queue_size number of items that can be stored in the queue without + * blocking + * @param obj_pool object pool that will be used to allocate items stored in the + * queue; the pool becomes owned by the queue + * @param callback that moves the contents between two data pointers + */ +ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, + ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)); +void tq_free(ThreadQueue **tq); + +/** + * Send an item for the given stream to the queue. + * + * @param data the item to send, its contents will be moved using the callback + * provided to tq_alloc(); on failure the item will be left + * untouched + * @return + * - 0 the item was successfully sent + * - AVERROR(ENOMEM) could not allocate an item for writing to the FIFO + * - AVERROR(EINVAL) the sending side has previously been marked as finished + * - AVERROR_EOF the receiving side has marked the given stream as finished + */ +int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data); +/** + * Mark the given stream finished from the sending side. + */ +void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx); + +/** + * Read the next item from the queue. + * + * @param stream_idx the index of the stream that was processed or -1 will be + * written here + * @param data the data item will be written here on success using the + * callback provided to tq_alloc() + * @return + * - 0 a data item was successfully read; *stream_idx contains a non-negative + * stream index + * - AVERROR_EOF When *stream_idx is non-negative, this signals that the sending + * side has marked the given stream as finished. This will happen at most once + * for each stream. When *stream_idx is -1, all streams are done. + */ +int tq_receive(ThreadQueue *tq, int *stream_idx, void *data); +/** + * Mark the given stream finished from the receiving side. + */ +void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx); + +#endif // FFTOOLS_THREAD_QUEUE_H