multiprocessing race condition on flushing stdout, deadlocks child on exit
Nadie ha tomado este issue todavía.
- Lenguaje dominante
- Python
- Estrellas
- 77.2k
- Forks
- 36k
- Métricas de merge de PR
- Métricas de PR pendientes
Descripción
I experienced deadlocks when using the logging example for multiprocessing, where a QueueHandler is used.
I found that sometimes a multiprocessing process is not terminating, when it has put an element in a Queue,
even if the parent process runs a thread to empty the queue and successfully retrieved the item.
The worker looks like this:
def worker(q):
q.put(current_process().name)
return
and while this works:
workers = []
for i in range(15):
p = Process(target=worker, args=(logq,), name=f"Worker {i+1}")
workers.append(p)
for p in workers:
p.start()
time.sleep(.1)
removing the sleep leads to a very high propability of deadlocking when I then try to join the processes, e.g,:
for w in workers:
print(f'trying to join on {w.name}, alive={w.is_alive()}, exitcode={w.exitcode}', w.name, w.is_alive(), w.exitcode)
w.join()
The Process is still marked as alive, hitting Ctrl+C gives this:
trying to join on Worker 14, alive=True, exitcode=None Worker 14 True None
^CTraceback (most recent call last):
File "/home/fls/pybug/deadlock.py", line 43, in <module>
w.join()
File "/usr/lib/python3.10/multiprocessing/process.py", line 149, in join
res = self._popen.wait(timeout)
File "/usr/lib/python3.10/multiprocessing/popen_fork.py", line 43, in wait
return self.poll(os.WNOHANG if timeout == 0.0 else 0)
File "/usr/lib/python3.10/multiprocessing/popen_fork.py", line 27, in poll
pid, sts = os.waitpid(self.pid, flag)
KeyboardInterrupt
Can reproduce using Python 3.9.10 and 3.10.4 on Linux:
import time
import threading
from multiprocessing import Process, Queue, current_process
def logger_thread(q: Queue):
while True:
record = q.get()
if record is None:
break
print("logrecord: ", record)
def worker(q):
q.put(current_process().name)
return
logq = Queue()
lp = threading.Thread(target=logger_thread, args=(logq,), daemon=True)
lp.start()
workers = []
for i in range(15):
p = Process(target=worker, args=(logq,), name=f"Worker {i+1}")
workers.append(p)
print("starting workers")
for p in workers:
p.start()
# no deadlock when added:
# time.sleep(.1)
print("waiting a bit")
time.sleep(1)
print("trying to join workers")
for w in workers:
print(f'trying to join on {w.name}, alive={w.is_alive()}, exitcode={w.exitcode}', w.name, w.is_alive(), w.exitcode)
w.join()
print(f'joined on {w.name}', w.name)
logq.put(None)
lp.join()
Guía de contribución
Primeros pasos
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- Abre un pull request que haga referencia al número del issue.
Línea de trabajo
Comience ejecutando la reproducción proporcionada de multiprocessing en Linux con las versiones de Python indicadas y, a continuación, inspeccione multiprocessing Queue y la ruta de salida de los procesos hijo, incluidos los archivos multiprocessing/process.py y multiprocessing/popen_fork.py mostrados en el traceback. Se considerará terminado cuando los workers hagan join de forma fiable después de recuperar los elementos de la Queue, con cobertura de regresión para el caso reproducido.
Escrito por el modelo de indexación a partir del texto del issue.
Evaluación
- Stack tecnológico
- python
- Área
- distributed-systems
- Tipo de issue
- Error
- Dificultad
- 4/5
- Tiempo estimado
- 3-5 días
- Estado de actividad
- Estancado
- Claridad
- Necesita aclaración
- Aptitud para principiantes
- 35/100