`aiogrpc` RuntimeError

[ShafSpecs]

Hello, so I was building a simple GCS. A simple UI on the front and a python server at the back.

Using websocket to handle commands and Server-Sent Events (SSE) for unidirectional information (telemetry). Whilst building out the telemetry aspect, I noticed that RuntimeError randomly got thrown.

Exception ignored in: <function WrappedIterator.__del__ at 0x00000202DCC768E0>
Traceback (most recent call last):
  File "C:\Users\user\app\.venv\Lib\site-packages\aiogrpc\utils.py", line 168, in __del__
    self._stream_executor.shutdown()
  File "C:\Python312\Lib\concurrent\futures\thread.py", line 238, in shutdown
    t.join()
  File "C:\Python312\Lib\threading.py", line 1144, in join
    raise RuntimeError("cannot join current thread")
RuntimeError: cannot join current thread

Depending on how I structured the SSE route, it got thrown frequently or once in a while. A basic structure:

# sse route
    async def event_generator():
        while True:
            if await request.is_disconnected():
                print("Gracefully stopping stream...")
                break

            # uncommenting these lines instead gets the error thrown more frequently
            #
            # battery = ensure_future(get_battery())
            # gps = ensure_future(get_gps())
            # position = ensure_future(get_position())
            # heading = ensure_future(get_heading())
            # flight_mode = ensure_future(get_flight_mode())
            # fw_metrics = ensure_future(get_fw_metrics())
            # # is_in_air = ensure_future(get_is_in_air())
            # # landed_state = ensure_future(get_landed_state())
            # # mission_progress = ensure_future(get_mission_progress())
            # home = ensure_future(get_home())
            # health = ensure_future(get_health())
            # # status = ensure_future(get_status_text())
            # rc_status = ensure_future(get_rc_status())

            battery = await get_battery()
            gps = await get_gps()
            position = await get_position()
            heading = await get_heading()
            flight_mode = await get_flight_mode()
            fw_metrics = await get_fw_metrics()
            # is_in_air = ensure_future(get_is_in_air())
            # landed_state = ensure_future(get_landed_state())
            # mission_progress = ensure_future(get_mission_progress())
            home = await get_home()
            health = await get_health()
            # status = ensure_future(get_status_text())
            rc_status = await get_rc_status()

            yield encode({
                "event": "telemetry",
                "data": {
                    "battery": battery,
                    "gps": gps,
                    "position": position,
                    "heading": heading,
                    "flight_mode": flight_mode,
                    "fw_metrics": fw_metrics,
                    # "is_in_air": await is_in_air,
                    # "landed_state": await landed_state,
                    # "mission_progress": await mission_progress,
                    "home": home,
                    "health": health,
                    # "info": info,
                    # "status": await status,
                    "rc_status": rc_status,
                }
            }).decode("utf-8")

            await sleep(.5)

    res = EventSourceResponse(
        content=event_generator(),
        status_code=200,
        media_type="text/event-stream",
        headers={
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Credentials": "true",
        }
    )
    return res

The error comes from mavsdk dependency: aiogrpc. I am not well-versed with multi-threading in python but I assume the issue stems from aiogrpc creating its own threads and attempting to join them. Is there any way I can disable that? Possibly use asyncio fully for running tasks on a separate thread? Or maybe find a fix I am overlooking. Thanks

[ShafSpecs]

Digging further, it seems the error is thrown (raised) internally as wrapping the whole code in a try...except block doesn't catch it.

[julianoes]

I have seen similar annoying exceptions when using it more recently.

Can you run mavsdk_server separate and check whether it doesn't crash or lockup?

[julianoes]

I ended up doing something like this, not sure if that would help your case:

# When mavsdk_server, which runs separately, crashes, we need to redo all
# mavsdk subscriptions. To do so, it's easiest just to restart this app
# entirely. We can detect that mavsdk_server has restarted by reading through
# the# aiogrpc.utils logs. Unfortunately, I didn't find a better way to catch
# this case.
class ErrorLogHandler(logging.Handler):
    def emit(self, record):
        log_entry = self.format(record)
        if "grpc exception" in log_entry and "StatusCode.UNAVAILABLE" in log_entry:
            print("Triggering restart due to gRPC failure.")
            loop = asyncio.get_event_loop()
            loop.stop()


aio_logger = logging.getLogger("aiogrpc.utils")
aio_logger.addHandler(ErrorLogHandler())

[ShafSpecs]

I have seen similar annoying exceptions when using it more recently.

Can you run mavsdk_server separate and check whether it doesn't crash or lockup?

That's what I am doing. I am using windows which means I have to start the server process right before each connection:

async def start_mavsdk_server(connection_url: str, port: int):
    cmd = [resource_path('include/mavsdk_server.exe'), "-p", str(port), connection_url]

    mavsdk_server_process = Popen(
        cmd,
        stdout=PIPE,
        stderr=PIPE,
        universal_newlines=True,
        start_new_session=True,
    )

    return mavsdk_server_process

Still raises the exception

[ShafSpecs]

I ended up doing something like this, not sure if that would help your case:

# When mavsdk_server, which runs separately, crashes, we need to redo all
# mavsdk subscriptions. To do so, it's easiest just to restart this app
# entirely. We can detect that mavsdk_server has restarted by reading through
# the# aiogrpc.utils logs. Unfortunately, I didn't find a better way to catch
# this case.
class ErrorLogHandler(logging.Handler):
    def emit(self, record):
        log_entry = self.format(record)
        if "grpc exception" in log_entry and "StatusCode.UNAVAILABLE" in log_entry:
            print("Triggering restart due to gRPC failure.")
            loop = asyncio.get_event_loop()
            loop.stop()


