Python 多处理管理器在烧瓶 API 中使用时显示错误

Posted 2023-02-16

【中文标题】Python 多处理管理器在烧瓶 API 中使用时显示错误

【英文标题】：Python multiprocessing manager showing error when used in flask API

【发布时间】：2021-12-05 12:59:35

【问题描述】：

我对做我想做的事情的最佳方式感到很困惑。

我想要什么？

对烧瓶应用程序的 API 调用 Flask 路由使用 Process 模块启动 4-5 个多进程，并使用共享的 Managers().list() 组合结果（在切片的 pandas 数据帧上） 将计算结果返回给客户端。

我的实现：

pos_iter_list = get_chunking_iter_list(len(position_records), 10000)

manager = Manager()
data_dict = manager.list()
processes = []
for i in range(len(pos_iter_list) - 1):
    temp_list = data_dict[pos_iter_list[i]:pos_iter_list[i + 1]]
    p = Process(
        target=transpose_dataset,
        args=(temp_list, name_space, align_namespace, measure_master_id, df_searchable, products,
              channels, all_cols, potential_col, adoption_col, final_segment, col_map, product_segments,
              data_dict)
    )
    p.start()
    processes.append(p)
for p in processes:
    p.join()

我的目录结构：

- main.py(flask entry point)
- helper.py(contains function where above code is executed & calls transpose_dataset function)

我在运行时遇到的错误？ RuntimeError：找不到提供的模块“mp_main”的根路径。这可能是因为模块来自不提供文件名信息的导入钩子，或者因为它是一个命名空间包。在这种情况下，需要显式提供根路径。

不确定这里出了什么问题，当使用 if __name__ == '__main__': 从 sample.py 文件调用管理器列表时，管理器列表工作正常

更新：同一段代码在我的 MacBook 上运行良好，但在 windows 操作系统上却没有。

烧瓶 API 调用示例：

@app.route(PREFIX + "ping", methods=['GET'])
def ping():
    man = mp.Manager()
    data = man.list()
    processes = []
    for i in range(0,5):
        pr = mp.Process(target=test_func, args=(data, i))
        pr.start()
        processes.append(pr)

    for pr in processes:
        pr.join()

    return json.dumps(list(data))

【问题讨论】：

完整的回溯会很有帮助。当使用“spawn”与“fork”时，这几乎可以肯定是关于可导入性的代码布局/结构问题（我假设你的 macos 有一个稍微旧的 python 版本仍然默认为“fork”）

你提到你正在使用if __name__ == "__main__":，但你必须确保基本上除了函数和类定义之外的所有内容都在其中。听起来有点像烧瓶试图在子进程中启动一个新的服务器实例，但失败了。

如果您使用 docs quickstart 中的样板文件，可能归结为 app = Flask(__name__).. 将它和您所有的 @app.route 函数定义放在 if __name__ == "__main__": 块中，以防止它尝试在子节点上构建另一个服务器流程导入。 可能可以更改为app = Flask("__main__")，但不知道这意味着什么。

@Aaron 这对我有用但是，理解为什么子进程尝试再次启动烧瓶应用程序有点令人困惑？ MacOS 不会发生同样的情况。

【参考方案1】：

Stack 有一个持续存在的错误，阻止我发表评论，所以我会写一个答案..

Python 有两种（主要）方式来启动新进程：“spawn”和“fork”。 fork 是仅在 *nix 中可用的系统命令（阅读：linux 或 macos），因此 spawn 是 windows 中的唯一选项。 3.8 之后 spawn 将是 MacOS 上的默认设置，但 fork 仍然可用。最大的不同是 fork 基本上复制了现有进程，而 spawn 启动了一个全新的进程（就像只是打开一个新的 cmd 窗口）。为什么和如何有很多细微差别，但是为了能够运行您希望子进程使用 spawn 运行的功能，子进程必须 import 主文件。导入文件就等于只执行该文件，然后通常将其命名空间绑定到变量：import flask 将运行 flask/__ini__.py 文件，并将其全局命名空间绑定到变量 flask。然而，通常有代码仅由主进程使用，不需要在子进程中导入/执行。在某些情况下，再次运行该代码实际上会破坏事情，因此您需要阻止它在主进程之外运行。考虑到这一点，“魔术”变量__name__ 仅等于主文件中的"__main__"（而不是在子进程中或在导入模块时）。

在您的特定情况下，您正在创建一个新的app = Flask(__name__)，它会在您运行服务器之前进行一些验证和检查。这是从子进程运行时遇到的这些设置/验证步骤之一。通过根本不让它运行来修复它是更清洁的解决方案，但您也可以通过给它一个不会绊倒的值来修复它，然后永远不要启动该辅助服务器（再次通过使用 if __name__ == "__main__": 保护它)