Path: csiph.com!v102.xanadu-bbs.net!xanadu-bbs.net!goblin2!goblin.stu.neva.ru!newsfeed.xs4all.nl!newsfeed3.news.xs4all.nl!xs4all!post.news.xs4all.nl!not-for-mail
MIME-Version: 1.0
In-Reply-To: <536BD338.4070004@andros.org.uk>
References: <536BD338.4070004@andros.org.uk>
From: Ian Kelly <ian.g.kelly@gmail.com>
Date: Thu, 8 May 2014 14:44:28 -0600
Subject: Re: Real-world use of concurrent.futures
To: Python <python-list@python.org>
Content-Type: multipart/alternative; boundary=047d7b15aba983be7d04f8e98e4c
Precedence: list
Newsgroups: comp.lang.python
Message-ID: <mailman.9797.1399581918.18130.python-list@python.org>
Lines: 104
NNTP-Posting-Host: 2001:888:2000:d::a6
Xref: csiph.com comp.lang.python:71129

--047d7b15aba983be7d04f8e98e4c
Content-Type: text/plain; charset=UTF-8

On May 8, 2014 12:57 PM, "Andrew McLean" <lists@andros.org.uk> wrote:
> So far so good. However, I thought this would be an opportunity to
> explore concurrent.futures and to see whether it offered any benefits
> over the more explicit approach discussed above. The problem I am having
> is that all the discussions I can find of the use of concurrent.futures
> show use with toy problems involving just a few tasks. The url
> downloader in the documentation is typical, it proceeds as follows:
>
> 1. Get an instance of concurrent.futuresThreadPoolExecutor
> 2. Submit a few tasks to the executer
> 3. Iterate over the results using concurrent.futures.as_completed
>
> That's fine, but I suspect that isn't a helpful pattern if I have a very
> large number of tasks. In my case I could run out of memory if I tried
> submitting all of the tasks to the executor before processing any of the
> results.

I thought that ThreadPoolExecutor.map would handle this transparently if
you passed it a lazy iterable such as a generator.  From my testing though,
that seems not to be the case; with a generator of 100000 items and a pool
of 2 workers, the entire generator was consumed before any results were
returned.

> I'm guessing what I want to do is, submit tasks in batches of perhaps a
> few hundred, iterate over the results until most are complete, then
> submit some more tasks and so on. I'm struggling to see how to do this
> elegantly without a lot of messy code just there to do "bookkeeping".
> This can't be an uncommon scenario. Am I missing something, or is this
> just not a job suitable for futures?

I don't think it needs to be "messy". Something like this should do the
trick, I think:

from concurrent.futures import *
from itertools import islice

def batched_pool_runner(f, iterable, pool, batch_size):
  it = iter(iterable)
  # Submit the first batch of tasks.
  futures = set(pool.submit(f, x) for x in islice(it, batch_size))
  while futures:
    done, futures = wait(futures, return_when=FIRST_COMPLETED)
    # Replenish submitted tasks up to the number that completed.
    futures.update(pool.submit(f, x) for x in islice(it, len(done)))
    yield from done

--047d7b15aba983be7d04f8e98e4c
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">On May 8, 2014 12:57 PM, &quot;Andrew McLean&quot; &lt;<a =
href=3D"mailto:lists@andros.org.uk">lists@andros.org.uk</a>&gt; wrote:<br>&=
gt; So far so good. However, I thought this would be an opportunity to<br>&=
gt; explore concurrent.futures and to see whether it offered any benefits<b=
r>

&gt; over the more explicit approach discussed above. The problem I am havi=
ng<br>&gt; is that all the discussions I can find of the use of concurrent.=
futures<br>&gt; show use with toy problems involving just a few tasks. The =
url<br>

&gt; downloader in the documentation is typical, it proceeds as follows:<br=
>&gt;<br>&gt; 1. Get an instance of concurrent.futuresThreadPoolExecutor<br=
>&gt; 2. Submit a few tasks to the executer<br>&gt; 3. Iterate over the res=
ults using concurrent.futures.as_completed<br>

&gt;<br>&gt; That&#39;s fine, but I suspect that isn&#39;t a helpful patter=
n if I have a very<br>&gt; large number of tasks. In my case I could run ou=
t of memory if I tried<br>&gt; submitting all of the tasks to the executor =
before processing any of the<br>

&gt; results.<br><br>I thought that ThreadPoolExecutor.map would handle thi=
s transparently if you passed it a lazy iterable such as a generator. =C2=
=A0From my testing though, that seems not to be the case; with a generator =
of 100000 items and a pool of 2 workers, the entire generator was consumed =
before any results were returned.<br>

<br>&gt; I&#39;m guessing what I want to do is, submit tasks in batches of =
perhaps a<br>&gt; few hundred, iterate over the results until most are comp=
lete, then<br>&gt; submit some more tasks and so on. I&#39;m struggling to =
see how to do this<br>

&gt; elegantly without a lot of messy code just there to do &quot;bookkeepi=
ng&quot;.<br>&gt; This can&#39;t be an uncommon scenario. Am I missing some=
thing, or is this<br>&gt; just not a job suitable for futures?<br><br>
I don&#39;t think it needs to be &quot;messy&quot;. Something like this sho=
uld do the trick, I think:<br>
<br>from concurrent.futures import *<br><div>from itertools import islice<b=
r><div><br></div><div>def batched_pool_runner(f, iterable, pool, batch_size=
):</div><div>=C2=A0 it =3D iter(iterable)</div><div>=C2=A0 # Submit the fir=
st batch of tasks.<br>

=C2=A0 futures =3D set(pool.submit(f, x) for x in islice(it, batch_size))<b=
r>=C2=A0 while futures:<br>=C2=A0 =C2=A0 done, futures =3D wait(futures, re=
turn_when=3DFIRST_COMPLETED)</div><div>=C2=A0 =C2=A0 # Replenish submitted =
tasks up to the number that completed.<br>

=C2=A0 =C2=A0 futures.update(pool.submit(f, x) for x in islice(it, len(done=
)))</div><div>=C2=A0 =C2=A0 yield from done<br><br></div></div></div>

--047d7b15aba983be7d04f8e98e4c--