2.0.0.2: -ffmpeg_producer: Implemented slice based multithreading.

git-svn-id: https://casparcg.svn.sourceforge.net/svnroot/casparcg/server/branches/2.0.0.2@785 362d55ac-95cf-4e76-9f9a-cbaa9c17b72d

diff --git a/modules/ffmpeg/ffmpeg.vcxproj b/modules/ffmpeg/ffmpeg.vcxproj
index 7b1c341900adc9ce1dd590cbe8d4b2ec343c9819..301e57434a1d3644580f9da9be558abe0ff7495b 100644 (file)
--- a/modules/ffmpeg/ffmpeg.vcxproj
+++ b/modules/ffmpeg/ffmpeg.vcxproj
@@ -304,6 +304,7 @@
       <PrecompiledHeader Condition="'$(Configuration)|$(Platform)'=='Develop|Win32'">Create</PrecompiledHeader>\r
       <PrecompiledHeader Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">Create</PrecompiledHeader>\r
     </ClCompile>\r
+    <ClCompile Include="tbb_avcodec.cpp" />\r
   </ItemGroup>\r
   <ItemGroup>\r
     <ClInclude Include="consumer\ffmpeg_consumer.h" />\r
@@ -315,6 +316,7 @@
     <ClInclude Include="producer\packet.h" />\r
     <ClInclude Include="producer\video\video_decoder.h" />\r
     <ClInclude Include="StdAfx.h" />\r
+    <ClInclude Include="tbb_avcodec.h" />\r
   </ItemGroup>\r
   <ItemGroup>\r
     <ProjectReference Include="..\..\common\common.vcxproj">\r

diff --git a/modules/ffmpeg/ffmpeg.vcxproj.filters b/modules/ffmpeg/ffmpeg.vcxproj.filters
index a5e59d97a0064d0aaac8f68b2964e9dec417b7dd..15184b6630893ab46b481059abc13363883b94e4 100644 (file)
--- a/modules/ffmpeg/ffmpeg.vcxproj.filters
+++ b/modules/ffmpeg/ffmpeg.vcxproj.filters
@@ -32,6 +32,7 @@
     </ClCompile>\r
     <ClCompile Include="ffmpeg.cpp" />\r
     <ClCompile Include="StdAfx.cpp" />\r
+    <ClCompile Include="tbb_avcodec.cpp" />\r
   </ItemGroup>\r
   <ItemGroup>\r
     <ClInclude Include="producer\ffmpeg_producer.h">\r
@@ -55,5 +56,6 @@
     <ClInclude Include="producer\packet.h">\r
       <Filter>producer</Filter>\r
     </ClInclude>\r
+    <ClInclude Include="tbb_avcodec.h" />\r
   </ItemGroup>\r
 </Project>
\ No newline at end of file

diff --git a/modules/ffmpeg/producer/input.cpp b/modules/ffmpeg/producer/input.cpp
index 8e0e0d76c4dedeecf7235566d7972999d60c4abf..aa1200a5b29716fec24e579720ba6d9a48d040ae 100644 (file)
--- a/modules/ffmpeg/producer/input.cpp
+++ b/modules/ffmpeg/producer/input.cpp
@@ -25,6 +25,7 @@
 \r
 #include "input.h"\r
 #include "../ffmpeg_error.h"\r
+#include "../tbb_avcodec.h"\r
 \r
 #include <core/video_format.h>\r
 \r
@@ -111,7 +112,7 @@ public:
                                boost::errinfo_errno(AVUNERROR(errn)));\r
                }\r
                \r
-               video_codec_context_ = open_stream(AVMEDIA_TYPE_VIDEO, video_s_index_, 4);\r
+               video_codec_context_ = open_stream(AVMEDIA_TYPE_VIDEO, video_s_index_, true);\r
                if(!video_codec_context_)\r
                        CASPAR_LOG(warning) << print() << " Could not open any video stream.";\r
                else\r
@@ -181,7 +182,7 @@ private:
                        context->time_base.num = static_cast<int>(std::pow(10.0, static_cast<int>(std::log10(static_cast<float>(context->time_base.den)))-1));\r
        }\r
 \r
