mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Rename `reactor` module to `io` 

The name was always confusing and didn't mean much in the contetx of the
service.
This commit is contained in:
Alexis Sellier 2023-05-08 12:01:07 +02:00
parent 8508eab8fa
commit 1d9ee81b6d
No known key found for this signature in database
7 changed files with 48 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
radicle-node/src/service.rs
View File

@ -3,8 +3,8 @@
#![allow(clippy::collapsible_if)] #![allow(clippy::collapsible_if)]
pub mod config; pub mod config;
pub mod filter; pub mod filter;
pub mod io;
pub mod message; pub mod message;
pub mod reactor;
pub mod session; pub mod session;
pub mod tracking; pub mod tracking;
@ -47,8 +47,8 @@ pub use crate::service::message::{Message, ZeroBytes};
pub use crate::service::session::Session; pub use crate::service::session::Session;
use self::gossip::Gossip; use self::gossip::Gossip;
use self::io::Outbox;
use self::message::InventoryAnnouncement; use self::message::InventoryAnnouncement;
use self::reactor::Reactor;
use self::tracking::NamespacesError; use self::tracking::NamespacesError;
/// Target number of peers to maintain connections to. /// Target number of peers to maintain connections to.
@ -194,8 +194,8 @@ pub struct Service<R, A, S, G> {
sessions: Sessions, sessions: Sessions,
/// Clock. Tells the time. /// Clock. Tells the time.
clock: LocalTime, clock: LocalTime,
/// Interface to the I/O reactor. /// I/O outbox.
reactor: Reactor, outbox: Outbox,
/// Source of entropy. /// Source of entropy.
rng: Rng, rng: Rng,
/// Fetch requests initiated by user, which are waiting for results. /// Fetch requests initiated by user, which are waiting for results.
@ -261,7 +261,7 @@ where
clock, clock,
routing, routing,
gossip: Gossip::default(), gossip: Gossip::default(),
reactor: Reactor::default(), outbox: Outbox::default(),
sessions, sessions,
fetch_reqs: HashMap::new(), fetch_reqs: HashMap::new(),
filter: Filter::empty(), filter: Filter::empty(),
@ -276,8 +276,8 @@ where
/// Return the next i/o action to execute. /// Return the next i/o action to execute.
#[allow(clippy::should_implement_trait)] #[allow(clippy::should_implement_trait)]
pub fn next(&mut self) -> Option<reactor::Io> { pub fn next(&mut self) -> Option<io::Io> {
self.reactor.next() self.outbox.next()
} }
/// Track a repository. /// Track a repository.
@ -356,9 +356,9 @@ where
Events::from(self.emitter.subscribe()) Events::from(self.emitter.subscribe())
} }
/// Get I/O reactor. /// Get I/O outbox.
pub fn reactor(&mut self) -> &mut Reactor { pub fn outbox(&mut self) -> &mut Outbox {
&mut self.reactor &mut self.outbox
} }
/// Lookup a project, both locally and in the routing table. /// Lookup a project, both locally and in the routing table.
@ -403,7 +403,7 @@ where
.filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)), .filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)),
); );
// Start periodic tasks. // Start periodic tasks.
self.reactor.wakeup(IDLE_INTERVAL); self.outbox.wakeup(IDLE_INTERVAL);
Ok(()) Ok(())
} }
@ -425,7 +425,7 @@ where
self.keep_alive(&now); self.keep_alive(&now);
self.disconnect_unresponsive_peers(&now); self.disconnect_unresponsive_peers(&now);
self.maintain_connections(); self.maintain_connections();
self.reactor.wakeup(IDLE_INTERVAL); self.outbox.wakeup(IDLE_INTERVAL);
self.last_idle = now; self.last_idle = now;
} }
if now - self.last_sync >= SYNC_INTERVAL { if now - self.last_sync >= SYNC_INTERVAL {
@ -434,7 +434,7 @@ where
if let Err(e) = self.fetch_missing_inventory() { if let Err(e) = self.fetch_missing_inventory() {
error!(target: "service", "Error fetching missing inventory: {e}"); error!(target: "service", "Error fetching missing inventory: {e}");
} }
self.reactor.wakeup(SYNC_INTERVAL); self.outbox.wakeup(SYNC_INTERVAL);
self.last_sync = now; self.last_sync = now;
} }
if now - self.last_announce >= ANNOUNCE_INTERVAL { if now - self.last_announce >= ANNOUNCE_INTERVAL {
@ -445,7 +445,7 @@ where
{ {
error!(target: "service", "Error announcing inventory: {}", err); error!(target: "service", "Error announcing inventory: {}", err);
} }
self.reactor.wakeup(ANNOUNCE_INTERVAL); self.outbox.wakeup(ANNOUNCE_INTERVAL);
self.last_announce = now; self.last_announce = now;
} }
if now - self.last_prune >= PRUNE_INTERVAL { if now - self.last_prune >= PRUNE_INTERVAL {
@ -454,7 +454,7 @@ where
if let Err(err) = self.prune_routing_entries(&now) { if let Err(err) = self.prune_routing_entries(&now) {
error!("Error pruning routing entries: {}", err); error!("Error pruning routing entries: {}", err);
} }
self.reactor.wakeup(PRUNE_INTERVAL); self.outbox.wakeup(PRUNE_INTERVAL);
self.last_prune = now; self.last_prune = now;
} }
@ -470,7 +470,7 @@ where
self.connect(nid, addr); self.connect(nid, addr);
} }
Command::Disconnect(nid) => { Command::Disconnect(nid) => {
self.reactor.disconnect(nid, DisconnectReason::Command); self.outbox.disconnect(nid, DisconnectReason::Command);
} }
Command::Seeds(rid, resp) => match self.seeds(&rid) { Command::Seeds(rid, resp) => match self.seeds(&rid) {
Ok(seeds) => { Ok(seeds) => {
@ -493,7 +493,7 @@ where
resp.send(tracked).ok(); resp.send(tracked).ok();
// Let all our peers know that we're interested in this repo from now on. // Let all our peers know that we're interested in this repo from now on.
self.reactor.broadcast( self.outbox.broadcast(
Message::subscribe(self.filter(), self.time(), Timestamp::MAX), Message::subscribe(self.filter(), self.time(), Timestamp::MAX),
self.sessions.connected().map(|(_, s)| s), self.sessions.connected().map(|(_, s)| s),
); );
@ -563,7 +563,7 @@ where
match self.tracking.namespaces_for(&self.storage, &rid) { match self.tracking.namespaces_for(&self.storage, &rid) {
Ok(namespaces) => { Ok(namespaces) => {
self.reactor.fetch(session, rid, namespaces); self.outbox.fetch(session, rid, namespaces);
} }
Err(err) => { Err(err) => {
error!(target: "service", "Error getting namespaces for {rid}: {err}"); error!(target: "service", "Error getting namespaces for {rid}: {err}");
@ -617,8 +617,7 @@ where
// For now, we only disconnect the remote in case of timeout. In the future, // For now, we only disconnect the remote in case of timeout. In the future,
// there may be other reasons to disconnect. // there may be other reasons to disconnect.
if err.is_timeout() { if err.is_timeout() {
self.reactor self.outbox.disconnect(remote, DisconnectReason::Fetch(err));
.disconnect(remote, DisconnectReason::Fetch(err));
} }
FetchResult::Failed { reason } FetchResult::Failed { reason }
} }
@ -692,7 +691,7 @@ where
if link.is_outbound() { if link.is_outbound() {
if let Some(peer) = self.sessions.get_mut(&remote) { if let Some(peer) = self.sessions.get_mut(&remote) {
let attempted = peer.to_connected(self.clock); let attempted = peer.to_connected(self.clock);
self.reactor.write_all(peer, msgs); self.outbox.write_all(peer, msgs);
if let Err(e) = self.addresses.connected(&remote, &attempted, self.time()) { if let Err(e) = self.addresses.connected(&remote, &attempted, self.time()) {
error!(target: "service", "Error updating address book with connection: {e}"); error!(target: "service", "Error updating address book with connection: {e}");
@ -714,7 +713,7 @@ where
self.clock, self.clock,
self.config.limits.clone(), self.config.limits.clone(),
)); ));
self.reactor.write_all(peer, msgs); self.outbox.write_all(peer, msgs);
} }
} }
} }
@ -756,7 +755,7 @@ where
debug!(target: "service", "Reconnecting to {remote} in {delay}.."); debug!(target: "service", "Reconnecting to {remote} in {delay}..");
self.reactor.wakeup(delay); self.outbox.wakeup(delay);
} else { } else {
self.sessions.remove(&remote); self.sessions.remove(&remote);
// Only re-attempt outbound connections, since we don't care if an inbound connection // Only re-attempt outbound connections, since we don't care if an inbound connection
@ -773,7 +772,7 @@ where
Err(err) => { Err(err) => {
// If there's an error, stop processing messages from this peer. // If there's an error, stop processing messages from this peer.
// However, we still relay messages returned up to this point. // However, we still relay messages returned up to this point.
self.reactor self.outbox
.disconnect(remote, DisconnectReason::Session(err)); .disconnect(remote, DisconnectReason::Session(err));
// FIXME: The peer should be set in a state such that we don't // FIXME: The peer should be set in a state such that we don't
@ -1089,7 +1088,7 @@ where
.filter(|(id, _)| *id != remote && *id != &announcer) .filter(|(id, _)| *id != remote && *id != &announcer)
.map(|(_, p)| p); .map(|(_, p)| p);
self.reactor.relay(ann, relay_to); self.outbox.relay(ann, relay_to);
return Ok(()); return Ok(());
} }
@ -1102,7 +1101,7 @@ where
// Don't send announcements authored by the remote, back to the remote. // Don't send announcements authored by the remote, back to the remote.
.filter(|ann| &ann.node != remote) .filter(|ann| &ann.node != remote)
{ {
self.reactor.write(peer, ann.into()); self.outbox.write(peer, ann.into());
} }
peer.subscribe = Some(subscribe); peer.subscribe = Some(subscribe);
} }
@ -1111,7 +1110,7 @@ where
if ponglen > Ping::MAX_PONG_ZEROES { if ponglen > Ping::MAX_PONG_ZEROES {
return Ok(()); return Ok(());
} }
self.reactor.write( self.outbox.write(
peer, peer,
Message::Pong { Message::Pong {
zeroes: ZeroBytes::new(ponglen), zeroes: ZeroBytes::new(ponglen),
@ -1238,7 +1237,7 @@ where
}); });
let ann = msg.signed(&self.signer); let ann = msg.signed(&self.signer);
self.reactor.broadcast(ann, peers); self.outbox.broadcast(ann, peers);
Ok(()) Ok(())
} }
@ -1266,7 +1265,7 @@ where
fn reconnect(&mut self, nid: NodeId, addr: Address) -> bool { fn reconnect(&mut self, nid: NodeId, addr: Address) -> bool {
if let Some(sess) = self.sessions.get_mut(&nid) { if let Some(sess) = self.sessions.get_mut(&nid) {
sess.to_initial(); sess.to_initial();
self.reactor.connect(nid, addr); self.outbox.connect(nid, addr);
return true; return true;
} }
@ -1292,7 +1291,7 @@ where
self.config.limits.clone(), self.config.limits.clone(),
), ),
); );
self.reactor.connect(nid, addr); self.outbox.connect(nid, addr);
true true
} }
@ -1358,7 +1357,7 @@ where
let time = self.time(); let time = self.time();
let inv = Message::inventory(gossip::inventory(time, inventory), &self.signer); let inv = Message::inventory(gossip::inventory(time, inventory), &self.signer);
for (_, sess) in self.sessions.connected() { for (_, sess) in self.sessions.connected() {
self.reactor.write(sess, inv.clone()); self.outbox.write(sess, inv.clone());
} }
Ok(()) Ok(())
} }
@ -1384,7 +1383,7 @@ where
.filter(|(_, session)| *now - session.last_active >= STALE_CONNECTION_TIMEOUT); .filter(|(_, session)| *now - session.last_active >= STALE_CONNECTION_TIMEOUT);
for (_, session) in stale { for (_, session) in stale {
self.reactor.disconnect( self.outbox.disconnect(
session.id, session.id,
DisconnectReason::Session(session::Error::Timeout), DisconnectReason::Session(session::Error::Timeout),
); );
@ -1399,7 +1398,7 @@ where
.filter(|(_, session)| *now - session.last_active >= KEEP_ALIVE_DELTA) .filter(|(_, session)| *now - session.last_active >= KEEP_ALIVE_DELTA)
.map(|(_, session)| session); .map(|(_, session)| session);
for session in inactive_sessions { for session in inactive_sessions {
session.ping(&mut self.reactor).ok(); session.ping(&mut self.outbox).ok();
} }
} }
@ -1488,7 +1487,7 @@ where
/// Maintain persistent peer connections. /// Maintain persistent peer connections.
fn maintain_persistent(&mut self) { fn maintain_persistent(&mut self) {
debug!(target: "service", "Maintaining persistent peers.."); trace!(target: "service", "Maintaining persistent peers..");
let now = self.local_time(); let now = self.local_time();
let mut reconnect = Vec::new(); let mut reconnect = Vec::new();

12
radicle-node/src/service/reactor.rs → radicle-node/src/service/io.rs
View File

@ -9,7 +9,7 @@ use crate::storage::Namespaces;
use super::message::{Announcement, AnnouncementMessage}; use super::message::{Announcement, AnnouncementMessage};
/// Output of a state transition. /// I/O operation to execute at the network/wire level.
#[derive(Debug)] #[derive(Debug)]
pub enum Io { pub enum Io {
/// There are some messages ready to be sent to a peer. /// There are some messages ready to be sent to a peer.
@ -31,14 +31,14 @@ pub enum Io {
Wakeup(LocalDuration), Wakeup(LocalDuration),
} }
/// Interface to the network reactor. /// Interface to the network.
#[derive(Debug, Default)] #[derive(Debug, Default)]
pub struct Reactor { pub struct Outbox {
/// Outgoing I/O queue. /// Outgoing I/O queue.
io: VecDeque<Io>, io: VecDeque<Io>,
} }
impl Reactor { impl Outbox {
/// Connect to a peer. /// Connect to a peer.
pub fn connect(&mut self, id: NodeId, addr: Address) { pub fn connect(&mut self, id: NodeId, addr: Address) {
self.io.push_back(Io::Connect(id, addr)); self.io.push_back(Io::Connect(id, addr));
@ -117,12 +117,12 @@ impl Reactor {
} }
#[cfg(any(test, feature = "test"))] #[cfg(any(test, feature = "test"))]
pub(crate) fn outbox(&mut self) -> &mut VecDeque<Io> { pub(crate) fn queue(&mut self) -> &mut VecDeque<Io> {
&mut self.io &mut self.io
} }
} }
impl Iterator for Reactor { impl Iterator for Outbox {
type Item = Io; type Item = Io;
fn next(&mut self) -> Option<Self::Item> { fn next(&mut self) -> Option<Self::Item> {

4
radicle-node/src/service/session.rs
View File

@ -4,7 +4,7 @@ use std::{fmt, mem};
use crate::service::config::Limits; use crate::service::config::Limits;
use crate::service::message; use crate::service::message;
use crate::service::message::Message; use crate::service::message::Message;
use crate::service::{Address, Id, LocalTime, NodeId, Reactor, Rng}; use crate::service::{Address, Id, LocalTime, NodeId, Outbox, Rng};
use crate::Link; use crate::Link;
#[derive(Debug, Copy, Clone, Default, PartialEq, Eq)] #[derive(Debug, Copy, Clone, Default, PartialEq, Eq)]
@ -295,7 +295,7 @@ impl Session {
} }
} }
pub fn ping(&mut self, reactor: &mut Reactor) -> Result<(), Error> { pub fn ping(&mut self, reactor: &mut Outbox) -> Result<(), Error> {
if let State::Connected { ping, .. } = &mut self.state { if let State::Connected { ping, .. } = &mut self.state {
let msg = message::Ping::new(&mut self.rng); let msg = message::Ping::new(&mut self.rng);
*ping = PingState::AwaitingResponse(msg.ponglen); *ping = PingState::AwaitingResponse(msg.ponglen);

8
radicle-node/src/test/peer.rs
View File

@ -18,8 +18,8 @@ use crate::node::routing;
use crate::prelude::*; use crate::prelude::*;
use crate::runtime::Emitter; use crate::runtime::Emitter;
use crate::service; use crate::service;
use crate::service::io::Io;
use crate::service::message::*; use crate::service::message::*;
use crate::service::reactor::Io;
use crate::service::tracking::{Policy, Scope}; use crate::service::tracking::{Policy, Scope};
use crate::service::*; use crate::service::*;
use crate::storage::git::transport::remote; use crate::storage::git::transport::remote;
@ -354,7 +354,7 @@ where
pub fn messages(&mut self, remote: NodeId) -> impl Iterator<Item = Message> { pub fn messages(&mut self, remote: NodeId) -> impl Iterator<Item = Message> {
let mut msgs = Vec::new(); let mut msgs = Vec::new();
self.service.reactor().outbox().retain(|o| match o { self.service.outbox().queue().retain(|o| match o {
Io::Write(a, messages) if *a == remote => { Io::Write(a, messages) if *a == remote => {
msgs.extend(messages.clone()); msgs.extend(messages.clone());
false false
@ -372,12 +372,12 @@ where
/// Get a draining iterator over the peer's I/O outbox. /// Get a draining iterator over the peer's I/O outbox.
pub fn outbox(&mut self) -> impl Iterator<Item = Io> + '_ { pub fn outbox(&mut self) -> impl Iterator<Item = Io> + '_ {
iter::from_fn(|| self.service.reactor().next()) iter::from_fn(|| self.service.outbox().next())
} }
/// Get a draining iterator over the peer's I/O outbox, which only returns fetches. /// Get a draining iterator over the peer's I/O outbox, which only returns fetches.
pub fn fetches(&mut self) -> impl Iterator<Item = (Id, NodeId, Namespaces)> + '_ { pub fn fetches(&mut self) -> impl Iterator<Item = (Id, NodeId, Namespaces)> + '_ {
iter::from_fn(|| self.service.reactor().next()).filter_map(|io| { iter::from_fn(|| self.service.outbox().next()).filter_map(|io| {
if let Io::Fetch { if let Io::Fetch {
rid, rid,
remote, remote,

2
radicle-node/src/test/simulator.rs
View File

@ -16,7 +16,7 @@ use log::*;
use crate::crypto::Signer; use crate::crypto::Signer;
use crate::git::raw as git; use crate::git::raw as git;
use crate::prelude::{Address, Id}; use crate::prelude::{Address, Id};
use crate::service::reactor::Io; use crate::service::io::Io;
use crate::service::{DisconnectReason, Event, Message, NodeId}; use crate::service::{DisconnectReason, Event, Message, NodeId};
use crate::storage::{Namespaces, RefUpdate}; use crate::storage::{Namespaces, RefUpdate};
use crate::storage::{WriteRepository, WriteStorage}; use crate::storage::{WriteRepository, WriteStorage};

2
radicle-node/src/tests.rs
View File

@ -18,8 +18,8 @@ use crate::prelude::*;
use crate::prelude::{LocalDuration, Timestamp}; use crate::prelude::{LocalDuration, Timestamp};
use crate::service::config::*; use crate::service::config::*;
use crate::service::filter::Filter; use crate::service::filter::Filter;
use crate::service::io::Io;
use crate::service::message::*; use crate::service::message::*;
use crate::service::reactor::Io;
use crate::service::ServiceState as _; use crate::service::ServiceState as _;
use crate::service::*; use crate::service::*;
use crate::storage::git::transport::{local, remote}; use crate::storage::git::transport::{local, remote};

2
radicle-node/src/wire/protocol.rs
View File

@ -27,7 +27,7 @@ use radicle::storage::WriteStorage;
use crate::crypto::Signer; use crate::crypto::Signer;
use crate::prelude::Deserializer; use crate::prelude::Deserializer;
use crate::service::reactor::Io; use crate::service::io::Io;
use crate::service::{session, DisconnectReason, Service}; use crate::service::{session, DisconnectReason, Service};
use crate::wire::frame; use crate::wire::frame;
use crate::wire::frame::{Frame, FrameData, StreamId}; use crate::wire::frame::{Frame, FrameData, StreamId};