Как я могу остановить родителя, если ребенок или работник переходит в исключение в мультиобработке?

Question

Я хочу добиться 2 вещей: 1) Поддерживать жизнь родителя, пока ребенок / работник не завершит работу, и 2) Если ребенок / работник переходит в исключение, тогда родитель должен остановиться после этого исключения. Не следует запускать оставшихся работников после исключения.

Я написал код, но не могу получить то, что ищу.

def waitTime(ds, index):
    try:
        if index == 3:
            index = 15
        ds[index] = 'WORKED!!!'
    except:
        global isException
        isException = True


if __name__ == '__main__':
    isException = False
    ds = multiprocessing.Manager().list([None]*5)
    new = []
    for i in range(1, len(ds)+1):
        temp = multiprocessing.Process(target=waitTime, args=(ds, i-1))
        new.append(temp)
        temp.start()
        temp.join()
        if isException:
            break

    print(ds)

мои ожидаемые результаты: ['WORKED !!!', 'WORKED !!!', 'РАБОТАЛИ !!!', Нет, Нет]

Мои реальные результаты: ['РАБОТАЛИ !!!', 'РАБОТАЛИ !!!', 'РАБОТАЛИ !!!', 'РАБОТАЛИ !!! ',' РАБОТАЛ !!! ']

Answer 1

Глобальная переменная не распределяется между многопроцессорными, но в потоке. Вы должны использовать Значение, Очередь, Блокировка, предоставляемые многопроцессорным модулем.

import multiprocessing
from multiprocessing import Value

v = Value('i', 0) # where i means integer type, 0 is default value


def waitTime(ds, index):
    try:
        if index == 3:
            index = 15
        ds[index] = 'WORKED!!!'
    except:
        v.value = 1 # v.value to assign integer


if __name__ == '__main__':
    isException = False
    ds = multiprocessing.Manager().list([None] * 5)
    new = []
    for i in range(1, len(ds) + 1):

       temp = multiprocessing.Process(target=waitTime, args=(ds, i - 1))
       new.append(temp)
       temp.start()
       temp.join()
       if v.value == 1:
           v.value = 0
           break

output:
['WORKED!!!', 'WORKED!!!', 'WORKED!!!', None, None]

Answer 2

У вас есть пара проблем здесь. Во-первых, вы на самом деле не вызываете исключение, поэтому предложение об исключении в вашем работнике никогда не срабатывает. Во-вторых, глобальные переменные не перемещаются между процессами - для этого нужен другой менеджер.

Учтите это:

def waitTime(ds, isException, index):
    try:
        if index == 3:
            index = 15
            raise IndexError
        ds[index] = 'WORKED!!!'
    except IndexError:
        isException.value = 1


if __name__ == '__main__':
    isException = multiprocessing.Manager().Value("h", 0)
    ds = multiprocessing.Manager().list([None]*5)
    new = []
    for i in range(1, len(ds)+1):
        temp = multiprocessing.Process(target=waitTime, args=(ds, isException, i-1))
        new.append(temp)
        temp.start()
        temp.join()
        if isException.value == 1:
            break

    print(ds)

Теперь мы вызываем IndexError и улавливаем это. Ваш isException теперь другой менеджер, используемый для переноса значения между вашими процессами. Это не обязательно очень элегантно, но это работает.