ffmpeg (-mt) и TBB

Question

Я только начал использовать последнюю сборку ffmpeg, в которую был объединен ffmpeg-mt.

Однако, так как мое приложение использует TBB (Intel Threading Building Blocks), реализация ffmpeg-mt с созданием нового потокаи синхронизация не совсем подходит, так как она может потенциально блокировать мои задачи TBB, выполняющие функции декодирования.Также это приведет к ненужной очистке кеша.

Я искал в pthread.c, который, кажется, реализует интерфейс, который ffmpeg использует для включения многопоточности.

Мой вопрос заключается в том, возможно ли создать tbb.c, который реализует те же функции, но использует задачи tbb вместо явных потоков?

У меня нет опыта работы с C, но я думаю,в том, что было бы невозможно легко скомпилировать tbb (который является C ++) в ffmpeg.Так что, может быть, каким-то образом можно перезаписать указатели на функции ffmpeg во время выполнения?

Буду признателен за любые предложения или комментарии относительно реализации TBB в API потоков ffmpeg.

Answer 1

Итак, я понял, как это сделать, прочитав код ffmpeg.

По сути, все, что вам нужно сделать, это включить приведенный ниже код и использовать tbb_avcodec_open/tbb_avcodec_close вместо ffmpegs 'avcodec_open/avcodec_close.

Это будет использовать задачи TBB для параллельного выполнения декодирования.

 // Author Robert Nagy

#include "tbb_avcodec.h"

#include <tbb/task.h>
#include <tbb/atomic.h>

extern "C" 
{
    #define __STDC_CONSTANT_MACROS
    #define __STDC_LIMIT_MACROS
    #include <libavformat/avformat.h>
}

int task_execute(AVCodecContext* s, std::function<int(void* arg, int arg_size, int jobnr, int threadnr)>&& func, void* arg, int* ret, int count, int size)
{   
    tbb::atomic<int> counter;
    counter = 0;

    // Execute s->thread_count number of tasks in parallel.
    tbb::parallel_for(0, s->thread_count, 1, [&](int threadnr) 
    {
        while(true)
        {
            int jobnr = counter++;
            if(jobnr >= count)
                break;

            int r = func(arg, size, jobnr, threadnr);
            if (ret)
                ret[jobnr] = r;
        }
    });

    return 0;
}

int thread_execute(AVCodecContext* s, int (*func)(AVCodecContext *c2, void *arg2), void* arg, int* ret, int count, int size)
{
    return task_execute(s, [&](void* arg, int arg_size, int jobnr, int threadnr) -> int
    {
        return func(s, reinterpret_cast<uint8_t*>(arg) + jobnr*size);
    }, arg, ret, count, size);
}

int thread_execute2(AVCodecContext* s, int (*func)(AVCodecContext* c2, void* arg2, int, int), void* arg, int* ret, int count)
{
    return task_execute(s, [&](void* arg, int arg_size, int jobnr, int threadnr) -> int
    {
        return func(s, arg, jobnr, threadnr);
    }, arg, ret, count, 0);
}

void thread_init(AVCodecContext* s)
{
    static const size_t MAX_THREADS = 16; // See mpegvideo.h
    static int dummy_opaque;

    s->active_thread_type = FF_THREAD_SLICE;
    s->thread_opaque      = &dummy_opaque; 
    s->execute            = thread_execute;
    s->execute2           = thread_execute2;
    s->thread_count       = MAX_THREADS; // We are using a task-scheduler, so use as many "threads/tasks" as possible.
}

void thread_free(AVCodecContext* s)
{
    s->thread_opaque = nullptr;
}

int tbb_avcodec_open(AVCodecContext* avctx, AVCodec* codec)
{
    avctx->thread_count = 1;
    if((codec->capabilities & CODEC_CAP_SLICE_THREADS) && (avctx->thread_type & FF_THREAD_SLICE))
        thread_init(avctx);
// ff_thread_init will not be executed since thread_opaque != nullptr || thread_count == 1.
    return avcodec_open(avctx, codec); 
}

int tbb_avcodec_close(AVCodecContext* avctx)
{
    thread_free(avctx);
    // ff_thread_free will not be executed since thread_opaque == nullptr.
    return avcodec_close(avctx); 
}

Answer 2

Повторно публикуем здесь мой ответ вам на форуме TBB , ради того, кто в SO может быть заинтересован.

Ваш код в ответе выше выглядит хорошо для меня; умный способ использовать TBB в контексте, который был разработан с учетом собственных потоков. Интересно, можно ли так сказать сделать еще больше TBBish. У меня есть несколько идей, которые вы можете попробовать, если у вас есть время и желание.

Следующие два элемента могут представлять интерес, если есть желание / необходимость контролировать количество потоков.

• в thread_init создайте выделенный в куче объект tbb::task_scheduler_init (TSI) и инициализируйте его таким количеством потоков, сколько необходимо (не обязательно MAX_THREADS). Сохраните адрес этого объекта в s->thread_opaque, если это возможно / разрешено; в противном случае возможное решение - глобальная карта, которая отображает AVCodecContext* на адрес соответствующего task_scheduler_init.
• соответственно в thread_free, получить и удалить объект TSI.

Независимо от вышесказанного, еще одно потенциальное изменение в том, как позвонить tbb::parallel_for. Вместо того, чтобы использовать его просто для создания достаточного количества потоков, нельзя ли использовать его по прямому назначению, как показано ниже?

int task_execute(AVCodecContext* s,
                 std::function<int(void*, int, int, int)>&& f,
                 void* arg, int* ret, int count, int size)   
{      
    tbb::atomic<int> counter;   
    counter = 0;   

    // Execute 'count' number of tasks in parallel.   
    tbb::parallel_for(tbb::blocked_range<int>(0, count, 2),
                      [&](const tbb::blocked_range<int> &r)    
    {   
        int threadnr = counter++;   
        for(int jobnr=r.begin(); jobnr!=r.end(); ++jobnr)
        {   
            int r = func(arg, size, jobnr, threadnr);   
            if (ret)   
                ret[jobnr] = r;   
        }
        --counter;
    });   

    return 0;   
}

Это может работать лучше, если count значительно больше, чем thread_count, потому что а) более параллельный провал означает, что TBB работает более эффективно (что вы, очевидно, знаете), и б) накладные расходы централизованного атомного счетчика распределены по больше итераций. Обратите внимание, что я выбрал размер зерна 2 для blocked_range; это связано с тем, что счетчик увеличивается и уменьшается внутри тела цикла, и поэтому для «подбора» вашего варианта необходимо как минимум две итерации на задачу (и, соответственно, count>=2*thread_count).