Use separate Tokio runtime for gossip persistence. #73
Conversation
|
LGTM, why is this draft? |
|
CI says "Cannot drop a runtime in a context where blocking is not allowed. This happens when a runtime is dropped from within an asynchronous context." |
|
Because of that CI issue that only happens in tests, though it runs fine. Just trying to get my local tests to fail the same way to then test my fix. Shouldn't take long. |
arik-so
force-pushed
the
arik/2024-03-tokio-runtime-split
branch
from
def546e
to
bf9d941
Compare
src/persistence.rs
Outdated
| let mut _db_schema = None; | ||
| #[cfg(test)] | ||
| { | ||
| _db_schema = Some(crate::tests::db_test_schema()); |
There was a problem hiding this comment.
Why cant this be in the #[test]?
There was a problem hiding this comment.
You mean pass db_schema as a parameter to persist_gossip?
There was a problem hiding this comment.
Ah, apologies, I missed the copy. Can we instead do this transparently as a part of the connection object?
src/persistence.rs
Outdated
| @@ -162,6 +166,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger { | |||
| let mut connections_set = connections_cache_ref.lock().await; | |||
| connections_set.push(client); | |||
| limiter_ref.add_permits(1); | |||
| drop(connections_set); | |||
There was a problem hiding this comment.
Shouldn't we do this right after we build the client? Otherwise we end up holding the lock for the duration of the insert, killing our parallelism.
There was a problem hiding this comment.
yeah, I'll clean it up for real now. Thanks for the concept ack
There was a problem hiding this comment.
Actually, minor followup first: I moved the connections_set creation into a scope that is dropped prior to the db insertion operations. Does that seem sufficient?
src/persistence.rs
Outdated
| @@ -118,6 +122,19 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger { | |||
| insert_limiter.acquire().await.unwrap().forget(); | |||
|
|
|||
| let limiter_ref = Arc::clone(&insert_limiter); | |||
| let client = { | |||
| let connections_cache_ref = Arc::clone(&connections_cache); | |||
There was a problem hiding this comment.
You probably don't need the Arc now :)
src/persistence.rs
Outdated
| @@ -162,6 +166,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger { | |||
| let mut connections_set = connections_cache_ref.lock().await; | |||
| connections_set.push(client); | |||
| limiter_ref.add_permits(1); | |||
| drop(connections_set); | |||
src/persistence.rs
Outdated
| @@ -162,6 +169,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger { | |||
| let mut connections_set = connections_cache_ref.lock().await; | |||
| connections_set.push(client); | |||
| limiter_ref.add_permits(1); | |||
| drop(connections_set); | |||
There was a problem hiding this comment.
This is dropping a lock taken two lines up at the end of a scope, you can just drop it :)
There was a problem hiding this comment.
Think about the femtoseconds, Matt.
arik-so
force-pushed
the
arik/2024-03-tokio-runtime-split
branch
from
2fab4ba
to
eac7646
Compare
src/persistence.rs
Outdated
| @@ -118,6 +121,18 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger { | |||
| insert_limiter.acquire().await.unwrap().forget(); | |||
|
|
|||
| let limiter_ref = Arc::clone(&insert_limiter); | |||
| let client = { | |||
| let connections_cache_ref = &connections_cache; | |||
arik-so
force-pushed
the
arik/2024-03-tokio-runtime-split
branch
from
414ec15
to
560dabd
Compare
arik-so
force-pushed
the
arik/2024-03-tokio-runtime-split
branch
from
560dabd
to
244ae25
Compare
arik-so
force-pushed
the
arik/2024-03-tokio-runtime-split
branch
from
244ae25
to
418d776
Compare
No description provided.