Python Multiprocessing: niektóre funkcje nie zwracają się, gdy są kompletne (materiał kolejki jest zbyt duży)

Question

Używam procesu i kolejki multiprocessingu. Rozpoczynam kilka funkcji równolegle i większość zachowuje się ładnie: kończą, ich wyjście przechodzi do kolejki i pojawiają się jako .is_alive() == Fałsz. Ale z jakiegoś powodu kilka funkcji nie działa. Zawsze pokazują one .is_alive() == Prawda, nawet po ostatnim wierszu funkcji (polecenie print "Finished") zostało zakończone. Dzieje się tak niezależnie od uruchamianych przeze mnie funkcji, nawet jeśli istnieje tylko jeden. Jeśli nie działają równolegle, funkcje zachowują się poprawnie i powracają normalnie. Co może być przyczyną problemu?

Oto ogólna funkcja, której używam do zarządzania zadaniami. Wszystko, czego nie pokazuję, to funkcje, które im przekazuję. Są długie, często używają matplotlib, czasami uruchamiają niektóre polecenia powłoki, ale nie wiem, co mają wspólnego te błędy.

def runFunctionsInParallel(listOf_FuncAndArgLists): 
    """ 
    Take a list of lists like [function, arg1, arg2, ...]. Run those functions in parallel, wait for them all to finish, and return the list of their return values, in order. 
    """ 
    from multiprocessing import Process, Queue 

    def storeOutputFFF(fff,theArgs,que): #add a argument to function for assigning a queue 
     print 'MULTIPROCESSING: Launching %s in parallel '%fff.func_name 
     que.put(fff(*theArgs)) #we're putting return value into queue 
     print 'MULTIPROCESSING: Finished %s in parallel! '%fff.func_name 
     # We get this far even for "bad" functions 
     return 

    queues=[Queue() for fff in listOf_FuncAndArgLists] #create a queue object for each function 
    jobs = [Process(target=storeOutputFFF,args=[funcArgs[0],funcArgs[1:],queues[iii]]) for iii,funcArgs in enumerate(listOf_FuncAndArgLists)] 
    for job in jobs: job.start() # Launch them all 
    import time 
    from math import sqrt 
    n=1 
    while any([jj.is_alive() for jj in jobs]): # debugging section shows progress updates 
     n+=1 
     time.sleep(5+sqrt(n)) # Wait a while before next update. Slow down updates for really long runs. 
     print('\n---------------------------------------------------\n'+ '\t'.join(['alive?','Job','exitcode','Func',])+ '\n---------------------------------------------------') 
     print('\n'.join(['%s:\t%s:\t%s:\t%s'%(job.is_alive()*'Yes',job.name,job.exitcode,listOf_FuncAndArgLists[ii][0].func_name) for ii,job in enumerate(jobs)])) 
     print('---------------------------------------------------\n') 
    # I never get to the following line when one of the "bad" functions is running. 
    for job in jobs: job.join() # Wait for them all to finish... Hm, Is this needed to get at the Queues? 
    # And now, collect all the outputs: 
    return([queue.get() for queue in queues]) 

Answer 1

porządku, wydaje się, że rura użyte do wypełnienia kolejki zostanie podłączony, gdy wyjście z funkcji jest zbyt duży (mój surowy zrozumienie? Jest to nierozwiązane/zamknięty bug? http://bugs.python.org/issue8237). Zmodyfikowałem kod w moim pytaniu, aby było trochę buforowania (kolejki są regularnie opróżniane podczas działania procesów), co rozwiązuje wszystkie moje problemy. Więc teraz bierze kolekcję zadań (funkcji i ich argumentów), uruchamia je i zbiera wyniki. Chciałbym, żeby było prostsze/czystsze.

Edytuj (2014 wrzesień; aktualizacja 2017 listopad: przepisany dla czytelności): Aktualizuję kod z ulepszeniami, które wprowadziłem od tego czasu. Nowy kod (ta sama funkcja, ale lepsze funkcje) znajduje się tutaj: https://github.com/cpbl/cpblUtilities/blob/master/parallel.py

Opis połączenia znajduje się również poniżej.

def runFunctionsInParallel(*args, **kwargs): 
    """ This is the main/only interface to class cRunFunctionsInParallel. See its documentation for arguments. 
    """ 
    return cRunFunctionsInParallel(*args, **kwargs).launch_jobs() 

########################################################################################### 
### 
class cRunFunctionsInParallel(): 
    ### 
    ####################################################################################### 
    """Run any list of functions, each with any arguments and keyword-arguments, in parallel. 
The functions/jobs should return (if anything) pickleable results. In order to avoid processes getting stuck due to the output queues overflowing, the queues are regularly collected and emptied. 
You can now pass os.system or etc to this as the function, in order to parallelize at the OS level, with no need for a wrapper: I made use of hasattr(builtinfunction,'func_name') to check for a name. 
Parameters 
---------- 
listOf_FuncAndArgLists : a list of lists 
    List of up-to-three-element-lists, like [function, args, kwargs], 
    specifying the set of functions to be launched in parallel. If an 
    element is just a function, rather than a list, then it is assumed 
    to have no arguments or keyword arguments. Thus, possible formats 
    for elements of the outer list are: 
     function 
     [function, list] 
     [function, list, dict] 
kwargs: dict 
    One can also supply the kwargs once, for all jobs (or for those 
    without their own non-empty kwargs specified in the list) 
names: an optional list of names to identify the processes. 
    If omitted, the function name is used, so if all the functions are 
    the same (ie merely with different arguments), then they would be 
    named indistinguishably 
offsetsSeconds: int or list of ints 
    delay some functions' start times 
expectNonzeroExit: True/False 
    Normal behaviour is to not proceed if any function exits with a 
    failed exit code. This can be used to override this behaviour. 
parallel: True/False 
    Whenever the list of functions is longer than one, functions will 
    be run in parallel unless this parameter is passed as False 
maxAtOnce: int 
    If nonzero, this limits how many jobs will be allowed to run at 
    once. By default, this is set according to how many processors 
    the hardware has available. 
showFinished : int 
    Specifies the maximum number of successfully finished jobs to show 
    in the text interface (before the last report, which should always 
    show them all). 
Returns 
------- 
Returns a tuple of (return codes, return values), each a list in order of the jobs provided. 
Issues 
------- 
Only tested on POSIX OSes. 
Examples 
-------- 
See the testParallel() method in this module 
    """