python multitraitement - processus se bloque sur rejoindre pour la grande file d'attente

Je suis en cours d'exécution python 2.7.3 et j'ai remarqué que le suivant comportement étrange. Considérons cet exemple minimal:

from multiprocessing import Process, Queue

def foo(qin, qout):
    while True:
        bar = qin.get()
        if bar is None:
            break
        qout.put({'bar': bar})

if __name__ == '__main__':
    import sys

    qin = Queue()
    qout = Queue()
    worker = Process(target=foo,args=(qin,qout))
    worker.start()

    for i in range(100000):
        print i
        sys.stdout.flush()
        qin.put(i**2)

    qin.put(None)
    worker.join()

Quand je boucle de plus de 10 000 habitants ou plus, mon script se bloque sur worker.join(). Il fonctionne très bien lorsque la boucle ne va que jusqu'à 1 000.

Des idées?

InformationsquelleAutor user545424 | 2014-02-08
  1. 32

    La qout file d'attente dans le processus secondaire est pleine. Les données que vous mettez dans de foo() ne rentre pas dans la mémoire tampon de l'OS, des tuyaux utilisés en interne, de sorte que le sous-processus blocs en essayant d'intégrer plus de données. Mais le processus parent n'est pas la lecture de ces données: il est tout simplement bloqué aussi, en attente pour le processus secondaire à la fin. Il s'agit d'une impasse.

    • Ce serait super si vous aussi offert un code solution au problème. I. e. comment vider la mémoire tampon, de sorte que le sous-processus ne bloque pas.
    InformationsquelleAutor Armin Rigo
  2. 4

    Il doit y avoir une limite sur la taille des files d'attente. Envisager la modification suivante:

    from multiprocessing import Process, Queue

def foo(qin,qout):
    while True:
        bar = qin.get()
        if bar is None:
            break
        #qout.put({'bar':bar})

if __name__=='__main__':
    import sys

    qin=Queue()
    qout=Queue()   ## POSITION 1
    for i in range(100):
        #qout=Queue()   ## POSITION 2
        worker=Process(target=foo,args=(qin,))
        worker.start()
        for j in range(1000):
            x=i*100+j
            print x
            sys.stdout.flush()
            qin.put(x**2)

        qin.put(None)
        worker.join()

    print 'Done!'

    Cela fonctionne-est (avec qout.put ligne en commentaire). Si vous essayez d'enregistrer tous les 100000 résultats, puis qout devient trop gros: si je décommentez la qout.put({'bar':bar}) dans foo, et de laisser la définition de qout en POSITION 1, le code se bloque. Si, toutefois, je me déplace qout définition de la POSITION 2, puis la fin du script.

    Donc en bref, vous devez être attentif à ce que ni qin ni qout devient trop grand. (Voir aussi: Multitraitement File d'attente maxsize limite est de 32767)

    InformationsquelleAutor amd
  3. 3

    J'ai eu le même problème sur python3 lorsque j'essaye de mettre les chaînes dans une file d'attente de la taille totale d'environ 5000 rselnal.

    Dans mon projet, il y avait un hôte de processus qui met en place une file d'attente et commence un processus secondaire, puis elle rejoint. Afrer join lit formulaire de la file d'attente. Lorsque le sous-processus producess trop de données, hôte pendus sur join. J'ai fixé ce, à l'aide de la fonction suivante à attendre pour les sous-processus du processus d'hôte:

    def yield_from_process(q, p):
    while p.is_alive():
        p.join(timeout=1)
        while True:
            try:
                yield q.get(block=False)
            except Empty:
                break

    J'ai lu à partir de la file d'attente dès qu'elle est pleine de sorte qu'il n'est jamais très grande

    InformationsquelleAutor Rugnar
  4. 0

    J'essayais de .get() un async travailleur après la piscine était fermée

    indentation erreur à l'extérieur d'un bloc

    j'ai eu ce

    with multiprocessing.Pool() as pool:
    async_results = list()
    for job in jobs:
        async_results.append(
            pool.apply_async(
                _worker_func,
                (job,),
            )
        )
# wrong
for async_result in async_results:
    yield async_result.get()

    j'ai besoin de cette

    with multiprocessing.Pool() as pool:
    async_results = list()
    for job in jobs:
        async_results.append(
            pool.apply_async(
                _worker_func,
                (job,),
            )
        )
    # right
    for async_result in async_results:
        yield async_result.get()
    InformationsquelleAutor ThorSummoner