python / python/cpython

multiprocessing.Barrier does not return if waiting process is terminated (tested on windows)

Aperta
#123,899 14 commenti 0 reazioni 1 assegnatario Vedi su GitHub

@Zheaoli ci sta già lavorando.

Dal 10/9/2024.

topic-multiprocessing type-bug
Lingua principale
Python
Stelle
77.2k
Fork
35.9k
Metriche di merge delle PR
Metriche PR in attesa

Descrizione

Bug report

Bug description:

I'm using multiprocessing.Barrier to synchronize processes. The individual processes might get terminated/killed by a GUI user interaction.
In case that a process, that already entered the waiting state, is killed, the barrier will never be released, a timeout is not taken into account.
When debugging this issue I realized that it is cause by https://github.com/python/cpython/blob/b52de7e02dba9e1f176d6d978d782fbd0509311e/Lib/multiprocessing/synchronize.py#L297

self._woken_count is not released by the terminated process and no timeout is specified in the call to self._woken_count.acquire()

If I add a (hardcoded...) timeout to self._woken_count.acquire(True, 2.0) the program continues as expected.
I hope that there a better approaches to fix this than using a hardcoded timeout...

Example script to reproduce error.

import logging
import time
import sys
import multiprocessing as mp
import multiprocessing.synchronize

from pathlib import Path

def process_main(process_index, barrier: multiprocessing.synchronize.Barrier):
    logging.debug('Process %d: Started', process_index)
    time.sleep(0.5)
    logging.debug('Process %d: Waiting for barrier', process_index)
    barrier.wait(timeout=5.0)
    logging.debug('Process %d: Barrier passed', process_index)
    time.sleep(0.5)
    logging.debug('Process %d: Terminated', process_index)

if __name__ == '__main__':
    # Set up logging
    logfile_name = Path(__file__).with_suffix('.log').name
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s %(levelname)s:%(name)s %(message)s',
        handlers=[
            logging.FileHandler(logfile_name, mode='w'),
            logging.StreamHandler(sys.stdout)
        ]    
    )

    instance_count = 4
    barrier = mp.Barrier(instance_count)

    processes = []
    for i in range(instance_count):
        runner_process = mp.Process(
            target=process_main, args=(i, barrier), daemon=True)
        processes.append(runner_process)

    for i, process in enumerate(processes):
        logging.debug('Starting process %d', i)
        process.start()
        time.sleep(0.200)

    # Terminate already waiting process
    logging.debug('Killing process 0')
    processes[0].kill()

    for process in processes:
        process.join()


CPython versions tested on:

3.12

Operating systems tested on:

Windows

Linked PRs
  • gh-125578

Guida per i contributori

Apri la guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.