-       std::shared_ptr<AVCodecContext> open_stream(int codec_type, int& s_index, int thread_count = -1) const\r
+       std::shared_ptr<AVCodecContext> open_stream(int codec_type, int& s_index, bool tbb = false) const\r
        {               \r
                const auto streams = boost::iterator_range<AVStream**>(format_context_->streams, format_context_->streams+format_context_->nb_streams);\r
                const auto stream = boost::find_if(streams, [&](AVStream* stream) \r
@@ -196,12 +197,21 @@ private:
                if(codec == nullptr)\r
                        return nullptr;\r
                        \r
-               if(avcodec_open((*stream)->codec, codec) < 0)           \r
-                       return nullptr;\r
-               \r
                s_index = (*stream)->index;\r
 \r
-               return std::shared_ptr<AVCodecContext>((*stream)->codec, avcodec_close);\r
+               if(!tbb)\r
+               {\r
+                       if(avcodec_open((*stream)->codec, codec) < 0)           \r
+                               return nullptr;\r
+                       return std::shared_ptr<AVCodecContext>((*stream)->codec, avcodec_close);\r
+               }\r
+               else\r
+               {\r
+                       if(tbb_avcodec_open((*stream)->codec, codec) < 0)               \r
+                               return nullptr;\r
+                       return std::shared_ptr<AVCodecContext>((*stream)->codec, tbb_avcodec_close);\r
+               }               \r
+\r
        }\r
        \r
        std::shared_ptr<AVCodecContext>& get_default_context()\r

diff --git a/modules/ffmpeg/tbb_avcodec.cpp b/modules/ffmpeg/tbb_avcodec.cpp
new file mode 100644 (file)
index 0000000..8c73e67
--- /dev/null
+++ b/modules/ffmpeg/tbb_avcodec.cpp
@@ -0,0 +1,107 @@
+// Author Robert Nagy\r
+\r
+#include "stdafx.h"\r
+\r
+#include "tbb_avcodec.h"\r
+\r
+#include <common/log/log.h>\r
+\r
+#include <tbb/task.h>\r
+#include <tbb/atomic.h>\r
+\r
+extern "C" \r
+{\r
+       #define __STDC_CONSTANT_MACROS\r
+       #define __STDC_LIMIT_MACROS\r
+       #include <libavformat/avformat.h>\r
+}\r
+\r
+namespace caspar {\r
+\r
+struct thread_context{};\r
+\r
+int task_execute(AVCodecContext* s, const std::function<int(void* arg, int arg_size, int jobnr, int threadnr)>& func, void* arg, int* ret, int count, int size)\r
+{      \r
+       // jobnr global for all threads? Doesn't order matter?\r
+    tbb::atomic<int> counter;\r
+       counter = 0;\r
+               \r
+       // Execute s->thread_count number of tasks in parallel.\r
+       tbb::parallel_for(0, s->thread_count, 1, [&](int threadnr) \r
+       {\r
+               while(true)\r
+               {\r
+                       int jobnr = counter++;\r
+                       if(jobnr >= count)\r
+                               break;\r
+\r
+                       int r = func(arg, size, jobnr, threadnr);\r
+                       if (ret)\r
+                               ret[jobnr] = r;\r
+               }\r
+       });\r
+       \r
+    return 0;\r
+}\r
+       \r
+int thread_execute(AVCodecContext* s, int (*func)(AVCodecContext *c2, void *arg2), void* arg, int* ret, int count, int size)\r
+{\r
+       return task_execute(s, [&](void* arg, int arg_size, int jobnr, int threadnr) -> int\r
+       {\r
+               return func(s, reinterpret_cast<uint8_t*>(arg) + jobnr*size);\r
+       }, arg, ret, count, size);\r
+}\r
+\r
+int thread_execute2(AVCodecContext* s, int (*func)(AVCodecContext* c2, void* arg2, int, int), void* arg, int* ret, int count)\r
+{\r
+       return task_execute(s, [&](void* arg, int arg_size, int jobnr, int threadnr) -> int\r
+       {\r
+               return func(s, arg, jobnr, threadnr);\r
+       }, arg, ret, count, 0);\r
+}\r
+\r
+int thread_init(AVCodecContext *s)\r
+{\r
+       // Only makes sense for slicing since decode is already called through task scheduler.\r
+    if(!(s->thread_type & FF_THREAD_SLICE))           \r
+       return 0;    \r
+\r
+       static const size_t MAX_THREADS = 16; // See mpegvideo.h\r
+\r
+    s->active_thread_type = FF_THREAD_SLICE;\r
+       s->thread_opaque          = malloc(sizeof(thread_context));      \r
+    s->execute                   = thread_execute;\r
+    s->execute2                          = thread_execute2;\r
+    s->thread_count              = MAX_THREADS; // We are using a taskscheduler, so use as many "threads/tasks" as possible. \r
+\r
+       CASPAR_LOG(info) << "Initialized ffmpeg tbb context.";\r
+\r
+    return 0;\r
+}\r
+\r
+void thread_free(AVCodecContext* s)\r
+{\r
+       if(!s->thread_opaque)\r
+               return;\r
+       \r
+       free(s->thread_opaque); \r
+       s->thread_opaque = nullptr;\r
+\r
+       CASPAR_LOG(info) << "Released ffmpeg tbb context.";\r
+}\r
+\r
+int tbb_avcodec_open(AVCodecContext* avctx, AVCodec* codec)\r
+{\r
+       thread_init(avctx);\r
+       // ff_thread_init will not be executed since thread_opaque != nullptr.\r
+       return avcodec_open(avctx, codec); \r
+}\r
+\r
+int tbb_avcodec_close(AVCodecContext* avctx)\r
+{\r
+       thread_free(avctx);\r
+       // ff_thread_free will not be executed since thread_opaque == nullptr.\r
+       return avcodec_close(avctx); \r
+}\r
+\r
+}
\ No newline at end of file

diff --git a/modules/ffmpeg/tbb_avcodec.h b/modules/ffmpeg/tbb_avcodec.h
new file mode 100644 (file)
index 0000000..e2aa3e9
--- /dev/null
+++ b/modules/ffmpeg/tbb_avcodec.h
@@ -0,0 +1,11 @@
+#pragma once\r
+\r
+struct AVCodecContext;\r
+struct AVCodec;\r
+\r
+namespace caspar {\r
+       \r
+int tbb_avcodec_open(AVCodecContext *avctx, AVCodec *codec);\r
+int tbb_avcodec_close(AVCodecContext *avctx);\r
+\r
+}
\ No newline at end of file