connect LIE state machine events to TIE subsystem

This commit adds a new broadcast event channel specifically for RIFT
level peering events. These events are distinct from the lower level
events that happen within the RIFT peering state machines, as those
should stay within the peering state machines and not leak out into
other subsystems. RIFT level peering events are specifically for entry
and exit of three way adjacencies.

When a three-way adjacency is accomplished, a peer-up event is emitted
that carries with it the peer information, including the LIE message
from the associated peer, and the link-local IPv6 interface that the
peer can be reached on.

The TIE subsystem listens for thse RIFT level peering events and
responds to them by entering into TIE exchanges with established peers.
Or in the case of peer-down events, clearing out routing state that is
associated with the peer that has been lost.

Author: rcgoodfellow

Cargo.lock

@@ -5,6 +5,7 @@
 
 use rift::Rift;
 use rift::config::Config;
+use rift_protocol::Level;
 use slog;
 use slog_term;
 use slog_async;
@@ -29,6 +30,7 @@ struct Opts {
     #[clap(short, long, parse(from_occurrences))]
     verbose: i32,
     id: u64,
+    level: Level,
 }
 
 
@@ -44,7 +46,10 @@ async fn main() -> Result<(), String> {
     let opts: Opts = Opts::parse();
 
     let ilu = Arc::new(Mutex::new(crate::platform::Illumos{log: log.clone()}));
-    let mut riftp = Rift::new(ilu, log.clone(), Config{id: opts.id});
+    let mut riftp = Rift::new(ilu, log.clone(), Config{
+        id: opts.id,
+        level: opts.level,
+    });
     match riftp.run().await {
         Ok(()) => warn!(log, "early exit?"),
         Err(e) => error!(log, "rift: {}", e),

illumos/src/main.rs

@@ -3,6 +3,7 @@
 pub mod error;
 
 use std::net;
+use std::hash::{Hash, Hasher};
 use error::Error;
 use icmpv6::RDPMessage;
 use tokio::sync::mpsc::{Sender, Receiver};
@@ -19,18 +20,21 @@ pub struct TIEPacketTx {
 }
 
 pub trait Platform {
+    // Local platform state
     fn get_links(&self) -> Result<Vec<LinkStatus>, Error>;
     fn get_link_status(&self, link_name: impl AsRef<str>) -> Result<LinkStatus, Error>;
     fn get_interface_v6ll(&self, interface: impl AsRef<str>) -> Result<Option<IpIfAddr>, Error>;
 
-
+    // IPv6 RDP
     fn solicit_rift_routers(&self, interface: Option<IpIfAddr>) -> Result<(), Error>;
     fn advertise_rift_router(&self, interface: Option<IpIfAddr>) -> Result<(), Error>;
     fn get_rdp_channel(&self, interface: Option<IpIfAddr>) -> Result<Receiver<RDPMessage>, Error>;
-    fn get_link_channel(&self, local: Ipv6Addr, peer: Ipv6Addr, local_ifx: i32) -> Result<
-        (Sender<LIEPacket>, Receiver<LIEPacket>),
-        Error
-    >;
+
+    // RIFT Link Element Exchange (LIE)
+    fn get_link_channel(&self, local: Ipv6Addr, peer: Ipv6Addr, local_ifx: i32) 
+    -> Result<(Sender<LIEPacket>, Receiver<LIEPacket>), Error>;
+
+    // RIFT Topology Element Exchange (TIE)
     fn get_topology_channel(&self) -> Result<(Sender<TIEPacketTx>, Receiver<TIEPacket>), Error>;
 }
 
@@ -40,6 +44,14 @@ pub struct IpIfAddr {
     pub if_index: i32,
 }
 
+impl Eq for IpIfAddr {}
+impl Hash for IpIfAddr {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.if_index.hash(state);
+        self.addr.hash(state);
+    }
+}
+
 #[derive(Debug, Copy, Clone, Deserialize, Serialize, JsonSchema, PartialEq)]
 pub enum LinkState {
     Unknown,

platform/src/lib.rs

@@ -1,8 +1,10 @@
 // Copyright 2021 Oxide Computer Company
 use serde::{Deserialize, Serialize};
 use schemars::JsonSchema;
+use rift_protocol::Level;
 
 #[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema)]
 pub struct Config {
     pub id: u64,
+    pub level: Level,
 }

rift/src/config.rs

@@ -10,14 +10,14 @@ use std::time::SystemTime;
 use crate::error::Error;
 use std::net::Ipv6Addr;
 use std::sync::Arc;
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, broadcast};
 use tokio::time::sleep;
 use std::time::{Duration};
 use std::collections::HashSet;
 use std::hash::{Hash, Hasher};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
-use platform::Platform;
+use platform::{Platform, IpIfAddr};
 use slog::{trace, info};
 use link::LinkSM;
 use rift_protocol::lie::{LIEPacket, Neighbor};
@@ -59,6 +59,12 @@ impl Peer {
     }
 }
 
+#[derive(Debug, Clone)]
+enum PeerEvent {
+    Up((Peer, IpIfAddr)),
+    Down((Peer, IpIfAddr)),
+}
+
 pub struct Rift<P: Platform + std::marker::Send + 'static> {
     platform: Arc::<Mutex::<P>>,
     links: Arc::<Mutex::<HashSet::<LinkSM>>>,
@@ -93,6 +99,17 @@ impl<P: Platform + std::marker::Send + std::marker::Sync> Rift<P> {
             p.get_links()?
         };
 
+        let (peer_event_tx, peer_event_rx) = broadcast::channel(32);
+
+        // start topology thread
+        topology::tie_entry(
+            self.log.clone(),
+            self.platform.clone(),
+            self.links.clone(),
+            self.config,
+            peer_event_rx,
+        ).await;
+
         // start link state machines
         for l in links.iter() {
             let mut sm = link::LinkSM::new(
@@ -101,19 +118,14 @@ impl<P: Platform + std::marker::Send + std::marker::Sync> Rift<P> {
                 l.state,
                 self.config,
             );
-            sm.run(self.platform.clone()).await;
+            sm.run(
+                self.platform.clone(),
+                peer_event_tx.clone(),
+            ).await;
             let mut lsms = self.links.lock().await;
             lsms.insert(sm);
         }
 
-        // start topology thread
-        if false {
-        topology::tie_entry(
-            self.log.clone(),
-            self.platform.clone(),
-        ).await;
-        }
-
         self.router_loop().await?;
 
         Ok(())