RuntimeError: Event loop is closed

[rnovatorov]

Hello @empicano, thank you for the library.

I have a Python program that spawns a bunch of concurrent asyncio.Tasks one of which uses aiomqtt to communicate with an MQTT broker. When interrupted by Ctrl+C, the program cancels all the tasks, terminates the event loop and exits. Usually it works just fine, but sometimes I get a RuntimeError indicating that aiomqtt tries to operate on a closed event loop.

Here are the versions of packages I use.

Package   Version
--------- -------
aiomqtt   2.4.0
paho-mqtt 2.1.0

Here's an example to reproduce the issue.

# script.py

import asyncio

import aiomqtt


async def main():
    task = asyncio.create_task(run_client())
    await asyncio.sleep(0)
    task.cancel()
    await task


async def run_client():
    async with aiomqtt.Client("test.mosquitto.org"):
        await asyncio.Event().wait()


if __name__ == "__main__":
    asyncio.run(main())

Here's a traceback I get running the example above.

Traceback (most recent call last):
  File "/app/script.py", line 19, in <module>
    asyncio.run(main())
    ~~~~~~~~~~~^^^^^^^^
  File "/usr/local/lib/python3.14/asyncio/runners.py", line 204, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "/usr/local/lib/python3.14/asyncio/runners.py", line 127, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "/usr/local/lib/python3.14/asyncio/base_events.py", line 719, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "/app/script.py", line 10, in main
    await task
  File "/app/script.py", line 14, in run_client
    async with aiomqtt.Client("test.mosquitto.org"):
               ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.14/site-packages/aiomqtt/client.py", line 750, in __aenter__
    await loop.run_in_executor(
    ...<9 lines>...
    )
asyncio.exceptions.CancelledError
Caught exception in on_socket_close: Event loop is closed
Exception ignored while calling deallocator <function Client.__del__ at 0x733090407320>:
Traceback (most recent call last):
  File "/usr/local/lib/python3.14/site-packages/paho/mqtt/client.py", line 880, in __del__
  File "/usr/local/lib/python3.14/site-packages/paho/mqtt/client.py", line 1139, in _reset_sockets
  File "/usr/local/lib/python3.14/site-packages/paho/mqtt/client.py", line 1132, in _sock_close
  File "/usr/local/lib/python3.14/site-packages/paho/mqtt/client.py", line 2867, in _call_socket_close
  File "/usr/local/lib/python3.14/site-packages/aiomqtt/client.py", line 705, in _on_socket_close
  File "/usr/local/lib/python3.14/asyncio/base_events.py", line 872, in call_soon_threadsafe
  File "/usr/local/lib/python3.14/asyncio/base_events.py", line 550, in _check_closed
RuntimeError: Event loop is closed
Task was destroyed but it is pending!
task: <Task pending name='Task-5' coro=<Client._misc_loop() running at /usr/local/lib/python3.14/site-packages/aiomqtt/client.py:738> wait_for=<Future pending cb=[Task.task_wakeup()]>>

As far as I understand aiomqtt.Client.__aenter__ begins setup, but gets cancelled, so aiomqtt.Client.__aexit__ doesn't have a chance to run and clean things up. Then paho.mqtt.client.Client calls aiomqtt.Client._on_socket_close while being garbage-collected. At that time the event loop has already been closed.

[rnovatorov]

I ended up creating a wrapper that protects aiomqtt.Client.__aenter__ from cancellation.

Code

@contextlib.asynccontextmanager
async def new_client(*args, **kwargs):
    client = aiomqtt.Client(*args, **kwargs)
    setup_task = asyncio.create_task(client.__aenter__())
    try:
        await asyncio.shield(setup_task)
    except asyncio.CancelledError as e:
        await setup_task
        await client.__aexit__(type(e), e, e.__traceback__)
        raise
    try:
        yield client
    except BaseException as e:
        await client.__aexit__(type(e), e, e.__traceback__)
        raise
    else:
        await client.__aexit__(None, None, None)

---

It doesn't protect __aexit__ from cancellation though. It might be useful to include information about cancellation safety in the docs.