Compare commits

...
4 Commits
6 changed files with 41 additions and 34 deletions
Generated
+1 -1
View File
@@ -1096,7 +1096,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
[[package]] [[package]]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.7.9" version = "0.7.10"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bitcoin_hashes", "bitcoin_hashes",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.7.9" version = "0.7.10"
edition = "2021" edition = "2021"
authors = ["Greg Heartsfield <scsibug@imap.cc>"] authors = ["Greg Heartsfield <scsibug@imap.cc>"]
description = "A relay implementation for the Nostr protocol" description = "A relay implementation for the Nostr protocol"
+2 -2
View File
@@ -78,8 +78,8 @@ reject_future_seconds = 1800
# defaults to unlimited (subject to subscription limits). # defaults to unlimited (subject to subscription limits).
#db_conns_per_client = 0 #db_conns_per_client = 0
# Limit blocking threads used for database connections. Defaults to 64. # Limit blocking threads used for database connections. Defaults to 16.
#max_blocking_threads = 64 #max_blocking_threads = 16
# Limit the maximum size of an EVENT message. Defaults to 128 KB. # Limit the maximum size of an EVENT message. Defaults to 128 KB.
# Set to 0 for unlimited. # Set to 0 for unlimited.
+1 -1
View File
@@ -219,7 +219,7 @@ impl Default for Settings {
messages_per_sec: None, messages_per_sec: None,
subscriptions_per_min: None, subscriptions_per_min: None,
db_conns_per_client: None, db_conns_per_client: None,
max_blocking_threads: 64, max_blocking_threads: 16,
max_event_bytes: Some(2 << 17), // 128K max_event_bytes: Some(2 << 17), // 128K
max_ws_message_bytes: Some(2 << 17), // 128K max_ws_message_bytes: Some(2 << 17), // 128K
max_ws_frame_bytes: Some(2 << 17), // 128K max_ws_frame_bytes: Some(2 << 17), // 128K
+15 -10
View File
@@ -636,9 +636,11 @@ pub async fn db_query(
query_tx: tokio::sync::mpsc::Sender<QueryResult>, query_tx: tokio::sync::mpsc::Sender<QueryResult>,
mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>, mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
) { ) {
task::spawn_blocking(move || {
let mut row_count: usize = 0;
let start = Instant::now(); let start = Instant::now();
task::spawn_blocking(move || {
debug!("moved DB query to thread in {:?}", start.elapsed());
let start = Instant::now();
let mut row_count: usize = 0;
// generate SQL query // generate SQL query
let (q, p) = query_from_sub(&sub); let (q, p) = query_from_sub(&sub);
debug!("SQL generated in {:?}", start.elapsed()); debug!("SQL generated in {:?}", start.elapsed());
@@ -647,16 +649,25 @@ pub async fn db_query(
// cutoff for displaying slow queries // cutoff for displaying slow queries
let slow_cutoff = Duration::from_millis(1000); let slow_cutoff = Duration::from_millis(1000);
let start = Instant::now(); let start = Instant::now();
let mut slow_first_event;
if let Ok(conn) = pool.get() { if let Ok(conn) = pool.get() {
// execute the query. Don't cache, since queries vary so much. // execute the query. Don't cache, since queries vary so much.
let mut stmt = conn.prepare(&q)?; let mut stmt = conn.prepare(&q)?;
let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?; let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
let mut first_result = true; let mut first_result = true;
while let Some(row) = event_rows.next()? { while let Some(row) = event_rows.next()? {
let first_event_elapsed = start.elapsed();
slow_first_event = first_event_elapsed >= slow_cutoff;
if first_result { if first_result {
let first_result_elapsed = start.elapsed(); debug!(
"first result in {:?} (cid: {}, sub: {:?})",
first_event_elapsed, client_id, sub.id
);
first_result = false;
}
// logging for slow queries; show sub and SQL // logging for slow queries; show sub and SQL
if first_result_elapsed >= slow_cutoff { //
if slow_first_event {
info!( info!(
"query req (slow): {:?} (cid: {}, sub: {:?})", "query req (slow): {:?} (cid: {}, sub: {:?})",
sub, client_id, sub.id sub, client_id, sub.id
@@ -679,12 +690,6 @@ pub async fn db_query(
sub.id sub.id
); );
} }
debug!(
"first result in {:?} (cid: {}, sub: {:?})",
first_result_elapsed, client_id, sub.id
);
first_result = false;
}
// check if this is still active // check if this is still active
// TODO: check every N rows // TODO: check every N rows
if abandon_query_rx.try_recv().is_ok() { if abandon_query_rx.try_recv().is_ok() {
+4 -2
View File
@@ -464,9 +464,11 @@ async fn nostr_server(
let cid = conn.get_client_prefix(); let cid = conn.get_client_prefix();
// Create a channel for receiving query results from the database. // Create a channel for receiving query results from the database.
// we will send out the tx handle to any query we generate. // we will send out the tx handle to any query we generate.
let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256); // this has capacity for some of the larger requests we see, which
// should allow the DB thread to release the handle earlier.
let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(20000);
// Create channel for receiving NOTICEs // Create channel for receiving NOTICEs
let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(32); let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(128);
// last time this client sent data (message, ping, etc.) // last time this client sent data (message, ping, etc.)
let mut last_message_time = Instant::now(); let mut last_message_time = Instant::now();