修正DirectProject事件队列队头清理
将运行态队列改为Vec加head索引\n仅清理可持久化事件的队头并保留未完成item事件\n增加定期compact避免已清理前缀长期占用内存
This commit is contained in:
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user