Path: csiph.com!v102.xanadu-bbs.net!xanadu-bbs.net!news.albasani.net!feeder.erje.net!eu.feeder.erje.net!newsfeed.xs4all.nl!newsfeed1a.news.xs4all.nl!xs4all!newsgate.cistron.nl!newsgate.news.xs4all.nl!post.news.xs4all.nl!not-for-mail
MIME-Version: 1.0
From: Valery Khamenya <khamenya@gmail.com>
Date: Sun, 3 Aug 2014 16:01:41 +0200
Subject: asyncio with map&reduce flavor and without flooding the event loop
To: python-list@python.org, Maxime Steisel <maximesteisel@gmail.com>
Content-Type: multipart/alternative; boundary=001a11c2ecea0f650904ffba11b8
Precedence: list
Newsgroups: comp.lang.python
Message-ID: <mailman.12584.1407074531.18130.python-list@python.org>
Lines: 121
NNTP-Posting-Host: 2001:888:2000:d::a6
Xref: csiph.com comp.lang.python:75598

--001a11c2ecea0f650904ffba11b8
Content-Type: text/plain; charset=UTF-8

Hi all

I am trying to use asyncio in real applications and it doesn't go that
easy, a help of asyncio gurus is needed badly.

Consider a task like crawling the web starting from some web-sites. Each
site leads to generation of new downloading tasks in exponential(!)
progression. However we don't want neither to flood the event loop nor to
overload our network. We'd like to control the task flow. This is what I
achieve well with modification of nice Maxime's solution proposed here:
https://mail.python.org/pipermail/python-list/2014-July/675048.html

Well, but I'd need as well a very natural thing, kind of map() & reduce()
or functools.reduce() if we are on python3 already. That is, I'd need to
call a "summarizing" function for all the downloading tasks completed on
links from a page. This is where i fail :(

I'd propose an oversimplified but still a nice test to model the use case:
Let's use fibonacci function implementation in its ineffective form.
That is, let the coro_sum() be our reduce() function and coro_fib be our
map().
Something like this:

@asyncio.coroutine
def coro_sum(x):
    return sum(x)

@asyncio.coroutine
def coro_fib(x):
    if x < 2:
        return 1
    res_coro =
executor_pool.spawn_task_when_arg_list_of_coros_ready(coro=coro_sum,

 arg_coro_list=[coro_fib(x - 1), coro_fib(x - 2)])
    return res_coro

So that we could run the following tests.

Test #1 on one worker:

  executor_pool = ExecutorPool(workers=1)
  executor_pool.as_completed( coro_fib(x) for x in range(20) )

Test #2 on two workers:
  executor_pool = ExecutorPool(workers=2)
  executor_pool.as_completed( coro_fib(x) for x in range(20) )

It would be very important that both each coro_fib() and coro_sum()
invocations are done via a Task on some worker, not just spawned implicitly
and unmanaged!

It would be cool to find asyncio gurus interested in this very natural goal.
Your help and ideas would be very much appreciated.

best regards
--
Valery

--001a11c2ecea0f650904ffba11b8
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hi all<div><br></div><div>I am trying to use asyncio in re=
al applications and it doesn&#39;t go that easy, a help of asyncio gurus is=
 needed badly.</div><div><br></div><div>Consider a task like crawling the w=
eb starting from some web-sites. Each site leads to generation of new downl=
oading tasks in exponential(!) progression. However we don&#39;t want neith=
er to flood the event loop nor to overload our network. We&#39;d like to co=
ntrol the task flow. This is what I achieve well with modification of nice =
Maxime&#39;s solution proposed here:</div>

<div><a href=3D"https://mail.python.org/pipermail/python-list/2014-July/675=
048.html">https://mail.python.org/pipermail/python-list/2014-July/675048.ht=
ml</a><br></div><div><br></div><div>Well, but I&#39;d need as well a very n=
atural thing, kind of map() &amp; reduce() or=C2=A0functools.reduce() if we=
 are on python3 already. That is, I&#39;d need to call a &quot;summarizing&=
quot; function for all the downloading tasks completed on links from a page=
. This is where i fail :(</div>

<div><br></div><div>I&#39;d propose an oversimplified but still a nice test=
 to model the use case:</div><div>Let&#39;s use fibonacci function implemen=
tation in its ineffective form.</div><div>That is, let the coro_sum() be ou=
r reduce() function and coro_fib be our map().</div>

<div>Something like this:</div><div><br></div><div><div>@asyncio.coroutine<=
/div><div>def coro_sum(x):</div><div>=C2=A0 =C2=A0 return sum(x)</div><div>=
<br></div><div>@asyncio.coroutine</div><div>def coro_fib(x):=C2=A0</div><di=
v>=C2=A0 =C2=A0 if x &lt; 2:</div>

<div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 return 1</div><div>=C2=A0 =C2=A0 res_coro =
=3D executor_pool.spawn_task_when_arg_list_of_coros_ready(coro=3Dcoro_sum,<=
/div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=
=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 =C2=A0 =C2=A0arg_coro_list=3D[coro_fib(x - 1), coro_fib(x - 2)])</di=
v>

<div>=C2=A0 =C2=A0 return res_coro<br></div><div>=C2=A0<br></div></div><div=
>So that we could run the following tests.</div><div><br></div><div>Test #1=
 on one worker:</div><div><br></div><div>=C2=A0 executor_pool =3D ExecutorP=
ool(workers=3D1)</div>

<div>=C2=A0 executor_pool.as_completed( coro_fib(x) for x in range(20) )<br=
></div><div><br></div><div>Test #2 on two workers:</div><div><div>=C2=A0 ex=
ecutor_pool =3D ExecutorPool(workers=3D2)</div><div>=C2=A0 executor_pool.as=
_completed( coro_fib(x) for x in range(20) )<br>

</div></div><div><br></div><div>It would be very important that both each c=
oro_fib() and coro_sum() invocations are done via a Task on some worker, no=
t just spawned implicitly and unmanaged!</div><div><br></div><div>It would =
be cool to find asyncio gurus interested in this very natural goal.</div>

<div>Your help and ideas would be very much appreciated.</div><div><br clea=
r=3D"all"><div>best regards<br>--<br>Valery=C2=A0</div>
</div></div>

--001a11c2ecea0f650904ffba11b8--