Python - Comment utiliser plusieurs processus pour résoudre le problème de l'imbrication lente des boucles ?

Question

Il existe un modèle de boucle dans une boucle.
Dans le corps de la boucle interne, les variables de la grande boucle et de la petite boucle doivent être utilisées en même temps.

Je l'ai simplifié ici en un modèle simple
Ce modèle sera super lent si la fonction est complexe
Je voudrais demander comment utiliser le multi-processus pour résoudre le problème de vitesse ?

Mon idée est d'utiliser le multi-processus uniquement pour les petites boucles,
écrivez le code multi-processus dans le corps de la boucle de la grande boucle,
mais il continue d'échouer,
veuillez demander à Dieu de donner le code correct.

Merci !

import random as r
list1=list(range(100))
i=0
reslist=[]
while i<2000:#大循环
    alist=[]#三个列表变量，每次循环开始时清空
    blist=[]
    clist=[]
    for each in list1:#小循环
        x=r.randint(i+30,i+60)+each#涉及到大、小循环变量的几个函数，这里用random示意
        y=r.randint(i+60,i+120)+each
        z=r.randint(i+60,i+180)+each
        
        res=2.5*x-y-z
        reslist.append(res)#对函数结果进行操作
        if res>=50:
            alist.append(each)
        if -50<res<50:
            blist.append(each)
        if res<=-50:
            clist.append(each)
            
    for each in alist:#在大循环中对小循环中得出的结果进行进一步其他操作
        print(each)
    for each in blist:
        print(each)
    for each in clist:
        print(each)
    
    i+=1

Answer 1

Tout d'abord, le calcul parallèle nécessite qu'il n'y ait pas de relation causale mutuelle entre les sous-programmes de diverses opérations parallèles.
Dans la petite boucle, res a une relation causale étroite avec x, y, z et alist, blist et clist, et il est difficile de les diviser en calculs parallèles.
Bien que le code posté par l'interrogateur ne soit pas le code original, je ne sais pas s'il existe une relation causale entre les grandes boucles dans le code original. Cependant, à en juger par le code schématique,
divisez la grande boucle en N threads. (aucun processus n'est nécessaire), cela devrait être Oui, chaque thread calcule 2000/N fois.
Par exemple, divisé en 8 threads, le thread 1 calcule i=0 à 249, le thread 2 calcule i=250 à 499, et ainsi de suite. . .
La taille de N ici peut être déterminée en fonction du nombre de cœurs de processeur. Si N dépasse le nombre de cœurs de processeur, cela n'aura pas beaucoup de sens, mais cela peut réduire l'efficacité.

Answer 2

Vous devriez utiliser elif au milieu. Il semble y avoir un problème avec l'indentation de for à la fin

.

Answer 3

Vous pouvez ouvrir plusieurs processus dans la grande boucle, par exemple, si la grande boucle est 2000 fois, si le nombre de cœurs de processeur est de 4, alors 4 processus seront ouverts et chaque processus sera responsable de l'exécution de 500

Une fois la petite boucle terminée, vous pouvez ouvrir un sous-thread pour effectuer les opérations suivantes suivantes, et la grande boucle continue de progresser

for each in alist:#在大循环中对小循环中得出的结果进行进一步其他操作
    print(each)
for each in blist:
    print(each)
for each in clist:
    print(each)

Answer 4

Vous pouvez utiliser des sous-processus pour gérer de petites boucles, mais dans ce cas, vous avez besoin de deux grandes boucles. Une boucle gère la petite boucle, et une fois cette boucle traitée, une grande boucle gère les choses suivantes

Comme ça

import random as r


def cumput(i, list1):
    alist = []
    blist = []
    clist = []
    reslist = []
    for each in list1:  # 小循环
        x = r.randint(i + 30, i + 60) + each  # 涉及到大、小循环变量的几个函数，这里用random示意
        y = r.randint(i + 60, i + 120) + each
        z = r.randint(i + 60, i + 180) + each

        res = 2.5 * x - y - z
        reslist.append(res)  # 对函数结果进行操作
        if res >= 50:
            alist.append(each)
        if -50 < res < 50:
            blist.append(each)
        if res <= -50:
            clist.append(each)
    return alist, blist, clist, reslist


if __name__ == '__main__':
    multiprocessing.freeze_support()
    list1 = list(range(100))
    i = 0
    pool = multiprocessing.Pool(2)
    res = {}
    while i < 2000:  # 大循环
        res[i]=pool.apply_async(cumput, (i, list1,))
        i += 1
    pool.close()
    pool.join()
    for i in res:
        for each in res[i].get()[0]:  # 在大循环中对小循环中得出的结果进行进一步其他操作
            print(each)
        for each in res[i].get()[1]:
            print(each)
        for each in res[i].get()[2]:
            print(each)

Answer 5

Si les fonctions exécutées dans la petite boucle prennent du temps, vous pouvez considérer le modèle producteur-consommateur

import random
from threading import Thread
from Queue import Queue

resqueue = Queue()
aqueue = Queue()
bqueue = Queue()
cqueue = Queue()

def producer():
    list1=list(range(100))
    
    for _ in range(2000):
        for each in list1:
            x=r.randint(i+30,i+60)+each
            y=r.randint(i+60,i+120)+each
            z=r.randint(i+60,i+180)+each
            
            res=2.5*x-y-z
            resqueue.put(res)
            
            if res>=50:
                aqueue.put(each)
            if -50<res<50:
                bqueue.put(each)
            if res<=-50:
                cqueue.put(each)

def consumer_a():
    while True:
        try:
            data = aqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            aqueue.task_done()
            
def consumer_b():
    while True:
        try:
            data = bqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            bqueue.task_done()
            
 def consumer_c():
    while True:
        try:
            data = cqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            cqueue.task_done()

 def consumer_res():
    while True:
        try:
            data = resqueue.get(timeout=5)
        except Queue.Empty:
            return
        else:
            # 耗时操作
            deal_data(data)
            resqueue.task_done()
            
if __name__ == "__main__":
    t1 = Thread(target=producer)
    t2 = Thread(target=consumer_a)
    ...
    
    t1.start()
    t2.start()
                       

Answer 6

Le questionneur doit-il d'abord concevoir l'entrée et la sortie du processus ? Si plusieurs processus effectuent du calcul parallèle, la communication entre les processus est la plus importante, pour autant que je sache, elle devrait être MPI, comme les boucles multicouches, et une partie. des données doivent être distribuées en premier. Accédez à chaque processus, chaque processus effectue des calculs, puis retourne au point d'intégration des données, puis fusionne les résultats et les génère.

Un autre point important est d'estimer le temps d'exécution de chaque processus. Après tout, s'il y a une communication inter-processus, le temps d'attente entraînera également une diminution de l'efficacité.

@daijianke a déclaré que votre imbrication n'est pas conforme aux règles de saisie du calcul parallèle. Vous pouvez jeter un œil à cet exemple

.

http://blog.csdn.net/zouxy09/...

J'ai déjà testé les exemples de l'article et il n'y a aucun problème si vous suivez ces étapes, vous devriez pouvoir le faire

.