Python 诅咒 : multiprocessing issue with Pool. map ？

Question

我在使用 Pool.map 与 Python 的 curses 模块时遇到问题。每当我使用 Pool.map 计算更大的工作负载时，我的 curses UI 就会中断:它不再对默认屏幕的 getch 使用react。我可以按任意数量的键，直到按回车键，而不是立即读取任何按下的键(并继续解析它)。有时(除此之外)甚至 UI 也会中断(例如显示正常 shell 的一小部分)。

<小时/>

Curses UI 包装

这是一个包装类(Screen)，为我处理诅咒 UI 内容:

# -*- coding: utf-8 -*-
import curses



class Screen(object):

    def __init__(self):
        # create a default screen
        self.__mainscr = curses.initscr()
        self.__stdscr = curses.newwin(curses.LINES - 2, curses.COLS - 2, 1, 1)
        self.__max_height, self.__max_width = self.__stdscr.getmaxyx()

        # start coloring
        curses.start_color()
        curses.use_default_colors()

        # define colors
        curses.init_pair(1, 197, -1)  # red
        curses.init_pair(2, 227, -1)  # yellow
        curses.init_pair(3, curses.COLOR_MAGENTA, -1)
        curses.init_pair(4, curses.COLOR_GREEN, -1)  # darkgreen
        curses.init_pair(5, curses.COLOR_BLUE, -1)
        curses.init_pair(6, curses.COLOR_BLACK, curses.COLOR_WHITE)
        curses.init_pair(7, curses.COLOR_WHITE, -1)
        curses.init_pair(8, curses.COLOR_CYAN, -1)
        curses.init_pair(9, 209, -1)  # orange
        curses.init_pair(10, 47, -1)  # green


    def add_str_to_scr(self, add_str: str, colorpair: int = 7):
        self.__stdscr.addstr(str(add_str), curses.color_pair(colorpair))


    def linebreak(self):
        self.__stdscr.addstr("\n")


    def clear_screen(self):
        self.__stdscr.clear()


    def refresh_screen(self):
        self.__stdscr.refresh()


    def wait_for_enter_or_esc(self):
        curses.noecho()
        while True:
            c = self.__stdscr.getch()
            if c == 10 or c == 27:  # 10: Enter, 27: ESC
                break
        curses.echo()


    def get_user_input_chr(self) -> str:
        return chr(self.__stdscr.getch())


    def get_user_input_str(self) -> str:
        return self.__stdscr.getstr().decode(encoding="utf-8")

<小时/>

实际程序

我写了一个小例子，因为当我在 curses UI 中组合 Pool.map 且工作负载较高时，总是会发生上述失败。该代码只是在 numpy 数组上计算一些无用的 mult 和 add 内容。

import curses
from screen import Screen
from multiprocessing import Pool, cpu_count
import numpy as np

s = Screen()           # initializing my Screen wrapper
np.random.seed(1234)   # setting the rng fixed to make results comparable

# worker function to simulate workload
def worker(arr):
    return arr * 2 + 1

s.clear_screen()    # cleans the screen
s.refresh_screen()  # displays current buffer's content

s.add_str_to_scr("Start processing data...")
s.linebreak()
s.linebreak()
s.refresh_screen()

# data to feed worker function with (sliced by rows)
data_arr = np.random.rand(8, int(1e7))   # <-- big array for high workload

with Pool(cpu_count()) as p:
    buffer = p.map(worker, [data_arr[row] for row in np.ndindex(data_arr.shape[0])])

s.add_str_to_scr("...finished processing:")
s.linebreak()
s.linebreak()
s.refresh_screen()
for row in buffer:
    s.add_str_to_scr(row[0:3])
    s.linebreak()
    s.refresh_screen()

# *Here* the program should wait until the user presses *any* key
# and continue INSTANTLY when any key gets pressed.
# However, for big workloads, it does not react to single key presses,
# but wait for any amount of keys pressed until you hit 'Enter'
s.get_user_input_chr()
curses.endwin()

<小时/>

现在，当我以高工作负载执行代码时(即处理形状为 (8, int(1e7) 等于 8 行、10,000,000 列的数组)curse的 getch 中断，我得到以下行为:

正如你所看到的，我可以随意按 q (或任何其他键)，但是 curse 的 getch没有反应。我必须按 Enter 键才能使其识别输入。

此外，由于某种原因，第一行被我原始 shell 的输出覆盖。

只有当 Pool.map 的计算大约需要 1 秒或更长时间时，才会发生此行为。

当我将 data_arr 设置为像 np.random.rand(8, 100) 这样的小数组时，一切都像魅力一样工作，但是一旦我提供大数组当计算时间 >= 1 秒时，就会出现这个奇怪的错误并破坏我的 curses UI。

有什么想法吗？

Pool.map 是否未以某种方式正确加入工作进程？

Answer 1

该程序正在执行您告诉它执行的操作:

• 您正在调用initscr(并忽略了curses创建顶级窗口的事实)，
• 然后创建一个覆盖大部分屏幕的子窗口，
• 在屏幕上打印几行，每行后刷新显示，并且
• 终于结束了，等待子窗口的输入。

但是，您的程序不会调用 cbreak、raw 等，这些函数可以让您读取未缓冲的(无需按“Enter”键)字符。此外，该程序不会关闭echo。如果负载很轻，您不会注意到，因为响应速度很快。但在重负载下，例如交换或高内存/CPU 使用率，当出现提示时它仍然会恢复。所以你注意到了。

关于屏幕尺寸，也许你的意思是

self.__stdscr = curses.newwin(curses.LINES - 1, curses.COLS - 1, 0, 0)

但是假设您打算在窗口周围留下“空白”空间，您可以通过这样做来改进事情

        self.__mainscr.refresh()

紧接着 initscr (这会删除屏幕)。