refactor: 串行锁按 topic_id 键化,支持多话题并发执行
将 chat_serial_locks 重命名为 topic_serial_locks,串行锁粒度从 chat_id 改为 topic_id。同一 topic 的消息处理仍串行执行,不同 topic 之间互不阻塞,为后续多话题并发修复铺路。 本提交仅重命名锁结构和访问方法,调用点仍用 chat_id 作为锁键 (回退兼容),不影响现有行为。
This commit is contained in:
parent
56612389ae
commit
3a8da51936
@ -200,15 +200,16 @@ impl AgentExecutionService {
|
|||||||
&self,
|
&self,
|
||||||
request: MessageExecutionRequest<'_>,
|
request: MessageExecutionRequest<'_>,
|
||||||
) -> Result<Vec<OutboundMessage>, AgentError> {
|
) -> Result<Vec<OutboundMessage>, AgentError> {
|
||||||
// 获取该 chat 的串行锁(通过短暂获取 session 锁)
|
// 获取该 topic 的串行锁(通过短暂获取 session 锁)
|
||||||
// 同一 chat 的消息处理必须串行执行,防止并发 loop 操作同一历史的不同快照
|
// 同一 topic 的消息处理必须串行执行,防止并发 loop 操作同一历史的不同快照
|
||||||
|
// 不同 topic 之间互不阻塞,支持多话题并发执行
|
||||||
let serial_lock = {
|
let serial_lock = {
|
||||||
let mut session_guard = request.session.lock().await;
|
let mut session_guard = request.session.lock().await;
|
||||||
session_guard.chat_serial_lock(request.chat_id)
|
session_guard.topic_serial_lock(request.chat_id)
|
||||||
};
|
};
|
||||||
|
|
||||||
// 等待该 chat 的前一条消息处理完成(含压缩)
|
// 等待该 topic 的前一条消息处理完成(含压缩)
|
||||||
// await 串行锁时不持有 session 锁,其他 chat 的消息可以正常处理
|
// await 串行锁时不持有 session 锁,其他 topic 的消息可以正常处理
|
||||||
let _serial_guard = serial_lock.lock().await;
|
let _serial_guard = serial_lock.lock().await;
|
||||||
|
|
||||||
let (history, agent, user_message, user_message_count, original_topic_id) = {
|
let (history, agent, user_message, user_message_count, original_topic_id) = {
|
||||||
@ -291,13 +292,13 @@ impl AgentExecutionService {
|
|||||||
&self,
|
&self,
|
||||||
request: ScheduledExecutionRequest<'_>,
|
request: ScheduledExecutionRequest<'_>,
|
||||||
) -> Result<Vec<OutboundMessage>, AgentError> {
|
) -> Result<Vec<OutboundMessage>, AgentError> {
|
||||||
// 获取该 chat 的串行锁(与普通消息路径共享,保证串行执行)
|
// 获取该 topic 的串行锁(与普通消息路径共享,保证串行执行)
|
||||||
let serial_lock = {
|
let serial_lock = {
|
||||||
let mut session_guard = request.session.lock().await;
|
let mut session_guard = request.session.lock().await;
|
||||||
session_guard.chat_serial_lock(request.chat_id)
|
session_guard.topic_serial_lock(request.chat_id)
|
||||||
};
|
};
|
||||||
|
|
||||||
// 等待该 chat 的前一条消息处理完成(含压缩)
|
// 等待该 topic 的前一条消息处理完成(含压缩)
|
||||||
let _serial_guard = serial_lock.lock().await;
|
let _serial_guard = serial_lock.lock().await;
|
||||||
|
|
||||||
let (history, mut agent, user_message, user_message_count, original_topic_id, store, session_id) = {
|
let (history, mut agent, user_message, user_message_count, original_topic_id, store, session_id) = {
|
||||||
@ -500,9 +501,9 @@ mod tests {
|
|||||||
assert!(should_display_message_to_user(true, &message));
|
assert!(should_display_message_to_user(true, &message));
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 对抗性测试:同一 chat 的串行锁被持有时,第二次获取应阻塞
|
/// 对抗性测试:同一 topic 的串行锁被持有时,第二次获取应阻塞
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_chat_serial_lock_blocks_concurrent_access() {
|
async fn test_topic_serial_lock_blocks_concurrent_access() {
|
||||||
let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
||||||
let _guard1 = lock.lock().await;
|
let _guard1 = lock.lock().await;
|
||||||
|
|
||||||
@ -516,9 +517,9 @@ mod tests {
|
|||||||
assert!(result.is_err(), "第二次获取同一锁应阻塞");
|
assert!(result.is_err(), "第二次获取同一锁应阻塞");
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 对抗性测试:不同 chat 的串行锁互不影响,可同时获取
|
/// 对抗性测试:不同 topic 的串行锁互不影响,可同时获取
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_different_chat_locks_independent() {
|
async fn test_different_topic_locks_independent() {
|
||||||
let lock_a = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
let lock_a = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
||||||
let lock_b = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
let lock_b = std::sync::Arc::new(tokio::sync::Mutex::new(()));
|
||||||
|
|
||||||
@ -531,7 +532,7 @@ mod tests {
|
|||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
assert!(result.is_ok(), "不同 chat 的锁应互不影响");
|
assert!(result.is_ok(), "不同 topic 的锁应互不影响");
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 对抗性测试:错误返回路径锁被正确释放(RAII 保证)
|
/// 对抗性测试:错误返回路径锁被正确释放(RAII 保证)
|
||||||
|
|||||||
@ -533,10 +533,11 @@ impl Session {
|
|||||||
&self.compressor
|
&self.compressor
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 获取该 chat 的串行化锁。
|
/// 获取该 topic 的串行化锁。
|
||||||
/// 同一 chat 的消息处理(agent loop + 压缩)共享此锁,保证串行执行。
|
/// 同一 topic 的消息处理(agent loop + 压缩)共享此锁,保证串行执行;
|
||||||
pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
/// 不同 topic 之间互不阻塞。
|
||||||
self.history.chat_serial_lock(chat_id)
|
pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
||||||
|
self.history.topic_serial_lock(topic_id)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
|
pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
|
||||||
|
|||||||
@ -21,10 +21,11 @@ pub(crate) struct SessionHistory {
|
|||||||
chat_topic_ids: HashMap<String, String>, // 每个 chat 的当前 topic
|
chat_topic_ids: HashMap<String, String>, // 每个 chat 的当前 topic
|
||||||
history_topic_ids: HashMap<String, String>, // 每个 chat 的历史所对应的话题
|
history_topic_ids: HashMap<String, String>, // 每个 chat 的历史所对应的话题
|
||||||
compression_in_flight: HashSet<String>,
|
compression_in_flight: HashSet<String>,
|
||||||
/// 按 chat_id 的串行化锁。
|
/// 按 topic_id 的串行化锁。
|
||||||
/// 同一 chat 的消息处理(agent loop + 压缩)必须串行执行,
|
/// 同一 topic 的消息处理(agent loop + 压缩)必须串行执行,
|
||||||
/// 防止并发 loop 操作同一历史的不同快照产生交错序列。
|
/// 防止并发 loop 操作同一历史的不同快照产生交错序列。
|
||||||
chat_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
|
/// 不同 topic 之间互不阻塞,支持多话题并发执行。
|
||||||
|
topic_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
|
||||||
conversations: Arc<dyn ConversationRepository>,
|
conversations: Arc<dyn ConversationRepository>,
|
||||||
skill_events: Arc<dyn SkillEventRepository>,
|
skill_events: Arc<dyn SkillEventRepository>,
|
||||||
}
|
}
|
||||||
@ -41,17 +42,18 @@ impl SessionHistory {
|
|||||||
chat_topic_ids: HashMap::new(),
|
chat_topic_ids: HashMap::new(),
|
||||||
history_topic_ids: HashMap::new(),
|
history_topic_ids: HashMap::new(),
|
||||||
compression_in_flight: HashSet::new(),
|
compression_in_flight: HashSet::new(),
|
||||||
chat_serial_locks: HashMap::new(),
|
topic_serial_locks: HashMap::new(),
|
||||||
conversations,
|
conversations,
|
||||||
skill_events,
|
skill_events,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 获取或创建该 chat 的串行化锁。
|
/// 获取或创建该 topic 的串行化锁。
|
||||||
/// 同一 chat 的所有消息处理共享同一个锁,保证串行执行。
|
/// 同一 topic 的所有消息处理共享同一个锁,保证串行执行;
|
||||||
pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
/// 不同 topic 之间互不阻塞,支持多话题并发执行。
|
||||||
self.chat_serial_locks
|
pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
||||||
.entry(chat_id.to_string())
|
self.topic_serial_locks
|
||||||
|
.entry(topic_id.to_string())
|
||||||
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
|
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
|
||||||
.clone()
|
.clone()
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user