Объект процесса Python3 никогда не присоединяется

1. Ваша цель - выполнить эту работу асинхронно. Почему бы не порождать асинхронные рабочие процессы подпроцесса из вашего основного процесса БЕЗ порождения дочернего процесса (класс Test)? Результаты будут доступны в вашем основном процессе, и вам не нужно будет ничего делать. Вы можете прекратить читать здесь, если решите это сделать. В противном случае читайте дальше.

2. Ваше соединение работает вечно, потому что есть два отдельных пула: один при создании объекта процесса (локальный для вашего основного процесса), а другой, когда вы разветвляете процесс, вызывая process.start () (локальный для порожденного процесса)

Например, это не работает:

def __init__(self, arg, shared):
    super(Test, self).__init__()
    self.arg = arg
    self.quest = Quest()
    self.shared = shared
    self.pool = Pool()

def run(self):
    iterable = list(range(10))
    self.shared.extend(self.pool.map_async(self.quest.doIt, iterable, error_callback=print).get())
    print("1" + str(self.shared))
    self.pool.close()

Однако это работает:

def __init__(self, arg, shared):
    super(Test, self).__init__()
    self.arg = arg
    self.quest = Quest()
    self.shared = shared

def run(self):
    pool = Pool()
    iterable = list(range(10))
    self.shared.extend(pool.map_async(self.quest.doIt, iterable, error_callback=print).get())
    print("1" + str(self.shared))
    pool.close()

Это связано с тем фактом, что когда вы создаете процесс, весь код, стек и сегменты кучи вашего процесса клонируются в процесс, так что ваш основной процесс и подпроцесс имеют разные контексты.

Итак, вы вызываете join () для объекта пула, созданного локально для вашего основного процесса, и который вызывает close () для пула. Затем в run () есть еще один объект пула, который был клонирован в подпроцесс при вызове start (), и этот пул никогда не был закрыт и не может быть присоединен таким образом, как вы это делаете. Проще говоря, ваш основной процесс не имеет ссылки на клонированный объект пула в подпроцессе.

This works fine if I make Test a subclass of threading.Thread instead of multiprocessing.Process. I'm really not sure why this breaks it in half.

Имеет смысл, потому что потоки отличаются от процессов тем, что они имеют независимые стеки вызовов, но совместно используют другие сегменты памяти, поэтому любые обновления, которые вы делаете для объекта, созданного в другом потоке, видны в вашем основном процессе (который является родительским для этих потоков. ) и наоборот.

Решением является создание объекта пула, локального для функции run (). Закройте объект пула в контексте подпроцесса и присоединитесь к подпроцессу в основном процессе. Что подводит нас к пункту 2 ...

2. Совместное состояние: есть эти объекты multiprocessing.Manager (), которые допускают какое-то волшебное безопасное совместное состояние между процессами. Не похоже, что менеджер позволяет переназначать ссылки на объекты, что имеет смысл, потому что, если вы переназначаете управляемое значение в подпроцессе, когда подпроцесс завершается, этот контекст процесса (код, стек, куча) исчезает, и ваш основной процесс никогда не видит это присвоение (поскольку оно было сделано с помощью ссылки на объект, локальный в контексте подпроцесса). Однако это может работать для примитивных значений ctype.

Если кто-то более опытный с Manager () захочет вмешаться в его внутренности, это было бы круто. Но следующий код дает вам ожидаемое поведение:

#!/usr/bin/env python3
from multiprocessing.pool import ThreadPool as Pool
from multiprocessing import Process, Manager
from sys import stdout

class Quest():
    def __init__(self):
        pass

    def doIt(self, i):
        return i

class Test(Process):

    def __init__(self, arg, shared):
        super(Test, self).__init__()
        self.arg = arg
        self.quest = Quest()
        self.shared = shared

    def run(self):
        with Pool() as pool:
            iterable = list(range(10))
            self.shared.extend(pool.map_async(self.quest.doIt, iterable, error_callback=print).get())
            print("1" + str(self.shared)) # can remove, just to make sure we've updated state

    def __str__(self):
        return str(self.arg)

with Manager() as manager:
    res = manager.list()
    test = Test("test", res)

    print(test) # should print 'test' (and does)

    test.start()
    test.join()

    print("2" + str(res)) # should print '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]'

Выходы:

rpg711$ python multiprocess_async_join.py 
test
1[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
2[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]