Многопроцессорность python: некоторые функции не возвращаются, когда они полны (слишком большой материал очереди)

Я использую многопроцессорный процесс и очередь. Я запускаю несколько функций параллельно, и большинство из них хорошо себя ведут: они заканчиваются, их выход переходит в их очередь, и они отображаются как .is_alive() == False. Но по какой-то причине несколько функций не ведут себя. Они всегда показывают .is_alive() == Истинно, даже после последней строки в функции (утверждение печати "Закончено" ) завершено. Это происходит независимо от набора функций, которые я запускаю, даже там только один. Если они не работают параллельно, функции ведут себя нормально и нормально возвращаются. Что это может быть проблемой?

Вот общая функция, которую я использую для управления заданиями. Все, что я не показываю, это функции, которые я передаю. Они длинные, часто используют matplotlib, иногда запускают некоторые команды оболочки, но я не могу понять, что у них есть недостающие.

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])

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 """

Ответ 1