Unsynchronize asyncio by using an ambient event loop in a separate thread.
- Mark all async functions with
@unsync. May also mark regular functions to execute in a separate thread.- All
@unsyncfunctions, async or not, return anUnfuture
- All
- All
Futuresmust beUnfutureswhich includes the result of an@unsyncfunction call, or wrappingUnfuture(asyncio.Future)orUnfuture(concurrent.Future).Unfuturecombines the behavior ofasyncio.Futureandconcurrent.Future:Unfuture.set_valueis threadsafe unlikeasyncio.FutureUnfutureinstances can be awaited, even if made fromconcurrent.FutureUnfuture.result()is a blocking operation except inunsync.loop/unsync.threadwhere it behaves likeasyncio.Future.resultand will throw an exception if the future is not done
- Functions will execute in different contexts:
@unsyncasync functions will execute in an event loop inunsync.thread@unsyncregular functions will execute inunsync.thread_executor, aThreadPoolExecutor@unsync(cpu_bound=True)regular functions will execute inunsync.process_executor, aProcessPoolExecutor
A simple sleeping example with asyncio:
asyncdefsync_async():
awaitasyncio.sleep(0.1)
return'I hate event loops'result=asyncio.run(sync_async())
print(result)Same example with unsync:
@unsyncasyncdefunsync_async():
awaitasyncio.sleep(0.1)
return'I like decorators'print(unsync_async().result())Synchronous functions can be made to run asynchronously by executing them in a concurrent.ThreadPoolExecutor.
This can be easily accomplished by marking the regular function @unsync.
@unsyncdefnon_async_function(seconds):
time.sleep(seconds)
return'Run in parallel!'start=time.time()
tasks= [non_async_function(0.1) for_inrange(10)]
print([task.result() fortaskintasks])
print('Executed in {} seconds'.format(time.time() -start))Which prints:
['Run in parallel!', 'Run in parallel!', ...]
Executed in 0.10807514190673828 seconds
Using Unfuture.then chains asynchronous calls and returns an Unfuture that wraps both the source, and continuation.
The continuation is invoked with the source Unfuture as the first argument.
Continuations can be regular functions (which will execute synchronously), or @unsync functions.
@unsyncasyncdefinitiate(request):
awaitasyncio.sleep(0.1)
returnrequest+1@unsyncasyncdefprocess(task):
awaitasyncio.sleep(0.1)
returntask.result() *2start=time.time()
print(initiate(3).then(process).result())
print('Executed in {} seconds'.format(time.time() -start))Which prints:
8
Executed in 0.20314741134643555 seconds
We'll start by converting a regular synchronous function into a threaded Unfuture which will begin our request.
@unsyncdefnon_async_function(num):
time.sleep(0.1)
returnnum, num+1We may want to refine the result in another function, so we define the following continuation.
@unsyncasyncdefresult_continuation(task):
awaitasyncio.sleep(0.1)
num, res=task.result()
returnnum, res*2We then aggregate all the results into a single dictionary in an async function.
@unsyncasyncdefresult_processor(tasks):
output= {}
fortaskintasks:
num, res=awaittaskoutput[num] =resreturnoutputExecuting the full chain of non_async_function→result_continuation→result_processor would look like:
start=time.time()
print(result_processor([non_async_function(i).then(result_continuation) foriinrange(10)]).result())
print('Executed in {} seconds'.format(time.time() -start))Which prints:
{0: 2, 1: 4, 2: 6, 3: 8, 4: 10, 5: 12, 6: 14, 7: 16, 8: 18, 9: 20}
Executed in 0.22115683555603027 seconds
As far as we know it is not possible to change the return type of a method or function using a decorator. Therefore, we need a workaround to properly use IntelliSense. You have three options in general:
Ignore type warnings.
Use a suppression statement where you reach the type warning.
A. When defining the unsynced method by changing the return type to an
Unfuture.B. When using the unsynced method.
Wrap the function without a decorator. Example:
deffunction_name(x: str) ->Unfuture[str]: async_method=unsync(__function_name_synced) returnasync_method(x) def__function_name_synced(x: str) ->str: returnx+'a'future_result=function_name('b') self.assertEqual('ba', future_result.result())