diff --git a/CHANGELOG.md b/CHANGELOG.md index dc19e71..824cc8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## 0.9.33 (2026-05-11) + +- Removed a deadlock that occurred when unsubscribing from an active subscription while a background thread attempted to acquire a subscription lock already held by the main subscribing thread. + ## 0.9.32 (2026-04-02) - Re-generate graph api. diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index 078207b..d54a151 100644 --- a/python/mujinwebstackclient/controllerwebclientraw.py +++ b/python/mujinwebstackclient/controllerwebclientraw.py @@ -602,7 +602,7 @@ async def _StopAllSubscriptions(self, error: Optional[ControllerGraphClientExcep subscription.GetSubscriptionCallbackFunction()(error=error, response=None) self._subscriptions.clear() - def SubscribeGraphAPI(self, query: str, callbackFunction: Callable[[Optional[ControllerGraphClientException], Optional[dict]], None], variables: Optional[dict] = None) -> Subscription: + def SubscribeGraphAPI(self, query: str, callbackFunction: Callable[[Optional[ControllerGraphClientException], Optional[dict]], None], variables: Optional[dict] = None, timeout: float = 5.0) -> Subscription: """Subscribes to changes on Mujin controller. Args: @@ -634,12 +634,18 @@ async def _Subscribe(): self._EnsureWebSocketConnection() # wait until the subscription is created - self._backgroundThread.RunCoroutine(_Subscribe()).result() + future = self._backgroundThread.RunCoroutine(_Subscribe()) + try: + # wait for the subscribe outside _subscriptionLock to avoid deadlocking + # with websocket callbacks that may acquire the same lock while resolving + future.result(timeout=timeout) + except Exception as e: + raise ControllerGraphClientException(f'Failed to subscribe within timeout: {e}') + with self._subscriptionLock: self._subscriptions[subscriptionId] = subscription + return subscription - return subscription - - def UnsubscribeGraphAPI(self, subscription: Subscription): + def UnsubscribeGraphAPI(self, subscription: Subscription, timeout: float = 5.0): """Unsubscribes to Mujin controller. Args: @@ -649,22 +655,14 @@ def UnsubscribeGraphAPI(self, subscription: Subscription): async def _Unsubscribe(): try: - # check if self._subscriptionIds has subscriptionId - if subscriptionId in self._subscriptions: - await self._webSocket.send( - json.dumps( - { - 'id': subscriptionId, - 'type': 'stop', - }, - ), - ) - # remove subscription - self._subscriptions.pop(subscriptionId, None) - - # close the websocket connection if no more subscribers are left - if len(self._subscriptions) == 0: - await self._CloseWebSocket() + await self._webSocket.send( + json.dumps( + { + 'id': subscriptionId, + 'type': 'stop', + }, + ), + ) except Exception as e: log.exception('caught WebSocket exception: %s', e) await self._StopAllSubscriptions(ControllerGraphClientException(_('Failed to unsubscribe: %s') % (e))) @@ -673,10 +671,21 @@ async def _Unsubscribe(): # nothing to do if websocket is not established if not self._IsWebSocketConnectionOpen(): return - - # check if the subscription exists at all - if subscription.GetSubscriptionID() not in self._subscriptions: + # check if self._subscriptionIds has subscriptionId + if subscriptionId not in self._subscriptions: return + # request unsubscribe under lock + future = self._backgroundThread.RunCoroutine(_Unsubscribe()) + try: + # wait for the async result outside the lock + future.result(timeout=timeout) + except Exception as e: + log.exception('timeout or error while unsubscribing: %s', e) + + # re-acquire lock to safely modify the dictionary and check for shutdown + with self._subscriptionLock: + self._subscriptions.pop(subscriptionId, None) - # actually unsubscribe and wait until there is a result - self._backgroundThread.RunCoroutine(_Unsubscribe()).result() + # close the websocket connection if no more subscribers are left + if len(self._subscriptions) == 0 and self._IsWebSocketConnectionOpen(): + self._backgroundThread.RunCoroutine(self._CloseWebSocket()) diff --git a/python/mujinwebstackclient/version.py b/python/mujinwebstackclient/version.py index bd57f99..efa7a71 100644 --- a/python/mujinwebstackclient/version.py +++ b/python/mujinwebstackclient/version.py @@ -1,3 +1,3 @@ -__version__ = '0.9.32' +__version__ = '0.9.33' # Do not forget to update CHANGELOG.md