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
use super::manager::RdfEventsStore;
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| {
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 self.insert_event(&event) {
Ok(_) => match event.kind {
Kind::Metadata
| Kind::ContactList
| Kind::RelayList => {
if let Err(e) =
self.delete_previous_events(&event)
{
eprintln!("{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));
}
}