Ability to adjust queue size in Executors

[postmasters]

BPO	14119
Nosy	@brianquinlan, @pitrou, @bitdancer, @postmasters, @thehesiod, @mhrivnak
Superseder	bpo-29595: Expose max_queue_size in ThreadPoolExecutor
Files	executor-queue-size.diff: patch to add queue size to ThreadPoolExecutor

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = 'https://github.com/brianquinlan'
closed_at = <Date 2018-03-31.17:04:32.560>
created_at = <Date 2012-02-25.00:19:01.762>
labels = ['type-feature', 'library']
title = 'Ability to adjust queue size in Executors'
updated_at = <Date 2018-03-31.17:04:32.543>
user = 'https://github.com/postmasters'

bugs.python.org fields:

activity = <Date 2018-03-31.17:04:32.543>
actor = 'pitrou'
assignee = 'bquinlan'
closed = True
closed_date = <Date 2018-03-31.17:04:32.560>
closer = 'pitrou'
components = ['Library (Lib)']
creation = <Date 2012-02-25.00:19:01.762>
creator = 'Nam.Nguyen'
dependencies = []
files = ['24666']
hgrepos = []
issue_num = 14119
keywords = ['patch']
message_count = 19.0
messages = ['154175', '154177', '162599', '162601', '162604', '162605', '162606', '162664', '162700', '207512', '207602', '207672', '207893', '207896', '253895', '274588', '280727', '305857', '314742']
nosy_count = 10.0
nosy_names = ['bquinlan', 'pitrou', 'r.david.murray', 'Nam.Nguyen', 'Victor.Varvariuc', 'thehesiod', 'Winterflower', 'Patrik Dufresne', 'Vinay Anantharaman', 'mhrivnak']
pr_nums = []
priority = 'normal'
resolution = 'duplicate'
stage = 'resolved'
status = 'closed'
superseder = '29595'
type = 'enhancement'
url = 'https://bugs.python.org/issue14119'
versions = ['Python 3.3']

[postmasters]

I am running into a memory consumption issue with concurrent.futures module. Its Executors do not have a public API to adjust their queue size. and the queues are created unbounded initially.

It would be helpful to have some public method or a parameter at construction time to limit this queue size.

[postmasters]

By the way, ProcessPoolExecutor actually sets its queue size to a reasonable number but ThreadPoolExecutor does not.

[bitdancer]

Brian: ping. Since this is an enhancement, if you are going to accept it it would be nice to get it into 3.3, which means committing it before June 23rd.

[brianquinlan]

Hey Nam,

I'm not sure that I understand. You want ThreadPoolExecutor.submit to block if there are too many work items in the queue? Are you sure that this happens currently with ProcessPoolExecutor? I can't see why it would.

[postmasters]

Currently, ProcessionPoolExecutor has this line in its constructor:

self._call_queue = multiprocessing.Queue(self._max_workers +
                                                 EXTRA_QUEUED_CALLS)

where EXTRA_QUEUED_CALLS is 1.

And yes, it would be best to export a method so that the app developer can set the queue size for themselves. In my case, I would want to limit the queue so that I dont run out of memory. Others might not want the queue to block, and hence would prefer an unlimited queue.

[brianquinlan]

The queue that you identified i.e.

self._call_queue = multiprocessing.Queue(self._max_workers +
                                         EXTRA_QUEUED_CALLS)

does not get considered during submit() - are you sure that it somehow causes submit() to block.

Could you explain what your use case is such that you can run out of memory?

[postmasters]

I used the ThreadPoolExecutor to process a large number (bounded) of input/output jobs. Because there were too many of them and the worker threads could not process them fast enough to drain them from the queue, the queue kept increasing in size.

It was okay for the script, though, to block, waiting for the queue to drain, before submitting new jobs. So I needed to limit the queue size.

[brianquinlan]

I've had people request that they be able control the order of processed work submissions. So a more general way to solve your problem might be to make the two executors take an optional Queue argument in their constructors.

You'd have to explain in detail in the document how the queues are used.

What do you think?

[postmasters]

+1

That was actually what I did. I replaced the internal queue with another one whose limit was properly set.

If you are busy to write one, let me find some time to create another patch.

[VictorVarvariuc]

Maybe I should have created another issue for this, but without this issue being solved, the new issue will not help much.
Here it is:
http://hg.python.org/cpython/file/37caaf21f827/Lib/concurrent/futures/thread.py#l63
After running an work item work_queue.task_done() is not called.
So it's difficult to know if worker threads have any more work to do.
http://stackoverflow.com/questions/20965754/determine-if-worker-threads-are-doing-any-work?noredirect=1#comment31495804_20965754

[brianquinlan]

Hi Victor,

I don't understand your problem. Could you be very specific in your description?

[VictorVarvariuc]

Hi Brian,

In one my projects I had to monkey-patch module concurrent.futures.thread:60( http://hg.python.org/cpython/file/37caaf21f827/Lib/concurrent/futures/thread.py#l60) with:

def _worker(executor_reference, work_queue):
    try:
        while True:
            work_item = work_queue.get(block=True)
            if work_item is not None:
                work_item.run()
                work_queue.task_done()  # <-- added this line
                continue
            executor = executor_reference()
            # Exit if:
            #   - The interpreter is shutting down OR
            #   - The executor that owns the worker has been collected OR
            #   - The executor that owns the worker has been shutdown.
            if futures_thread._shutdown or executor is None or executor._shutdown:
                # Notice other workers
                work_queue.put(None)
                return
            del executor
    except BaseException:
        futures_thread._base.LOGGER.critical('Exception in worker', exc_info=True)

This helps me to control the state of the work queue -- I can see if there are any work items still being processed:

if executor._work_queue.unfinished_tasks:
    # executor is still producing something
    ...