Skip to content

Commit

Permalink
Use separate tokio runtime for gossip persistence.
Browse files Browse the repository at this point in the history
  • Loading branch information
arik-so committed Mar 12, 2024
1 parent eaebdf7 commit 81cb3b3
Showing 1 changed file with 6 additions and 2 deletions.
8 changes: 6 additions & 2 deletions src/persistence.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ use lightning::log_info;
use lightning::routing::gossip::NetworkGraph;
use lightning::util::logger::Logger;
use lightning::util::ser::Writeable;
use tokio::runtime::Runtime;
use tokio::sync::{mpsc, Mutex, Semaphore};

use crate::config;
Expand All @@ -19,16 +20,19 @@ const INSERT_PARALELLISM: usize = 16;
pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
network_graph: Arc<NetworkGraph<L>>,
tokio_runtime: Runtime,
logger: L
}

impl<L: Deref> GossipPersister<L> where L::Target: Logger {
pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
let (gossip_persistence_sender, gossip_persistence_receiver) =
mpsc::channel::<GossipMessage>(100);
let runtime = Runtime::new().unwrap();
(GossipPersister {
gossip_persistence_receiver,
network_graph,
tokio_runtime: runtime,
logger
}, gossip_persistence_sender)
}
Expand Down Expand Up @@ -127,7 +131,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
let mut announcement_signed = Vec::new();
announcement.write(&mut announcement_signed).unwrap();

let _task = tokio::spawn(async move {
let _task = self.tokio_runtime.spawn(async move {
let client;
{
let mut connections_set = connections_cache_ref.lock().await;
Expand Down Expand Up @@ -219,7 +223,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
// this may not be used outside test cfg
let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;

let _task = tokio::spawn(async move {
let _task = self.tokio_runtime.spawn(async move {
let client;
{
let mut connections_set = connections_cache_ref.lock().await;
Expand Down

0 comments on commit 81cb3b3

Please sign in to comment.