Comment puis-je récupérer la valeur de retour d'une fonction passée à multiprocessing.Process ?

Question

Dans l'exemple de code ci-dessous, j'aimerais récupérer la valeur de retour de la fonction worker . Comment puis-je m'y prendre ? Où cette valeur est-elle stockée ?

Exemple de code :

import multiprocessing

def worker(procnum):
    '''worker function'''
    print str(procnum) + ' represent!'
    return procnum

if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        jobs.append(p)
        p.start()

    for proc in jobs:
        proc.join()
    print jobs

Sortie :

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[<Process(Process-1, stopped)>, <Process(Process-2, stopped)>, <Process(Process-3, stopped)>, <Process(Process-4, stopped)>, <Process(Process-5, stopped)>]

Je ne parviens pas à trouver l'attribut correspondant dans les objets stockés dans la base de données de l'UE. jobs .

Answer 1

Il semble que vous devriez utiliser le multiprocessing.pool à la place et utiliser les méthodes .apply() .apply_async(), map()

http://docs.python.org/library/multiprocessing.html?highlight=pool#multiprocessing.pool.AsyncResult

Answer 2

Vous pouvez utiliser le exit intégré pour définir le code de sortie d'un processus. Il peut être obtenu à partir de la fonction exitcode attribut du processus :

import multiprocessing

def worker(procnum):
    print str(procnum) + ' represent!'
    exit(procnum)

if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        jobs.append(p)
        p.start()

    result = []
    for proc in jobs:
        proc.join()
        result.append(proc.exitcode)
    print result

Sortie :

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[0, 1, 2, 3, 4]

Answer 3

En caillou a une abstraction intéressante qui tire parti de multiprocessing.Pipe ce qui rend les choses assez simples :

from pebble import concurrent

@concurrent.process
def function(arg, kwarg=0):
    return arg + kwarg

future = function(1, kwarg=1)

print(future.result())

Exemple de : https://pythonhosted.org/Pebble/#concurrent-decorators

Answer 4

J'ai pensé simplifier les exemples les plus simples copiés de l'exemple ci-dessus, qui fonctionnent pour moi sur Py3.6. Le plus simple est multiprocessing.Pool :

import multiprocessing
import time

def worker(x):
    time.sleep(1)
    return x

pool = multiprocessing.Pool()
print(pool.map(worker, range(10)))

Vous pouvez définir le nombre de processus dans le pool avec, par exemple, Pool(processes=5) . Cependant, il s'agit par défaut du nombre de CPU, alors laissez-le vide pour les tâches liées au CPU. (Les tâches liées aux E/S conviennent souvent aux threads de toute façon, car les threads sont principalement en attente et peuvent donc partager un cœur de CPU). Pool s'applique également optimisation des jonctions .

(Notez que la méthode worker ne peut pas être imbriquée dans une autre méthode. J'ai initialement défini ma méthode worker à l'intérieur de la méthode qui fait l'appel à la méthode pool.map Mais les processus ne pouvaient pas l'importer et lançaient "AttributeError : Can't pickle local object outer_method..inner_method". Plus d'informations sur ici . Il peut être à l'intérieur d'une classe).

(Appréciez la question originale spécifiant l'impression 'represent!' plutôt que time.sleep() mais sans cela, je pensais que du code s'exécutait en même temps alors que ce n'était pas le cas).

---

Py3's ProcessPoolExecutor est également de deux lignes ( .map renvoie un générateur, vous avez donc besoin de l'option list() ):

from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor() as executor:
    print(list(executor.map(worker, range(10))))

---

Avec des Process es :

import multiprocessing
import time

def worker(x, queue):
    time.sleep(1)
    queue.put(x)

queue = multiprocessing.SimpleQueue()
tasks = range(10)

for task in tasks:
    multiprocessing.Process(target=worker, args=(task, queue,)).start()

for _ in tasks:
    print(queue.get())

Utilisez SimpleQueue si tout ce dont vous avez besoin est put y get . La première boucle démarre tous les processus, avant que la seconde n'effectue le blocage. queue.get appels. Je ne pense pas qu'il y ait une raison d'appeler p.join() aussi.

Answer 5

Une solution simple :

import multiprocessing

output=[]
data = range(0,10)

def f(x):
    return x**2

def handler():
    p = multiprocessing.Pool(64)
    r=p.map(f, data)
    return r

if __name__ == '__main__':
    output.append(handler())

print(output[0])

Sortie :

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]