我需要运行一个初始化成本较高的Python脚本（约0.5秒）。为了降低初始化时间，会对多个解释器进行池化，当需要执行时，从池中取出一个解释器，立刻运行给定的脚本。

def createInterpreters():
    for i in range(coreCount):
        createInterpreter().whenInitializationComplete(addToPool)

def runScript(code):
    interp = getFromPool()
    interp.runScript(code)

这种方法在一般情况下工作良好，但在脚本涉及UI（如PyQt、Tk）时会有一些问题，因为实际的解释器是一个C#进程，它通过pythonnet动态加载Python，进行昂贵的初始化，并将结果反馈给面向用户的应用程序。我想重写解释器的创建方式：启动一个Python进程，该进程会预加载C#基础设施并等待用户脚本。

initCsharp()
path = waitForScriptPath()
runScript(path)

从Python的角度来看，它应该像下面这样：

initCsharp()
path = waitForScriptPath()
runScript(path)

"Advanced search engines" recommend redirect stdin/out/err, write each line, flush, read output

proc = subprocess.Popen(
    ['python3'],  # Use 'python' if python3 isn't needed
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,  # Use text mode for easier string handling (Python 3.7+)
    bufsize=0   # Unbuffered for immediate interaction
)

init_script = "open('asd.txt', 'wb')" # perform side-effect to see whether it's working
proc.stdin.write(init_script + '\n')
proc.stdin.flush()
# file isn't created here, even if I wait for couple of seconds

user_script = "print(1)"
proc.stdin.write(user_script + '\n')
proc.stdin.flush()  # Critical: ensures line is sent immediately

proc.stdin.close()
proc.wait()

这种方法的问题在于，脚本只有在标准输入被关闭时才会运行，因此初始化是在用户启动脚本时就完成了。

# startup.py
import sys

sys.ps1 = ""
sys.ps2 = ""
sys.__interactivehook__ = lambda : ()

del sys

proc = subprocess.Popen(
    ['python3', '-q', '-i'],
    stdin=subprocess.PIPE,
    text=True,
    bufsize=0,
    env={ "PYTHONSTARTUP": "workdir/startup.py"})

code = """
print(1)
print(__name__)
raise Exception()
""".splitlines()

for line in code:
    proc.stdin.write(line + '\n')

proc.stdin.close()
proc.wait()

它确实可行，但抛出异常时，行号总是等于1。

所以问题是：如何在保留行信息的前提下，将初始化与实际脚本运行分离？

解决方案

你可以尝试 multiprocessing.Pool with initializer 这一专门用于

do costly init once then receive tasks

from multiprocessing import Pool
import runpy

def init_worker():# runs once per worker process before any task

def run_script(path):
    runpy.run_path(path, run_name="__main__")  # line numbers fully intact

# pool pre warms n processes with init already done
with Pool(processes=coreCount, initializer=init_worker) as pool:
    pool.apply(run_script, args=('/path/to/user_script.py',))

initCsharp() - 在每个工作进程开始处理任务之前执行一次

initializer参数指定一个在工作进程开始接受任务之前用于初始化每个工作进程的函数，这正是你需要的“预热”模式。

对于jupyer Kernals

# warmup.py
import sys

initCsharp()

script_path = sys.stdin.readline().strip()

# compile with the real filename #tracebacks show correct lines
with open(script_path) as f:
    source = f.read()

code = compile(source, script_path, 'exec')
exec(code, {'__name__': '__main__', '__file__': script_path})

initCsharp( - 在用户脚本到达之前，这里完成成本高昂的初始化