bluez: syncronize disconnect and cleanup

When the following sequence of events occured, `assert self._bus is None`
would be hit.

- A "Connected": False property change signal is sent over D-Bus.
- Bleak handles the event and calls the `_cleanup_all()` method.
- The `_cleanup_all()` sends a D-Bus message to unsubscribe from signals and awaits it.
- While the previous method is still awaiting, the disconnect method is called.
- This method hits an assertation because the connected property is false
  but there is still a D-Bus bus connection since `_cleanup_all()` is still using it.

This fixes the issue by making `_cleanup_all()` a syncronous method.
Since we are closing the bus connection, we should not need to remove
signal watchers first.

Fixes #641.

CHANGELOG.rst

@@ -43,6 +43,7 @@ Fixed
 * Fixed race on disconnect and cleanup of BlueZ matches when device disconnects early. Fixes #603.
 * Fixed memory leaks on Windows.
 * Fixed protocol error code descriptions on WinRT backend. Fixes #532.
+* Fixed race condition hitting assertation in BlueZ ``disconect()`` method. Fixes #641.
 
 
 `0.12.1`_ (2021-07-07)

bleak/backends/bluezdbus/client.py

@@ -22,7 +22,7 @@
 from bleak.backends.bluezdbus.descriptor import BleakGATTDescriptorBlueZDBus
 from bleak.backends.bluezdbus.scanner import BleakScannerBlueZDBus
 from bleak.backends.bluezdbus.service import BleakGATTServiceBlueZDBus
-from bleak.backends.bluezdbus.signals import MatchRules, add_match, remove_match
+from bleak.backends.bluezdbus.signals import MatchRules, add_match
 from bleak.backends.bluezdbus.utils import (
     assert_reply,
     extract_service_handle_from_path,
@@ -83,8 +83,6 @@ def __init__(self, address_or_ble_device: Union[BLEDevice, str], **kwargs):
 
         # used to override mtu_size property
         self._mtu_size: Optional[int] = None
-        # setup a lock to handle concurrency
-        self._cleanup_lock = asyncio.Lock()
 
         # get BlueZ version
         p = subprocess.Popen(["bluetoothctl", "--version"], stdout=subprocess.PIPE)
@@ -371,39 +369,12 @@ async def _disconnect_monitor(self) -> None:
             except Exception:
                 pass
 
-    async def _remove_signal_handlers(self) -> None:
+    def _cleanup_all(self) -> None:
         """
-        Remove all pending notifications of the client. This method is used to
-        free the DBus matches that have been established.
-        """
-        logger.debug(f"_remove_signal_handlers({self._device_path})")
-
-        if self._bus is None:
-            logger.debug("no bus object")
-            return
-
-        self._bus.remove_message_handler(self._parse_msg)
-
-        for rule in self._rules:
-            try:
-                await remove_match(self._bus, rule)
-            except Exception as e:
-                logger.error(
-                    f"Failed to remove match {rule.member} ({self._device_path}): {e}"
-                )
-        self._rules = []
-
-        # There is no need to call stop_notify() here since the the device is
-        # already disconnected and we are about to close the D-Bus message bus
-        # which has the effect of clearing all notifications in BlueZ.
-        self._subscriptions = []
-
-    def _disconnect_message_bus(self) -> None:
-        """
-        Free the resources allocated for both the DBus bus.
-        Use this method upon final disconnection.
+        Free all the allocated resource in DBus. Use this method to
+        eventually cleanup all otherwise leaked resources.
         """
-        logger.debug(f"_disconnect_message_bus({self._device_path})")
+        logger.debug(f"_cleanup_all({self._device_path})")
 
         if not self._bus:
             logger.debug(f"already disconnected ({self._device_path})")
@@ -422,14 +393,10 @@ def _disconnect_message_bus(self) -> None:
             # a stuck client.
             self._bus = None
 
-    async def _cleanup_all(self) -> None:
-        """
-        Free all the allocated resource in DBus. Use this method to
-        eventually cleanup all otherwise leaked resources.
-        """
-        async with self._cleanup_lock:
-            await self._remove_signal_handlers()
-            self._disconnect_message_bus()
+            # since we closed the bus, any notification subscripts no longer
+            # need to be removed.
+            self._rules = []
+            self._subscriptions = []
 
             # Reset all stored services.
             self.services = BleakGATTServiceCollection()
@@ -1098,11 +1065,9 @@ def _parse_msg(self, message: Message):
                         self._disconnect_monitor_event.set()
                         self._disconnect_monitor_event = None
 
-                    task = asyncio.ensure_future(self._cleanup_all())
+                    self._cleanup_all()
                     if self._disconnected_callback is not None:
-                        task.add_done_callback(
-                            lambda _: self._disconnected_callback(self)
-                        )
+                        self._disconnected_callback(self)
                     disconnecting_event = self._disconnecting_event
                     if disconnecting_event:
-                        task.add_done_callback(lambda _: disconnecting_event.set())
+                        disconnecting_event.set()