python - 如何使用带有 Mutliprocessing 的 PYTHON 打开程序，并从主进程向其发送字符串？

Question

我有一个程序使用类似于以下内容的方式将帧作为字符串发送到 FFMPEG:

当前在 Ubuntu 上不使用多处理模块进行流式传输的工作脚本

#!/usr/bin/python
import sys, os
import subprocess as sp
import pygame
from pygame.locals import QUIT, KEYUP, K_ESCAPE
import pygame.display

pygame.init()
os.environ['SDL_VIDEODRIVER'] = 'dummy'
pygame.display.init()
Display_Surface = pygame.display.set_mode([1280,720], 0, 32)

# FFMPEG command and settings
command = ['ffmpeg', '-framerate', '25', '-s', '1280x720', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
           '-f', 'lavfi', '-i', 'anullsrc=cl=mono',
           '-pix_fmt', 'yuv420p','-s', 'hd720', '-r', '25', '-g', '50',
           '-f', 'flv', 'rtmp://a.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx']

pipe = sp.Popen(command, bufsize=0, stdin=sp.PIPE)

while True:
    # Quit event handling
    for event in pygame.event.get():
        if event.type == QUIT or (event.type == KEYUP and event.key == K_ESCAPE):
            pygame.quit()
            sys.exit()

    pipe.stdin.write(pygame.image.tostring(Display_Surface, "RGBA"))

pipe.stdin.close()
pygame.display.quit()
os._exit()

这工作正常，除了它会占用我的 CPU，这反过来会导致我的实时流经常卡住。愚蠢的 GIL 不会让 FFMPEG 在另一个 CPU/核心上运行，而我有 3 个非常好的核心什么都不做。

我刚刚编写了一些代码以在另一个进程中打开 FFMPEG。 (顺便说一句，我熟悉 threading.Thread，但不熟悉 Multiprocessing)。

import os
import subprocess as sp
import multiprocessing

class FFMPEG_Consumer():

    def __init__(self):
        proc = multiprocessing.Process(target=self.start_ffmpeg)
        proc.start()

    def start_ffmpeg(self):
        command = ['ffmpeg','-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
                   '-f, 'lavfi', '-i', 'anullsrc=channel_layout=stereo:sample_rate=44100',
                   '-pix_fmt', 'yuv420p','-s', 'hd720', '-f', 'flv', 'rtmp://example.com']

        pipe = sp.Popen(command, bufsize=-1, stdin=sp.PIPE)

    def send_down_the_pipe(self, frame):
        pipe.stdin.write(frame)

ffmpeg = FFMPEG_Consumer()

对于任何知道如何使用多处理的人，我相信您会立即发现这是行不通的，因为我不能以这种方式跨进程共享变量。但是，它确实在另一个核心上打开了 FFMPEG。

大多数在线教程和资源都侧重于创建工作人员池和队列，以向这些工作人员发送需要处理的内容，直到作业完成。然而，我试图通过每次迭代向 FFMPEG 重复发送一个新字符串。

如何将我的字符串通过管道传输到 FFMPEG 的进程/实例？

或者我想做的是不可能的？

这是可行的解决方案(简化了 FFMPEG 设置):

#!/usr/bin/python
import sys, os, multiprocessing
import subprocess as sp
import pygame
from pygame.locals import QUIT, KEYUP, K_ESCAPE
import pygame.display

pygame.init()
os.environ['SDL_VIDEODRIVER'] = 'dummy'
pygame.display.init()
Display_Surface = pygame.display.set_mode([1280,720], 0, 32)

class FFMPEGConsumer(object):
    def __init__(self):
        self._r, self._w = multiprocessing.Pipe()
        self.reader = os.fdopen(self._r.fileno(), 'r')
        self.writer = os.fdopen(self._w.fileno(), 'w', 0)
        self.proc = None

    def start_ffmpeg(self):

        command = ['ffmpeg', '-framerate', '25', '-s', '1280x720', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
           '-f', 'lavfi', '-i', 'anullsrc=cl=mono',
           '-pix_fmt', 'yuv420p','-s', 'hd720', '-r', '25', '-g', '50',
           '-f', 'flv', 'rtmp://a.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx']
        
        self.proc = sp.Popen(command, bufsize=-1, stdin=self.reader)

    def send_down_the_pipe(self, frame):
        self.writer.write(frame)
        #print self._stdin.read()

    def __del__(self):
        self.reader.close()
        self.writer.close()

ffmpeg = FFMPEGConsumer()

