multitraitement et Queues

Question

`Ce code est une tentative d'utiliser une file d'attente pour alimenter tâches à un certain nombre processus de travail .

Je voulais mesurer la différence de vitesse entre différents nombres de processus et différentes méthodes de traitement des données.

Mais la sortie ne fait pas ce que je pensais.

from multiprocessing import Process, Queue 
import time 
result = [] 

base = 2 

data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 23, 45, 76, 4567, 65423, 45, 4, 3, 21] 

# create queue for new tasks 
new_tasks = Queue(maxsize=0) 

# put tasks in queue 
print('Putting tasks in Queue') 
for i in data: 
    new_tasks.put(i) 

# worker function definition 
def f(q, p_num): 
    print('Starting process: {}'.format(p_num)) 
    while not q.empty(): 
     # mimic some process being done 
     time.sleep(0.05) 
     print(q.get(), p_num) 
    print('Finished', p_num) 

print('initiating processes') 
processes = [] 
for i in range(0, 2): 
    if __name__ == '__main__': 
     print('Creating process {}'.format(i)) 
     p = Process(target=f, args=(new_tasks, i)) 
     processes.append(p) 
#record start time 
start = time.time() 

# start process 
for p in processes: 
    p.start() 

# wait for processes to finish processes 
for p in processes: 
    p.join() 

#record end time 
end = time.time() 

# print time result 
print('Time taken: {}'.format(end-start)) 

Je attends ceci:

Putting tasks in Queue 
initiating processes 
Creating process 0 
Creating process 1 
Starting process: 1 
Starting process: 0 
1 1 
2 0 
3 1 
4 0 
5 1 
6 0 
7 1 
8 0 
9 1 
10 0 
11 1 
23 0 
45 1 
76 0 
4567 1 
65423 0 
45 1 
4 0 
3 1 
21 0 
Finished 1 
Finished 0 
Time taken: <some-time> 

Mais au lieu que je en fait obtenir ceci:

Putting tasks in Queue 
initiating processes 
Creating process 0 
Creating process 1 
Time taken: 0.01000523567199707 
Putting tasks in Queue 
Putting tasks in Queue 
initiating processes 
Time taken: 0.0 
Starting process: 1 
initiating processes 
Time taken: 0.0 
Starting process: 0 
1 1 
2 0 
3 1 
4 0 
5 1 
6 0 
7 1 
8 0 
9 1 
10 0 
11 1 
23 0 
45 1 
76 0 
4567 1 
65423 0 
45 1 
4 0 
3 1 
21 0 
Finished 0 

Il semble y avoir deux problèmes majeurs, je ne suis pas sûr lié ils sont:

1. Les états d'impression tels que: Putting tasks in Queue initiating processes Time taken: 0.0 sont répétées systématiquement si le code - je dis systématiquement becasue ils répètent exactement à chaque fois.

2. Le second processus ne se termine jamais, il reconnaît jamais la file d'attente est vide et ne permet donc pas quitter

Answer 1

1) Je ne peux pas reproduire ce.

2) Regardez le code suivant:

while not q.empty(): 
    time.sleep(0.05) 
    print(q.get(), p_num) 

Chaque ligne peut être exécuté dans un ordre quelconque par tout proces. Maintenant, considérons q ayant un seul élément et deux processus A et B. Considérons maintenant l'ordre suivant d'exécution:

# A runs 
while not q.empty(): 
    time.sleep(0.05) 

# B runs 
while not q.empty(): 
    time.sleep(0.05) 

# A runs 
print(q.get(), p_num) # Removes and prints the last element of q 

# B runs 
print(q.get(), p_num) # q is now empty so q.get() blocks forever 

Permutation l'ordre de time.sleep et q.get supprime le blocage dans toutes mes courses, mais il est encore possible d'avoir plus d'un processus entrent dans la boucle avec un seul élément gauche.

La façon de résoudre ce problème est d'utiliser un appel get non bloquant et attraper l'exception queue.Empty:

import queue 

while True: 
    time.sleep(0.05) 
    try: 
     print(q.get(False), p_num) 
    except queue.Empty: 
     break 

Answer 2

Vos threads de travail devrait être comme ceci:

def f(q, p_num): 
    print('Starting process: {}'.format(p_num)) 
    while True: 
     value = q.get() 
     if value is None: 
      break 
     # mimic some process being done 
     time.sleep(0.05) 
     print(value, p_num) 
    print('Finished', p_num) 

Et la file d'attente devrait être rempli de marqueurs après les données réelles:

for i in data: 
    new_tasks.put(i) 
for _ in range(num_of_threads): 
    new_tasks.put(None)