B F.\@sdZddlZddlmZddlmZddlmZddlmZddl m Z dd l m Z m Z dd l mZdd lmZd d ZGddde ZGddde jZee jZdS)zAExperimental interface for asyncio, may disappear without warningN)wraps)merge)BaseAsyncIOLoop)to_asyncio_future)client)ClientFuture)Variable)ignoringc stfdd}|S)z;Converts Tornado gen.coroutines and futures to asyncio onescsrt|}t||S)N)rr)argskwargs)default_kwargsfn2lib/python3.7/site-packages/distributed/asyncio.pyconverts zto_asyncio..convert)r)rrrr)rrr to_asynciosrcsleZdZdZfddZddZddZdd Zd d Ze e j d d Z e e j Z e e j Z e e jZZS) AioClienta` Connect to and drive computation on a distributed Dask cluster This class provides an asyncio compatible async/await interface for dask.distributed. The Client connects users to a dask.distributed compute cluster. It provides an asynchronous user interface around functions and futures. This class resembles executors in ``concurrent.futures`` but also allows ``Future`` objects within ``submit/map`` calls. AioClient is an **experimental** interface for distributed and may disappear without warning! Parameters ---------- address: string, or Cluster This can be the address of a ``Scheduler`` server like a string ``'127.0.0.1:8786'`` or a cluster object like ``LocalCluster()`` Examples -------- Provide cluster's scheduler address on initialization:: client = AioClient('127.0.0.1:8786') Start the client:: async def start_the_client(): client = await AioClient() # Use the client.... await client.close() An ``async with`` statement is a more convenient way to start and shut down the client:: async def start_the_client(): async with AioClient() as client: # Use the client within this block. pass Use the ``submit`` method to send individual computations to the cluster, and await the returned future to retrieve the result:: async def add_two_numbers(): async with AioClient() as client: a = client.submit(add, 1, 2) result = await a Continue using submit or map on results to build up larger computations, and gather results with the ``gather`` method:: async def gather_some_results(): async with AioClient() as client: a = client.submit(add, 1, 2) b = client.submit(add, 10, 20) c = client.submit(add, a, b) result = await client.gather([c]) See Also -------- distributed.client.Client: Blocking Client distributed.scheduler.Scheduler: Internal scheduler cs,t}t|}tj||dd|dS)NT)loopZ asynchronous)asyncioZget_event_looprsuper__init__)selfr r rZioloop) __class__rrr_szAioClient.__init__cCs tddS)Nz2Use AioClient in an 'async with' block, not 'with') RuntimeError)rrrr __enter__dszAioClient.__enter__cst|jIdH|S)N)r_started)rrrr __aenter__gszAioClient.__aenter__cst|IdHdS)N)rZ_close)rtypevalue tracebackrrr __aexit__kszAioClient.__aexit__cCst|jS)N)rr __await__)rrrrr#nszAioClient.__await__F)sync)__name__ __module__ __qualname____doc__rrrr"r#rrgetr$closeZshutdown __classcell__rr)rrrsA   rc@seZdZeejjZdS) as_completedN)r%r&r'rrr, __anext__rrrrr,wsr,)r(r functoolsrZtoolzrZtornado.platform.asynciorrrrr Zvariabler Zutilsr rrr,Z_waitwaitrrrrs        Z