executor.map hangs at seemly random iteration when iterable is large

[tburke02]

Bug report

Bug description:

Minimum working example of issue:

import numpy as np
import concurrent.futures

with concurrent.futures.ProcessPoolExecutor() as executor:
 executor.map(print, np.arange(1000000))

Each time this is run it seems to hang at a different iteration with no apparent CPU or memory usage. May be related to #74028

CPython versions tested on:

3.9

Operating systems tested on:

Linux

Linked PRs

• gh-74028: Introduce a prefetch parameter to Executor.map to handle large iterators #114975

[gaogaotiantian]

Is this reproducible without numpy? np.arange is simple enough but it would be nice to rule out all the possible externel dependencies in the example.

[tburke02]

Is this reproducible without numpy? np.arange is simple enough but it would be nice to rule out all the possible externel dependencies in the example.

Yes, same behavior when using range.

[terryjreedy]

Win10, main, Command Prompt, with range: I get multiple duplicate tracebacks.

 File "C:\Programs\Python313\Lib\multiprocessing\spawn.py", line 164, in get_preparation_data
 _check_not_importing_main()
 ~~~~~~~~~~~~~~~~~~~~~~~~~^^
 File "C:\Programs\Python313\Lib\multiprocessing\spawn.py", line 140, in _check_not_importing_main
 raise RuntimeError('''
 ...<16 lines>...
 ''')
 File "C:\Programs\Python313\Lib\multiprocessing\spawn.py", line 140, in _check_not_importing_main
 raise RuntimeError('''
 ...<16 lines>...
 ''')
RuntimeError:
 An attempt has been made to start a new process before the
 current process has finished its bootstrapping phase.

 This probably means that you are not using fork to start your
 child processes and you have forgotten to use the proper idiom
 in the main module:

 if __name__ == '__main__':
 ...

With the needed (on Windows) clause added, it prints 0 to 12 on separate lines and hangs. ^C and I get 12 interlaced tracebacks with some of the following lines

 File "C:\Programs\Python313\Lib\multiprocessing\process.py", line 314, in _bootstrap
 self.run()
 File "C:\Programs\Python313\Lib\multiprocessing\process.py", line 108, in run
 self._target(*self._args, **self._kwargs)
 File "C:\Programs\Python313\Lib\concurrent\futures\process.py", line 251, in _process_worker
 call_item = call_queue.get(block=True)
 ~~~~~~~~~~~~~~^^^^^^^^^^^^
 File "C:\Programs\Python313\Lib\multiprocessing\queues.py", line 102, in get
 with self._rlock:
 res = self._recv_bytes()
 File "C:\Programs\Python313\Lib\multiprocessing\queues.py", line 103, in get
 res = self._recv_bytes()
 File "C:\Programs\Python313\Lib\multiprocessing\synchronize.py", line 95, in __enter__
 return self._semlock.__enter__()
KeyboardInterrupt 

[Jason-Y-Z]

Thanks all for the discussion and feedback. I've started a PR which should hopefully address/help with this and a couple of related issues.

[gaogaotiantian]

This does not hang infinitely, it just takes a long time. (~1min on my comupter).

This is the main process timeline for a "normal" map operation, where the input is range(100).

The MainThread is doing the map, which submit the work item one by one to a dict. In each submit, two thing happen:

1. a wakeup signal is sent to self.thread_wakeup queue
2. a WorkItem's id is added to self._work_ids queue

During the first call of submit, an executor manager thread is brought up (Thread-1 in the graph), which is in charge of actually distributing the tasks.

It will get the WorkItems from self._work_ids and put them in a process queue that has a size of MAX_WORKERS + 1 for the process workers. Because the queue has a maximum size, the manager thread will not be able to put all the tasks in it.

When the task queue is full, the manager waits for a broken or wakeup signal in wait_result_broken_or_wakeup, and this is where the issue happened.

In this function, it needs to read all the pending wakeup signals in self.thread_wakeup queue mentioned above. Remember, the submit will not be blocked, it's not lazily evaluated, so each task sends a wakeup signal. This function needs to process ALL of them to be able to send more WorkItems to the worker process.

Each poll & recv takes about 50 us on my computer:

So if you have 1e6 tasks like in the example, you'll need 50s to process all the wake up signals before you can send out more WorkItems to the process workers.

So it's not a real dead lock, it just takes a long time. However, I do agree that if we can do something about it, that would be nice. But it won't make this go any faster even with lazy evaluation - it will make this smoother, as each wakeup signal always need to take 50us and it needs to be processed anyway (but it can be parallelized so if the task itself takes more than that, it should be ignorable if it's not blocking the task distribution).

[Jason-Y-Z]

@gaogaotiantian Thanks for the profiling, please see #114975 for an attempt to improve the large iterator use case

[gaogaotiantian]

Ah okay this is a false alarm. What you should do is to specify a chunksize to batchify you data.

The docs says:

For very long iterables, using a large value for chunksize can significantly improve performance compared to the default size of 1

executor.map(print, np.arange(1000000), chunksize=10000)

should just solve the issue.