From 067b53215401a90e4554287e196c438c5715e7a6 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Wed, 18 Jan 2023 15:15:52 +0100 Subject: [PATCH] node: Get replication test to pass Signed-off-by: Alexis Sellier --- radicle-node/src/client.rs | 22 ++++++++++------- radicle-node/src/client/handle.rs | 14 ++++++++--- radicle-node/src/tests/e2e.rs | 3 ++- radicle-node/src/wire.rs | 2 +- radicle-node/src/wire/protocol.rs | 39 ++++++++++++++++++++----------- radicle-node/src/worker.rs | 10 ++++++-- 6 files changed, 60 insertions(+), 30 deletions(-) diff --git a/radicle-node/src/client.rs b/radicle-node/src/client.rs index 79d40402..fd6b0da8 100644 --- a/radicle-node/src/client.rs +++ b/radicle-node/src/client.rs @@ -1,5 +1,6 @@ use std::{io, net, thread, time}; +use crossbeam_channel as chan; use cyphernet::{Cert, EcSign}; use netservices::resource::NetAccept; use radicle::profile::Home; @@ -13,6 +14,7 @@ use crate::control; use crate::crypto::Signature; use crate::node::NodeId; use crate::service::{routing, tracking}; +use crate::wire; use crate::wire::Wire; use crate::worker::{WorkerPool, WorkerReq}; use crate::{crypto, service, LocalTime}; @@ -57,7 +59,7 @@ pub struct Runtime { pub id: NodeId, pub handle: Handle, pub control: thread::JoinHandle>, - pub reactor: Reactor, + pub reactor: Reactor, pub pool: WorkerPool, pub local_addrs: Vec, } @@ -113,14 +115,7 @@ impl Runtime { sig: EcSign::sign(&signer, id.as_slice()), }; - let (worker_send, worker_recv) = crossbeam_channel::unbounded::>(); - let pool = WorkerPool::with( - 8, - time::Duration::from_secs(9), - storage, - worker_recv, - id.to_human(), - ); + let (worker_send, worker_recv) = chan::unbounded::>(); let mut wire = Wire::new(service, worker_send, cert, signer, proxy, clock); let mut local_addrs = Vec::new(); @@ -140,6 +135,15 @@ impl Runtime { move || control::listen(node_sock, handle) }); + let pool = WorkerPool::with( + 8, + time::Duration::from_secs(9), + storage, + worker_recv, + handle.clone(), + id.to_human(), + ); + Ok(Runtime { id, control, diff --git a/radicle-node/src/client/handle.rs b/radicle-node/src/client/handle.rs index c1a3951b..65cf953d 100644 --- a/radicle-node/src/client/handle.rs +++ b/radicle-node/src/client/handle.rs @@ -10,6 +10,7 @@ use crate::profile::Home; use crate::service; use crate::service::{CommandError, FetchLookup, QueryState}; use crate::service::{NodeId, Sessions}; +use crate::wire; /// An error resulting from a handle method. #[derive(Error, Debug)] @@ -51,7 +52,7 @@ impl From> for Error { pub struct Handle { pub(crate) home: Home, - pub(crate) controller: reactor::Controller, + pub(crate) controller: reactor::Controller, } impl Clone for Handle { @@ -64,12 +65,19 @@ impl Clone for Handle { } impl Handle { - pub fn new(home: Home, controller: reactor::Controller) -> Self { + pub fn new(home: Home, controller: reactor::Controller) -> Self { Self { home, controller } } + pub fn wakeup(&mut self) -> Result<(), Error> { + // TODO: Handle channel disconnect error correctly. + // This just returns `BrokenPipe`. + self.controller.cmd(wire::Control::Wakeup)?; + Ok(()) + } + fn command(&self, cmd: service::Command) -> Result<(), Error> { - self.controller.cmd(cmd)?; + self.controller.cmd(wire::Control::Command(cmd))?; Ok(()) } } diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index 485139c5..0c852bcd 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/radicle-node/src/tests/e2e.rs @@ -317,7 +317,6 @@ fn test_inventory_sync_star() { } #[test] -#[ignore] fn test_replication() { logger::init(log::Level::Debug); @@ -353,6 +352,8 @@ fn test_replication() { assert_eq!(from, bob.id); assert_eq!(updated, vec![]); + log::debug!(target: "test", "Fetch complete with {}", from); + let inventory = alice.handle.inventory().unwrap(); let alice_refs = alice .storage diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index 151f87a6..8a3d31cf 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -2,7 +2,7 @@ mod message; mod protocol; pub use message::{AddressType, MessageType}; -pub use protocol::{Wire, WireReader, WireSession, WireWriter}; +pub use protocol::{Control, Wire, WireReader, WireSession, WireWriter}; use std::collections::BTreeMap; use std::convert::TryFrom; diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index a82ca8ff..2976c9dd 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -31,6 +31,12 @@ use crate::worker::{WorkerReq, WorkerResp}; use crate::Link; use crate::{address, service}; +#[derive(Debug)] +pub enum Control { + Command(service::Command), + Wakeup, +} + /// Peer session type. pub type WireSession = CypherSession; /// Peer session type (read-only). @@ -307,6 +313,20 @@ where } } + fn wakeup(&mut self) { + let mut completed = Vec::new(); + for peer in self.peers.values() { + if let Peer::Upgraded { response, .. } = peer { + if let Ok(resp) = response.try_recv() { + completed.push(resp); + } + } + } + for resp in completed { + self.fetch_complete(resp); + } + } + fn fetch_complete(&mut self, resp: WorkerResp) { log::debug!(target: "transport", "Fetch completed: {:?}", resp.result); @@ -344,23 +364,11 @@ where { type Listener = NetAccept>; type Transport = NetTransport>; - type Command = service::Command; + type Command = Control; fn tick(&mut self, _time: Duration) { // FIXME: Change this once a proper timestamp is passed into the function. self.service.tick(LocalTime::from(SystemTime::now())); - - let mut completed = Vec::new(); - for peer in self.peers.values() { - if let Peer::Upgraded { response, .. } = peer { - if let Ok(resp) = response.try_recv() { - completed.push(resp); - } - } - } - for resp in completed { - self.fetch_complete(resp); - } } fn handle_wakeup(&mut self) { @@ -487,7 +495,10 @@ where } fn handle_command(&mut self, cmd: Self::Command) { - self.service.command(cmd); + match cmd { + Control::Command(cmd) => self.service.command(cmd), + Control::Wakeup => self.wakeup(), + } } fn handle_error( diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index d776c171..9c51803e 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -13,6 +13,7 @@ use radicle::storage::{ReadRepository, RefUpdate, WriteRepository, WriteStorage} use radicle::{git, Storage}; use reactor::poller::popol; +use crate::client::handle::Handle; use crate::service::reactor::Fetch; use crate::service::{FetchError, FetchResult}; use crate::wire::{WireReader, WireSession, WireWriter}; @@ -36,19 +37,20 @@ struct Worker { storage: Storage, tasks: chan::Receiver>, timeout: time::Duration, + handle: Handle, } impl Worker { /// Waits for tasks and runs them. Blocks indefinitely unless there is an error receiving /// the next task. - fn run(self) -> Result<(), chan::RecvError> { + fn run(mut self) -> Result<(), chan::RecvError> { loop { let task = self.tasks.recv()?; self.process(task); } } - fn process(&self, task: WorkerReq) { + fn process(&mut self, task: WorkerReq) { let WorkerReq { fetch, session, @@ -71,6 +73,8 @@ impl Worker { if channel.send(WorkerResp { result, session }).is_err() { log::error!("Unable to report fetch result: worker channel disconnected"); + } else { + self.handle.wakeup().unwrap(); } } @@ -249,6 +253,7 @@ impl WorkerPool { timeout: time::Duration, storage: Storage, tasks: chan::Receiver>, + handle: Handle, name: String, ) -> Self { let mut pool = Vec::with_capacity(capacity); @@ -256,6 +261,7 @@ impl WorkerPool { let worker = Worker { tasks: tasks.clone(), storage: storage.clone(), + handle: handle.clone(), timeout, }; let thread = thread::Builder::new()