AioClient update

[misterpilou]

In the documentation of AioClient, it tells that it is experimental and it could disappear without warning. I feel the way it works is quite odd, at first i was expecting it to work as aiomultiprocess from jreese repository. As it could disappear i wonder if i could enhancing it. I guess i will have a lot of questions about Nanny. Also as dask is based on tornado, i'm afraid there is a lot of work wrapping process pools with asyncio. So my question: Can i mess up with AioClient?

[mrocklin]

I'm inclined to remove AioClient. In recent versions Tornado is now pretty asyncio native. I suspect that you could use the normal Client(..., asynchronous=True) and be fine.

[misterpilou]

But wouldn't a AioClient that would map/submit i mean handle every operation but with async function would be interesting? I need to say that i'm not an expert but from my benchmark client.sync with tornado vs aiomultiprocess, dask is 5 times slower. No bashing here but i really think dask is a well suited framework for a lots of tasks and supporting async/await would be amazing in my opinion. I know it would surely implied a lots of work and architecture considerations, but i got time and contributing to dask would be a pleasure

[mrocklin]

The Client already supports async/await: http://distributed.dask.org/en/latest/asynchronous.html#basic-operation You wouldn't use client.sync. You just await the normal functions. I would be surprised to learn that Tornado is 5x slower than asyncio when used in this way.

…

On Mon, May 6, 2019 at 1:32 PM Becaree ***@***.***> wrote: But wouldn't a AioClient that would map/submit i mean handle every operation but with async function would be interesting? I need to say that i'm not an expert but from my benchmark client.sync with tornado or asyncio is 5 times slower. No bashing here but i really think dask is a well suited framework for a lots of tasks and supporting async/await would be amazing in my opinion. I know it would surely implied a lots of work and architecture considerations, but i got time and contributing to dask would be a pleasure — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2661 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AACKZTFTQRJUFWBRY24ZG2DPUB2VBANCNFSM4HK75N3A> .

[misterpilou]

Thanks for the information, i was gonna say that i'm maybe misunderstanding dask.distributed API in my 2nd message, it happen that it is. You rock @mrocklin =)

[misterpilou]

So i tried a whole lots of things, i always ending up with "Can't pickle coroutines objects".

Little sample of my code:

async def run_crawler():
    client = await Client(asynchronous=True)
    resps = client.map(fetch, urls)
    res = await client.gather(resps)

async def fetch(url):
    """just a coroutine"""

if name == "__main__":
     loop = asyncio.get_event_loop()
     loop.run_until_complete(run_crawler())

[misterpilou]

with the traceback:

Traceback (most recent call last):
  File "/home/misterpilou/.local/share/virtualenvs/playweb-tEmYZhqM/lib/python3.6/site-packages/distributed/protocol/serialize.py", line 146, in serialize
    header, frames = dumps(x, context=context) if wants_context else dumps(x)
  File "/home/misterpilou/.local/share/virtualenvs/playweb-tEmYZhqM/lib/python3.6/site-packages/distributed/protocol/serialize.py", line 54, in pickle_dumps
    return {"serializer": "pickle"}, [pickle.dumps(x)]
  File "/home/misterpilou/.local/share/virtualenvs/playweb-tEmYZhqM/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 53, in dumps
    return cloudpickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
  File "/home/misterpilou/.local/share/virtualenvs/playweb-tEmYZhqM/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 963, in dumps
    cp.dump(obj)
  File "/home/misterpilou/.local/share/virtualenvs/playweb-tEmYZhqM/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 267, in dump
    return Pickler.dump(self, obj)
  File "/home/misterpilou/.pyenv/versions/3.6.8/lib/python3.6/pickle.py", line 409, in dump
    self.save(obj)
  File "/home/misterpilou/.pyenv/versions/3.6.8/lib/python3.6/pickle.py", line 496, in save
    rv = reduce(self.proto)
TypeError: can't pickle coroutine objects

[mrocklin]

I've just added an async-await example notebook. Perhaps it will help? If you have a chance to take a look at it and see if it resolves your problem then that would be helpful.

dask/dask-examples#71

[misterpilou]

I guess if def inc() was an async def inc() it would fit my case or a gen.coroutine. Anyway, i will try to work on it, maybe by example it will be more explicit. But it will be a challenge for me.