1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
use super::manager::RdfEventsStore;
use crate::parser::hybrid_parser;
use crate::rdfify::ntify_event_sync_with_parser;
use nostr::{Event, Kind};
use std::sync::Arc;
use std::{thread::sleep, time::Duration};
use thread_priority::*;
impl RdfEventsStore {
/// Start the events processing threads
pub fn start(
self: Arc<Self>,
thr_count: Option<usize>,
sleep_ms: Option<u64>,
) {
let selfc = Arc::clone(&self);
std::thread::spawn(move || {
selfc.events_process_threadpool(
thr_count.unwrap_or(4),
sleep_ms.unwrap_or(350),
);
});
}
/// Events processing threadpool
pub fn events_process_threadpool(
&self,
thread_count: usize,
proc_sleep_ms: u64,
) {
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(thread_count)
.build()
.unwrap();
pool.scope(|s| {
// Use the same JSON-LD parser in all threads
let parser = Arc::new(hybrid_parser());
s.spawn_broadcast(move |_, _ctx| {
// Set a low thread priority
if let Err(e) = set_current_thread_priority(ThreadPriority::Min)
{
eprintln!("Error setting thread priority: {e}");
}
loop {
if let Ok((event, _prio)) = self.event_rx.try_recv() {
match ntify_event_sync_with_parser(&event, &parser) {
Ok(nt) => {
// Purge similar but older events matching this
// event before storing the new event
match event.kind {
Kind::Metadata
| Kind::ContactList
| Kind::RelayList => {
if let Err(e) =
self.delete_previous_events(&event)
{
eprintln!("{e}");
}
}
_ => {}
}
if let Err(e) = self.store_event_nt(&event, nt)
{
eprintln!("Error storing event: {e}");
}
}
Err(e) => {
eprintln!("Error processing event: {e:?}");
}
}
sleep(Duration::from_millis(proc_sleep_ms));
} else {
sleep(Duration::from_millis(proc_sleep_ms * 2));
}
}
});
});
}
/// Send an [`Event`] for processing, with an optional priority
pub fn process_event(&self, event: Event, priority: Option<i32>) {
let _ = self.event_tx.try_send(event, priority.unwrap_or(0));
}
}