tracks lexicons and how many times they appeared on the jetstream
refactor(server): use rclite and parking_lot, replace the scc::Queue with a Arc<Mutex<Vec<_>>>
+67
-25
+35
server/Cargo.lock
+35
server/Cargo.lock
···
203
203
]
204
204
205
205
[[package]]
206
+
name = "branches"
207
+
version = "0.2.2"
208
+
source = "registry+https://github.com/rust-lang/crates.io-index"
209
+
checksum = "a918aa7a861caeba57e502465c30e3a0d74ae02ee0b9db2933602fdb6a3a90e5"
210
+
dependencies = [
211
+
"rustc_version",
212
+
]
213
+
214
+
[[package]]
206
215
name = "brotli"
207
216
version = "8.0.1"
208
217
source = "registry+https://github.com/rust-lang/crates.io-index"
···
1177
1186
]
1178
1187
1179
1188
[[package]]
1189
+
name = "rclite"
1190
+
version = "0.2.7"
1191
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1192
+
checksum = "2f528dfeba924f5fc67bb84a17fe043451d1b392758016ce2d9e9116649b0f35"
1193
+
dependencies = [
1194
+
"branches",
1195
+
]
1196
+
1197
+
[[package]]
1180
1198
name = "redox_syscall"
1181
1199
version = "0.5.15"
1182
1200
source = "registry+https://github.com/rust-lang/crates.io-index"
···
1295
1313
checksum = "357703d41365b4b27c590e3ed91eabb1b663f07c4c084095e60cbed4362dff0d"
1296
1314
1297
1315
[[package]]
1316
+
name = "rustc_version"
1317
+
version = "0.4.1"
1318
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1319
+
checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92"
1320
+
dependencies = [
1321
+
"semver",
1322
+
]
1323
+
1324
+
[[package]]
1298
1325
name = "rustix"
1299
1326
version = "1.0.8"
1300
1327
source = "registry+https://github.com/rust-lang/crates.io-index"
···
1462
1489
checksum = "0f7d95a54511e0c7be3f51e8867aa8cf35148d7b9445d44de2f943e2b206e749"
1463
1490
1464
1491
[[package]]
1492
+
name = "semver"
1493
+
version = "1.0.26"
1494
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1495
+
checksum = "56e6fa9c48d24d85fb3de5ad847117517440f6beceb7798af16b4a87d616b8d0"
1496
+
1497
+
[[package]]
1465
1498
name = "serde"
1466
1499
version = "1.0.219"
1467
1500
source = "registry+https://github.com/rust-lang/crates.io-index"
···
1528
1561
"futures-util",
1529
1562
"itertools",
1530
1563
"ordered-varint",
1564
+
"parking_lot",
1531
1565
"quanta",
1532
1566
"rayon",
1567
+
"rclite",
1533
1568
"rkyv",
1534
1569
"rustls",
1535
1570
"scc",
+3
-1
server/Cargo.toml
+3
-1
server/Cargo.toml
···
8
8
async-trait = "0.1"
9
9
tracing-subscriber = {version = "0.3", features = ["env-filter"]}
10
10
tracing = "0.1"
11
-
tokio = { version = "1", features = ["full"] }
11
+
tokio = { version = "1", features = ["full", "parking_lot"] }
12
12
tokio-util = { version = "0.7", features = ["tracing"] }
13
13
rustls = { version = "0.23", default-features = false, features = ["log", "ring", "std"] }
14
14
tokio-websockets = { version = "0.12", features = ["client", "rustls-platform-verifier", "getrandom", "ring"] }
···
29
29
itertools = "0.14.0"
30
30
byteview = "0.6.1"
31
31
rayon = "1.10.0"
32
+
parking_lot = { version = "0.12", features = ["send_guard", "hardware-lock-elision"] }
33
+
rclite = "0.2.7"
+1
-1
server/src/api.rs
+1
-1
server/src/api.rs
···
3
3
fmt::Display,
4
4
net::SocketAddr,
5
5
ops::{Bound, Deref, RangeBounds},
6
-
sync::Arc,
7
6
time::Duration,
8
7
};
9
8
···
16
15
routing::get,
17
16
};
18
17
use axum_tws::{Message, WebSocketUpgrade};
18
+
use rclite::Arc;
19
19
use serde::{Deserialize, Serialize};
20
20
use smol_str::SmolStr;
21
21
use tokio_util::sync::CancellationToken;
+4
server/src/db/block.rs
+4
server/src/db/block.rs
+22
-22
server/src/db/mod.rs
+22
-22
server/src/db/mod.rs
···
2
2
io::Cursor,
3
3
ops::{Bound, Deref, RangeBounds},
4
4
path::Path,
5
-
sync::{
6
-
Arc,
7
-
atomic::{AtomicU64, AtomicUsize, Ordering as AtomicOrdering},
8
-
},
5
+
sync::atomic::{AtomicU64, AtomicUsize, Ordering as AtomicOrdering},
9
6
time::Duration,
10
7
};
11
8
12
9
use byteview::ByteView;
13
10
use fjall::{Config, Keyspace, Partition, PartitionCreateOptions, Slice};
14
11
use itertools::{Either, Itertools};
12
+
use parking_lot::Mutex;
15
13
use rayon::iter::{IndexedParallelIterator, IntoParallelIterator, ParallelIterator};
14
+
use rclite::Arc;
16
15
use rkyv::{Archive, Deserialize, Serialize, rancor::Error};
17
16
use smol_str::SmolStr;
18
17
use tokio::sync::broadcast;
···
85
84
pub struct LexiconHandle {
86
85
tree: Partition,
87
86
nsid: SmolStr,
88
-
buf: Arc<scc::Queue<EventRecord>>,
89
-
// this is stored here since scc::Queue does not have O(1) length
90
-
buf_len: AtomicUsize, // seqcst
87
+
buf: Arc<Mutex<Vec<EventRecord>>>,
91
88
last_insert: AtomicU64, // relaxed
92
89
eps: DefaultRateTracker,
93
90
}
···
99
96
tree: keyspace.open_partition(nsid, opts).unwrap(),
100
97
nsid: nsid.into(),
101
98
buf: Default::default(),
102
-
buf_len: AtomicUsize::new(0),
103
99
last_insert: AtomicU64::new(0),
104
100
eps: RateTracker::new(Duration::from_secs(10)),
105
101
}
106
102
}
107
103
108
104
fn item_count(&self) -> usize {
109
-
self.buf_len.load(AtomicOrdering::SeqCst)
105
+
self.buf.lock().len()
110
106
}
111
107
112
108
fn since_last_activity(&self) -> u64 {
···
118
114
}
119
115
120
116
fn insert(&self, event: EventRecord) {
121
-
self.buf.push(event);
122
-
self.buf_len.fetch_add(1, AtomicOrdering::SeqCst);
117
+
self.buf.lock().push(event);
123
118
self.last_insert.store(CLOCK.raw(), AtomicOrdering::Relaxed);
124
119
self.eps.observe();
125
120
}
126
121
122
+
fn compact(&self) {}
123
+
127
124
fn encode_block(&self, item_count: usize) -> AppResult<Block> {
128
125
let mut writer = ItemEncoder::new(
129
126
WritableByteView::with_size(ItemEncoder::encoded_len(item_count)),
···
132
129
let mut start_timestamp = None;
133
130
let mut end_timestamp = None;
134
131
let mut written = 0_usize;
135
-
while let Some(event) = self.buf.pop() {
132
+
for event in self.buf.lock().drain(..) {
136
133
let item = Item::new(
137
134
event.timestamp,
138
135
&NsidHit {
···
157
154
.into());
158
155
}
159
156
if let (Some(start_timestamp), Some(end_timestamp)) = (start_timestamp, end_timestamp) {
160
-
self.buf_len.store(0, AtomicOrdering::SeqCst);
161
157
let value = writer.finish()?;
162
158
let key = varints_unsigned_encoded([start_timestamp, end_timestamp]);
163
159
return Ok(Block {
···
208
204
})
209
205
}
210
206
207
+
#[inline(always)]
211
208
pub fn shutting_down(&self) -> impl Future<Output = ()> {
212
209
self.cancel_token.cancelled()
213
210
}
214
211
212
+
#[inline(always)]
215
213
pub fn is_shutting_down(&self) -> bool {
216
214
self.cancel_token.is_cancelled()
215
+
}
216
+
217
+
#[inline(always)]
218
+
pub fn eps(&self) -> usize {
219
+
self.eps.rate() as usize
220
+
}
221
+
222
+
#[inline(always)]
223
+
pub fn new_listener(&self) -> broadcast::Receiver<(SmolStr, NsidCounts)> {
224
+
self.event_broadcaster.subscribe()
217
225
}
218
226
219
227
pub fn sync(&self, all: bool) -> AppResult<()> {
···
284
292
Ok(())
285
293
}
286
294
287
-
#[inline(always)]
288
-
pub fn eps(&self) -> usize {
289
-
self.eps.rate() as usize
290
-
}
291
-
292
-
#[inline(always)]
293
-
pub fn new_listener(&self) -> broadcast::Receiver<(SmolStr, NsidCounts)> {
294
-
self.event_broadcaster.subscribe()
295
-
}
295
+
pub fn compact(&self) {}
296
296
297
297
#[inline(always)]
298
298
fn maybe_run_in_nsid_tree<T>(