how to throttle a large number of tasks without using all workers
<p>Imagine I have a dask grid with 10 workers &amp; 40 cores totals. This is a shared grid, so I don't want to fully saturate it with my work. I have 1000 tasks to do, and I want to submit (and have actively running) a maximum of 20 tasks at a time.</p>

<p>To be concrete, </p>

<pre><code>from time import sleep
from random import random

def inc(x):
    from random import random
    sleep(random() * 2)
    return x + 1

def double(x):
    from random import random
    sleep(random())
    return 2 * x

&gt;&gt;&gt; from distributed import Executor
&gt;&gt;&gt; e = Executor('127.0.0.1:8786')
&gt;&gt;&gt; e
&lt;Executor: scheduler=127.0.0.1:8786 workers=10 threads=40&gt;
</code></pre>

<p>If I setup a system of Queues</p>

<pre><code>&gt;&gt;&gt; from queue import Queue
&gt;&gt;&gt; input_q = Queue()
&gt;&gt;&gt; remote_q = e.scatter(input_q)
&gt;&gt;&gt; inc_q = e.map(inc, remote_q)
&gt;&gt;&gt; double_q = e.map(double, inc_q)
</code></pre>

<p>This will work, BUT, this will just dump ALL of my tasks to the grid, saturating it. Ideally I could:</p>

<pre><code>e.scatter(input_q, max_submit=20)
</code></pre>

<p>It seems that the example from the docs <a href="http://distributed.readthedocs.io/en/latest/queues.html#example">here</a> would allow me to use a <code>maxsize</code> queue. But that looks like from a user-perspective I would still have to deal with the backpressure. Ideally  <code>dask</code> would automatically take care of this. </p>