2

I need to fetch really big data chunks by iterating over the data. In total I need a few million iterations. So I'd thought subpressing would speed up my process and it almost did. I use subprocess.Queue to invoke different Threads, which actually works fine, but when I call *subprocess.Queue.get()`the programm takes forever to get the results. Maybe I did something wrong. Here is my minimal example:

def get_losses(self, tags=None):
    return_dict = {}
    output_list = multiprocessing.Queue()
    process_list = []

    # Create quese definition
    for experiment, path in self.tf_board_dicts.items():
        t = multiprocessing.Process(target=self._load_vec_from_tfboard, args=(path, tags, experiment))
        process_list.append(t)
    print("Starting subprocesse with a total of {} workers. \n These are  {}".format(len(process_list),
                                                                                         process_list))
    # Run processes
    for p in process_list:
        p.start()

    # Exit the finished threads
    for p in process_list:
        p.join()
    print("All subprocesses are termianted")

    # Get results
    results = [output_list.get() for p in process_list]
    print("All losses are gathered: {}".format([tup[0] for tup in results]))

    # Create dict
    for experiment_losses in results:
         return_dict[experiment_losses[0]] = experiment_losses[1]

    return return_dict
2
  • 1
    For debugging try this: results = [output_list.get(timeout=0.1) for p in process_list]
    – kantal
    Commented Dec 2, 2018 at 16:09
  • Thanks, I did and in fact that showed me that the return list was not filled properly. Unfortunately I found the reason, why I does not work. The bottleneck is the loading process and somehow it does not multiprocess this step. But I think this is another question.
    – Lau
    Commented Dec 4, 2018 at 8:54

1 Answer 1

1

You can find the answer for infinite time issue with queue here : Python Processes not joining

This happens because the Queue uses a buffer internally when a lot of data is being pushed into it. The process writing to the Queue can't exit until that buffer gets flushed, which won't happen until you start pulling things out of the Queue. So, because you're trying to join all the processes before you pull anything out of the Queue objects they're writing to, they can't exit, so the join hangs. You could fix the issue by draining the Queue prior to calling join on the processes. – dano Sep 25 '14 at 16:16

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.