This is an automated email from the ASF dual-hosted git repository.
hsluoyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/casbin-pycasbin.git
The following commit(s) were added to refs/heads/master by this push:
new 4f42fb7 fix: notify watcher after releasing the enforcer lock (#408)
4f42fb7 is described below
commit 4f42fb759d32d3b9d64403546c2b014cb32266c2
Author: Yang Luo <[email protected]>
AuthorDate: Thu Aug 6 21:40:32 2026 +0800
fix: notify watcher after releasing the enforcer lock (#408)
---
casbin/synced_enforcer.py | 126 +++++++++++++++++++++++++++++++++++--
tests/test_synced_enforcer.py | 143 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 265 insertions(+), 4 deletions(-)
diff --git a/casbin/synced_enforcer.py b/casbin/synced_enforcer.py
index 39b11a5..a10a57f 100644
--- a/casbin/synced_enforcer.py
+++ b/casbin/synced_enforcer.py
@@ -35,6 +35,112 @@ class AtomicBool:
self._value = value
+class _DeferredNotifier:
+ """Queues the watcher notifications raised inside a locked section and
delivers them
+ once the lock has been released.
+
+ Watchers have a lock of their own, and some of them (the redis watcher,
for one) hold
+ it while running the update callback, which calls back into the enforcer
to reload the
+ policy. Notifying from inside the enforcer lock would let the two threads
take those
+ two locks in opposite orders and deadlock:
+ https://github.com/casbin/pycasbin/issues/408
+ """
+
+ def __init__(self):
+ self._local = threading.local()
+
+ def _stack(self):
+ stack = getattr(self._local, "stack", None)
+ if stack is None:
+ stack = []
+ self._local.stack = stack
+ return stack
+
+ def defer(self, func, args, kwargs):
+ """queues a call, and returns whether it was queued. False means no
lock is held
+ by this thread, so the caller has to run the call itself."""
+ stack = self._stack()
+ if not stack:
+ return False
+ stack[-1].append((func, args, kwargs))
+ return True
+
+ def begin(self):
+ """starts collecting, called right after the lock has been taken."""
+ self._stack().append([])
+
+ def end(self, suppress_errors=False):
+ """delivers what was collected, called right after the lock has been
released."""
+ stack = self._stack()
+ queued = stack.pop()
+ if stack:
+ # an outer lock is still held, so hand the calls over to it
+ stack[-1].extend(queued)
+ return
+
+ for func, args, kwargs in queued:
+ try:
+ func(*args, **kwargs)
+ except Exception:
+ # the caller is already unwinding an exception of its own, and
hiding it
+ # behind a notification failure would only make the original
harder to find
+ if not suppress_errors:
+ raise
+
+
+class _NotifyAfterUnlock:
+ """wraps one of the enforcer locks so that _DeferredNotifier collects
notifications
+ for as long as the lock is held."""
+
+ def __init__(self, lock, notifier):
+ self._lock = lock
+ self._notifier = notifier
+
+ def __enter__(self):
+ self._lock.__enter__()
+ self._notifier.begin()
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ try:
+ self._lock.__exit__(exc_type, exc_value, traceback)
+ finally:
+ self._notifier.end(suppress_errors=exc_type is not None)
+ return False
+
+
+class _DeferredWatcher:
+ """stands in for the user's watcher on the wrapped enforcer, and routes
every call it
+ makes through a _DeferredNotifier."""
+
+ def __init__(self, watcher, notifier):
+ self._watcher = watcher
+ self._notifier = notifier
+
+ def __bool__(self):
+ return bool(self._watcher)
+
+ def __getattr__(self, name):
+ watcher = self.__dict__.get("_watcher")
+ if watcher is None:
+ raise AttributeError(name)
+
+ attr = getattr(watcher, name)
+ if not callable(attr):
+ return attr
+
+ notifier = self._notifier
+
+ def deferred(*args, **kwargs):
+ # a deferred call has no result to hand back yet, but the enforcer
ignores
+ # what the watcher returns anyway
+ if notifier.defer(attr, args, kwargs):
+ return None
+ return attr(*args, **kwargs)
+
+ return deferred
+
+
class SyncedEnforcer:
"""SyncedEnforcer wraps Enforcer and provides synchronized access.
It's also a drop-in replacement for Enforcer"""
@@ -42,8 +148,10 @@ class SyncedEnforcer:
def __init__(self, model=None, adapter=None):
self._e = self._new_enforcer(model, adapter)
self._rwlock = RWLockWrite()
- self._rl = self._rwlock.gen_rlock()
- self._wl = self._rwlock.gen_wlock()
+ self._notifier = _DeferredNotifier()
+ self._rl = _NotifyAfterUnlock(self._rwlock.gen_rlock(), self._notifier)
+ self._wl = _NotifyAfterUnlock(self._rwlock.gen_wlock(), self._notifier)
+ self._watcher = None
self._auto_loading = AtomicBool(False)
self._auto_loading_thread = None
@@ -115,9 +223,19 @@ class SyncedEnforcer:
self._e.set_adapter(adapter)
def set_watcher(self, watcher):
- """sets the current watcher."""
+ """sets the current watcher.
+
+ It is notified only once the enforcer lock has been released, so a
watcher that
+ holds a lock of its own while running the update callback cannot
deadlock against
+ the enforcer."""
with self._wl:
- self._e.set_watcher(watcher)
+ self._watcher = watcher
+ self._e.set_watcher(None if watcher is None else
_DeferredWatcher(watcher, self._notifier))
+
+ def get_watcher(self):
+ """gets the watcher that was passed to set_watcher()."""
+ with self._rl:
+ return self._watcher
def set_effector(self, eft):
"""sets the current effector."""
diff --git a/tests/test_synced_enforcer.py b/tests/test_synced_enforcer.py
new file mode 100644
index 0000000..862baa9
--- /dev/null
+++ b/tests/test_synced_enforcer.py
@@ -0,0 +1,143 @@
+# Copyright 2026 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from unittest import TestCase
+
+import casbin
+from tests.test_enforcer import get_examples
+
+
+class LockingWatcher:
+ """mimics the redis watcher, which holds a lock of its own both while
notifying the
+ other nodes and while running the update callback."""
+
+ def __init__(self):
+ self.mutex = threading.Lock()
+ self.callback = None
+ self.updates = []
+
+ def set_update_callback(self, callback):
+ self.callback = callback
+
+ def update(self):
+ with self.mutex:
+ self.updates.append("update")
+ return True
+
+ def close(self):
+ pass
+
+
+class TestSyncedEnforcerWatcher(TestCase):
+ """the lock the enforcer takes and the lock the watcher takes have to be
taken in the
+ same order everywhere, or the two deadlock. See
+ https://github.com/casbin/pycasbin/issues/408""";
+
+ def get_enforcer(self, adapter):
+ return casbin.SyncedEnforcer(get_examples("basic_model.conf"), adapter)
+
+ def test_watcher_is_notified(self):
+ w = LockingWatcher()
+ e = self.get_enforcer(casbin.persist.Adapter())
+ e.set_watcher(w)
+
+ e.add_policy("alice", "data1", "read")
+
+ # the notification is deferred, but only until the lock is released:
it still has
+ # to have been delivered by the time add_policy() returns
+ self.assertEqual(w.updates, ["update"])
+ self.assertTrue(e.enforce("alice", "data1", "read"))
+
+ def test_get_watcher_returns_the_watcher_that_was_set(self):
+ w = LockingWatcher()
+ e = self.get_enforcer(casbin.persist.Adapter())
+ e.set_watcher(w)
+
+ self.assertIs(e.get_watcher(), w)
+
+ def test_watcher_is_notified_outside_the_enforcer_lock(self):
+ w = LockingWatcher()
+ e = self.get_enforcer(casbin.persist.Adapter())
+ e.set_watcher(w)
+
+ reader_finished = threading.Event()
+
+ def update():
+ # a reader can only get through here if the write lock has been
released
+ reader = threading.Thread(target=lambda: (e.get_policy(),
reader_finished.set()), daemon=True)
+ reader.start()
+ reader.join(timeout=5)
+
+ w.update = update
+ e.add_policy("alice", "data1", "read")
+
+ self.assertTrue(reader_finished.is_set(), "the watcher was notified
while the enforcer was still locked")
+
+ def test_save_policy_notifies_outside_the_enforcer_lock(self):
+ """save_policy() notifies the watcher too, and it holds the read lock
while doing
+ so, so the same deferral has to apply there."""
+ w = LockingWatcher()
+ e = self.get_enforcer(casbin.persist.Adapter())
+ e.set_watcher(w)
+
+ writer_finished = threading.Event()
+
+ def update():
+ # a writer can only get through here if the read lock has been
released
+ writer = threading.Thread(target=lambda: (e.load_policy(),
writer_finished.set()), daemon=True)
+ writer.start()
+ writer.join(timeout=5)
+
+ w.update = update
+ e.save_policy()
+
+ self.assertTrue(writer_finished.is_set(), "the watcher was notified
while the enforcer was still locked")
+
+ def test_callback_holding_the_watcher_lock_does_not_deadlock(self):
+ """reproduces the issue: the subscribe thread holds the watcher lock
and wants the
+ enforcer lock, while the thread changing the policy holds the enforcer
lock and
+ wants the watcher lock."""
+ w = LockingWatcher()
+ enforcer_locked = threading.Event()
+ callback_reached = threading.Event()
+
+ class SlowAdapter(casbin.persist.Adapter):
+ def add_policy(self, sec, ptype, rule):
+ # the write lock is held for as long as we stay in here, so
this is where
+ # the subscribe thread gets its chance to ask for it
+ enforcer_locked.set()
+ callback_reached.wait(timeout=5)
+
+ e = self.get_enforcer(SlowAdapter())
+ e.set_watcher(w)
+ w.set_update_callback(e.load_policy)
+
+ def subscribe():
+ with w.mutex:
+ enforcer_locked.wait(timeout=5)
+ callback_reached.set()
+ w.callback()
+
+ subscriber = threading.Thread(target=subscribe, daemon=True)
+ subscriber.start()
+
+ added = threading.Event()
+ writer = threading.Thread(target=lambda: (e.add_policy("alice",
"data1", "read"), added.set()), daemon=True)
+ writer.start()
+
+ self.assertTrue(added.wait(timeout=10), "add_policy deadlocked against
the watcher lock")
+ subscriber.join(timeout=10)
+ self.assertFalse(subscriber.is_alive(), "the update callback
deadlocked against the enforcer lock")
+ self.assertEqual(w.updates, ["update"])
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
