bmf是否有办法对中间的处理过程并行

[Serenagirl]

比如这个超分的例子里https://colab.research.google.com/github/BabitMF/bmf/blob/master/bmf/demo/video_enhance/bmf-enhance-demo.ipynb
我希望bmf可以把视频的所有帧提取出来（假设1s的视频30帧），然后我使用30个cpu同时对这30张图像做超分，然后再还原成视频，怎么实现？目前我的超分算法内部可以控制使用单核

[github-actions]

Thank you for your feedback! We have received your question and will deal with it later.
If any developer is interested in this issue, please leave a message "claim this issue" below the issue. Welcome to participate in open source co-construction!
We are very grateful to every contributor.
To communicate with official developers, you can download the Feishu app and scan the QR code to join the Feishu group！
Thank you again for your support!

感谢您的反馈！我们已经收到您的问题，并会在稍后处理。
如果有开发者对这个问题感兴趣，请在问题下方留言“认领此问题”。欢迎参与开源共建！
如需与官方开发者沟通，可以下载飞书应用并扫描二维码加入飞书群！
再次感谢您的支持！

[sfeiwong]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node
该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

[Serenagirl]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node 该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

我试了一下这个例子，但是报错：(python310) [root@localhost build]# ./cpp demo
[2025-03-25 14:29:34.488] [info] BMF Version: 0.0.12
[2025-03-25 14:29:34.488] [info] BMF Commit: 2b42a8a
[2025-03-25 14:29:34.488] [info] start init graph
[2025-03-25 14:29:34.488] [info] scheduler count7
debug queue size, node O, queue size: 5
[2025-03-25 14:29:34.488] [info] c_ffmpeg_decoder c++ /home/Lilu/bmf-master/bmf-master/output/bmf/lib/libbuiltin _modules.so
libbuiltin modules.CFFDecoder
ш-Jшq/JЭqsew-Jwq/n1T1/эWOU/ Japopa(H3O sa1npow uTa1TnqqT1 ++0 Japooap Badwyy o оJuT ainpoW [oJuT] [887 7E:67:VT_SZ-£O-SZOZ]
aster/output/bmf/lib/libbuiltin modules.so
[2025-03-25 14:29:34.490] [info] Constructing c++ module
terminate called after throwing an instance of 'bmf_sdk::Exception'
what(): BMF(0.0.12) /home/Lilu/bmf-master/bmf-master/bmf/c_modules/src/ffmpeg_decoder.cpp:823: error: (-224:BMF Transcod
e Error) avformat_open_input failed: No such file or directory in function 'init_input'
Aborted (core dumped) 我也没找到init_input在哪

[JackLau1222]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node 该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

我试了一下这个例子，但是报错：(python310) [root@localhost build]# ./cpp demo [2025-03-25 14:29:34.488] [info] BMF Version: 0.0.12 [2025-03-25 14:29:34.488] [info] BMF Commit: 2b42a8a [2025-03-25 14:29:34.488] [info] start init graph [2025-03-25 14:29:34.488] [info] scheduler count7 debug queue size, node O, queue size: 5 [2025-03-25 14:29:34.488] [info] c_ffmpeg_decoder c++ /home/Lilu/bmf-master/bmf-master/output/bmf/lib/libbuiltin _modules.so libbuiltin modules.CFFDecoder ш-Jшq/JЭqsew-Jwq/n1T1/эWOU/ Japopa(H3O sa1npow uTa1TnqqT1 ++0 Japooap Badwyy o оJuT ainpoW [oJuT] [887 7E:67:VT_SZ-£O-SZOZ] aster/output/bmf/lib/libbuiltin modules.so [2025-03-25 14:29:34.490] [info] Constructing c++ module terminate called after throwing an instance of 'bmf_sdk::Exception' what(): BMF(0.0.12) /home/Lilu/bmf-master/bmf-master/bmf/c_modules/src/ffmpeg_decoder.cpp:823: error: (-224:BMF Transcod e Error) avformat_open_input failed: No such file or directory in function 'init_input' Aborted (core dumped) 我也没找到init_input在哪

init_input可以在cpp_demo中手动设置哈

bmf/bmf/test/distributed_node/cpp_demo.cpp

Line 25 in 1dcb1de

{"input_path", "../../../files/big_bunny_10s_30fps.mp4"}

demo中用到的文件big_bunny_10s_30fps.mp4来自于https://github.com/BabitMF/bmf/releases/download/files/files.tar.gz

[Serenagirl]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node 该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

我试了一下这个例子，但是报错：(python310) [root@localhost build]# ./cpp demo [2025-03-25 14:29:34.488] [info] BMF Version: 0.0.12 [2025-03-25 14:29:34.488] [info] BMF Commit: 2b42a8a [2025-03-25 14:29:34.488] [info] start init graph [2025-03-25 14:29:34.488] [info] scheduler count7 debug queue size, node O, queue size: 5 [2025-03-25 14:29:34.488] [info] c_ffmpeg_decoder c++ /home/Lilu/bmf-master/bmf-master/output/bmf/lib/libbuiltin _modules.so libbuiltin modules.CFFDecoder ш-Jшq/JЭqsew-Jwq/n1T1/эWOU/ Japopa(H3O sa1npow uTa1TnqqT1 ++0 Japooap Badwyy o оJuT ainpoW [oJuT] [887 7E:67:VT_SZ-£O-SZOZ] aster/output/bmf/lib/libbuiltin modules.so [2025-03-25 14:29:34.490] [info] Constructing c++ module terminate called after throwing an instance of 'bmf_sdk::Exception' what(): BMF(0.0.12) /home/Lilu/bmf-master/bmf-master/bmf/c_modules/src/ffmpeg_decoder.cpp:823: error: (-224:BMF Transcod e Error) avformat_open_input failed: No such file or directory in function 'init_input' Aborted (core dumped) 我也没找到init_input在哪

init_input可以在cpp_demo中手动设置哈

bmf/bmf/test/distributed_node/cpp_demo.cpp

Line 25 in 1dcb1de

{"input_path", "../../../files/big_bunny_10s_30fps.mp4"}

demo中用到的文件big_bunny_10s_30fps.mp4来自于https://github.com/BabitMF/bmf/releases/download/files/files.tar.gz

好的谢谢，已经跑通了，另外请问下使用bmf的分布式并行，设置的dist_num原理是啥？我本身的视频只有100帧，开了dist_num=160的情况下竟然还需要跑多轮，时间比单核单实例的时间长了4倍左右

[JackLau1222]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node 该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

我试了一下这个例子，但是报错：(python310) [root@localhost build]# ./cpp demo [2025-03-25 14:29:34.488] [info] BMF Version: 0.0.12 [2025-03-25 14:29:34.488] [info] BMF Commit: 2b42a8a [2025-03-25 14:29:34.488] [info] start init graph [2025-03-25 14:29:34.488] [info] scheduler count7 debug queue size, node O, queue size: 5 [2025-03-25 14:29:34.488] [info] c_ffmpeg_decoder c++ /home/Lilu/bmf-master/bmf-master/output/bmf/lib/libbuiltin _modules.so libbuiltin modules.CFFDecoder ш-Jшq/JЭqsew-Jwq/n1T1/эWOU/ Japopa(H3O sa1npow uTa1TnqqT1 ++0 Japooap Badwyy o оJuT ainpoW [oJuT] [887 7E:67:VT_SZ-£O-SZOZ] aster/output/bmf/lib/libbuiltin modules.so [2025-03-25 14:29:34.490] [info] Constructing c++ module terminate called after throwing an instance of 'bmf_sdk::Exception' what(): BMF(0.0.12) /home/Lilu/bmf-master/bmf-master/bmf/c_modules/src/ffmpeg_decoder.cpp:823: error: (-224:BMF Transcod e Error) avformat_open_input failed: No such file or directory in function 'init_input' Aborted (core dumped) 我也没找到init_input在哪

init_input可以在cpp_demo中手动设置哈
bmf/bmf/test/distributed_node/cpp_demo.cpp
Line 25 in 1dcb1de
{"input_path", "../../../files/big_bunny_10s_30fps.mp4"}
demo中用到的文件big_bunny_10s_30fps.mp4来自于https://github.com/BabitMF/bmf/releases/download/files/files.tar.gz

好的谢谢，已经跑通了，另外请问下使用bmf的分布式并行，设置的dist_num原理是啥？我本身的视频只有100帧，开了dist_num=160的情况下竟然还需要跑多轮，时间比单核单实例的时间长了4倍左右

这个原理其实就是多线程并行计算 当dist_num设置大于逻辑处理器数量，效果就会逐渐递减，多出来的时间花在了线程调度上

[Serenagirl]

可以试试这个功能：https://github.com/BabitMF/bmf/tree/master/bmf/test/distributed_node 该功能主要实现了frame级别的分布处理，比较接近你想要的效果，当然了，你可以在你的算法模块中去控制选定cpu的策略。

我试了一下这个例子，但是报错：(python310) [root@localhost build]# ./cpp demo [2025-03-25 14:29:34.488] [info] BMF Version: 0.0.12 [2025-03-25 14:29:34.488] [info] BMF Commit: 2b42a8a [2025-03-25 14:29:34.488] [info] start init graph [2025-03-25 14:29:34.488] [info] scheduler count7 debug queue size, node O, queue size: 5 [2025-03-25 14:29:34.488] [info] c_ffmpeg_decoder c++ /home/Lilu/bmf-master/bmf-master/output/bmf/lib/libbuiltin _modules.so libbuiltin modules.CFFDecoder ш-Jшq/JЭqsew-Jwq/n1T1/эWOU/ Japopa(H3O sa1npow uTa1TnqqT1 ++0 Japooap Badwyy o оJuT ainpoW [oJuT] [887 7E:67:VT_SZ-£O-SZOZ] aster/output/bmf/lib/libbuiltin modules.so [2025-03-25 14:29:34.490] [info] Constructing c++ module terminate called after throwing an instance of 'bmf_sdk::Exception' what(): BMF(0.0.12) /home/Lilu/bmf-master/bmf-master/bmf/c_modules/src/ffmpeg_decoder.cpp:823: error: (-224:BMF Transcod e Error) avformat_open_input failed: No such file or directory in function 'init_input' Aborted (core dumped) 我也没找到init_input在哪

init_input可以在cpp_demo中手动设置哈
bmf/bmf/test/distributed_node/cpp_demo.cpp
Line 25 in 1dcb1de
{"input_path", "../../../files/big_bunny_10s_30fps.mp4"}
demo中用到的文件big_bunny_10s_30fps.mp4来自于https://github.com/BabitMF/bmf/releases/download/files/files.tar.gz

好的谢谢，已经跑通了，另外请问下使用bmf的分布式并行，设置的dist_num原理是啥？我本身的视频只有100帧，开了dist_num=160的情况下竟然还需要跑多轮，时间比单核单实例的时间长了4倍左右

这个原理其实就是多线程并行计算 当dist_num设置大于逻辑处理器数量，效果就会逐渐递减，多出来的时间花在了线程调度上

我试了一下这个和我需要的还是不一样，使用的是多线程，但是我想要的是多进程，一个cpu核绑定计算一张图像，而不是由线程分配，线程分配的话同一时间每个cpu的利用率是不一样的，占不满，我让ai写了一个方法，但是processed_video = graph.module这一段不确定参数是不是合理的，是否有文档可以看下都能加哪些参数

生成CPU绑定配置（每个物理核绑定两个逻辑线程）

cpu_groups = [f"{i2}-{i2+1}" for i in range(num_physical_cores)]

构建BMF Graph

graph = bmf.graph()

解码输入视频

input_stream = graph.decode({"input_path": input_video_path})

创建分布式处理模块

processed_video = graph.module(
[input_stream["video"]],
option={
"dist_num": num_physical_cores,
"scheduler": "local",
"input_manager": "server",
"module_confs": {
"cpu_groups": cpu_groups, # 改为字典结构
"worker_script": "process_worker.py",
"model_path": "/home/lilu/BasicSR-master/acl_fp161280.onnx"
},
"use_cmd": True # 启用命令行模式
}
)

编码输出

bmf.encode(
processed_video[0],
input_stream['audio'],
{
"video_params": {
"codec": "h264",
"width": 1920,
"height": 1080,
"crf": 23,
"preset": "fast"
},
"audio_params": {
"codec": "aac",
"bit_rate": 128000,
"sample_rate": 44100,
"channels": 2
},
"output_path": output_video_path
}
).run()

[sfeiwong]

“我试了一下这个和我需要的还是不一样，使用的是多线程，但是我想要的是多进程，一个cpu核绑定计算一张图像”

你可以基于这个功能，在你的模块中fork多进程就实现绑核处理了。

[Serenagirl]

在你的模块中fork多进程就实现绑核处理了

能举个例子吗？没太看懂

[sfeiwong]

1. 首先你需要写一个你的超分模块，或者用bmf demo中的enhance模块：https://github.com/BabitMF/bmf/blob/master/bmf/demo/video_enhance/enhance_module.py
2. 假如你使用了python模块，结合下面的多进程代码去改造一下原有的bmf模块：

import multiprocessing
import time
import psutil
import os
shared_counter = multiprocessing.Value('i', 0)

class EnhanceModule(Module):
    def __init__(self):
        self.task_queue = multiprocessing.Queue()
        self.cpu_core_id = -1
        self.fork_process = multiprocessing.Process(target=self.exe_procedure, args=(self.task_queue,))          
        self.fork_process.start()

    def exe_procedure(self, task_queue):
        if self.cpu_core_id == -1:
            current_process = psutil.Process(os.getpid())
            with shared_counter.get_lock():
                current_process.cpu_affinity([shared_counter.value])
                print(f"进程 {os.getpid()} 已绑定到CPU核心 {shared_counter.value}")
                self.cpu_core_id = shared_counter.value
                shared_counter.value += 1
        while True:
            task = task_queue.get()
            if task is None:
                break
            print(f"子进程 {multiprocessing.current_process().pid} 开始执行任务: {task}")
            # **你要执行的增强算法**
            time.sleep(2)
            print(f"子进程 {multiprocessing.current_process().pid} 完成任务: {task}")

    def process(self, task):
        self.task_queue.put(task)

    def stop(self):
        self.task_queue.put(None)
        self.fork_process.join()

    def clean(self):
        self.stop()

[Serenagirl]

1. 首先你需要写一个你的超分模块，或者用bmf demo中的enhance模块：https://github.com/BabitMF/bmf/blob/master/bmf/demo/video_enhance/enhance_module.py
2. 假如你使用了python模块，结合下面的多进程代码去改造一下原有的bmf模块：

import multiprocessing
import time
import psutil
import os
shared_counter = multiprocessing.Value('i', 0)

class EnhanceModule(Module):
    def __init__(self):
        self.task_queue = multiprocessing.Queue()
        self.cpu_core_id = -1
        self.fork_process = multiprocessing.Process(target=self.exe_procedure, args=(self.task_queue,))          
        self.fork_process.start()

    def exe_procedure(self, task_queue):
        if self.cpu_core_id == -1:
            current_process = psutil.Process(os.getpid())
            with shared_counter.get_lock():
                current_process.cpu_affinity([shared_counter.value])
                print(f"进程 {os.getpid()} 已绑定到CPU核心 {shared_counter.value}")
                self.cpu_core_id = shared_counter.value
                shared_counter.value += 1
        while True:
            task = task_queue.get()
            if task is None:
                break
            print(f"子进程 {multiprocessing.current_process().pid} 开始执行任务: {task}")
            # **你要执行的增强算法**
            time.sleep(2)
            print(f"子进程 {multiprocessing.current_process().pid} 完成任务: {task}")

    def process(self, task):
        self.task_queue.put(task)

    def stop(self):
        self.task_queue.put(None)
        self.fork_process.join()

    def clean(self):
        self.stop()

好的感谢，另外问一下bmf并行处理过程中出现segmentation fault(core dumped)可能是啥原因，我发现一样的代码有时候报错有时候不报错，有时候我设置dist_num=2也报错，htop查看memory远没有到瓶颈，硬件CPU核数也是dist_num两倍左右

[JackLau1222]

1. 首先你需要写一个你的超分模块，或者用bmf demo中的enhance模块：https://github.com/BabitMF/bmf/blob/master/bmf/demo/video_enhance/enhance_module.py
2. 假如你使用了python模块，结合下面的多进程代码去改造一下原有的bmf模块：

import multiprocessing
import time
import psutil
import os
shared_counter = multiprocessing.Value('i', 0)

class EnhanceModule(Module):
    def __init__(self):
        self.task_queue = multiprocessing.Queue()
        self.cpu_core_id = -1
        self.fork_process = multiprocessing.Process(target=self.exe_procedure, args=(self.task_queue,))          
        self.fork_process.start()

    def exe_procedure(self, task_queue):
        if self.cpu_core_id == -1:
            current_process = psutil.Process(os.getpid())
            with shared_counter.get_lock():
                current_process.cpu_affinity([shared_counter.value])
                print(f"进程 {os.getpid()} 已绑定到CPU核心 {shared_counter.value}")
                self.cpu_core_id = shared_counter.value
                shared_counter.value += 1
        while True:
            task = task_queue.get()
            if task is None:
                break
            print(f"子进程 {multiprocessing.current_process().pid} 开始执行任务: {task}")
            # **你要执行的增强算法**
            time.sleep(2)
            print(f"子进程 {multiprocessing.current_process().pid} 完成任务: {task}")

    def process(self, task):
        self.task_queue.put(task)

    def stop(self):
        self.task_queue.put(None)
        self.fork_process.join()

    def clean(self):
        self.stop()

好的感谢，另外问一下bmf并行处理过程中出现segmentation fault(core dumped)可能是啥原因，我发现一样的代码有时候报错有时候不报错，有时候我设置dist_num=2也报错，htop查看memory远没有到瓶颈，硬件CPU核数也是dist_num两倍左右

可能是个bug 我回头再测试一下