From 0857a8ff116d8f83a831b1669effe5bd11a17d4c Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Fri, 8 May 2026 15:05:11 +0900 Subject: [PATCH 1/7] remove deadlocks due to main thread unsubscription and background thread websocket listener on unsubscribe event --- .../controllerwebclientraw.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index 078207b..af95401 100644 --- a/python/mujinwebstackclient/controllerwebclientraw.py +++ b/python/mujinwebstackclient/controllerwebclientraw.py @@ -634,10 +634,16 @@ 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=5.0) + 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): """Unsubscribes to Mujin controller. @@ -679,4 +685,6 @@ async def _Unsubscribe(): return # actually unsubscribe and wait until there is a result - self._backgroundThread.RunCoroutine(_Unsubscribe()).result() + future = self._backgroundThread.RunCoroutine(_Unsubscribe()) + # Wait outside _subscriptionLock to avoid deadlocks with async unsubscribe handling + future.result(timeout=5.0) From c11b39b2dba6ea6a0d68fc543fd12d9f9d66f599 Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Fri, 8 May 2026 15:33:32 +0900 Subject: [PATCH 2/7] change the shared dictionary onlu under lock to avouid disasters --- .../controllerwebclientraw.py | 46 ++++++++++++------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index af95401..52202c1 100644 --- a/python/mujinwebstackclient/controllerwebclientraw.py +++ b/python/mujinwebstackclient/controllerwebclientraw.py @@ -655,26 +655,40 @@ 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))) + with self._subscriptionLock: + if not self._IsWebSocketConnectionOpen(): + return + # 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=5.0) + 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) + + # close the websocket connection if no more subscribers are left + if len(self._subscriptions) == 0 and self._IsWebSocketConnectionOpen(): + self._backgroundThread.RunCoroutine(self._CloseWebSocket()) + with self._subscriptionLock: # nothing to do if websocket is not established if not self._IsWebSocketConnectionOpen(): From 43519914975227c8e8232a3beb4e2ac625a4a5d3 Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Fri, 8 May 2026 16:26:49 +0900 Subject: [PATCH 3/7] remove dead code --- .../mujinwebstackclient/controllerwebclientraw.py | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index 52202c1..062fcc6 100644 --- a/python/mujinwebstackclient/controllerwebclientraw.py +++ b/python/mujinwebstackclient/controllerwebclientraw.py @@ -688,17 +688,3 @@ async def _Unsubscribe(): # close the websocket connection if no more subscribers are left if len(self._subscriptions) == 0 and self._IsWebSocketConnectionOpen(): self._backgroundThread.RunCoroutine(self._CloseWebSocket()) - - with self._subscriptionLock: - # 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: - return - - # actually unsubscribe and wait until there is a result - future = self._backgroundThread.RunCoroutine(_Unsubscribe()) - # Wait outside _subscriptionLock to avoid deadlocks with async unsubscribe handling - future.result(timeout=5.0) From 83c6b8f0011a9d8091da5e6cbd755b26da77f1f9 Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Fri, 8 May 2026 16:57:51 +0900 Subject: [PATCH 4/7] configurable timeouts --- python/mujinwebstackclient/controllerwebclientraw.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index 062fcc6..1246543 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: @@ -638,14 +638,14 @@ async def _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=5.0) + 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 - def UnsubscribeGraphAPI(self, subscription: Subscription): + def UnsubscribeGraphAPI(self, subscription: Subscription, timeout: float = 5.0): """Unsubscribes to Mujin controller. Args: @@ -677,7 +677,7 @@ async def _Unsubscribe(): future = self._backgroundThread.RunCoroutine(_Unsubscribe()) try: # wait for the async result outside the lock - future.result(timeout=5.0) + future.result(timeout=timeout) except Exception as e: log.exception('timeout or error while unsubscribing: %s', e) From 84f127bdac41ac2c0ebc787e1e224b68695e06bb Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Mon, 11 May 2026 16:00:16 +0900 Subject: [PATCH 5/7] fix comments and capitalization --- python/mujinwebstackclient/controllerwebclientraw.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/python/mujinwebstackclient/controllerwebclientraw.py b/python/mujinwebstackclient/controllerwebclientraw.py index 1246543..d54a151 100644 --- a/python/mujinwebstackclient/controllerwebclientraw.py +++ b/python/mujinwebstackclient/controllerwebclientraw.py @@ -636,7 +636,7 @@ async def _Subscribe(): # wait until the subscription is created future = self._backgroundThread.RunCoroutine(_Subscribe()) try: - # Wait for the subscribe outside _subscriptionLock to avoid deadlocking + # 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: @@ -668,6 +668,7 @@ async def _Unsubscribe(): await self._StopAllSubscriptions(ControllerGraphClientException(_('Failed to unsubscribe: %s') % (e))) with self._subscriptionLock: + # nothing to do if websocket is not established if not self._IsWebSocketConnectionOpen(): return # check if self._subscriptionIds has subscriptionId @@ -681,7 +682,7 @@ async def _Unsubscribe(): 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 + # re-acquire lock to safely modify the dictionary and check for shutdown with self._subscriptionLock: self._subscriptions.pop(subscriptionId, None) From c1eeca9b4c6b6a35a6ce036589779e5431129eee Mon Sep 17 00:00:00 2001 From: "mohnish.wadhwa" Date: Mon, 11 May 2026 16:04:16 +0900 Subject: [PATCH 6/7] update changelog and version --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) 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. From 5fff123fea9803ddb4017d81b22ea3572a0c1a5c Mon Sep 17 00:00:00 2001 From: Barkin Simsek Date: Tue, 12 May 2026 10:55:51 +0900 Subject: [PATCH 7/7] Bump the version as well. --- python/mujinwebstackclient/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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