Ecosystem/serai
Ecosystem/serai
1
0
Fork 0
mirror of https://github.com/serai-dex/serai.git synced 2025-01-10 21:04:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Have every node respond to Heartbeat so they don't download the messages over the net

Browse source
This commit is contained in:
Luke Parker 2023-10-14 14:57:46 -04:00
parent f414735be5
commit 863a7842ca
No known key found for this signature in database
3 changed files with 12 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
coordinator/src/p2p.rs
View file

@ -493,6 +493,7 @@ pub async fn handle_p2p_task<D: Db, P: P2p>(
} }
*/ */
/*
// Have up to three nodes respond // Have up to three nodes respond
let responders = u64::from(spec.n().min(3)); let responders = u64::from(spec.n().min(3));
@ -521,7 +522,13 @@ pub async fn handle_p2p_task<D: Db, P: P2p>(
} }
log::debug!("received heartbeat and selected to respond"); log::debug!("received heartbeat and selected to respond");
*/
// Have every node respond
// While we could only have a subset respond, LibP2P will sync all messages
// it isn't aware of
// It's cheaper to be aware from our disk than from over the network
// TODO: Spawn a dedicated topic for this heartbeat response?
let mut latest = msg.msg[.. 32].try_into().unwrap(); let mut latest = msg.msg[.. 32].try_into().unwrap();
while let Some(next) = reader.block_after(&latest) { while let Some(next) = reader.block_after(&latest) {
let mut res = reader.block(&next).unwrap().serialize(); let mut res = reader.block(&next).unwrap().serialize();

6
coordinator/src/tests/tributary/handle_p2p.rs
View file

@ -3,8 +3,6 @@ use std::sync::Arc;
use rand_core::OsRng; use rand_core::OsRng;
use ciphersuite::{Ciphersuite, Ristretto};
use tokio::{sync::broadcast, time::sleep}; use tokio::{sync::broadcast, time::sleep};
use serai_db::MemDb; use serai_db::MemDb;
@ -30,11 +28,11 @@ async fn handle_p2p_test() {
let mut tributary_senders = vec![]; let mut tributary_senders = vec![];
let mut tributary_arcs = vec![]; let mut tributary_arcs = vec![];
for (i, (p2p, tributary)) in tributaries.drain(..).enumerate() { for (p2p, tributary) in tributaries.drain(..) {
let tributary = Arc::new(tributary); let tributary = Arc::new(tributary);
tributary_arcs.push(tributary.clone()); tributary_arcs.push(tributary.clone());
let (new_tributary_send, new_tributary_recv) = broadcast::channel(5); let (new_tributary_send, new_tributary_recv) = broadcast::channel(5);
tokio::spawn(handle_p2p_task(Ristretto::generator() * *keys[i], p2p, new_tributary_recv)); tokio::spawn(handle_p2p_task(p2p, new_tributary_recv));
new_tributary_send new_tributary_send
.send(TributaryEvent::NewTributary(ActiveTributary { spec: spec.clone(), tributary })) .send(TributaryEvent::NewTributary(ActiveTributary { spec: spec.clone(), tributary }))
.map_err(|_| "failed to send ActiveTributary") .map_err(|_| "failed to send ActiveTributary")

7
coordinator/src/tests/tributary/sync.rs
View file

@ -38,12 +38,11 @@ async fn sync_test() {
let mut tributary_senders = vec![]; let mut tributary_senders = vec![];
let mut tributary_arcs = vec![]; let mut tributary_arcs = vec![];
let mut p2p_threads = vec![]; let mut p2p_threads = vec![];
for (i, (p2p, tributary)) in tributaries.drain(..).enumerate() { for (p2p, tributary) in tributaries.drain(..) {
let tributary = Arc::new(tributary); let tributary = Arc::new(tributary);
tributary_arcs.push(tributary.clone()); tributary_arcs.push(tributary.clone());
let (new_tributary_send, new_tributary_recv) = broadcast::channel(5); let (new_tributary_send, new_tributary_recv) = broadcast::channel(5);
let thread = let thread = tokio::spawn(handle_p2p_task(p2p, new_tributary_recv));
tokio::spawn(handle_p2p_task(Ristretto::generator() * *keys[i], p2p, new_tributary_recv));
new_tributary_send new_tributary_send
.send(TributaryEvent::NewTributary(ActiveTributary { spec: spec.clone(), tributary })) .send(TributaryEvent::NewTributary(ActiveTributary { spec: spec.clone(), tributary }))
.map_err(|_| "failed to send ActiveTributary") .map_err(|_| "failed to send ActiveTributary")
@ -78,7 +77,7 @@ async fn sync_test() {
let syncer_key = Ristretto::generator() * *syncer_key; let syncer_key = Ristretto::generator() * *syncer_key;
let syncer_tributary = Arc::new(syncer_tributary); let syncer_tributary = Arc::new(syncer_tributary);
let (syncer_tributary_send, syncer_tributary_recv) = broadcast::channel(5); let (syncer_tributary_send, syncer_tributary_recv) = broadcast::channel(5);
tokio::spawn(handle_p2p_task(syncer_key, syncer_p2p.clone(), syncer_tributary_recv)); tokio::spawn(handle_p2p_task(syncer_p2p.clone(), syncer_tributary_recv));
syncer_tributary_send syncer_tributary_send
.send(TributaryEvent::NewTributary(ActiveTributary { .send(TributaryEvent::NewTributary(ActiveTributary {
spec: spec.clone(), spec: spec.clone(),