Conversation
…ent and fix loop shutdown - Replace list-based queue with sorted set for better dead client cleanup - Add zombie cleanup buffer to handle expired queue entries - Fix potential None loop reference in graceful shutdown - Add task start time to write_message_task result - Update lock acquisition script to use ZSET operations - Remove unused queue cleanup scripts - Ensure proper lock release and renewal failure handling
Contributor
Reviewer's Guide重构 RedisFairLock，使其使用 Redis 有序集合（sorted set）和单个获取锁的 Lua 脚本（带“僵尸”条目清理）来管理队列，改进续期与释放行为，并加固 Celery 写入任务循环的生命周期与结果元数据。 使用 ZSET 和 ACQUIRE_SCRIPT 的 RedisFairLock.acquire 时序图sequenceDiagram
participant Client
participant RedisFairLock
participant RedisServer as Redis
Client->>RedisFairLock: acquire()
activate RedisFairLock
RedisFairLock->>RedisServer: EVAL ACQUIRE_SCRIPT(queue_key, key, value, expire, timeout)
activate RedisServer
RedisServer-->>RedisFairLock: 1 (lock acquired) or 0 (not acquired)
deactivate RedisServer
alt ok == 1
RedisFairLock->>RedisFairLock: _locked = True
alt auto_renewal is True
RedisFairLock->>RedisFairLock: _start_renewal()
end
RedisFairLock-->>Client: True
else ok == 0 and not timed out
RedisFairLock->>RedisFairLock: sleep(retry_interval)
RedisFairLock->>RedisServer: EVAL ACQUIRE_SCRIPT(...)
RedisServer-->>RedisFairLock: 1 or 0
RedisFairLock->>Client: (loop continues)
else timeout reached
RedisFairLock->>RedisServer: ZREM queue_key value
RedisFairLock-->>Client: False
end
deactivate RedisFairLock
Celery 写入任务循环生命周期与 RedisFairLock 使用的时序图sequenceDiagram
actor Worker
participant Task as MemoryWriteTask
participant RedisFairLock
participant EventLoop as AsyncioLoop
Worker->>Task: execute()
activate Task
Task->>RedisFairLock: create instance(key, redis_client,...)
Task->>Task: task_start_time = time()
Task->>EventLoop: set_asyncio_event_loop()
activate EventLoop
Task->>EventLoop: run_until_complete(_run())
EventLoop-->>Task: result
deactivate EventLoop
alt redis_client is not None
Task->>RedisFairLock: acquire() (context manager entry)
RedisFairLock-->>Task: lock acquired
Task->>Task: perform work
Task->>RedisFairLock: release() (context manager exit)
end
Task->>Task: compute elapsed_time
Task-->>Worker: status, result, start_at, end_user_id, config_id, elapsed_time
alt loop is not None
Task->>EventLoop: _shutdown_loop_gracefully(loop)
end
deactivate Task
重构后 RedisFairLock 的类图classDiagram
class RedisFairLock {
+int CLEANUP_BUFFER
-str key
-str queue_key
-str value
-int expire
-float retry_interval
-float timeout
-bool auto_renewal
-redis.StrictRedis redis
-bool _locked
-threading.Event _stop_renew
-threading.Thread _renew_thread
+RedisFairLock(key, redis_client, expire, retry_interval, timeout, auto_renewal)
+bool acquire()
-void _renewal_loop()
-void _start_renewal()
-void _stop_renewal_loop()
+void release()
+RedisFairLock __enter__()
+void __exit__(exc_type, exc_value, traceback)
}
Redis ACQUIRE_SCRIPT 获取锁与僵尸条目清理的流程图flowchart TD
A[Start ACQUIRE_SCRIPT] --> B[Read queue_key and lock_key]
B --> C[Set local client_id, expire, time_out]
C --> D[Get now from redis.call time]
D --> E{ZSET contains client_id?}
E -- No --> F[ZADD queue_key now client_id]
E -- Yes --> G[Skip enqueue]
F --> H
G --> H
H[Find expired entries with ZRANGEBYSCORE queue_key 0 now - time_out] --> I[For each expired value, ZREM queue_key value]
I --> J[Get first client in ZSET with ZRANGE queue_key 0 0]
J --> K{first == client_id?}
K -- No --> L[Return 0]
K -- Yes --> M{SET lock_key client_id NX EX expire success?}
M -- Yes --> N[ZREM queue_key client_id and return 1]
M -- No --> O{GET lock_key == client_id?}
O -- Yes --> P[EXPIRE lock_key expire and return 1]
O -- No --> Q[Return 0]
L --> R[End]
N --> R
P --> R
Q --> R
文件级变更
Tips and commands与 Sourcery 交互
自定义你的体验访问你的 dashboard 以：
获取帮助Original review guide in EnglishReviewer's GuideRefactors RedisFairLock to manage its queue using a Redis sorted set and a single acquire Lua script (with zombie cleanup), improves renewal and release behavior, and hardens the Celery write task loop lifecycle and result metadata. Sequence diagram for RedisFairLock.acquire using ZSET and ACQUIRE_SCRIPTsequenceDiagram
participant Client
participant RedisFairLock
participant RedisServer as Redis
Client->>RedisFairLock: acquire()
activate RedisFairLock
RedisFairLock->>RedisServer: EVAL ACQUIRE_SCRIPT(queue_key, key, value, expire, timeout)
activate RedisServer
RedisServer-->>RedisFairLock: 1 (lock acquired) or 0 (not acquired)
deactivate RedisServer
alt ok == 1
RedisFairLock->>RedisFairLock: _locked = True
alt auto_renewal is True
RedisFairLock->>RedisFairLock: _start_renewal()
end
RedisFairLock-->>Client: True
else ok == 0 and not timed out
RedisFairLock->>RedisFairLock: sleep(retry_interval)
RedisFairLock->>RedisServer: EVAL ACQUIRE_SCRIPT(...)
RedisServer-->>RedisFairLock: 1 or 0
RedisFairLock->>Client: (loop continues)
else timeout reached
RedisFairLock->>RedisServer: ZREM queue_key value
RedisFairLock-->>Client: False
end
deactivate RedisFairLock
Sequence diagram for Celery write task loop lifecycle and RedisFairLock usagesequenceDiagram
actor Worker
participant Task as MemoryWriteTask
participant RedisFairLock
participant EventLoop as AsyncioLoop
Worker->>Task: execute()
activate Task
Task->>RedisFairLock: create instance(key, redis_client,...)
Task->>Task: task_start_time = time()
Task->>EventLoop: set_asyncio_event_loop()
activate EventLoop
Task->>EventLoop: run_until_complete(_run())
EventLoop-->>Task: result
deactivate EventLoop
alt redis_client is not None
Task->>RedisFairLock: acquire() (context manager entry)
RedisFairLock-->>Task: lock acquired
Task->>Task: perform work
Task->>RedisFairLock: release() (context manager exit)
end
Task->>Task: compute elapsed_time
Task-->>Worker: status, result, start_at, end_user_id, config_id, elapsed_time
alt loop is not None
Task->>EventLoop: _shutdown_loop_gracefully(loop)
end
deactivate Task
Class diagram for refactored RedisFairLockclassDiagram
class RedisFairLock {
+int CLEANUP_BUFFER
-str key
-str queue_key
-str value
-int expire
-float retry_interval
-float timeout
-bool auto_renewal
-redis.StrictRedis redis
-bool _locked
-threading.Event _stop_renew
-threading.Thread _renew_thread
+RedisFairLock(key, redis_client, expire, retry_interval, timeout, auto_renewal)
+bool acquire()
-void _renewal_loop()
-void _start_renewal()
-void _stop_renewal_loop()
+void release()
+RedisFairLock __enter__()
+void __exit__(exc_type, exc_value, traceback)
}
Flow diagram for Redis ACQUIRE_SCRIPT lock acquisition and zombie cleanupflowchart TD
A[Start ACQUIRE_SCRIPT] --> B[Read queue_key and lock_key]
B --> C[Set local client_id, expire, time_out]
C --> D[Get now from redis.call time]
D --> E{ZSET contains client_id?}
E -- No --> F[ZADD queue_key now client_id]
E -- Yes --> G[Skip enqueue]
F --> H
G --> H
H[Find expired entries with ZRANGEBYSCORE queue_key 0 now - time_out] --> I[For each expired value, ZREM queue_key value]
I --> J[Get first client in ZSET with ZRANGE queue_key 0 0]
J --> K{first == client_id?}
K -- No --> L[Return 0]
K -- Yes --> M{SET lock_key client_id NX EX expire success?}
M -- Yes --> N[ZREM queue_key client_id and return 1]
M -- No --> O{GET lock_key == client_id?}
O -- Yes --> P[EXPIRE lock_key expire and return 1]
O -- No --> Q[Return 0]
L --> R[End]
N --> R
P --> R
Q --> R
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我已经审查了你的更改，一切看起来都很棒！
帮我变得更有用！请在每条评论上点👍或👎，我会根据你的反馈来改进后续的评审。
Original comment in English
Hey - I've reviewed your changes and they look great!
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
重构
RedisFairLock，使用基于有序集合（sorted set）的锁获取脚本，并改进清理和失败处理；同时调整任务执行逻辑，以暴露任务开始时间并安全地关闭事件循环。New Features:
write_message_task的结果负载中包含任务开始时间戳。Bug Fixes:
Enhancements:
Original summary in English
Summary by Sourcery
Refactor RedisFairLock to use a sorted-set-based acquisition script with improved cleanup and failure handling, and adjust task execution to expose start time and safely shut down the event loop.
New Features:
Bug Fixes:
Enhancements: