diff --git a/lib/tests/data/v210_flow_01.json b/lib/tests/data/v210_flow_01.json new file mode 100644 index 000000000..b0e2e8829 --- /dev/null +++ b/lib/tests/data/v210_flow_01.json @@ -0,0 +1,43 @@ +{ + "$copyright": "SPDX-FileCopyrightText: 2025 Contributors to the Media eXchange Layer project.", + "$license": "SPDX-License-Identifier: Apache-2.0", + "description": "MXL Test Flow, 1080p29", + "id": "6fbec3b1-1b0f-417d-9059-8b94a47197ed", + "tags": { + "urn:x-nmos:tag:grouphint/v1.0": [ + "Media Function XYZ:Video" + ] + }, + "format": "urn:x-nmos:format:video", + "label": "MXL Test Flow, 1080p29", + "parents": [], + "media_type": "video/v210", + "grain_rate": { + "numerator": 30000, + "denominator": 1001 + }, + "frame_width": 1920, + "frame_height": 1080, + "interlace_mode": "progressive", + "colorspace": "BT709", + "components": [ + { + "name": "Y", + "width": 1920, + "height": 1080, + "bit_depth": 10 + }, + { + "name": "Cb", + "width": 960, + "height": 1080, + "bit_depth": 10 + }, + { + "name": "Cr", + "width": 960, + "height": 1080, + "bit_depth": 10 + } + ] +} \ No newline at end of file diff --git a/lib/tests/data/v210_flow_01.json.license b/lib/tests/data/v210_flow_01.json.license new file mode 100644 index 000000000..acfd4c533 --- /dev/null +++ b/lib/tests/data/v210_flow_01.json.license @@ -0,0 +1,3 @@ +SPDX-FileCopyrightText: 2025 Contributors to the Media eXchange Layer project. + +SPDX-License-Identifier: Apache-2.0 diff --git a/tools/mxl-gst/sink.cpp b/tools/mxl-gst/sink.cpp index 518fdcc3d..e732b59a9 100644 --- a/tools/mxl-gst/sink.cpp +++ b/tools/mxl-gst/sink.cpp @@ -5,9 +5,12 @@ #include #include #include +#include +#include #include #include #include +#include #include #include #include @@ -150,7 +153,7 @@ namespace } } - void launchPipeline(std::string const& pipelineDescription) + void launchPipeline(std::string const& pipelineDescription, std::string const& appSourceName = "appsource") { GError* error = nullptr; _pipeline = ::gst_parse_launch(pipelineDescription.c_str(), &error); @@ -165,7 +168,7 @@ namespace throw std::runtime_error{"Gstreamer pipeline could not be created."}; } - _appSource = ::gst_bin_get_by_name(GST_BIN(_pipeline), "appsource"); + _appSource = ::gst_bin_get_by_name(GST_BIN(_pipeline), appSourceName.c_str()); if (_appSource == nullptr) { throw std::runtime_error{"Well-known application source element could not be found in the gstreamer pipeline."}; @@ -191,6 +194,20 @@ namespace } } + public: + [[nodiscard]] + constexpr GstElement* getPipeline() noexcept + { + return _pipeline; + } + + [[nodiscard]] + constexpr GstElement const* getPipeline() const noexcept + { + return _pipeline; + } + + protected: [[nodiscard]] constexpr GstElement* getAppSource() noexcept { @@ -203,7 +220,6 @@ namespace return _appSource; } - private: GstElement* _pipeline; GstElement* _appSource; std::uint64_t _mxlBaseTime; @@ -236,7 +252,7 @@ namespace _config.frameRate.denominator, _config.offset); - MXL_INFO("Generating following GStreamer video pipeline -> {}", pipelineDesc); + MXL_INFO("Generating following Video gsteamer pipeline -> {}", pipelineDesc); launchPipeline(pipelineDesc); } @@ -270,7 +286,7 @@ namespace generateMixMatrix(), _config.offset); - MXL_INFO("Generating following GStreamer audio pipeline -> {}", pipelineDesc); + MXL_INFO("Generating following Audio gsteamer pipeline -> {}", pipelineDesc); launchPipeline(pipelineDesc); { @@ -289,6 +305,7 @@ namespace auto const caps = ::gst_audio_info_to_caps(_audioInfo); ::g_object_set(G_OBJECT(getAppSource()), "caps", caps, nullptr); + ::gst_caps_unref(caps); } } @@ -348,6 +365,9 @@ namespace GstAudioInfo* _audioInfo; }; + // Forward declaration for MxlReader::run() overload + class MultiviewerPipeline; + class MxlReader { public: @@ -374,13 +394,48 @@ namespace ~MxlReader() { - if (_reader != nullptr) + close(); + } + + // A reader can't be duplicated or copy-assigned since that would mess with the internal state of the flow reader. + MxlReader(MxlReader const&) = delete; + void operator=(MxlReader const&) = delete; + + MxlReader(MxlReader&& other) noexcept + : MxlReader{} + { + // *this is fully initialized through the delegating constructor, + // then takes ownership from other via swap. + swap(other); + } + + MxlReader& operator=(MxlReader other) noexcept + { + swap(other); + + return *this; + } + + void swap(MxlReader& other) noexcept + { + using std::swap; + swap(_instance, other._instance); + swap(_reader, other._reader); + swap(_configInfo, other._configInfo); + swap(_highestLatencyNs, other._highestLatencyNs); + } + + void close() + { + if (_reader) { ::mxlReleaseFlowReader(_instance, _reader); + _reader = nullptr; } - if (_instance != nullptr) + if (_instance) { ::mxlDestroyInstance(_instance); + _instance = nullptr; } } @@ -688,6 +743,9 @@ namespace } } + // Declaration only - implementation after MultiviewerPipeline class definition + void run(MultiviewerPipeline& gstPipeline, std::size_t sourceIndex, std::int64_t readDelay); + private: /** * This default constructor is private, because the only reason why it @@ -839,6 +897,425 @@ namespace std::uint64_t _highestLatencyNs; }; + [[maybe_unused]] + void swap(MxlReader& lhs, MxlReader& rhs) noexcept + { + lhs.swap(rhs); + } + + class MultiviewerPipeline : public GstreamerPipeline + { + public: + MultiviewerPipeline(std::vector const& configs) + : GstreamerPipeline{} + , _configs{configs} + { + if (configs.empty() || configs.size() > 4) + { + throw std::invalid_argument{"Multiviewer supports 1 to 4 video flows"}; + } + + MXL_INFO("Creating multiviewer pipeline with {} video flow(s)", configs.size()); + + // Build the compositor layout pipeline + auto const pipelineDesc = buildMultiviewerPipeline(); + MXL_INFO("Generating multiviewer gstreamer pipeline -> {}", pipelineDesc); + launchPipeline(pipelineDesc, "appsource0"); + + // Configure all appsrc elements to use time format + for (auto i = std::size_t{0}; i < _configs.size(); ++i) + { + auto const sourceName = fmt::format("appsource{}", i); + auto* const appSrc = ::gst_bin_get_by_name(GST_BIN(getPipeline()), sourceName.c_str()); + if (appSrc) + { + ::g_object_set(G_OBJECT(appSrc), "format", GST_FORMAT_TIME, nullptr); + ::gst_object_unref(appSrc); + MXL_INFO("Configured {} with GST_FORMAT_TIME", sourceName); + } + else + { + MXL_WARN("Could not find {} in pipeline", sourceName); + } + } + + // Get compositor element and configure sink pad positions + auto const compositor = ::gst_bin_get_by_name(GST_BIN(getPipeline()), "c"); + if (compositor) + { + MXL_INFO("Retrieved compositor element: {}", GST_ELEMENT_NAME(compositor)); + MXL_INFO("Compositor type: {}", G_OBJECT_TYPE_NAME(compositor)); + + // Get compositor properties + gint background; + ::g_object_get(compositor, "background", &background, nullptr); + MXL_INFO("Compositor background mode: {}", background); + + // Get number of sink pads + GValue val = G_VALUE_INIT; + ::g_value_init(&val, G_TYPE_UINT); + ::g_object_get_property(G_OBJECT(compositor), "n-pads", &val); + auto const numPads = ::g_value_get_uint(&val); + ::g_value_unset(&val); + MXL_INFO("Compositor has {} sink pads", numPads); + + configureCompositorLayout(compositor); + ::gst_object_unref(compositor); + } + else + { + MXL_WARN("Could not retrieve compositor element from pipeline"); + } + } + + [[nodiscard]] + std::vector const& configs() const noexcept + { + return _configs; + } + + private: + std::string buildMultiviewerPipeline() + { + auto pipelineDesc = std::ostringstream{}; + + // Create appsrc for each video flow + for (auto i = std::size_t{0}; i < _configs.size(); ++i) + { + if (i > 0) + { + pipelineDesc << " "; + } + + auto const& config = _configs[i]; + pipelineDesc << fmt::format("appsrc name=appsource{} is-live=true ! " + "video/x-raw,format=v210,width={},height={},framerate={}/{} ! " + "videoconvert ! videoscale ! c.sink_{}", + i, + config.frameWidth, + config.frameHeight, + config.frameRate.numerator, + config.frameRate.denominator, + i); + } + + // Determine output resolution based on number of inputs + auto [outputWidth, outputHeight] = getOutputResolution(_configs.size()); + MXL_INFO("Compositor output resolution: {}x{}", outputWidth, outputHeight); + + pipelineDesc << fmt::format(" compositor name=c background=black ! " + "video/x-raw,width={},height={} ! " + "videoconvert ! videoscale ! autovideosink ts-offset={}", + outputWidth, + outputHeight, + _configs[0].offset); + + return pipelineDesc.str(); + } + + constexpr std::pair getOutputResolution(std::size_t flowCount) const noexcept + { + switch (flowCount) + { + case 1: return {1280, 960}; // Single video at full resolution + case 2: return {1280, 480}; // Side-by-side + case 3: return {1280, 720}; // 2 top, 1 bottom + case 4: return {1280, 960}; // 2x2 grid + default: return {1280, 960}; + } + } + + void configureCompositorLayout(GstElement* compositor) + { + auto const numFlows = _configs.size(); + + MXL_INFO("Configuring compositor layout for {} video flows", numFlows); + + // Iterate through all pads to see what's available + auto* padIter = ::gst_element_iterate_sink_pads(compositor); + GValue item = G_VALUE_INIT; + auto padCount = 0; + while (::gst_iterator_next(padIter, &item) == GST_ITERATOR_OK) + { + auto* const pad = static_cast(::g_value_get_object(&item)); + MXL_INFO("Found compositor pad: {}", GST_PAD_NAME(pad)); + ::g_value_reset(&item); + padCount++; + } + ::g_value_unset(&item); + ::gst_iterator_free(padIter); + MXL_INFO("Total sink pads found: {}", padCount); + + for (auto i = std::size_t{0}; i < numFlows; ++i) + { + // Try multiple naming schemes + auto sinkPadName = fmt::format("sink_{}", i); + auto sinkPad = ::gst_element_get_static_pad(compositor, sinkPadName.c_str()); + + if (!sinkPad) + { + // Try alternative naming + sinkPadName = fmt::format("sink%d", i); + sinkPad = ::gst_element_get_static_pad(compositor, sinkPadName.c_str()); + } + + if (sinkPad) + { + auto [xpos, ypos, width, height] = getLayoutPosition(numFlows, i); + + MXL_INFO("Setting compositor sink pad {}: xpos={} ypos={} width={} height={}", GST_PAD_NAME(sinkPad), xpos, ypos, width, height); + + g_object_set(sinkPad, + "xpos", + static_cast(xpos), + "ypos", + static_cast(ypos), + "width", + static_cast(width), + "height", + static_cast(height), + "alpha", + 1.0, + NULL); + + // Verify the properties were set + gint verifyX, verifyY, verifyW, verifyH; + ::g_object_get(sinkPad, "xpos", &verifyX, "ypos", &verifyY, "width", &verifyW, "height", &verifyH, nullptr); + MXL_INFO("Verified compositor sink pad {}: xpos={} ypos={} width={} height={}", + GST_PAD_NAME(sinkPad), + verifyX, + verifyY, + verifyW, + verifyH); + + ::gst_object_unref(sinkPad); + } + else + { + MXL_WARN("Could not find sink pad {} in compositor", sinkPadName); + } + } + } + + constexpr std::tuple getLayoutPosition(std::size_t flowCount, + std::size_t index) const noexcept + { + switch (flowCount) + { + case 1: return {0, 0, 1280, 960}; + + case 2: + // Side-by-side: 640x480 each + if (index == 0) + { + return {0, 0, 640, 480}; + } + else + { + return {640, 0, 640, 480}; + } + + case 3: + // 2 top, 1 bottom centered + if (index == 0) + { + return {0, 0, 640, 360}; + } + else if (index == 1) + { + return {640, 0, 640, 360}; + } + else + { + return {320, 360, 640, 360}; + } + + case 4: + // 2x2 grid + { + auto row = index / 2; + auto col = index % 2; + return {col * 640, row * 480, 640, 480}; + } + + default: return {0, 0, 1280, 960}; + } + } + + private: + std::vector _configs; + }; + + // MxlReader::run() implementation for MultiviewerPipeline - defined here after MultiviewerPipeline class is complete + void MxlReader::run(MultiviewerPipeline& gstPipeline, std::size_t sourceIndex, std::int64_t readDelay) + { + if (_configInfo.common.format != MXL_DATA_FORMAT_VIDEO) + { + throw std::domain_error{"Attempt to feed a gstreamer video pipeline from a non-video MXL flow."}; + } + + // Get the specific appsource for this reader by index + auto const sourceName = fmt::format("appsource{}", sourceIndex); + auto* appSource = ::gst_bin_get_by_name(GST_BIN(gstPipeline.getPipeline()), sourceName.c_str()); + + auto appSourceRef = std::unique_ptr{GST_OBJECT(appSource), + [](GstObject* obj) + { + if (obj != nullptr) + { + ::gst_object_unref(obj); + } + }}; + + if (appSource == nullptr) + { + throw std::runtime_error{fmt::format("Could not find appsource {} in multiviewer pipeline", sourceIndex)}; + } + + // Only start the pipeline once (first reader) + static auto startFlag = std::once_flag{}; + std::call_once(startFlag, [&]() { gstPipeline.start(); }); + + auto const rate = _configInfo.common.grainRate; + auto const slicesPerBatch = _configInfo.common.maxSyncBatchSizeHint; + auto const& config = gstPipeline.configs()[sourceIndex]; + auto const sliceReadMode = (slicesPerBatch < config.frameHeight); + + if (slicesPerBatch > config.frameHeight) + { + throw std::invalid_argument{"slicesPerBatch cannot be greater than frame height."}; + } + + MXL_INFO("Starting multiviewer flow {} reading at rate {}/{} and slices per batch {}", + sourceIndex, + rate.numerator, + rate.denominator, + slicesPerBatch); + + auto cursor = Cursor{rate, 1U, readDelay}; + auto const mxlBaseTime = ::mxlGetTime(); + + initializeHighestLatency(fmt::format("Video{}", sourceIndex).c_str(), rate, cursor.currentIndex(), cursor.requestedIndex()); + + auto expectedSlices = slicesPerBatch; + while (!g_exit_requested) + { + auto ret = mxlStatus{}; + auto grainInfo = mxlGrainInfo{}; + uint8_t* payload; + + auto const iterationStartTime = ::mxlGetTime(); + auto const iterationTimeoutNs = + (iterationStartTime < cursor.deliveryDeadline()) ? (cursor.deliveryDeadline() - iterationStartTime) : 0ULL; + + if (sliceReadMode) + { + ret = ::mxlFlowReaderGetGrainSlice(_reader, cursor.requestedIndex(), expectedSlices, iterationTimeoutNs, &grainInfo, &payload); + } + else + { + ret = ::mxlFlowReaderGetGrain(_reader, cursor.requestedIndex(), iterationTimeoutNs, &grainInfo, &payload); + } + + if (ret == MXL_STATUS_OK) + { + if (grainInfo.validSlices >= grainInfo.totalSlices) + { + if ((grainInfo.flags & MXL_GRAIN_FLAG_INVALID) == 0) + { + updateHighestLatency( + fmt::format("Video{}", sourceIndex).c_str(), ::mxlIndexToTimestamp(&rate, cursor.requestedIndex()), ::mxlGetTime()); + + auto const buffer = ::gst_buffer_new_allocate(nullptr, grainInfo.grainSize, nullptr); + auto map = GstMapInfo{}; + + ::gst_buffer_map(buffer, &map, GST_MAP_WRITE); + std::memcpy(map.data, payload, grainInfo.grainSize); + ::gst_buffer_unmap(buffer, &map); + + // Push to the specific appsource for this reader + GST_BUFFER_PTS(buffer) = ::mxlIndexToTimestamp(&rate, cursor.currentIndex() + 1U) - mxlBaseTime; + int pushRet; + ::g_signal_emit_by_name(appSource, "push-buffer", buffer, &pushRet); + if (pushRet != GST_FLOW_OK) + { + MXL_ERROR("Could not push buffer to appsource{}", sourceIndex); + } + + ::gst_buffer_unref(buffer); + } + + cursor.next(); + expectedSlices = slicesPerBatch; + } + else if (sliceReadMode) + { + expectedSlices = std::min(grainInfo.totalSlices, grainInfo.validSlices + slicesPerBatch); + } + } + else if (ret == MXL_ERR_FLOW_INVALID) + { + if (handleInvalidFlow(cursor.requestedIndex())) + { + cursor.realign(iterationStartTime); + } + } + else if (ret == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) + { + auto runtimeInfo = ::mxlFlowRuntimeInfo{}; + (void)::mxlFlowReaderGetRuntimeInfo(_reader, &runtimeInfo); + MXL_WARN("Flow {}: Failed to get grain at index {}: TOO EARLY. Last published {}", + sourceIndex, + cursor.requestedIndex(), + runtimeInfo.headIndex); + } + else if (ret == MXL_ERR_OUT_OF_RANGE_TOO_LATE) + { + auto runtimeInfo = ::mxlFlowRuntimeInfo{}; + (void)::mxlFlowReaderGetRuntimeInfo(_reader, &runtimeInfo); + MXL_TRACE("Flow {}: Failed to get grain at index {}: TOO LATE. Last published {}", + sourceIndex, + cursor.requestedIndex(), + runtimeInfo.headIndex); + cursor.realign(iterationStartTime); + } + else + { + if (_reader == nullptr) + { + auto const flowId = uuids::to_string(_configInfo.common.id); + if (auto const ret = ::mxlCreateFlowReader(_instance, flowId.c_str(), "", &_reader); ret != MXL_STATUS_OK) + { + MXL_TRACE("Flow {}: Failed to reopen video flow reader with status code {}.", sourceIndex, static_cast(ret)); + std::this_thread::sleep_for(std::chrono::milliseconds{500}); + } + else + { + MXL_INFO("Flow {}: Reconnected to video flowId {}.", sourceIndex, flowId); + if (auto const ret = mxlFlowReaderGetConfigInfo(_reader, &_configInfo); ret != MXL_STATUS_OK) + { + MXL_ERROR("Flow {}: Failed to get flow config info with status code {}. Exiting.", sourceIndex, static_cast(ret)); + return; + } + } + + if (_reader != nullptr) + { + cursor.realign(iterationStartTime); + } + } + else + { + MXL_ERROR("Flow {}: Unexpected error when reading grain {} with status {}. Exiting...", + sourceIndex, + cursor.requestedIndex(), + static_cast(ret)); + return; + } + } + } + } + std::string readFlowDescriptor(std::string const& domain, std::string const& flowID) { auto const opts = "{}"; @@ -854,10 +1331,13 @@ namespace if (mxlGetFlowDef(instance, flowID.c_str(), fourKBuffer, &requiredBufferSize) != MXL_STATUS_OK) { + ::mxlDestroyInstance(instance); throw std::runtime_error{"Failed to get flow definition for flow id " + flowID}; } - return std::string{fourKBuffer, requiredBufferSize - 1}; + auto const flowDescriptor = std::string{fourKBuffer, requiredBufferSize - 1}; + ::mxlDestroyInstance(instance); + return flowDescriptor; } int real_main(int argc, char** argv, void*) @@ -867,8 +1347,8 @@ namespace auto app = CLI::App{"mxl-gst-sink"}; - auto videoFlowID = std::string{}; - app.add_option("-v, --video-flow-id", videoFlowID, "The video flow ID"); + auto videoFlowIDs = std::vector{}; + app.add_option("-v, --video-flow-id", videoFlowIDs, "The video flow ID(s) (supports multiple for multiviewer mode)"); auto audioFlowID = std::string{}; app.add_option("-a, --audio-flow-id", audioFlowID, "The audio flow ID"); @@ -918,48 +1398,136 @@ namespace auto threads = std::vector{}; - if (!videoFlowID.empty()) + // Handle video flows - single or multiviewer mode + if (!videoFlowIDs.empty()) { - threads.emplace_back( - [&]() - { - try + if (videoFlowIDs.size() == 1) + { + // Single video mode - existing behavior + threads.emplace_back( + [&]() { - auto reader = MxlReader{domain, videoFlowID}; + try + { + auto reader = MxlReader{domain, videoFlowIDs[0]}; - auto const flowDescriptor = readFlowDescriptor(domain, videoFlowID); - auto const flowNmos = json_utils::parseBuffer(flowDescriptor); + auto const flowDescriptor = readFlowDescriptor(domain, videoFlowIDs[0]); + auto const flowNmos = json_utils::parseBuffer(flowDescriptor); - if (json_utils::getField(flowNmos, "interlace_mode") != "progressive") + if (json_utils::getField(flowNmos, "interlace_mode") != "progressive") + { + throw std::invalid_argument{"This application does not support interlaced flows."}; + } + + auto const grainRate = json_utils::getRational(flowNmos, "grain_rate"); + auto const frameWidth = static_cast(json_utils::getField(flowNmos, "frame_width")); + auto const frameHeight = static_cast(json_utils::getField(flowNmos, "frame_height")); + + auto videoConfig = VideoPipelineConfig{ + .frameRate = grainRate, + .frameWidth = static_cast(frameWidth), + .frameHeight = static_cast(frameHeight), + .offset = playbackDelay + ((audioVideoOffset < 0) ? -audioVideoOffset : 0LL), + }; + + auto pipeline = VideoPipeline{videoConfig}; + reader.run(pipeline, readDelay); + + MXL_INFO("Video pipeline finished"); + } + catch (std::exception const& e) + { + MXL_ERROR("Error while processing video pipeline: {}", e.what()); + } + catch (...) { - throw std::invalid_argument{"This application does not support interlaced flows."}; + MXL_ERROR("Encountered unknown error while processing video pipeline."); } + }); + } + else + { + // Multiviewer mode - multiple videos + threads.emplace_back( + [&]() + { + try + { + auto configs = std::vector{}; + auto readers = std::vector{}; - auto const grainRate = json_utils::getRational(flowNmos, "grain_rate"); - auto const frameWidth = static_cast(json_utils::getField(flowNmos, "frame_width")); - auto const frameHeight = static_cast(json_utils::getField(flowNmos, "frame_height")); + // Create readers and configs for all video flows + for (auto i = std::size_t{0}; i < videoFlowIDs.size(); ++i) + { + auto const& flowID = videoFlowIDs[i]; + + MXL_INFO("Creating reader {} for flow ID: {}", i, flowID); + readers.emplace_back(domain, flowID); + + auto const flowDescriptor = readFlowDescriptor(domain, flowID); + auto const flowNmos = json_utils::parseBuffer(flowDescriptor); + + auto const interlaceMode = json_utils::getField(flowNmos, "interlace_mode"); + MXL_INFO(" Reader[{}] interlace_mode: {}", i, interlaceMode); + + if (interlaceMode != "progressive") + { + throw std::invalid_argument{"This application does not support interlaced flows."}; + } + + auto const grainRate = json_utils::getRational(flowNmos, "grain_rate"); + auto const frameWidth = static_cast(json_utils::getField(flowNmos, "frame_width")); + auto const frameHeight = static_cast(json_utils::getField(flowNmos, "frame_height")); + + MXL_INFO(" Reader[{}] resolution: {}x{} @ {}/{} fps", + i, + frameWidth, + frameHeight, + grainRate.numerator, + grainRate.denominator); + + configs.emplace_back(VideoPipelineConfig{ + .frameRate = grainRate, + .frameWidth = static_cast(frameWidth), + .frameHeight = static_cast(frameHeight), + .offset = playbackDelay + ((audioVideoOffset < 0) ? -audioVideoOffset : 0LL), + }); + } - auto videoConfig = VideoPipelineConfig{ - .frameRate = grainRate, - .frameWidth = static_cast(frameWidth), - .frameHeight = static_cast(frameHeight), - .offset = playbackDelay + ((audioVideoOffset < 0) ? -audioVideoOffset : 0LL), - }; + // Print all configs + MXL_INFO("Creating MultiviewerPipeline with {} configs:", configs.size()); + for (auto i = std::size_t{0}; i < configs.size(); ++i) + { + MXL_INFO(" Config[{}]: {}", i, configs[i].display()); + } - auto pipeline = VideoPipeline{videoConfig}; - reader.run(pipeline, readDelay); + auto pipeline = MultiviewerPipeline{configs}; + // Run all readers in parallel for multiviewer + auto readerThreads = std::vector{}; + MXL_INFO("Multiviewer pipeline started with {} video flows", videoFlowIDs.size()); + MXL_INFO("readers.size(): {}", readers.size()); + for (auto i = std::size_t{0}; i < readers.size(); ++i) + { + readerThreads.emplace_back([&readers, &pipeline, i, readDelay]() { readers[i].run(pipeline, i, readDelay); }); + } + MXL_INFO("readerThreads.size(): {}", readerThreads.size()); + for (auto& t : readerThreads) + { + t.join(); + } - MXL_INFO("Video pipeline finished"); - } - catch (std::exception const& e) - { - MXL_ERROR("Error while processing video pipeline: {}", e.what()); - } - catch (...) - { - MXL_ERROR("Encountered unknown error while processing video pipeline."); - } - }); + MXL_INFO("Multiviewer pipeline finished"); + } + catch (std::exception const& e) + { + MXL_ERROR("Error while processing multiviewer pipeline: {}", e.what()); + } + catch (...) + { + MXL_ERROR("Encountered unknown error while processing multiviewer pipeline."); + } + }); + } } if (!audioFlowID.empty()) @@ -1007,7 +1575,6 @@ namespace return 0; } - } int main(int argc, char* argv[])