Exception when awaiting `call_async()`

[bananajr]

Generated by Generative AI

No response

Operating System:

ros:jazzy-ros-core (ubuntu:noble)

ROS version or commit hash:

Jazzy 7.1.9, Rolling 10.0.3

RMW implementation (if applicable):

rmw_cyclonedds_cpp

RMW Configuration (if applicable):

No response

Client library (if applicable):

rclpy

'ros2 doctor --report' output

No response

Steps to reproduce issue

1. Create a service that does not return immediately
2. Create a client of the service
3. Call the service via asyncio w/ await service.call_async(request)

Expected behavior

Service call is awaited in the client and returns successfully.

Actual behavior

All code that assumes call_async() is awaitable will, as of 7.1.9, exit on exception RuntimeError: Task got bad yield: <rclpy.task.Future object at 0x74d4e2623ef0> if the Future pends.

Additional information

This is a regression in Jazzy 7.1.9 due to the change to Future.__await__ in #1560, backported from #1469, Rolling 10.0.3 @bjsowa.

rclpy.Future implements __await__ indicating it is intended for use with asyncio. Previously, rclpy.Future.__await__ yielded None, which asyncio responds to by checking the future again at the next iteration of the event loop. As of #1560, it yields itself. Because rclpy.Future is not a "Future-like object" due to missing get_loop() and _asyncio_future_blocking apis (as discussed in #1399) asyncio throws the exception.

Noting that #1560 is a backport of #1469 into Rolling.

[bjsowa]

My understanding was that rclpy Future and Task object only align with the api of asyncio but rclpy does not officially support running tasks or awaiting futures in asyncio event loop. @nadavelkabets ?

[nadavelkabets]

My understanding was that rclpy Future and Task object only align with the api of asyncio but rclpy does not officially support running tasks or awaiting futures in asyncio event loop. @nadavelkabets ?

Indeed. They were never designed to work with asyncio, only with the Executor.
I wouldn’t even call this a regression since previously if someone were to await the rclpy Future using asyncio that would result in a busy loop and cause the asyncio event loop to hog the cpu at 100% utilization.

To work around this, you can chain the returned rclpy Future to an asyncio Future, similar to the native asyncio wrap of the concurrent.futures Future object.
For this to work the event loop and rclpy Executor must run in the same thread, otherwise asyncio calls from the Executor threads must be wrapped with loop.call_soon_threadsafe().

    import asyncio
    from rclpy.task import Future
    def wrap_future(rclpy_future: Future) -> asyncio.Future:
        """
        Chain two futures so that when one completes, so does the other.

        The result (or exception) of source will be copied to destination.
        If destination is cancelled, source gets cancelled too.
        """
        asyncio_future = asyncio.get_running_loop().create_future()

        def _call_check_cancel(_: asyncio.Future):
            if asyncio_future.cancelled():
                rclpy_future.cancel()

        def _call_set_state(_: Future):
            if asyncio_future.cancelled():
                return
            if rclpy_future.cancelled():
                asyncio_future.cancel()
            else:
                exception = rclpy_future.exception()
                if exception is not None:
                    asyncio_future.set_exception(exception)
                else:
                    result = rclpy_future.result()
                    asyncio_future.set_result(result)

        asyncio_future.add_done_callback(_call_check_cancel)
        rclpy_future.add_done_callback(_call_set_state)
        return asyncio_future

[bananajr]

Thanks for the quick reply and workaround

My understanding was that rclpy Future and Task object only align with the api of asyncio but rclpy does not officially support running tasks or awaiting futures in asyncio event loop

I'm curious why the API to rclpy.Future was made to align with asyncio, including the dunder method __await__; one would need to go quite far out of their way to call this other than via asyncio await. It also seems that test_create_task_coroutine_wake_from_another_thread awaits on an rclpy.Future. This does make it appear that rclpy.Future supports await properly, and I worry that there are many codebases out there poised to break in the way mine did.

I ran unit tests from 7.1.9 against python 3.12.3 and test_create_task_coroutine_wake_from_another_thread passes without exception. My code also runs python 3.12.3 -- the same asyncio code that introspects the values yielded from rclpy.Future.__await__() which appears to just check if the Future has the _asyncio_future_blocking attribute -- but crashes on the RuntimeError: bad yield exception.

[bjsowa]

I'm curious why the API to rclpy.Future was made to align with asyncio, including the dunder method __await__; one would need to go quite far out of their way to call this other than via asyncio await. It also seems that test_create_task_coroutine_wake_from_another_thread awaits on an rclpy.Future. This does make it appear that rclpy.Future supports await properly, and I worry that there are many codebases out there poised to break in the way mine did.

await is not strictly tied to asyncio. rclpy Executor supports running coroutines as callbacks (async def instead of def) which await on other Tasks or Futures running on the same executor. Here's an example of how I utilized coroutines in rclpy.

EDIT: Just one thing worth remembering when using async/await in rclpy is to create separate callback groups for each coroutine. Otherwise await will block other callbacks from executing.

[bananajr]

Again thanks. TIL a lot more about how await & asyncio work. I will implement the much appreciated workaround given by @nadavelkabets along with your advice re: callback groups.

I believe the point still stands that others may have made the same mistake as me of assuming that rclpy.Future implements the "Future-like object" API sufficiently that they can use it with asyncio, and will encounter this same RuntimeError when they upgrade to 7.1.9.