aio_logger = logging.getLogger("aiogrpc.utils")
aio_logger.addHandler(ErrorLogHandler())

Would try adding this to my event generator and seeing if it helps 👍. Does this restart the entire app or just the mavsdk server? Due to the nature of sockets, restarting the entire app would pose an issue due to socket client behaviour.

Would I also need to restart mavsdk server manually after stopping the event loop?

[julianoes]

For me this restarts the full app which is a quart server in my case.

[dp1140a]

I get a similar error when running the examples. However it seems to be intermittent (The worst kind of error :) ) Takeoff and Land:

#!/usr/bin/env python3

import asyncio
from mavsdk import System


async def run():

    drone = System()
    await drone.connect(system_address="udp://:14540")

    status_text_task = asyncio.ensure_future(print_status_text(drone))

    print("Waiting for drone to connect...")
    async for state in drone.core.connection_state():
        if state.is_connected:
            print(f"-- Connected to drone!")
            break

    print("Waiting for drone to have a global position estimate...")
    async for health in drone.telemetry.health():
        if health.is_global_position_ok and health.is_home_position_ok:
            print("-- Global position estimate OK")
            break

    print("-- Arming")
    await drone.action.arm()

    print("-- Taking off")
    await drone.action.takeoff()

    await asyncio.sleep(10)

    print("-- Landing")
    await drone.action.land()

    #status_text_task.cancel()


async def print_status_text(drone):
    try:
        async for status_text in drone.telemetry.status_text():
            print(f"Status: {status_text.type}: {status_text.text}")
    except asyncio.CancelledError:
        return


if __name__ == "__main__":
    # Run the asyncio loop
    asyncio.run(run())

It runs seemingly ok. Drone takes off and lands. but then it throws the error:

python test1.py 
Waiting for drone to connect...
-- Connected to drone!
Waiting for drone to have a global position estimate...
-- Global position estimate OK
-- Arming
-- Taking off
Status: INFO: Armed by external command	
Status: INFO: Using default takeoff altitude: 10.0 m	
Status: INFO: [logger] ./log/2024-05-20/23_47_00.ulg	
Status: INFO: Using default takeoff altitude: 10.0 m	
Status: INFO: Takeoff detected	
-- Landing
Exception ignored in: <function WrappedIterator.__del__ at 0x710c78765750>
Traceback (most recent call last):
  File "/home/dave/.local/lib/python3.10/site-packages/aiogrpc/utils.py", line 168, in __del__
    self._stream_executor.shutdown()
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 235, in shutdown
    t.join()
  File "/usr/lib/python3.10/threading.py", line 1093, in join
    raise RuntimeError("cannot join current thread")
RuntimeError: cannot join current thread

[julianoes]

@JonasVautherin any ideas?

[ShafSpecs]

Getting a new error, still stemming from aiogrpc:

__anext__ grpc exception
Traceback (most recent call last):
  File "C:\Users\user\project\.venv\Lib\site-packages\aiogrpc\utils.py", line 145, in cb
    fut.result()
  File "C:\Python312\Lib\concurrent\futures\thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\project\.venv\Lib\site-packages\aiogrpc\utils.py", line 131, in _next
    return next(self._iterator)
           ^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\project\.venv\Lib\site-packages\grpc\_channel.py", line 542, in __next__
    return self._next()
           ^^^^^^^^^^^^
  File "C:\Users\user\project\.venv\Lib\site-packages\grpc\_channel.py", line 968, in _next
    raise self
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
	status = StatusCode.UNKNOWN
	details = "Unexpected error in RPC handling"
	debug_error_string = "UNKNOWN:Error received from peer  {created_time:"2024-05-28T11:11:28.0840526+00:00", grpc_status:2, grpc_message:"Unexpected error in RPC handling"}"
>

This occurs more rarely, but it gives my systems a bit of unpredictability when trying to read telemetry consistently

Any ideas? cc @JonasVautherin

[ShafSpecs]

Overall, I am thinking if it's possible to dump aiogrpc (which hasn't been updated in 4 years) and roll with something like: grpclib which still receives updates and works natively with asyncio? I am not sure how tightly coupled this project is with aiogrpc. Would take a look at the feasibility of such conversion 👍

[ShafSpecs]

Overall, I am thinking if it's possible to dump aiogrpc (which hasn't been updated in 4 years) and roll with something like: grpclib which still receives updates and works natively with asyncio? I am not sure how tightly coupled this project is with aiogrpc. Would take a look at the feasibility of such conversion 👍

It doesn't look too complicated to facilitate a switch, but I do not know what I am doing yet. aiogrpc is used in just the async plugin manager. Not sure about the pbs files generation though, as grpclib does have a protoc plugin: https://github.com/vmagamedov/grpclib#protoc-plugin

[julianoes]

I didn't realize we could swap out aiogrpc. That sounds reasonable.

[ShafSpecs]

I didn't realize we could swap out aiogrpc. That sounds reasonable.

Is this something you intend on working on in the near future?

[julianoes]

Probably not in the next couple of weeks. Is this urgent for you, or are you asking to try yourself?

[ShafSpecs]

Probably not in the next couple of weeks. Is this urgent for you, or are you asking to try yourself?

Both. It's a bit urgent, and I am wondering if I can do it. Asides changing imports in the async plugin manager, I am not sure if I would need to re-generate any file or the current proto files generation are good to go