difference in calling async code ?

[zcattacz]

async def wrapped_fn()
     # do_something
    ...

async def _warper1()
    await wrapped_fn()

def _wrapper2()
    asyncio.get_event_loop().run_until_complete(wrapped_fn())

async def wrapper():
    await _wrapp1() # method 1
    _wrapp2() # method 2

In actual execution, Is there any difference between method1 and method2 ?
I want to create a rpc wrapper for both async and sync function, since async code is very contagious, I found it hard to share common processing code between sync and async interface. Currently I am doing method1, I wonder if I use method2, if there will be any implication ?

[palonza]

Hi @zcattacz,

if you, use the _wrapper2, that is a synchronous function, it will block the execution of the code until wrapped_fn completes. It would be more appropriate to use method 1. This way, you can handle the asynchronous code properly and ensure the execution flow is not blocked.

[zcattacz]

I think they work the same, here is the toy code I used to test my assumption.

The inner asyncio.sleep seems to pause code execution up to the coro task from which the intermediate synchronous function is invoked from, which is what I need. It seems to work just like called from a chain of async functions.
But the test is simple, asyncio is still new to me. I am not sure if there are implications I should know.

import uasyncio as asyncio

async def wrapped():
    for i in range(0,10):
        print(i,end="")
        await asyncio.sleep_ms(200)

def sync_wrapper(fn, aloop=None):
    # some parameter validation, context setup in the wrapper
    # ...
    for i in range(0,3):
        if aloop:
            aloop.run_until_complete(wrapped())
        else:
            if wrapped.__class__.__name__ == "generator":
                raise Exception()
            else:
                wrapped()

async def async_wrapper1(fn):
    # copy paste sync_wrapper processing code here :-(
    for i in range(0,3):
        await fn()

async def async_wrapper2(fn):
    aloop = asyncio.get_event_loop()
    sync_wrapper(fn, aloop=aloop)

async def main():
    asyncio.create_task(async_wrapper1(wrapped)) # method1
    asyncio.create_task(async_wrapper2(wrapped)) # method2
    await asyncio.sleep(0)
    for i in range(0,20):
        await asyncio.sleep_ms(500)
        print('M',end="")
    print()
asyncio.run(main())

#output :  
#001122M3344M556677M8899M001122M3344M556677M8899M001122M3344M556677M8899MMMMMMMMM

[zcattacz]

With CPython this only lead to RuntimeError('This event loop is already running')

Task exception was never retrieved
future: <Task finished name='Task-3' coro=<async_wrapper2() done, defined at /home/user/repos/micropython/ujrpc/test_async.py:36> exception=RuntimeError('This event loop is already running')>
Traceback (most recent call last):
  File "/home/user/repos/micropython/ujrpc/test_async.py", line 38, in async_wrapper2
    sync_wrapper(fn, aloop=aloop)
  File "/home/user/repos/micropython/ujrpc/test_async.py", line 20, in sync_wrapper
    aloop.run_until_complete(wrapped())
  File "/usr/lib/python3.9/asyncio/base_events.py", line 618, in run_until_complete
    self._check_running()
  File "/usr/lib/python3.9/asyncio/base_events.py", line 578, in _check_running
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running
/usr/lib/python3.9/asyncio/base_events.py:1875: RuntimeWarning: coroutine 'wrapped' was never awaited
  handle = self._ready.popleft()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

[zcattacz]

OK, I saw discussion in this SO post, According to the comments:

run_until_complete is not for running any number of arbitrary async functions synchronously, it is for running the main entry point of your entire async program. This constraint is not immediately apparent from the docs.

I guess the above code is not supposed to work, though currently mpy implementation made it appears to work.

The above code works in cpython with asyncio.ensure_future(wrapped(), loop=aloop).

[zcattacz]

