修正DirectProject事件队列队头清理

将运行态队列改为Vec加head索引\n仅清理可持久化事件的队头并保留未完成item事件\n增加定期compact避免已清理前缀长期占用内存
This commit is contained in:
2026-09-15 17:33:31 +08:00
committed by kdletters
parent af53a43aa9
commit cbad0ee1c0
@@ -6,7 +6,7 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, HashSet, VecDeque};
use std::collections::{HashMap, HashSet};
use std::sync::{Mutex, OnceLock};
use uuid::Uuid;
@@ -75,7 +75,8 @@ struct SubscriberState {
#[derive(Clone, Debug)]
struct ThreadState {
next_seq: u64,
events: VecDeque<StoredEvent>,
events: Vec<StoredEvent>,
head: usize,
total_bytes: usize,
active_items: HashSet<String>,
unresolved_requests: HashSet<String>,
@@ -88,7 +89,8 @@ impl Default for ThreadState {
fn default() -> Self {
Self {
next_seq: 0,
events: VecDeque::new(),
events: Vec::new(),
head: 0,
total_bytes: 0,
active_items: HashSet::new(),
unresolved_requests: HashSet::new(),
@@ -144,7 +146,7 @@ impl DirectThreadManager {
.map(|value| value.len())
.unwrap_or_default();
thread.total_bytes = thread.total_bytes.saturating_add(bytes);
thread.events.push_back(StoredEvent {
thread.events.push(StoredEvent {
event: event.clone(),
bytes,
cleanable,
@@ -165,6 +167,7 @@ impl DirectThreadManager {
let mut events = thread
.events
.iter()
.skip(thread.head)
.filter(|stored| Self::is_bootstrap_event(thread, &stored.event, stored.cleanable))
.map(|stored| stored.event.clone())
.collect::<Vec<_>>();
@@ -206,7 +209,7 @@ impl DirectThreadManager {
.expect("subscriber checked above");
let oldest_seq = thread
.events
.front()
.get(thread.head)
.map(|stored| stored.event.seq)
.unwrap_or(thread.next_seq.saturating_add(1));
if cursor.saturating_add(1) < oldest_seq {
@@ -216,6 +219,7 @@ impl DirectThreadManager {
let events = thread
.events
.iter()
.skip(thread.head)
.filter(|stored| stored.event.seq > cursor)
.map(|stored| stored.event.clone())
.collect::<Vec<_>>();
@@ -235,7 +239,7 @@ impl DirectThreadManager {
fn thread_debug(&self, thread_id: &str) -> Option<(usize, usize, usize)> {
self.threads.get(thread_id).map(|thread| {
(
thread.events.len(),
thread.events.len().saturating_sub(thread.head),
thread.total_bytes,
thread.subscribers.len(),
)
@@ -322,21 +326,34 @@ impl DirectThreadManager {
loop {
let can_pop = thread
.events
.front()
.get(thread.head)
.is_some_and(|stored| stored.event.seq <= min_cursor && stored.cleanable);
if !can_pop {
break;
}
if let Some(stored) = thread.events.pop_front() {
thread.total_bytes = thread.total_bytes.saturating_sub(stored.bytes);
}
let bytes = thread
.events
.get(thread.head)
.map(|stored| stored.bytes)
.unwrap_or_default();
thread.total_bytes = thread.total_bytes.saturating_sub(bytes);
thread.head = thread.head.saturating_add(1);
}
Self::compact(thread);
}
fn compact(thread: &mut ThreadState) {
if thread.head >= 1024 && thread.head.saturating_mul(2) >= thread.events.len() {
thread.events.drain(..thread.head);
thread.head = 0;
}
}
fn evict(&mut self, thread_id: &str) {
loop {
let over_limit = self.threads.get(thread_id).is_some_and(|thread| {
thread.events.len() > self.max_events || thread.total_bytes > self.max_bytes
thread.events.len().saturating_sub(thread.head) > self.max_events
|| thread.total_bytes > self.max_bytes
});
if !over_limit {
if let Some(thread) = self.threads.get_mut(thread_id) {