Exception in on_socket_open: Non-thread-safe operation invoked on an event loop other than the current one

[flyte]

When running with the PYTHONASYNCIODEBUG=1 environment variable set, using the following code:

import asyncio

from asyncio_mqtt import Client


async def main():
    client = Client("test.mosquitto.org")

    print("Connecting...")
    await client.connect()
    print("Connected!")

    print("Sleeping...")
    await asyncio.sleep(5)
    print("Slept!")

    print("Disconnecting...")
    await client.disconnect()
    print("Disconnected!")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()
        loop.stop()

Connecting to an MQTT server raises the following exceptions:

$ PYTHONASYNCIODEBUG=1 PYTHONTRACEMALLOC=1 python3.8 test.py 
Connecting...
Caught exception in on_socket_open: Non-thread-safe operation invoked on an event loop other than the current one
Traceback (most recent call last):
  File "test.py", line 25, in <module>
    loop.run_until_complete(main())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 608, in run_until_complete
    return future.result()
  File "test.py", line 10, in main
    await client.connect()
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/asyncio_mqtt/client.py", line 79, in connect
    await loop.run_in_executor(None, self._client.connect, self._hostname, self._port, 60)
  File "/usr/lib/python3.8/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 941, in connect
    return self.reconnect()
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 1117, in reconnect
    self._call_socket_open()
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 2071, in _call_socket_open
    self.on_socket_open(self, self._userdata, self._sock)
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/asyncio_mqtt/client.py", line 303, in _on_socket_open
    self._misc_task = self._loop.create_task(self._misc_loop())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 427, in create_task
    task = tasks.Task(coro, loop=self, name=name)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 713, in call_soon
    self._check_thread()
  File "/usr/lib/python3.8/asyncio/base_events.py", line 750, in _check_thread
    raise RuntimeError(
RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one
Caught exception in on_socket_close: 'NoneType' object has no attribute 'cancel'
Exception ignored in: <function Client.__del__ at 0x7f298a709dc0>
Traceback (most recent call last):
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 660, in __del__
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 704, in _reset_sockets
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 698, in _sock_close
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 2105, in _call_socket_close
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/asyncio_mqtt/client.py", line 308, in _on_socket_close
AttributeError: 'NoneType' object has no attribute 'cancel'
sys:1: RuntimeWarning: coroutine 'Client._misc_loop' was never awaited
Task was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "/usr/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.8/concurrent/futures/thread.py", line 80, in _worker
    work_item.run()
  File "/usr/lib/python3.8/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 941, in connect
    return self.reconnect()
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 1117, in reconnect
    self._call_socket_open()
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/paho/mqtt/client.py", line 2071, in _call_socket_open
    self.on_socket_open(self, self._userdata, self._sock)
  File "/home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/asyncio_mqtt/client.py", line 303, in _on_socket_open
    self._misc_task = self._loop.create_task(self._misc_loop())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 427, in create_task
    task = tasks.Task(coro, loop=self, name=name)
task: <Task pending name='Task-2' coro=<Client._misc_loop() running at /home/flyte/workspaces/asyncio-mqtt/ve/lib/python3.8/site-packages/asyncio_mqtt/client.py:318> created at /usr/lib/python3.8/asyncio/base_events.py:427>

[flyte]

It's because the self._client.connect() call is done in an executor (because it blocks on socket creation).

[flyte]

Added #41 to fix the issue in the second exception too.

[cmur2]

I'm experiencing the same problem - thanks for already finding a fix!

Is the recommended workaround to use the master version until there is a new release?

[frederikaalund]

@cmur2 Version 0.9.0 is now live. It includes the fix for this issue. 👍