ptr.pet
tracks lexicons and how many times they appeared on the jetstream
refactor(server): use relaxed ordering for keeping track of lexicon buf len / last activity / block size since we don't really need any strict ordering
+9
-9
server/src/db/mod.rs
+9
-9
server/src/db/mod.rs
···
76
pub struct LexiconHandle {
77
tree: Partition,
78
buf: Arc<scc::Queue<EventRecord>>,
79
-
buf_len: AtomicUsize,
80
-
last_insert: AtomicInstant,
81
eps: Rate,
82
-
block_size: AtomicUsize,
83
}
84
85
impl LexiconHandle {
···
96
}
97
98
fn item_count(&self) -> usize {
99
-
self.buf_len.load(AtomicOrdering::Acquire)
100
}
101
102
fn last_insert(&self) -> Instant {
103
-
self.last_insert.load(AtomicOrdering::Acquire)
104
}
105
106
fn suggested_block_size(&self) -> usize {
···
109
110
fn insert(&self, event: EventRecord) {
111
self.buf.push(event);
112
-
self.buf_len.fetch_add(1, AtomicOrdering::Release);
113
self.last_insert
114
-
.store(Instant::now(), AtomicOrdering::Release);
115
self.eps.observe(&(), 1);
116
let rate = self.eps.rate(&()) as usize;
117
if rate != 0 {
···
144
written += 1;
145
}
146
if let (Some(start_timestamp), Some(end_timestamp)) = (start_timestamp, end_timestamp) {
147
-
self.buf_len.store(0, AtomicOrdering::Release);
148
let value = writer.finish()?;
149
let mut key = Vec::with_capacity(size_of::<u64>() * 2);
150
key.write_varint(start_timestamp)?;
···
161
// hits is tree per nsid: varint start time + varint end time -> block of hits
162
pub struct Db {
163
inner: Keyspace,
164
hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,
165
syncpool: threadpool::ThreadPool,
166
-
counts: Partition,
167
event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,
168
eps: Rate,
169
shutting_down: AtomicBool,
···
76
pub struct LexiconHandle {
77
tree: Partition,
78
buf: Arc<scc::Queue<EventRecord>>,
79
+
buf_len: AtomicUsize, // relaxed
80
+
last_insert: AtomicInstant, // relaxed
81
+
block_size: AtomicUsize, // relaxed
82
eps: Rate,
83
}
84
85
impl LexiconHandle {
···
96
}
97
98
fn item_count(&self) -> usize {
99
+
self.buf_len.load(AtomicOrdering::Relaxed)
100
}
101
102
fn last_insert(&self) -> Instant {
103
+
self.last_insert.load(AtomicOrdering::Relaxed)
104
}
105
106
fn suggested_block_size(&self) -> usize {
···
109
110
fn insert(&self, event: EventRecord) {
111
self.buf.push(event);
112
+
self.buf_len.fetch_add(1, AtomicOrdering::Relaxed);
113
self.last_insert
114
+
.store(Instant::now(), AtomicOrdering::Relaxed);
115
self.eps.observe(&(), 1);
116
let rate = self.eps.rate(&()) as usize;
117
if rate != 0 {
···
144
written += 1;
145
}
146
if let (Some(start_timestamp), Some(end_timestamp)) = (start_timestamp, end_timestamp) {
147
+
self.buf_len.store(0, AtomicOrdering::Relaxed);
148
let value = writer.finish()?;
149
let mut key = Vec::with_capacity(size_of::<u64>() * 2);
150
key.write_varint(start_timestamp)?;
···
161
// hits is tree per nsid: varint start time + varint end time -> block of hits
162
pub struct Db {
163
inner: Keyspace,
164
+
counts: Partition,
165
hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,
166
syncpool: threadpool::ThreadPool,
167
event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,
168
eps: Rate,
169
shutting_down: AtomicBool,