| Message ID | TYSPR06MB64330156947AEE87488F6682AA14A@TYSPR06MB6433.apcprd06.prod.outlook.com |
|---|---|
| State | New |
| Headers | show |
| Series | [FFmpeg-devel,v4] vvcdec: add thread executor | expand |
| Context | Check | Description |
|---|---|---|
| yinshiyou/make_loongarch64 | success | Make finished |
| yinshiyou/make_fate_loongarch64 | success | Make fate finished |
| andriy/make_x86 | success | Make finished |
| andriy/make_fate_x86 | success | Make fate finished |
On Tue, Aug 15, 2023 at 10:50:13PM +0800, Nuo Mi wrote: > The executor design pattern was inroduced by java > <https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/concurrent/Executor.html> > it also adapted by python > <https://docs.python.org/3/library/concurrent.futures.html> > Compared to handcrafted thread pool management, it greatly simplifies the thread code. > --- > libavutil/Makefile | 2 + > libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ > libavutil/executor.h | 67 +++++++++++++++ > 3 files changed, 270 insertions(+) > create mode 100644 libavutil/executor.c > create mode 100644 libavutil/executor.h needs doc/APIchanges addition should be ok otherwise thx
Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : > The executor design pattern was inroduced by java > <https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc > urrent/Executor.html> it also adapted by python > <https://docs.python.org/3/library/concurrent.futures.html> > Compared to handcrafted thread pool management, it greatly simplifies the > thread code. --- > libavutil/Makefile | 2 + > libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ > libavutil/executor.h | 67 +++++++++++++++ > 3 files changed, 270 insertions(+) > create mode 100644 libavutil/executor.c > create mode 100644 libavutil/executor.h > > diff --git a/libavutil/Makefile b/libavutil/Makefile > index 7828c94dc5..4711f8cde8 100644 > --- a/libavutil/Makefile > +++ b/libavutil/Makefile > @@ -31,6 +31,7 @@ HEADERS = adler32.h > \ encryption_info.h > \ error.h \ eval.h > \ + > executor.h \ fifo.h > \ file.h > \ frame.h > \ @@ -127,6 +128,7 @@ OBJS = adler32.o > \ encryption_info.o > \ error.o > \ eval.o > \ + executor.o > \ fifo.o > \ file.o \ > file_open.o \ diff > --git a/libavutil/executor.c b/libavutil/executor.c > new file mode 100644 > index 0000000000..6b47ba0bf2 > --- /dev/null > +++ b/libavutil/executor.c > @@ -0,0 +1,201 @@ > +/* > + * Copyright (C) 2023 Nuo Mi > + * > + * 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 "internal.h" > +#include "mem.h" > +#include "thread.h" > + > +#include "executor.h" > + > +#if !HAVE_THREADS > +#define pthread_create(t, a, s, ar) 0 Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. > +#define pthread_join(t, r) do {} while(0) ...and this should be an unreachable(), assert(0), abort() or something along those lines. > + > +#define pthread_cond_init(c, a) 0 > +#define pthread_cond_broadcast(c) do {} while(0) > +#define pthread_cond_signal(c) do {} while(0) > +#define pthread_cond_wait(c, m) do {} while(0) > +#define pthread_cond_destroy(c) do {} while(0) > + > +#define pthread_mutex_init(m, a) 0 > +#define pthread_mutex_lock(l) do {} while(0) > +#define pthread_mutex_unlock(l) do {} while(0) > +#define pthread_mutex_destroy(l) do {} while(0) > +#endif > + > +typedef struct ThreadInfo { > + AVExecutor *e; > + pthread_t thread; Does this even work if !HAVE_THREAD ? > +} ThreadInfo; > + > +struct AVExecutor { > + AVTaskCallbacks cb; > + int thread_count; > + > + ThreadInfo *threads; > + uint8_t *local_contexts; > + > + pthread_mutex_t lock; > + pthread_cond_t cond; > + int die; > + > + AVTask *tasks; > +}; > + > +static AVTask* remove_task(AVTask **prev, AVTask *t) > +{ > + *prev = t->next; > + t->next = NULL; > + return t; > +} > + > +static void add_task(AVTask **prev, AVTask *t) > +{ > + t->next = *prev; > + *prev = t; > +} > + > +static int run_one_task(AVExecutor *e, void *lc) > +{ > + AVTaskCallbacks *cb = &e->cb; > + AVTask **prev; > + > + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); prev > = &(*prev)->next) > + /* nothing */; > + if (*prev) { > + AVTask *t = remove_task(prev, *prev); > + pthread_mutex_unlock(&e->lock); > + cb->run(t, lc, cb->user_data); > + pthread_mutex_lock(&e->lock); > + return 1; > + } > + return 0; > +} > + > +#if HAVE_THREADS > +static void *executor_worker_task(void *data) > +{ > + ThreadInfo *ti = (ThreadInfo*)data; > + AVExecutor *e = ti->e; > + void *lc = e->local_contexts > + (ti - e->threads) * > e->cb.local_context_size; + > + pthread_mutex_lock(&e->lock); > + while (1) { > + if (e->die) break; > + > + if (!run_one_task(e, lc)) { > + //no task in one loop > + pthread_cond_wait(&e->cond, &e->lock); > + } > + } > + pthread_mutex_unlock(&e->lock); > + return NULL; > +} > +#endif > + > +static void executor_free(AVExecutor *e, const int has_lock, const int > has_cond) +{ > + if (e->thread_count) { > + //signal die > + pthread_mutex_lock(&e->lock); > + e->die = 1; > + pthread_cond_broadcast(&e->cond); > + pthread_mutex_unlock(&e->lock); > + > + for (int i = 0; i < e->thread_count; i++) > + pthread_join(e->threads[i].thread, NULL); > + } > + if (has_cond) > + pthread_cond_destroy(&e->cond); > + if (has_lock) > + pthread_mutex_destroy(&e->lock); > + > + av_free(e->threads); > + av_free(e->local_contexts); > + > + av_free(e); > +} > + > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int thread_count) > +{ > + AVExecutor *e; > + int has_lock = 0, has_cond = 0; > + if (!cb || !cb->user_data || !cb->ready || !cb->run || > !cb->priority_higher) + return NULL; > + > + e = av_calloc(1, sizeof(*e)); > + if (!e) > + return NULL; > + e->cb = *cb; > + > + e->local_contexts = av_calloc(thread_count, e->cb.local_context_size); > + if (!e->local_contexts) > + goto free_executor; > + > + e->threads = av_calloc(thread_count, sizeof(*e->threads)); > + if (!e->threads) > + goto free_executor; > + > + has_lock = !pthread_mutex_init(&e->lock, NULL); > + has_cond = !pthread_cond_init(&e->cond, NULL); > + > + if (!has_lock || !has_cond) > + goto free_executor; > + > + for (/* nothing */; e->thread_count < thread_count; e->thread_count++) > { + ThreadInfo *ti = e->threads + e->thread_count; > + ti->e = e; > + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) > + goto free_executor; > + } > + return e; > + > +free_executor: > + executor_free(e, has_lock, has_cond); > + return NULL; > +} > + > +void av_executor_free(AVExecutor **executor) > +{ > + if (!executor || !*executor) > + return; > + executor_free(*executor, 1, 1); > + *executor = NULL; > +} > + > +void av_executor_execute(AVExecutor *e, AVTask *t) > +{ > + AVTaskCallbacks *cb = &e->cb; > + AVTask **prev; > + > + pthread_mutex_lock(&e->lock); > + if (t) { > + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); prev > = &(*prev)->next) + /* nothing */; > + add_task(prev, t); > + } > + pthread_cond_signal(&e->cond); > + pthread_mutex_unlock(&e->lock); > + > +#if !HAVE_THREADS > + // We are running in a single-threaded environment, so we must handle > all tasks ourselves > + while (run_one_task(e, e->local_contexts)) > + /* nothing */; > +#endif > +} > diff --git a/libavutil/executor.h b/libavutil/executor.h > new file mode 100644 > index 0000000000..c3fe44bd0e > --- /dev/null > +++ b/libavutil/executor.h > @@ -0,0 +1,67 @@ > +/* > + * Copyright (C) 2023 Nuo Mi > + * > + * 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 AVUTIL_EXECUTOR_H > +#define AVUTIL_EXECUTOR_H > + > +typedef struct AVExecutor AVExecutor; > +typedef struct AVTask AVTask; > + > +struct AVTask { > + AVTask *next; > +}; > + > +typedef struct AVTaskCallbacks { > + void *user_data; > + > + int local_context_size; > + > + // return 1 if a's priority > b's priority > + int (*priority_higher)(const AVTask *a, const AVTask *b); > + > + // task is ready for run > + int (*ready)(const AVTask *t, void *user_data); > + > + // run the task > + int (*run)(AVTask *t, void *local_context, void *user_data); > +} AVTaskCallbacks; > + > +/** > + * Alloc executor > + * @param callbacks callback strucutre for executor > + * @param thread_count worker thread number > + * @return return the executor > + */ > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int > thread_count); + > +/** > + * Free executor > + * @param e pointer to executor > + */ > +void av_executor_free(AVExecutor **e); > + > +/** > + * Add task to executor > + * @param e pointer to executor > + * @param t pointer to task. If NULL, it will wakeup one work thread > + */ > +void av_executor_execute(AVExecutor *e, AVTask *t); > + > +#endif //AVUTIL_EXECUTOR_H
On 8/16/2023 1:42 PM, Rémi Denis-Courmont wrote: > Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : >> The executor design pattern was inroduced by java >> <https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc >> urrent/Executor.html> it also adapted by python >> <https://docs.python.org/3/library/concurrent.futures.html> >> Compared to handcrafted thread pool management, it greatly simplifies the >> thread code. --- >> libavutil/Makefile | 2 + >> libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ >> libavutil/executor.h | 67 +++++++++++++++ >> 3 files changed, 270 insertions(+) >> create mode 100644 libavutil/executor.c >> create mode 100644 libavutil/executor.h >> >> diff --git a/libavutil/Makefile b/libavutil/Makefile >> index 7828c94dc5..4711f8cde8 100644 >> --- a/libavutil/Makefile >> +++ b/libavutil/Makefile >> @@ -31,6 +31,7 @@ HEADERS = adler32.h >> \ encryption_info.h >> \ error.h \ eval.h >> \ + >> executor.h \ fifo.h >> \ file.h >> \ frame.h >> \ @@ -127,6 +128,7 @@ OBJS = adler32.o >> \ encryption_info.o >> \ error.o >> \ eval.o >> \ + executor.o >> \ fifo.o >> \ file.o \ >> file_open.o \ diff >> --git a/libavutil/executor.c b/libavutil/executor.c >> new file mode 100644 >> index 0000000000..6b47ba0bf2 >> --- /dev/null >> +++ b/libavutil/executor.c >> @@ -0,0 +1,201 @@ >> +/* >> + * Copyright (C) 2023 Nuo Mi >> + * >> + * 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 "internal.h" >> +#include "mem.h" >> +#include "thread.h" >> + >> +#include "executor.h" >> + >> +#if !HAVE_THREADS >> +#define pthread_create(t, a, s, ar) 0 > > Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. pthread_create() is used below even if !HAVE_THREADS, so it should not fail. > >> +#define pthread_join(t, r) do {} while(0) > > ...and this should be an unreachable(), assert(0), abort() or something along > those lines. This is also used below. It being a no-op should be ok. > >> + >> +#define pthread_cond_init(c, a) 0 >> +#define pthread_cond_broadcast(c) do {} while(0) >> +#define pthread_cond_signal(c) do {} while(0) >> +#define pthread_cond_wait(c, m) do {} while(0) >> +#define pthread_cond_destroy(c) do {} while(0) >> + >> +#define pthread_mutex_init(m, a) 0 >> +#define pthread_mutex_lock(l) do {} while(0) >> +#define pthread_mutex_unlock(l) do {} while(0) >> +#define pthread_mutex_destroy(l) do {} while(0) >> +#endif >> + >> +typedef struct ThreadInfo { >> + AVExecutor *e; >> + pthread_t thread; > > Does this even work if !HAVE_THREAD ? > >> +} ThreadInfo; >> + >> +struct AVExecutor { >> + AVTaskCallbacks cb; >> + int thread_count; >> + >> + ThreadInfo *threads; >> + uint8_t *local_contexts; >> + >> + pthread_mutex_t lock; >> + pthread_cond_t cond; >> + int die; >> + >> + AVTask *tasks; >> +}; >> + >> +static AVTask* remove_task(AVTask **prev, AVTask *t) >> +{ >> + *prev = t->next; >> + t->next = NULL; >> + return t; >> +} >> + >> +static void add_task(AVTask **prev, AVTask *t) >> +{ >> + t->next = *prev; >> + *prev = t; >> +} >> + >> +static int run_one_task(AVExecutor *e, void *lc) >> +{ >> + AVTaskCallbacks *cb = &e->cb; >> + AVTask **prev; >> + >> + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); prev >> = &(*prev)->next) >> + /* nothing */; >> + if (*prev) { >> + AVTask *t = remove_task(prev, *prev); >> + pthread_mutex_unlock(&e->lock); >> + cb->run(t, lc, cb->user_data); >> + pthread_mutex_lock(&e->lock); >> + return 1; >> + } >> + return 0; >> +} >> + >> +#if HAVE_THREADS >> +static void *executor_worker_task(void *data) >> +{ >> + ThreadInfo *ti = (ThreadInfo*)data; >> + AVExecutor *e = ti->e; >> + void *lc = e->local_contexts >> + (ti - e->threads) * >> e->cb.local_context_size; + >> + pthread_mutex_lock(&e->lock); >> + while (1) { >> + if (e->die) break; >> + >> + if (!run_one_task(e, lc)) { >> + //no task in one loop >> + pthread_cond_wait(&e->cond, &e->lock); >> + } >> + } >> + pthread_mutex_unlock(&e->lock); >> + return NULL; >> +} >> +#endif >> + >> +static void executor_free(AVExecutor *e, const int has_lock, const int >> has_cond) +{ >> + if (e->thread_count) { >> + //signal die >> + pthread_mutex_lock(&e->lock); >> + e->die = 1; >> + pthread_cond_broadcast(&e->cond); >> + pthread_mutex_unlock(&e->lock); >> + >> + for (int i = 0; i < e->thread_count; i++) >> + pthread_join(e->threads[i].thread, NULL); >> + } >> + if (has_cond) >> + pthread_cond_destroy(&e->cond); >> + if (has_lock) >> + pthread_mutex_destroy(&e->lock); >> + >> + av_free(e->threads); >> + av_free(e->local_contexts); >> + >> + av_free(e); >> +} >> + >> +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int thread_count) >> +{ >> + AVExecutor *e; >> + int has_lock = 0, has_cond = 0; >> + if (!cb || !cb->user_data || !cb->ready || !cb->run || >> !cb->priority_higher) + return NULL; >> + >> + e = av_calloc(1, sizeof(*e)); >> + if (!e) >> + return NULL; >> + e->cb = *cb; >> + >> + e->local_contexts = av_calloc(thread_count, e->cb.local_context_size); >> + if (!e->local_contexts) >> + goto free_executor; >> + >> + e->threads = av_calloc(thread_count, sizeof(*e->threads)); >> + if (!e->threads) >> + goto free_executor; >> + >> + has_lock = !pthread_mutex_init(&e->lock, NULL); >> + has_cond = !pthread_cond_init(&e->cond, NULL); >> + >> + if (!has_lock || !has_cond) >> + goto free_executor; >> + >> + for (/* nothing */; e->thread_count < thread_count; e->thread_count++) >> { + ThreadInfo *ti = e->threads + e->thread_count; >> + ti->e = e; >> + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) >> + goto free_executor; >> + } >> + return e; >> + >> +free_executor: >> + executor_free(e, has_lock, has_cond); >> + return NULL; >> +} >> + >> +void av_executor_free(AVExecutor **executor) >> +{ >> + if (!executor || !*executor) >> + return; >> + executor_free(*executor, 1, 1); >> + *executor = NULL; >> +} >> + >> +void av_executor_execute(AVExecutor *e, AVTask *t) >> +{ >> + AVTaskCallbacks *cb = &e->cb; >> + AVTask **prev; >> + >> + pthread_mutex_lock(&e->lock); >> + if (t) { >> + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); prev >> = &(*prev)->next) + /* nothing */; >> + add_task(prev, t); >> + } >> + pthread_cond_signal(&e->cond); >> + pthread_mutex_unlock(&e->lock); >> + >> +#if !HAVE_THREADS >> + // We are running in a single-threaded environment, so we must handle >> all tasks ourselves >> + while (run_one_task(e, e->local_contexts)) >> + /* nothing */; >> +#endif >> +} >> diff --git a/libavutil/executor.h b/libavutil/executor.h >> new file mode 100644 >> index 0000000000..c3fe44bd0e >> --- /dev/null >> +++ b/libavutil/executor.h >> @@ -0,0 +1,67 @@ >> +/* >> + * Copyright (C) 2023 Nuo Mi >> + * >> + * 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 AVUTIL_EXECUTOR_H >> +#define AVUTIL_EXECUTOR_H >> + >> +typedef struct AVExecutor AVExecutor; >> +typedef struct AVTask AVTask; >> + >> +struct AVTask { >> + AVTask *next; >> +}; >> + >> +typedef struct AVTaskCallbacks { >> + void *user_data; >> + >> + int local_context_size; >> + >> + // return 1 if a's priority > b's priority >> + int (*priority_higher)(const AVTask *a, const AVTask *b); >> + >> + // task is ready for run >> + int (*ready)(const AVTask *t, void *user_data); >> + >> + // run the task >> + int (*run)(AVTask *t, void *local_context, void *user_data); >> +} AVTaskCallbacks; >> + >> +/** >> + * Alloc executor >> + * @param callbacks callback strucutre for executor >> + * @param thread_count worker thread number >> + * @return return the executor >> + */ >> +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int >> thread_count); + >> +/** >> + * Free executor >> + * @param e pointer to executor >> + */ >> +void av_executor_free(AVExecutor **e); >> + >> +/** >> + * Add task to executor >> + * @param e pointer to executor >> + * @param t pointer to task. If NULL, it will wakeup one work thread >> + */ >> +void av_executor_execute(AVExecutor *e, AVTask *t); >> + >> +#endif //AVUTIL_EXECUTOR_H > >
On 8/15/2023 11:50 AM, Nuo Mi wrote: > The executor design pattern was inroduced by java > <https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/concurrent/Executor.html> > it also adapted by python > <https://docs.python.org/3/library/concurrent.futures.html> > Compared to handcrafted thread pool management, it greatly simplifies the thread code. > --- > libavutil/Makefile | 2 + > libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ > libavutil/executor.h | 67 +++++++++++++++ > 3 files changed, 270 insertions(+) > create mode 100644 libavutil/executor.c > create mode 100644 libavutil/executor.h > > diff --git a/libavutil/Makefile b/libavutil/Makefile > index 7828c94dc5..4711f8cde8 100644 > --- a/libavutil/Makefile > +++ b/libavutil/Makefile > @@ -31,6 +31,7 @@ HEADERS = adler32.h \ > encryption_info.h \ > error.h \ > eval.h \ > + executor.h \ > fifo.h \ > file.h \ > frame.h \ > @@ -127,6 +128,7 @@ OBJS = adler32.o \ > encryption_info.o \ > error.o \ > eval.o \ > + executor.o \ > fifo.o \ > file.o \ > file_open.o \ > diff --git a/libavutil/executor.c b/libavutil/executor.c > new file mode 100644 > index 0000000000..6b47ba0bf2 > --- /dev/null > +++ b/libavutil/executor.c > @@ -0,0 +1,201 @@ > +/* > + * Copyright (C) 2023 Nuo Mi > + * > + * 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 "internal.h" > +#include "mem.h" > +#include "thread.h" > + > +#include "executor.h" > + > +#if !HAVE_THREADS > +#define pthread_create(t, a, s, ar) 0 > +#define pthread_join(t, r) do {} while(0) > + > +#define pthread_cond_init(c, a) 0 > +#define pthread_cond_broadcast(c) do {} while(0) > +#define pthread_cond_signal(c) do {} while(0) > +#define pthread_cond_wait(c, m) do {} while(0) > +#define pthread_cond_destroy(c) do {} while(0) > + > +#define pthread_mutex_init(m, a) 0 > +#define pthread_mutex_lock(l) do {} while(0) > +#define pthread_mutex_unlock(l) do {} while(0) > +#define pthread_mutex_destroy(l) do {} while(0) You have ff_mutex_* and AVMutex for this. I sent a patch adding ff_cond_* and AVCond, as well as ff_thread_create/join and AVThread to thread.h to simplify things. > +#endif > + > +typedef struct ThreadInfo { > + AVExecutor *e; > + pthread_t thread; You will need to use AVThread, else this will not be defined when !HAVE_THREADS. > +} ThreadInfo; > + > +struct AVExecutor { > + AVTaskCallbacks cb; > + int thread_count; > + > + ThreadInfo *threads; > + uint8_t *local_contexts; > + > + pthread_mutex_t lock; > + pthread_cond_t cond; Same here. AVMutex and AVCond. > + int die; > + > + AVTask *tasks; > +}; > + > +static AVTask* remove_task(AVTask **prev, AVTask *t) > +{ > + *prev = t->next; > + t->next = NULL; > + return t; > +} > + > +static void add_task(AVTask **prev, AVTask *t) > +{ > + t->next = *prev; > + *prev = t; > +} > + > +static int run_one_task(AVExecutor *e, void *lc) > +{ > + AVTaskCallbacks *cb = &e->cb; > + AVTask **prev; > + > + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); prev = &(*prev)->next) > + /* nothing */; > + if (*prev) { > + AVTask *t = remove_task(prev, *prev); > + pthread_mutex_unlock(&e->lock); > + cb->run(t, lc, cb->user_data); > + pthread_mutex_lock(&e->lock); > + return 1; > + } > + return 0; > +} > + > +#if HAVE_THREADS > +static void *executor_worker_task(void *data) > +{ > + ThreadInfo *ti = (ThreadInfo*)data; > + AVExecutor *e = ti->e; > + void *lc = e->local_contexts + (ti - e->threads) * e->cb.local_context_size; > + > + pthread_mutex_lock(&e->lock); > + while (1) { > + if (e->die) break; > + > + if (!run_one_task(e, lc)) { > + //no task in one loop > + pthread_cond_wait(&e->cond, &e->lock); > + } > + } > + pthread_mutex_unlock(&e->lock); > + return NULL; > +} > +#endif > + > +static void executor_free(AVExecutor *e, const int has_lock, const int has_cond) > +{ > + if (e->thread_count) { > + //signal die > + pthread_mutex_lock(&e->lock); > + e->die = 1; > + pthread_cond_broadcast(&e->cond); > + pthread_mutex_unlock(&e->lock); > + > + for (int i = 0; i < e->thread_count; i++) > + pthread_join(e->threads[i].thread, NULL); > + } > + if (has_cond) > + pthread_cond_destroy(&e->cond); > + if (has_lock) > + pthread_mutex_destroy(&e->lock); > + > + av_free(e->threads); > + av_free(e->local_contexts); > + > + av_free(e); > +} > + > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int thread_count) > +{ > + AVExecutor *e; > + int has_lock = 0, has_cond = 0; > + if (!cb || !cb->user_data || !cb->ready || !cb->run || !cb->priority_higher) > + return NULL; > + > + e = av_calloc(1, sizeof(*e)); av_mallocz() > + if (!e) > + return NULL; > + e->cb = *cb; > + > + e->local_contexts = av_calloc(thread_count, e->cb.local_context_size); > + if (!e->local_contexts) > + goto free_executor; > + > + e->threads = av_calloc(thread_count, sizeof(*e->threads)); > + if (!e->threads) > + goto free_executor; > + > + has_lock = !pthread_mutex_init(&e->lock, NULL); > + has_cond = !pthread_cond_init(&e->cond, NULL); > + > + if (!has_lock || !has_cond) > + goto free_executor; > + > + for (/* nothing */; e->thread_count < thread_count; e->thread_count++) { > + ThreadInfo *ti = e->threads + e->thread_count; > + ti->e = e; > + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) > + goto free_executor; > + } > + return e; > + > +free_executor: > + executor_free(e, has_lock, has_cond); > + return NULL; > +} > + > +void av_executor_free(AVExecutor **executor) > +{ > + if (!executor || !*executor) > + return; > + executor_free(*executor, 1, 1); > + *executor = NULL; > +} > + > +void av_executor_execute(AVExecutor *e, AVTask *t) > +{ > + AVTaskCallbacks *cb = &e->cb; > + AVTask **prev; > + > + pthread_mutex_lock(&e->lock); > + if (t) { > + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); prev = &(*prev)->next) > + /* nothing */; > + add_task(prev, t); > + } > + pthread_cond_signal(&e->cond); > + pthread_mutex_unlock(&e->lock); > + > +#if !HAVE_THREADS > + // We are running in a single-threaded environment, so we must handle all tasks ourselves > + while (run_one_task(e, e->local_contexts)) > + /* nothing */; > +#endif > +} > diff --git a/libavutil/executor.h b/libavutil/executor.h > new file mode 100644 > index 0000000000..c3fe44bd0e > --- /dev/null > +++ b/libavutil/executor.h > @@ -0,0 +1,67 @@ > +/* > + * Copyright (C) 2023 Nuo Mi > + * > + * 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 AVUTIL_EXECUTOR_H > +#define AVUTIL_EXECUTOR_H > + > +typedef struct AVExecutor AVExecutor; > +typedef struct AVTask AVTask; > + > +struct AVTask { > + AVTask *next; > +}; > + > +typedef struct AVTaskCallbacks { > + void *user_data; > + > + int local_context_size; > + > + // return 1 if a's priority > b's priority > + int (*priority_higher)(const AVTask *a, const AVTask *b); > + > + // task is ready for run > + int (*ready)(const AVTask *t, void *user_data); > + > + // run the task > + int (*run)(AVTask *t, void *local_context, void *user_data); > +} AVTaskCallbacks; > + > +/** > + * Alloc executor > + * @param callbacks callback strucutre for executor > + * @param thread_count worker thread number > + * @return return the executor > + */ > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int thread_count); > + > +/** > + * Free executor > + * @param e pointer to executor > + */ > +void av_executor_free(AVExecutor **e); > + > +/** > + * Add task to executor > + * @param e pointer to executor > + * @param t pointer to task. If NULL, it will wakeup one work thread > + */ > +void av_executor_execute(AVExecutor *e, AVTask *t); > + > +#endif //AVUTIL_EXECUTOR_H
On 8/18/2023 1:17 PM, James Almer wrote: > On 8/15/2023 11:50 AM, Nuo Mi wrote: >> The executor design pattern was inroduced by java >> <https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/concurrent/Executor.html> >> it also adapted by python >> <https://docs.python.org/3/library/concurrent.futures.html> >> Compared to handcrafted thread pool management, it greatly simplifies >> the thread code. >> --- >> libavutil/Makefile | 2 + >> libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ >> libavutil/executor.h | 67 +++++++++++++++ >> 3 files changed, 270 insertions(+) >> create mode 100644 libavutil/executor.c >> create mode 100644 libavutil/executor.h >> >> diff --git a/libavutil/Makefile b/libavutil/Makefile >> index 7828c94dc5..4711f8cde8 100644 >> --- a/libavutil/Makefile >> +++ b/libavutil/Makefile >> @@ -31,6 +31,7 @@ HEADERS = >> adler32.h \ >> >> encryption_info.h \ >> >> error.h \ >> >> eval.h \ >> + >> executor.h \ >> >> fifo.h \ >> >> file.h \ >> >> frame.h \ >> @@ -127,6 +128,7 @@ OBJS = >> adler32.o \ >> >> encryption_info.o \ >> >> error.o \ >> >> eval.o \ >> + >> executor.o \ >> >> fifo.o \ >> >> file.o \ >> >> file_open.o \ >> diff --git a/libavutil/executor.c b/libavutil/executor.c >> new file mode 100644 >> index 0000000000..6b47ba0bf2 >> --- /dev/null >> +++ b/libavutil/executor.c >> @@ -0,0 +1,201 @@ >> +/* >> + * Copyright (C) 2023 Nuo Mi >> + * >> + * 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 "internal.h" >> +#include "mem.h" >> +#include "thread.h" >> + >> +#include "executor.h" >> + >> +#if !HAVE_THREADS >> +#define pthread_create(t, a, s, ar) 0 >> +#define pthread_join(t, r) do {} while(0) >> + >> +#define pthread_cond_init(c, a) 0 >> +#define pthread_cond_broadcast(c) do {} while(0) >> +#define pthread_cond_signal(c) do {} while(0) >> +#define pthread_cond_wait(c, m) do {} while(0) >> +#define pthread_cond_destroy(c) do {} while(0) >> + >> +#define pthread_mutex_init(m, a) 0 >> +#define pthread_mutex_lock(l) do {} while(0) >> +#define pthread_mutex_unlock(l) do {} while(0) >> +#define pthread_mutex_destroy(l) do {} while(0) > > You have ff_mutex_* and AVMutex for this. I sent a patch adding > ff_cond_* and AVCond, as well as ff_thread_create/join and AVThread to > thread.h to simplify things. > >> +#endif >> + >> +typedef struct ThreadInfo { >> + AVExecutor *e; >> + pthread_t thread; > > You will need to use AVThread, else this will not be defined when > !HAVE_THREADS. I ultimately didn't add AVThread, so you'll have to wrap this field. and the calls to pthread_create() and pthread_join(). > >> +} ThreadInfo; >> + >> +struct AVExecutor { >> + AVTaskCallbacks cb; >> + int thread_count; >> + >> + ThreadInfo *threads; >> + uint8_t *local_contexts; >> + >> + pthread_mutex_t lock; >> + pthread_cond_t cond; > > Same here. AVMutex and AVCond. > >> + int die; >> + >> + AVTask *tasks; >> +}; >> + >> +static AVTask* remove_task(AVTask **prev, AVTask *t) >> +{ >> + *prev = t->next; >> + t->next = NULL; >> + return t; >> +} >> + >> +static void add_task(AVTask **prev, AVTask *t) >> +{ >> + t->next = *prev; >> + *prev = t; >> +} >> + >> +static int run_one_task(AVExecutor *e, void *lc) >> +{ >> + AVTaskCallbacks *cb = &e->cb; >> + AVTask **prev; >> + >> + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); >> prev = &(*prev)->next) >> + /* nothing */; >> + if (*prev) { >> + AVTask *t = remove_task(prev, *prev); >> + pthread_mutex_unlock(&e->lock); >> + cb->run(t, lc, cb->user_data); >> + pthread_mutex_lock(&e->lock); >> + return 1; >> + } >> + return 0; >> +} >> + >> +#if HAVE_THREADS >> +static void *executor_worker_task(void *data) >> +{ >> + ThreadInfo *ti = (ThreadInfo*)data; >> + AVExecutor *e = ti->e; >> + void *lc = e->local_contexts + (ti - e->threads) * >> e->cb.local_context_size; >> + >> + pthread_mutex_lock(&e->lock); >> + while (1) { >> + if (e->die) break; >> + >> + if (!run_one_task(e, lc)) { >> + //no task in one loop >> + pthread_cond_wait(&e->cond, &e->lock); >> + } >> + } >> + pthread_mutex_unlock(&e->lock); >> + return NULL; >> +} >> +#endif >> + >> +static void executor_free(AVExecutor *e, const int has_lock, const >> int has_cond) >> +{ >> + if (e->thread_count) { >> + //signal die >> + pthread_mutex_lock(&e->lock); >> + e->die = 1; >> + pthread_cond_broadcast(&e->cond); >> + pthread_mutex_unlock(&e->lock); >> + >> + for (int i = 0; i < e->thread_count; i++) >> + pthread_join(e->threads[i].thread, NULL); >> + } >> + if (has_cond) >> + pthread_cond_destroy(&e->cond); >> + if (has_lock) >> + pthread_mutex_destroy(&e->lock); >> + >> + av_free(e->threads); >> + av_free(e->local_contexts); >> + >> + av_free(e); >> +} >> + >> +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int >> thread_count) >> +{ >> + AVExecutor *e; >> + int has_lock = 0, has_cond = 0; >> + if (!cb || !cb->user_data || !cb->ready || !cb->run || >> !cb->priority_higher) >> + return NULL; >> + >> + e = av_calloc(1, sizeof(*e)); > > av_mallocz() > >> + if (!e) >> + return NULL; >> + e->cb = *cb; >> + >> + e->local_contexts = av_calloc(thread_count, >> e->cb.local_context_size); >> + if (!e->local_contexts) >> + goto free_executor; >> + >> + e->threads = av_calloc(thread_count, sizeof(*e->threads)); >> + if (!e->threads) >> + goto free_executor; >> + >> + has_lock = !pthread_mutex_init(&e->lock, NULL); >> + has_cond = !pthread_cond_init(&e->cond, NULL); >> + >> + if (!has_lock || !has_cond) >> + goto free_executor; >> + >> + for (/* nothing */; e->thread_count < thread_count; >> e->thread_count++) { >> + ThreadInfo *ti = e->threads + e->thread_count; >> + ti->e = e; >> + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) executor_worker_task() doesn't exist if !HAVE_THREADS, so like i said above, wrap this for loop. >> + goto free_executor; >> + } >> + return e; >> + >> +free_executor: >> + executor_free(e, has_lock, has_cond); >> + return NULL; >> +} >> + >> +void av_executor_free(AVExecutor **executor) >> +{ >> + if (!executor || !*executor) >> + return; >> + executor_free(*executor, 1, 1); >> + *executor = NULL; >> +} >> + >> +void av_executor_execute(AVExecutor *e, AVTask *t) >> +{ >> + AVTaskCallbacks *cb = &e->cb; >> + AVTask **prev; >> + >> + pthread_mutex_lock(&e->lock); >> + if (t) { >> + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, >> t); prev = &(*prev)->next) >> + /* nothing */; >> + add_task(prev, t); >> + } >> + pthread_cond_signal(&e->cond); >> + pthread_mutex_unlock(&e->lock); >> + >> +#if !HAVE_THREADS >> + // We are running in a single-threaded environment, so we must >> handle all tasks ourselves >> + while (run_one_task(e, e->local_contexts)) >> + /* nothing */; >> +#endif >> +} >> diff --git a/libavutil/executor.h b/libavutil/executor.h >> new file mode 100644 >> index 0000000000..c3fe44bd0e >> --- /dev/null >> +++ b/libavutil/executor.h >> @@ -0,0 +1,67 @@ >> +/* >> + * Copyright (C) 2023 Nuo Mi >> + * >> + * 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 AVUTIL_EXECUTOR_H >> +#define AVUTIL_EXECUTOR_H >> + >> +typedef struct AVExecutor AVExecutor; >> +typedef struct AVTask AVTask; >> + >> +struct AVTask { >> + AVTask *next; >> +}; >> + >> +typedef struct AVTaskCallbacks { >> + void *user_data; >> + >> + int local_context_size; >> + >> + // return 1 if a's priority > b's priority >> + int (*priority_higher)(const AVTask *a, const AVTask *b); >> + >> + // task is ready for run >> + int (*ready)(const AVTask *t, void *user_data); >> + >> + // run the task >> + int (*run)(AVTask *t, void *local_context, void *user_data); >> +} AVTaskCallbacks; >> + >> +/** >> + * Alloc executor >> + * @param callbacks callback strucutre for executor >> + * @param thread_count worker thread number >> + * @return return the executor >> + */ >> +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int >> thread_count); >> + >> +/** >> + * Free executor >> + * @param e pointer to executor >> + */ >> +void av_executor_free(AVExecutor **e); >> + >> +/** >> + * Add task to executor >> + * @param e pointer to executor >> + * @param t pointer to task. If NULL, it will wakeup one work thread >> + */ >> +void av_executor_execute(AVExecutor *e, AVTask *t); >> + >> +#endif //AVUTIL_EXECUTOR_H
On Thu, Aug 17, 2023 at 12:42 AM Rémi Denis-Courmont <remi@remlab.net> wrote: > Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : > > The executor design pattern was inroduced by java > > < > https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc > > urrent/Executor.html> it also adapted by python > > <https://docs.python.org/3/library/concurrent.futures.html> > > Compared to handcrafted thread pool management, it greatly simplifies the > > thread code. --- > > libavutil/Makefile | 2 + > > libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ > > libavutil/executor.h | 67 +++++++++++++++ > > 3 files changed, 270 insertions(+) > > create mode 100644 libavutil/executor.c > > create mode 100644 libavutil/executor.h > > > > diff --git a/libavutil/Makefile b/libavutil/Makefile > > index 7828c94dc5..4711f8cde8 100644 > > --- a/libavutil/Makefile > > +++ b/libavutil/Makefile > > @@ -31,6 +31,7 @@ HEADERS = adler32.h > > > \ encryption_info.h > > > \ error.h \ > eval.h > > \ + > > executor.h \ fifo.h > > > \ file.h > > > \ frame.h > > > \ @@ -127,6 +128,7 @@ OBJS = adler32.o > > > \ encryption_info.o > > > \ error.o > > > \ eval.o > > > \ + executor.o > > > \ fifo.o > > > \ file.o > \ > > file_open.o \ diff > > --git a/libavutil/executor.c b/libavutil/executor.c > > new file mode 100644 > > index 0000000000..6b47ba0bf2 > > --- /dev/null > > +++ b/libavutil/executor.c > > @@ -0,0 +1,201 @@ > > +/* > > + * Copyright (C) 2023 Nuo Mi > > + * > > + * 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 "internal.h" > > +#include "mem.h" > > +#include "thread.h" > > + > > +#include "executor.h" > > + > > +#if !HAVE_THREADS > > +#define pthread_create(t, a, s, ar) 0 > > Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. > > > +#define pthread_join(t, r) do {} while(0) > > ...and this should be an unreachable(), assert(0), abort() or something > along > those lines. > > > + > > +#define pthread_cond_init(c, a) 0 > > +#define pthread_cond_broadcast(c) do {} while(0) > > +#define pthread_cond_signal(c) do {} while(0) > > +#define pthread_cond_wait(c, m) do {} while(0) > > +#define pthread_cond_destroy(c) do {} while(0) > > + > > +#define pthread_mutex_init(m, a) 0 > > +#define pthread_mutex_lock(l) do {} while(0) > > +#define pthread_mutex_unlock(l) do {} while(0) > > +#define pthread_mutex_destroy(l) do {} while(0) > > +#endif > > + > > +typedef struct ThreadInfo { > > + AVExecutor *e; > > + pthread_t thread; > > Does this even work if !HAVE_THREAD ? Yes. The definition provided by the following include chain: "mem.h"->"avutil.h"->"common.h"->"<stdlib.h>"->"<sys/types.h>"->"<bits/pthreadtypes.h>" > > > +} ThreadInfo; > > + > > +struct AVExecutor { > > + AVTaskCallbacks cb; > > + int thread_count; > > + > > + ThreadInfo *threads; > > + uint8_t *local_contexts; > > + > > + pthread_mutex_t lock; > > + pthread_cond_t cond; > > + int die; > > + > > + AVTask *tasks; > > +}; > > + > > +static AVTask* remove_task(AVTask **prev, AVTask *t) > > +{ > > + *prev = t->next; > > + t->next = NULL; > > + return t; > > +} > > + > > +static void add_task(AVTask **prev, AVTask *t) > > +{ > > + t->next = *prev; > > + *prev = t; > > +} > > + > > +static int run_one_task(AVExecutor *e, void *lc) > > +{ > > + AVTaskCallbacks *cb = &e->cb; > > + AVTask **prev; > > + > > + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); > prev > > = &(*prev)->next) > > + /* nothing */; > > + if (*prev) { > > + AVTask *t = remove_task(prev, *prev); > > + pthread_mutex_unlock(&e->lock); > > + cb->run(t, lc, cb->user_data); > > + pthread_mutex_lock(&e->lock); > > + return 1; > > + } > > + return 0; > > +} > > + > > +#if HAVE_THREADS > > +static void *executor_worker_task(void *data) > > +{ > > + ThreadInfo *ti = (ThreadInfo*)data; > > + AVExecutor *e = ti->e; > > + void *lc = e->local_contexts > > + (ti - e->threads) * > > e->cb.local_context_size; + > > + pthread_mutex_lock(&e->lock); > > + while (1) { > > + if (e->die) break; > > + > > + if (!run_one_task(e, lc)) { > > + //no task in one loop > > + pthread_cond_wait(&e->cond, &e->lock); > > + } > > + } > > + pthread_mutex_unlock(&e->lock); > > + return NULL; > > +} > > +#endif > > + > > +static void executor_free(AVExecutor *e, const int has_lock, const int > > has_cond) +{ > > + if (e->thread_count) { > > + //signal die > > + pthread_mutex_lock(&e->lock); > > + e->die = 1; > > + pthread_cond_broadcast(&e->cond); > > + pthread_mutex_unlock(&e->lock); > > + > > + for (int i = 0; i < e->thread_count; i++) > > + pthread_join(e->threads[i].thread, NULL); > > + } > > + if (has_cond) > > + pthread_cond_destroy(&e->cond); > > + if (has_lock) > > + pthread_mutex_destroy(&e->lock); > > + > > + av_free(e->threads); > > + av_free(e->local_contexts); > > + > > + av_free(e); > > +} > > + > > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int > thread_count) > > +{ > > + AVExecutor *e; > > + int has_lock = 0, has_cond = 0; > > + if (!cb || !cb->user_data || !cb->ready || !cb->run || > > !cb->priority_higher) + return NULL; > > + > > + e = av_calloc(1, sizeof(*e)); > > + if (!e) > > + return NULL; > > + e->cb = *cb; > > + > > + e->local_contexts = av_calloc(thread_count, > e->cb.local_context_size); > > + if (!e->local_contexts) > > + goto free_executor; > > + > > + e->threads = av_calloc(thread_count, sizeof(*e->threads)); > > + if (!e->threads) > > + goto free_executor; > > + > > + has_lock = !pthread_mutex_init(&e->lock, NULL); > > + has_cond = !pthread_cond_init(&e->cond, NULL); > > + > > + if (!has_lock || !has_cond) > > + goto free_executor; > > + > > + for (/* nothing */; e->thread_count < thread_count; > e->thread_count++) > > { + ThreadInfo *ti = e->threads + e->thread_count; > > + ti->e = e; > > + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) > > + goto free_executor; > > + } > > + return e; > > + > > +free_executor: > > + executor_free(e, has_lock, has_cond); > > + return NULL; > > +} > > + > > +void av_executor_free(AVExecutor **executor) > > +{ > > + if (!executor || !*executor) > > + return; > > + executor_free(*executor, 1, 1); > > + *executor = NULL; > > +} > > + > > +void av_executor_execute(AVExecutor *e, AVTask *t) > > +{ > > + AVTaskCallbacks *cb = &e->cb; > > + AVTask **prev; > > + > > + pthread_mutex_lock(&e->lock); > > + if (t) { > > + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); > prev > > = &(*prev)->next) + /* nothing */; > > + add_task(prev, t); > > + } > > + pthread_cond_signal(&e->cond); > > + pthread_mutex_unlock(&e->lock); > > + > > +#if !HAVE_THREADS > > + // We are running in a single-threaded environment, so we must > handle > > all tasks ourselves > > + while (run_one_task(e, e->local_contexts)) > > + /* nothing */; > > +#endif > > +} > > diff --git a/libavutil/executor.h b/libavutil/executor.h > > new file mode 100644 > > index 0000000000..c3fe44bd0e > > --- /dev/null > > +++ b/libavutil/executor.h > > @@ -0,0 +1,67 @@ > > +/* > > + * Copyright (C) 2023 Nuo Mi > > + * > > + * 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 AVUTIL_EXECUTOR_H > > +#define AVUTIL_EXECUTOR_H > > + > > +typedef struct AVExecutor AVExecutor; > > +typedef struct AVTask AVTask; > > + > > +struct AVTask { > > + AVTask *next; > > +}; > > + > > +typedef struct AVTaskCallbacks { > > + void *user_data; > > + > > + int local_context_size; > > + > > + // return 1 if a's priority > b's priority > > + int (*priority_higher)(const AVTask *a, const AVTask *b); > > + > > + // task is ready for run > > + int (*ready)(const AVTask *t, void *user_data); > > + > > + // run the task > > + int (*run)(AVTask *t, void *local_context, void *user_data); > > +} AVTaskCallbacks; > > + > > +/** > > + * Alloc executor > > + * @param callbacks callback strucutre for executor > > + * @param thread_count worker thread number > > + * @return return the executor > > + */ > > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int > > thread_count); + > > +/** > > + * Free executor > > + * @param e pointer to executor > > + */ > > +void av_executor_free(AVExecutor **e); > > + > > +/** > > + * Add task to executor > > + * @param e pointer to executor > > + * @param t pointer to task. If NULL, it will wakeup one work thread > > + */ > > +void av_executor_execute(AVExecutor *e, AVTask *t); > > + > > +#endif //AVUTIL_EXECUTOR_H > > > -- > Rémi Denis-Courmont > http://www.remlab.net/ > > > > _______________________________________________ > ffmpeg-devel mailing list > ffmpeg-devel@ffmpeg.org > https://ffmpeg.org/mailman/listinfo/ffmpeg-devel > > To unsubscribe, visit link above, or email > ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe". >
> > >> + if (pthread_create(&ti->thread, NULL, executor_worker_task, > ti)) > > executor_worker_task() doesn't exist if !HAVE_THREADS, so like i said > above, wrap this for loop. > Not needed since pthread_create is defined to 0 > >
> > > > > + > > +#if !HAVE_THREADS > > +#define pthread_create(t, a, s, ar) 0 > > +#define pthread_join(t, r) do {} while(0) > > + > > +#define pthread_cond_init(c, a) 0 > > +#define pthread_cond_broadcast(c) do {} while(0) > > +#define pthread_cond_signal(c) do {} while(0) > > +#define pthread_cond_wait(c, m) do {} while(0) > > +#define pthread_cond_destroy(c) do {} while(0) > > + > > +#define pthread_mutex_init(m, a) 0 > > +#define pthread_mutex_lock(l) do {} while(0) > > +#define pthread_mutex_unlock(l) do {} while(0) > > +#define pthread_mutex_destroy(l) do {} while(0) > > You have ff_mutex_* and AVMutex for this. I sent a patch adding > ff_cond_* and AVCond, as well as ff_thread_create/join and AVThread to > thread.h to simplify things. Will do it for AVMutex and AVCond. But it's hard to add a wrapper for pthread_t. Because we do not have a way to detect typedef and I do not want to add "#ifdef _BITS_PTHREADTYPES_COMMON_H" > > > +#endif > > + > > +typedef struct ThreadInfo { > > + AVExecutor *e; > > + pthread_t thread; > > +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int > thread_count) > > +{ > > + AVExecutor *e; > > + int has_lock = 0, has_cond = 0; > > + if (!cb || !cb->user_data || !cb->ready || !cb->run || > !cb->priority_higher) > > + return NULL; > > + > > + e = av_calloc(1, sizeof(*e)); > > av_mallocz() > will do > > > + if (!e) > > + return NULL; > > + e->cb = *cb; > > + > > + e->local_contexts = av_calloc(thread_count, > e->cb.local_context_size); > > + if (!e->local_contexts) > >
On 8/22/2023 9:03 AM, Nuo Mi wrote: > On Thu, Aug 17, 2023 at 12:42 AM Rémi Denis-Courmont <remi@remlab.net> > wrote: > >> Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : >>> The executor design pattern was inroduced by java >>> < >> https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc >>> urrent/Executor.html> it also adapted by python >>> <https://docs.python.org/3/library/concurrent.futures.html> >>> Compared to handcrafted thread pool management, it greatly simplifies the >>> thread code. --- >>> libavutil/Makefile | 2 + >>> libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ >>> libavutil/executor.h | 67 +++++++++++++++ >>> 3 files changed, 270 insertions(+) >>> create mode 100644 libavutil/executor.c >>> create mode 100644 libavutil/executor.h >>> >>> diff --git a/libavutil/Makefile b/libavutil/Makefile >>> index 7828c94dc5..4711f8cde8 100644 >>> --- a/libavutil/Makefile >>> +++ b/libavutil/Makefile >>> @@ -31,6 +31,7 @@ HEADERS = adler32.h >> >>> \ encryption_info.h >> >>> \ error.h \ >> eval.h >>> \ + >>> executor.h \ fifo.h >> >>> \ file.h >> >>> \ frame.h >> >>> \ @@ -127,6 +128,7 @@ OBJS = adler32.o >> >>> \ encryption_info.o >> >>> \ error.o >> >>> \ eval.o >> >>> \ + executor.o >> >>> \ fifo.o >> >>> \ file.o >> \ >>> file_open.o \ diff >>> --git a/libavutil/executor.c b/libavutil/executor.c >>> new file mode 100644 >>> index 0000000000..6b47ba0bf2 >>> --- /dev/null >>> +++ b/libavutil/executor.c >>> @@ -0,0 +1,201 @@ >>> +/* >>> + * Copyright (C) 2023 Nuo Mi >>> + * >>> + * 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 "internal.h" >>> +#include "mem.h" >>> +#include "thread.h" >>> + >>> +#include "executor.h" >>> + >>> +#if !HAVE_THREADS >>> +#define pthread_create(t, a, s, ar) 0 >> >> Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. >> >>> +#define pthread_join(t, r) do {} while(0) >> >> ...and this should be an unreachable(), assert(0), abort() or something >> along >> those lines. >> >>> + >>> +#define pthread_cond_init(c, a) 0 >>> +#define pthread_cond_broadcast(c) do {} while(0) >>> +#define pthread_cond_signal(c) do {} while(0) >>> +#define pthread_cond_wait(c, m) do {} while(0) >>> +#define pthread_cond_destroy(c) do {} while(0) >>> + >>> +#define pthread_mutex_init(m, a) 0 >>> +#define pthread_mutex_lock(l) do {} while(0) >>> +#define pthread_mutex_unlock(l) do {} while(0) >>> +#define pthread_mutex_destroy(l) do {} while(0) >>> +#endif >>> + >>> +typedef struct ThreadInfo { >>> + AVExecutor *e; >>> + pthread_t thread; >> >> Does this even work if !HAVE_THREAD ? > > Yes. The definition provided by the following include chain: > "mem.h"->"avutil.h"->"common.h"->"<stdlib.h>"->"<sys/types.h>"->"<bits/pthreadtypes.h>" Why would bits/pthreadtypes.h be available if there's no pthreads?
On Tue, Aug 22, 2023 at 8:23 PM James Almer <jamrial@gmail.com> wrote: > On 8/22/2023 9:03 AM, Nuo Mi wrote: > > On Thu, Aug 17, 2023 at 12:42 AM Rémi Denis-Courmont <remi@remlab.net> > > wrote: > > > >> Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : > >>> The executor design pattern was inroduced by java > >>> < > >> > https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc > >>> urrent/Executor.html> it also adapted by python > >>> <https://docs.python.org/3/library/concurrent.futures.html> > >>> Compared to handcrafted thread pool management, it greatly simplifies > the > >>> thread code. --- > >>> libavutil/Makefile | 2 + > >>> libavutil/executor.c | 201 > +++++++++++++++++++++++++++++++++++++++++++ > >>> libavutil/executor.h | 67 +++++++++++++++ > >>> 3 files changed, 270 insertions(+) > >>> create mode 100644 libavutil/executor.c > >>> create mode 100644 libavutil/executor.h > >>> > >>> diff --git a/libavutil/Makefile b/libavutil/Makefile > >>> index 7828c94dc5..4711f8cde8 100644 > >>> --- a/libavutil/Makefile > >>> +++ b/libavutil/Makefile > >>> @@ -31,6 +31,7 @@ HEADERS = adler32.h > >> > >>> \ encryption_info.h > >> > >>> \ error.h \ > >> eval.h > >>> \ + > >>> executor.h \ fifo.h > >> > >>> \ file.h > >> > >>> \ frame.h > >> > >>> \ @@ -127,6 +128,7 @@ OBJS = adler32.o > >> > >>> \ encryption_info.o > >> > >>> \ error.o > >> > >>> \ eval.o > >> > >>> \ + executor.o > >> > >>> \ fifo.o > >> > >>> \ file.o > >> \ > >>> file_open.o \ diff > >>> --git a/libavutil/executor.c b/libavutil/executor.c > >>> new file mode 100644 > >>> index 0000000000..6b47ba0bf2 > >>> --- /dev/null > >>> +++ b/libavutil/executor.c > >>> @@ -0,0 +1,201 @@ > >>> +/* > >>> + * Copyright (C) 2023 Nuo Mi > >>> + * > >>> + * 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 "internal.h" > >>> +#include "mem.h" > >>> +#include "thread.h" > >>> + > >>> +#include "executor.h" > >>> + > >>> +#if !HAVE_THREADS > >>> +#define pthread_create(t, a, s, ar) 0 > >> > >> Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. > >> > >>> +#define pthread_join(t, r) do {} while(0) > >> > >> ...and this should be an unreachable(), assert(0), abort() or something > >> along > >> those lines. > >> > >>> + > >>> +#define pthread_cond_init(c, a) 0 > >>> +#define pthread_cond_broadcast(c) do {} while(0) > >>> +#define pthread_cond_signal(c) do {} while(0) > >>> +#define pthread_cond_wait(c, m) do {} while(0) > >>> +#define pthread_cond_destroy(c) do {} while(0) > >>> + > >>> +#define pthread_mutex_init(m, a) 0 > >>> +#define pthread_mutex_lock(l) do {} while(0) > >>> +#define pthread_mutex_unlock(l) do {} while(0) > >>> +#define pthread_mutex_destroy(l) do {} while(0) > >>> +#endif > >>> + > >>> +typedef struct ThreadInfo { > >>> + AVExecutor *e; > >>> + pthread_t thread; > >> > >> Does this even work if !HAVE_THREAD ? > > > > Yes. The definition provided by the following include chain: > > > "mem.h"->"avutil.h"->"common.h"->"<stdlib.h>"->"<sys/types.h>"->"<bits/pthreadtypes.h>" > > Why would bits/pthreadtypes.h be available if there's no pthreads? > I guess it is related to POSIX. If you check <sys/types.h>, you will find the following code: /* Now add the thread types. */ #if defined __USE_POSIX199506 || defined __USE_UNIX98 # include <bits/pthreadtypes.h> #endif > _______________________________________________ > ffmpeg-devel mailing list > ffmpeg-devel@ffmpeg.org > https://ffmpeg.org/mailman/listinfo/ffmpeg-devel > > To unsubscribe, visit link above, or email > ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe". >
On 8/22/2023 9:30 AM, Nuo Mi wrote: > On Tue, Aug 22, 2023 at 8:23 PM James Almer <jamrial@gmail.com> wrote: > >> On 8/22/2023 9:03 AM, Nuo Mi wrote: >>> On Thu, Aug 17, 2023 at 12:42 AM Rémi Denis-Courmont <remi@remlab.net> >>> wrote: >>> >>>> Le tiistaina 15. elokuuta 2023, 17.50.13 EEST Nuo Mi a écrit : >>>>> The executor design pattern was inroduced by java >>>>> < >>>> >> https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/conc >>>>> urrent/Executor.html> it also adapted by python >>>>> <https://docs.python.org/3/library/concurrent.futures.html> >>>>> Compared to handcrafted thread pool management, it greatly simplifies >> the >>>>> thread code. --- >>>>> libavutil/Makefile | 2 + >>>>> libavutil/executor.c | 201 >> +++++++++++++++++++++++++++++++++++++++++++ >>>>> libavutil/executor.h | 67 +++++++++++++++ >>>>> 3 files changed, 270 insertions(+) >>>>> create mode 100644 libavutil/executor.c >>>>> create mode 100644 libavutil/executor.h >>>>> >>>>> diff --git a/libavutil/Makefile b/libavutil/Makefile >>>>> index 7828c94dc5..4711f8cde8 100644 >>>>> --- a/libavutil/Makefile >>>>> +++ b/libavutil/Makefile >>>>> @@ -31,6 +31,7 @@ HEADERS = adler32.h >>>> >>>>> \ encryption_info.h >>>> >>>>> \ error.h \ >>>> eval.h >>>>> \ + >>>>> executor.h \ fifo.h >>>> >>>>> \ file.h >>>> >>>>> \ frame.h >>>> >>>>> \ @@ -127,6 +128,7 @@ OBJS = adler32.o >>>> >>>>> \ encryption_info.o >>>> >>>>> \ error.o >>>> >>>>> \ eval.o >>>> >>>>> \ + executor.o >>>> >>>>> \ fifo.o >>>> >>>>> \ file.o >>>> \ >>>>> file_open.o \ diff >>>>> --git a/libavutil/executor.c b/libavutil/executor.c >>>>> new file mode 100644 >>>>> index 0000000000..6b47ba0bf2 >>>>> --- /dev/null >>>>> +++ b/libavutil/executor.c >>>>> @@ -0,0 +1,201 @@ >>>>> +/* >>>>> + * Copyright (C) 2023 Nuo Mi >>>>> + * >>>>> + * 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 "internal.h" >>>>> +#include "mem.h" >>>>> +#include "thread.h" >>>>> + >>>>> +#include "executor.h" >>>>> + >>>>> +#if !HAVE_THREADS >>>>> +#define pthread_create(t, a, s, ar) 0 >>>> >>>> Err, this should probably return an error value, e.g. ENOSYS or ENOTSUP. >>>> >>>>> +#define pthread_join(t, r) do {} while(0) >>>> >>>> ...and this should be an unreachable(), assert(0), abort() or something >>>> along >>>> those lines. >>>> >>>>> + >>>>> +#define pthread_cond_init(c, a) 0 >>>>> +#define pthread_cond_broadcast(c) do {} while(0) >>>>> +#define pthread_cond_signal(c) do {} while(0) >>>>> +#define pthread_cond_wait(c, m) do {} while(0) >>>>> +#define pthread_cond_destroy(c) do {} while(0) >>>>> + >>>>> +#define pthread_mutex_init(m, a) 0 >>>>> +#define pthread_mutex_lock(l) do {} while(0) >>>>> +#define pthread_mutex_unlock(l) do {} while(0) >>>>> +#define pthread_mutex_destroy(l) do {} while(0) >>>>> +#endif >>>>> + >>>>> +typedef struct ThreadInfo { >>>>> + AVExecutor *e; >>>>> + pthread_t thread; >>>> >>>> Does this even work if !HAVE_THREAD ? >>> >>> Yes. The definition provided by the following include chain: >>> >> "mem.h"->"avutil.h"->"common.h"->"<stdlib.h>"->"<sys/types.h>"->"<bits/pthreadtypes.h>" >> >> Why would bits/pthreadtypes.h be available if there's no pthreads? >> > I guess it is related to POSIX. If you check <sys/types.h>, you will find > the following code: > > /* Now add the thread types. */ > #if defined __USE_POSIX199506 || defined __USE_UNIX98 > # include <bits/pthreadtypes.h> > #endif Well, not all targets are POSIX. If you look at the win32 and os/2 wrappers, they need to define pthread_t. So it's not guaranteed to be defined when !HAVE_THREADS.
On Tue, Aug 22, 2023 at 8:32 PM James Almer <jamrial@gmail.com> wrote: > > >>> > >> > "mem.h"->"avutil.h"->"common.h"->"<stdlib.h>"->"<sys/types.h>"->"<bits/pthreadtypes.h>" > >> > >> Why would bits/pthreadtypes.h be available if there's no pthreads? > >> > > I guess it is related to POSIX. If you check <sys/types.h>, you will find > > the following code: > > > > /* Now add the thread types. */ > > #if defined __USE_POSIX199506 || defined __USE_UNIX98 > > # include <bits/pthreadtypes.h> > > #endif > > Well, not all targets are POSIX. If you look at the win32 and os/2 > wrappers, they need to define pthread_t. So it's not guaranteed to be > defined when !HAVE_THREADS. > Done by v5 > _______________________________________________ > ffmpeg-devel mailing list > ffmpeg-devel@ffmpeg.org > https://ffmpeg.org/mailman/listinfo/ffmpeg-devel > > To unsubscribe, visit link above, or email > ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe". >
On Thu, Aug 17, 2023 at 12:36 AM Michael Niedermayer <michael@niedermayer.cc> wrote: > On Tue, Aug 15, 2023 at 10:50:13PM +0800, Nuo Mi wrote: > > The executor design pattern was inroduced by java > > < > https://docs.oracle.com/en/java/javase/20/docs/api/java.base/java/util/concurrent/Executor.html > > > > it also adapted by python > > <https://docs.python.org/3/library/concurrent.futures.html> > > Compared to handcrafted thread pool management, it greatly simplifies > the thread code. > > --- > > libavutil/Makefile | 2 + > > libavutil/executor.c | 201 +++++++++++++++++++++++++++++++++++++++++++ > > libavutil/executor.h | 67 +++++++++++++++ > > 3 files changed, 270 insertions(+) > > create mode 100644 libavutil/executor.c > > create mode 100644 libavutil/executor.h > > needs doc/APIchanges addition > should be ok otherwise > Done by v5 > > thx > > > -- > Michael GnuPG fingerprint: 9FF2128B147EF6730BADF133611EC787040B0FAB > > The worst form of inequality is to try to make unequal things equal. > -- Aristotle > _______________________________________________ > ffmpeg-devel mailing list > ffmpeg-devel@ffmpeg.org > https://ffmpeg.org/mailman/listinfo/ffmpeg-devel > > To unsubscribe, visit link above, or email > ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe". >
diff --git a/libavutil/Makefile b/libavutil/Makefile index 7828c94dc5..4711f8cde8 100644 --- a/libavutil/Makefile +++ b/libavutil/Makefile @@ -31,6 +31,7 @@ HEADERS = adler32.h \ encryption_info.h \ error.h \ eval.h \ + executor.h \ fifo.h \ file.h \ frame.h \ @@ -127,6 +128,7 @@ OBJS = adler32.o \ encryption_info.o \ error.o \ eval.o \ + executor.o \ fifo.o \ file.o \ file_open.o \ diff --git a/libavutil/executor.c b/libavutil/executor.c new file mode 100644 index 0000000000..6b47ba0bf2 --- /dev/null +++ b/libavutil/executor.c @@ -0,0 +1,201 @@ +/* + * Copyright (C) 2023 Nuo Mi + * + * 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 "internal.h" +#include "mem.h" +#include "thread.h" + +#include "executor.h" + +#if !HAVE_THREADS +#define pthread_create(t, a, s, ar) 0 +#define pthread_join(t, r) do {} while(0) + +#define pthread_cond_init(c, a) 0 +#define pthread_cond_broadcast(c) do {} while(0) +#define pthread_cond_signal(c) do {} while(0) +#define pthread_cond_wait(c, m) do {} while(0) +#define pthread_cond_destroy(c) do {} while(0) + +#define pthread_mutex_init(m, a) 0 +#define pthread_mutex_lock(l) do {} while(0) +#define pthread_mutex_unlock(l) do {} while(0) +#define pthread_mutex_destroy(l) do {} while(0) +#endif + +typedef struct ThreadInfo { + AVExecutor *e; + pthread_t thread; +} ThreadInfo; + +struct AVExecutor { + AVTaskCallbacks cb; + int thread_count; + + ThreadInfo *threads; + uint8_t *local_contexts; + + pthread_mutex_t lock; + pthread_cond_t cond; + int die; + + AVTask *tasks; +}; + +static AVTask* remove_task(AVTask **prev, AVTask *t) +{ + *prev = t->next; + t->next = NULL; + return t; +} + +static void add_task(AVTask **prev, AVTask *t) +{ + t->next = *prev; + *prev = t; +} + +static int run_one_task(AVExecutor *e, void *lc) +{ + AVTaskCallbacks *cb = &e->cb; + AVTask **prev; + + for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); prev = &(*prev)->next) + /* nothing */; + if (*prev) { + AVTask *t = remove_task(prev, *prev); + pthread_mutex_unlock(&e->lock); + cb->run(t, lc, cb->user_data); + pthread_mutex_lock(&e->lock); + return 1; + } + return 0; +} + +#if HAVE_THREADS +static void *executor_worker_task(void *data) +{ + ThreadInfo *ti = (ThreadInfo*)data; + AVExecutor *e = ti->e; + void *lc = e->local_contexts + (ti - e->threads) * e->cb.local_context_size; + + pthread_mutex_lock(&e->lock); + while (1) { + if (e->die) break; + + if (!run_one_task(e, lc)) { + //no task in one loop + pthread_cond_wait(&e->cond, &e->lock); + } + } + pthread_mutex_unlock(&e->lock); + return NULL; +} +#endif + +static void executor_free(AVExecutor *e, const int has_lock, const int has_cond) +{ + if (e->thread_count) { + //signal die + pthread_mutex_lock(&e->lock); + e->die = 1; + pthread_cond_broadcast(&e->cond); + pthread_mutex_unlock(&e->lock); + + for (int i = 0; i < e->thread_count; i++) + pthread_join(e->threads[i].thread, NULL); + } + if (has_cond) + pthread_cond_destroy(&e->cond); + if (has_lock) + pthread_mutex_destroy(&e->lock); + + av_free(e->threads); + av_free(e->local_contexts); + + av_free(e); +} + +AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int thread_count) +{ + AVExecutor *e; + int has_lock = 0, has_cond = 0; + if (!cb || !cb->user_data || !cb->ready || !cb->run || !cb->priority_higher) + return NULL; + + e = av_calloc(1, sizeof(*e)); + if (!e) + return NULL; + e->cb = *cb; + + e->local_contexts = av_calloc(thread_count, e->cb.local_context_size); + if (!e->local_contexts) + goto free_executor; + + e->threads = av_calloc(thread_count, sizeof(*e->threads)); + if (!e->threads) + goto free_executor; + + has_lock = !pthread_mutex_init(&e->lock, NULL); + has_cond = !pthread_cond_init(&e->cond, NULL); + + if (!has_lock || !has_cond) + goto free_executor; + + for (/* nothing */; e->thread_count < thread_count; e->thread_count++) { + ThreadInfo *ti = e->threads + e->thread_count; + ti->e = e; + if (pthread_create(&ti->thread, NULL, executor_worker_task, ti)) + goto free_executor; + } + return e; + +free_executor: + executor_free(e, has_lock, has_cond); + return NULL; +} + +void av_executor_free(AVExecutor **executor) +{ + if (!executor || !*executor) + return; + executor_free(*executor, 1, 1); + *executor = NULL; +} + +void av_executor_execute(AVExecutor *e, AVTask *t) +{ + AVTaskCallbacks *cb = &e->cb; + AVTask **prev; + + pthread_mutex_lock(&e->lock); + if (t) { + for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); prev = &(*prev)->next) + /* nothing */; + add_task(prev, t); + } + pthread_cond_signal(&e->cond); + pthread_mutex_unlock(&e->lock); + +#if !HAVE_THREADS + // We are running in a single-threaded environment, so we must handle all tasks ourselves + while (run_one_task(e, e->local_contexts)) + /* nothing */; +#endif +} diff --git a/libavutil/executor.h b/libavutil/executor.h new file mode 100644 index 0000000000..c3fe44bd0e --- /dev/null +++ b/libavutil/executor.h @@ -0,0 +1,67 @@ +/* + * Copyright (C) 2023 Nuo Mi + * + * 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 AVUTIL_EXECUTOR_H +#define AVUTIL_EXECUTOR_H + +typedef struct AVExecutor AVExecutor; +typedef struct AVTask AVTask; + +struct AVTask { + AVTask *next; +}; + +typedef struct AVTaskCallbacks { + void *user_data; + + int local_context_size; + + // return 1 if a's priority > b's priority + int (*priority_higher)(const AVTask *a, const AVTask *b); + + // task is ready for run + int (*ready)(const AVTask *t, void *user_data); + + // run the task + int (*run)(AVTask *t, void *local_context, void *user_data); +} AVTaskCallbacks; + +/** + * Alloc executor + * @param callbacks callback strucutre for executor + * @param thread_count worker thread number + * @return return the executor + */ +AVExecutor* av_executor_alloc(const AVTaskCallbacks *callbacks, int thread_count); + +/** + * Free executor + * @param e pointer to executor + */ +void av_executor_free(AVExecutor **e); + +/** + * Add task to executor + * @param e pointer to executor + * @param t pointer to task. If NULL, it will wakeup one work thread + */ +void av_executor_execute(AVExecutor *e, AVTask *t); + +#endif //AVUTIL_EXECUTOR_H