Renderfarm开发适配：subprocess执行逻辑与多命令顺序需求

当前存在的问题 #

• 使用subprocess.Popen(self.data)可执行多条命令，但无执行顺序，测试函数Le_Printes()会立即触发，无法完成后续渲染文件迁移操作；
• 使用subprocess.run/call/check_call可顺序执行命令，但服务器仅支持单进程处理，一次只能处理一个任务，必须重启才能接收新命令。

需求 #

需要实现类似subprocess.Popen(self.data, afterFunction= MyTransferFunction())的回调功能，或者让subprocess.call支持多线程，同时接收并处理多条命令。

客户端代码 #

ShotName = MayaRenderPath.split("/")[-3]
DesktopMayaRenderPath = HostName + "/MayaRenders/" + ShotName
print("HostDesktopMayaRenderPathName = {}".format(DesktopMayaRenderPath))

RoyalCommand = Constant + Constant2 + DesktopMayaRenderPath + " " + FullMayaPath
print("RoyalCommand = {}".format(RoyalCommand))

##########################################
#os.system(RoyalCommand)
HOST, PORT = "10.100.20.139", 9999
#data = " ".join(sys.argv[1:])
data = RoyalCommand
dmrp = DesktopMayaRenderPath
MayaPateshestvie = r"{}".format(MayaRenderPath)

# 创建TCP Socket连接
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    # 连接服务器并发送数据
    sock.connect((HOST, PORT))
    sock.sendall(bytes(data, "utf-8"))
    sock.sendall(bytes("\n".join([dmrp, MayaPateshestvie]), "utf-8"))

    # 接收服务器响应
    received = str(sock.recv(2048), "utf-8")

print("Sent:     {}".format(data))
print("Received: {}".format(received))

##########################################
#def MyTransferFunction():
#for folder in os.listdir(DesktopMayaRenderPath):
#    shutil.move(DesktopMayaRenderPath + "/" + folder,MayaRenderPath)
#    shutil.move(os.path.join(DesktopMayaRenderPath,folder),MayaRenderPath)
#    try:
#        shutil.rmtree(os.path.join(MayaRenderPath,folder),ignore_errors=True)
#        shutil.move(os.path.join(DesktopMayaRenderPath,folder),MayaRenderPath)
#    except: 
#        pass

服务端代码 #

from asyncio import constants
from asyncio.subprocess import STDOUT
from operator import sub
import socketserver
import os,subprocess
class MyTCPHandler(socketserver.BaseRequestHandler):
    """
    服务器请求处理类，每个客户端连接实例化一次，
    必须重写handle()方法实现与客户端的通信逻辑。
    """
    def Le_Printes(self):
        print("Le Finished")

    def handle(self):
        # self.request是与客户端连接的TCP Socket
        self.data = self.request.recv(2048).strip().decode('utf-8')
        self.dmrp, self.mrp = [i for i in self.request.recv(2048).decode('utf-8').split('\n')]
        print("{} wrote:".format(self.client_address[0]))
        print("RoyalCommand = {}".format(self.data))
        print("DesktopMayaRenderPath = {}".format(self.dmrp))
        print("MayaRenderPath = {}".format(self.mrp))

        self.HostName = os.path.expanduser("~")
        self.HostName = self.HostName.replace("\\","/") + "/Desktop"
        print("HostName = {}".format(self.HostName))

        subprocess.Popen(self.data)
        self.Le_Printes()
        

if __name__ == "__main__":
    HOST, PORT = "10.100.20.139", 9999

    # 创建服务器，绑定到指定IP和端口
    with socketserver.TCPServer((HOST, PORT), MyTCPHandler) as server:
        # 启动服务器，持续运行直到Ctrl-C中断
        server.serve_forever()

解决方案 #

方案1：为Popen添加回调逻辑 #

通过线程等待Popen进程执行完毕，再触发文件迁移函数，既保留多进程处理能力，又保证操作顺序。修改服务端代码如下：

首先添加依赖和迁移函数：

import threading
import shutil

def MyTransferFunction(dmrp, mrp):
    """渲染完成后执行文件迁移"""
    for folder in os.listdir(dmrp):
        src_folder = os.path.join(dmrp, folder)
        dst_folder = os.path.join(mrp, folder)
        try:
            shutil.rmtree(dst_folder, ignore_errors=True)
            shutil.move(src_folder, mrp)
            print(f"已迁移文件夹: {src_folder} -> {mrp}")
        except Exception as e:
            print(f"迁移文件出错: {str(e)}")

def run_with_callback(cmd, callback, *args):
    """执行命令后调用回调函数"""
    proc = subprocess.Popen(cmd, shell=True)
    proc.wait()  # 等待命令执行完成
    callback(*args)

然后修改handle方法：

def handle(self):
    # 原有接收数据的代码保持不变

    # 启动独立线程执行命令和回调，不阻塞服务器接收新连接
    threading.Thread(
        target=run_with_callback,
        args=(self.data, MyTransferFunction, self.dmrp, self.mrp)
    ).start()
    self.Le_Printes()
    # 可选：向客户端发送任务已启动的响应
    self.request.sendall(b"任务已启动，将在完成后自动迁移文件")

方案2：使用多线程版TCPServer #

默认TCPServer是单线程的，改用ThreadingTCPServer即可同时处理多个客户端连接，配合subprocess.run实现单任务内的顺序执行：

修改服务端主函数：

if __name__ == "__main__":
    HOST, PORT = "10.100.20.139", 9999

    # 使用ThreadingTCPServer替代单线程TCPServer
    with socketserver.ThreadingTCPServer((HOST, PORT), MyTCPHandler) as server:
        server.serve_forever()

修改handle方法，使用subprocess.run顺序执行：

def handle(self):
    # 原有接收数据的代码保持不变

    try:
        # 顺序执行渲染命令，完成后执行迁移
        subprocess.run(self.data, shell=True, check=True)
        MyTransferFunction(self.dmrp, self.mrp)
        self.request.sendall(b"渲染及文件迁移完成")
    except subprocess.CalledProcessError as e:
        self.request.sendall(f"渲染命令执行失败: {str(e)}".encode('utf-8'))
    except Exception as e:
        self.request.sendall(f"任务执行出错: {str(e)}".encode('utf-8'))
    self.Le_Printes()

注意事项 #

• 使用shell=True时需确保命令来源安全，避免命令注入风险；如果命令由多个参数组成，建议拆分列表形式传入subprocess（如subprocess.Popen(["maya", "-render", ...])）。
• 方案1适合高并发场景，每个任务在独立线程中执行；方案2实现更简单，适合对任务执行顺序要求严格的场景。

内容的提问来源于stack exchange，提问作者Collegia Titanica