simplify redis code, less redis calls
This commit is contained in:
parent
39cdce6960
commit
15f1411bfb
@ -3182,14 +3182,6 @@ class StateManagerRedis(StateManager):
|
|||||||
"e" # For evicted events (i.e. maxmemory exceeded)
|
"e" # For evicted events (i.e. maxmemory exceeded)
|
||||||
)
|
)
|
||||||
|
|
||||||
# These events indicate that a lock is no longer held
|
|
||||||
_redis_keyspace_lock_release_events: Set[bytes] = {
|
|
||||||
b"del",
|
|
||||||
b"expire",
|
|
||||||
b"expired",
|
|
||||||
b"evicted",
|
|
||||||
}
|
|
||||||
|
|
||||||
async def _get_parent_state(
|
async def _get_parent_state(
|
||||||
self, token: str, state: BaseState | None = None
|
self, token: str, state: BaseState | None = None
|
||||||
) -> BaseState | None:
|
) -> BaseState | None:
|
||||||
@ -3467,20 +3459,16 @@ class StateManagerRedis(StateManager):
|
|||||||
raise
|
raise
|
||||||
async with self.redis.pubsub() as pubsub:
|
async with self.redis.pubsub() as pubsub:
|
||||||
await pubsub.psubscribe(lock_key_channel)
|
await pubsub.psubscribe(lock_key_channel)
|
||||||
while not state_is_locked:
|
|
||||||
# wait for the lock to be released
|
# wait for the lock to be released
|
||||||
while True:
|
while True:
|
||||||
if not await self.redis.exists(lock_key):
|
# fast path
|
||||||
break # key was removed, try to get the lock again
|
if await self._try_get_lock(lock_key, lock_id):
|
||||||
message = await pubsub.get_message(
|
return
|
||||||
|
# wait for lock events
|
||||||
|
_ = await pubsub.get_message(
|
||||||
ignore_subscribe_messages=True,
|
ignore_subscribe_messages=True,
|
||||||
timeout=self.lock_expiration / 1000.0,
|
timeout=self.lock_expiration / 1000.0,
|
||||||
)
|
)
|
||||||
if message is None:
|
|
||||||
continue
|
|
||||||
if message["data"] in self._redis_keyspace_lock_release_events:
|
|
||||||
break
|
|
||||||
state_is_locked = await self._try_get_lock(lock_key, lock_id)
|
|
||||||
|
|
||||||
@contextlib.asynccontextmanager
|
@contextlib.asynccontextmanager
|
||||||
async def _lock(self, token: str):
|
async def _lock(self, token: str):
|
||||||
|
Loading…
Reference in New Issue
Block a user