while True:
    # Quit event handling
    for event in pygame.event.get():
        if event.type == QUIT or (event.type == KEYUP and event.key == K_ESCAPE):
            pygame.quit()
            sys.exit()

    ffmpeg.send_down_the_pipe(pygame.image.tostring(Display_Surface, "RGBA"))
    proc.join()

pipe.stdin.close()
pygame.display.quit()
os._exit()

所有核心都在发射，到目前为止没有滞后!!!

Answer 1

您可以使用 multiprocessing.Pipe 与子进程通信。

multiprocess.Pipe 为您提供管道的两端，其中一端可用于写入，另一端用于读取，反之亦然(如果双工标志设置为 True)。这两端的事情是那些只是文件描述符。这就是为什么你需要打开它们，就像文件一样，然后读/写。要打开它们，您必须使用 os.fdopen。从此就像读/写文件一样。

像这样的东西应该有帮助:

import os
import subprocess as sp
import multiprocessing



class FFMPEG_Consumer():

    def __init__(self):
        r, w = multiprocessing.Pipe()
        self._stdin = os.fdopen(r.fileno(), 'r')
        self._stdout = os.fdopen(w.fileno(), 'w')
        proc = multiprocessing.Process(target=self.start_ffmpeg)
        proc.start()

    def start_ffmpeg(self):
        command = ['ffmpeg', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-', '-f',
                   'lavfi', '-i', 'anullsrc=channel_layout=stereo:sample_rate=44100',
                   '-pix_fmt', 'yuv420p', '-s', 'hd720', '-f', 'flv', 'rtmp://example.com']

        sp.Popen(command, bufsize=-1, stdin=self._stdin)

    def send_down_the_pipe(self, frame):
        self._stdout.write(frame)


ffmpeg = FFMPEG_Consumer()
ffmpeg.send_down_the_pipe(frame)

编辑 1.

您可以调整缓冲以更好地满足您的需求。当您使用 os.fdopen 时，第三个参数是缓冲区大小。 0是完全不缓冲打开

编辑2.

我不想出于教育目的删除旧版本。我之前的脚本有什么问题？

class FFMPEG_Consumer():

    def __init__(self):
        r, w = multiprocessing.Pipe()
        self._stdin = os.fdopen(r.fileno(), 'r')
        self._stdout = os.fdopen(w.fileno(), 'w')
        ...

__init__ 方法执行后，r 和w 文件描述符被垃圾回收。现在，您有 self._stdin 和 self._stdout 仍然存在并指向不存在的文件描述符。实例化 FFMPEG_Consumer 后，您会收到以下错误:

close failed in file object destructor:
IOError: [Errno 9] Bad file descriptor
close failed in file object destructor:
IOError: [Errno 9] Bad file descriptor

如何解决？您必须确保在实例化 FFMPEG_Consumer 之后保留文件描述符。

在展示如何解决这个问题的示例脚本之前，让我给你几个建议:

1. 你说你使用 Python2.7。这就是为什么您在定义类时可能需要继承 object 的原因。您将在示例脚本中看到它。
2. 命名类时，惯例是使用驼峰式大小写。所以 FFMPEGConsumer 优于 FFMPEG_Consumer。

现在进入工作示例。

工作示例

这是我使用的两个脚本，可以让任何人在不使用 ffmpeg 的情况下测试它的工作原理。 注意:此为Python2.7版本

脚本 so6_reader.py 需要用户输入:

#!/usr/bin/env python2.7

if __name__ == "__main__":
    for i in range(2):
        a = raw_input('Enter the string: ')
        print 'You entered %s' % a

启动子进程的脚本，该子进程启动 so6_reader.py 并写入其标准输入。

#!/usr/bin/env python2.7
import os
import multiprocessing
import subprocess as sp


class FFMPEGConsumer(object):

    def __init__(self):
        self._r, self._w = multiprocessing.Pipe()
        self.reader = os.fdopen(self._r.fileno(), 'r')
        self.writer = os.fdopen(self._w.fileno(), 'w', 0)
        self.proc = None

    def start_ffmpeg(self):
        command = ['python', 'so6_reader.py']
        self.proc = sp.Popen(command, bufsize=0, stdin=self.reader)

    def send_down_the_pipe(self, frame):
        self.writer.write(frame)

    def __del__(self):
        self.reader.close()
        self.writer.close()


ffmpeg = FFMPEGConsumer()

proc = multiprocessing.Process(target=ffmpeg.start_ffmpeg)
proc.start()

ffmpeg.send_down_the_pipe('test 001\n')
ffmpeg.send_down_the_pipe('test 002\n')

proc.join()