False claim, not supposed to work (check issue #8515), will cause unpredicated behaviour.

[zcattacz]

When handle_rpca is coded like async_wrapper2 method, publish will never happen, as if it has never been called.
but when handle_rpca is coded like async_wrapper1, everything work just ok like before ...
I wonder what could happen to mess up the next await. ?

    async def _handle_rpc(self, json_string):
        jrpc.debug=False
        ret = await jrpc.handle_rpca(json_string)
        print("agent._handle_rpc: ret:", ret) # ret is confirmed to be correctly returned json string
        await self.cln.publish(self.tpc_rpc+"/out", ret) # but here nothing happened.
        # asyncio.create_task(self.cln.publish(self.tpc_rpc+"/out", ret)) this works as expected.

[peterhinch]

In general there is no need to use asyncio.get_event_loop() or asyncio.run_until_complete. They are part of the old paradigm before the advent of asyncio.run(). I have yet to encounter a valid reason to use these.

[zcattacz]

I have a jsonrpc wrapper for both synchronous/asynchronous calls
I tried to share the common code between sync/async handling, but I found the following structures impossible to share, though they are almost identical.

    def handle_rpc(self, request): # handle from synchronous call
        _r = self.rsp2.copy()
        data = self._hndl_parsing(request, _r)
        if isinstance(data, dict):
            self._hndl_rpc(data,_r)
        elif isinstance(data, list):
            rets = []
            for batched_rpc in data:
                _r = self.rsp2.copy()
                self._hndl_rpc(batched_rpc,_r)
                rets.append(_r)
            return json.dumps(rets) if self.ret_str else rets
        return json.dumps(_r) if self.ret_str else _r
    
    async def handle_rpca(self, request): # handle from asynchronous call
        _r = self.rsp2.copy()
        data = self._hndl_parsing(request, _r)
        if isinstance(data, dict):
            await self._hndl_rpca(data,_r)
        elif isinstance(data, list):
            rets = []
            for batched_rpc in data:
                _r = self.rsp2.copy()
                await self._hndl_rpca(batched_rpc,_r)
                rets.append(_r)
            return json.dumps(rets) if self.ret_str else rets
        return json.dumps(_r) if self.ret_str else _r

with ensure_future, they can be reduced to

    def handle_rpc(self, request, aloop=None): # handle from synchronous call
        _r = self.rsp2.copy()
        data = self._hndl_parsing(request, _r)
        if isinstance(data, dict):
            self._hndl_rpc(data,_r, aloop)
        elif isinstance(data, list):
            rets = []
            for batched_rpc in data:
                _r = self.rsp2.copy()
                self._hndl_rpc(batched_rpc,_r, aloop)
                rets.append(_r)
            return json.dumps(rets) if self.ret_str else rets
        return json.dumps(_r) if self.ret_str else _r
    
    async def handle_rpca(self, request): # handle from asynchronous call
        aloop = asyncio.get_event_loop()
        return handle_rpc(request, aloop=aloop)

   # later.  in _hndl_rpc, I can check if aloop is None, 
   # and do asyncio.ensure_future(async_fn, aloop) for asynchronous callers.

The major problem with redundant code structure is testing, I have to do double testing for almost identical code. only because there is some await down there somewhere :-(.

[bixb922]

Yes, it's difficult to mix async with sync, and rightly so, async is best if used with async. For testing purposes I have wrapped async functions in asyncio.run(coro())

[zcattacz]

The ensure_future() is a really neat feature, it allows sync code to sit between async caller and async callee in my use case, allow code structure sharing between async and sync handling. Hope mpy can add future support some day.

[peterhinch]

Having followed this thread I still don't understand your overall system design. Evidently your application runs under uasyncio. Yet you're calling a handle_rpc method which evidently blocks, holding up all other tasks for the duration. Why?

[zcattacz]

Oh, no not necessarily under asyncio, I got the question in the process of updating the jrpc lib which is a utility class.
I was updating it to make the same lib also usable in async code.

If the caller is async and target function is async, the jrpc lib as a wrapper has to be async, all handling code in-between has to be plagued with async/await keywords. though as a utility class, the JSON RPC protocol wrapper itself has nothing to do with asyncio.

Then an async version of the lib will be a mere duplicate of the original synchronous version with some await/async keywords. This make the code maintenance difficult, since the same future changes have to be applied to both copies and tested seperately.

The ensure_future() in cpython makes the async interface handle_rpca() share code with the synchronous interface handle_rpc() without creating separate identical code.

[peterhinch]

tl;dr I can't think of a solution in MicroPython but I'll keep giving it some thought.

The following is offered in case it triggers an idea.

I have a similar situation with display drivers which may be used in synchronous or asynchronous code. These have a synchronous .show method which blocks. The driver detects whether uasyncio is running. If it is, .show clears an Event, launches a ._as_show task and returns immediately. ._as_show sets the Event when complete. The asynchronous coder can test the Event for completion or have a task wait on the Event, thus avoiding the error condition of two concurrent ._as_show tasks.

The crucial difference is that in my case the task is "fire and forget". In your case the synchronous caller must wait on the outcome of the asynchronous task. It's not obvious how this can be achieved without blocking (and stalling the task). If you think of a solution, please say.

[zcattacz]

Yes, the way I come up would be as follows :-D after future (pun intended) is available

async def showa():
    aloop = asyncio.get_event_loop()
    show(_aloop=aloop)

def show(_aloop=None):
    ... all other logic...
    _send(cmd, data, _aloop)
    ... all other logic ...
    for  ... :
        _send(cmd, data, _aloop)
    if ...:
        _send(cmd, data, _aloop)
    else:
        _send(cmd, data, _aloop)

def _send(_aloop=None):
    if _aloop:
        asyncio.ensure_future(as_socket.send(cmd_data), loop=_aloop)
    else:
        socket.send(cmd_data)