introduce a link-state route algo
This commit is contained in:
@@ -98,7 +98,7 @@ impl ForeignNetworkClient {
|
||||
let peers = peer_map.list_peers().await;
|
||||
let mut tasks = JoinSet::new();
|
||||
if !peers.is_empty() {
|
||||
tracing::warn!(?peers, "collect next hop in foreign network");
|
||||
tracing::warn!(?peers, my_peer_id = ?peer_rpc.my_peer_id(), "collect next hop in foreign network");
|
||||
}
|
||||
for peer in peers {
|
||||
let peer_rpc = peer_rpc.clone();
|
||||
|
||||
@@ -297,7 +297,7 @@ mod tests {
|
||||
create_mock_peer_manager_with_mock_stun, replace_stun_info_collector,
|
||||
},
|
||||
peers::{
|
||||
peer_manager::PeerManager,
|
||||
peer_manager::{PeerManager, RouteAlgoType},
|
||||
tests::{connect_peer_manager, wait_route_appear},
|
||||
},
|
||||
rpc::NatType,
|
||||
@@ -308,6 +308,7 @@ mod tests {
|
||||
async fn create_mock_peer_manager_for_foreign_network(network: &str) -> Arc<PeerManager> {
|
||||
let (s, _r) = tokio::sync::mpsc::channel(1000);
|
||||
let peer_mgr = Arc::new(PeerManager::new(
|
||||
RouteAlgoType::Ospf,
|
||||
get_mock_global_ctx_with_network(Some(NetworkIdentity {
|
||||
network_name: network.to_string(),
|
||||
network_secret: network.to_string(),
|
||||
@@ -359,7 +360,7 @@ mod tests {
|
||||
.list_peers()
|
||||
.await
|
||||
);
|
||||
wait_route_appear(pma_net1.clone(), pmb_net1.my_peer_id())
|
||||
wait_route_appear(pma_net1.clone(), pmb_net1.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(1, pma_net1.list_routes().await.len());
|
||||
@@ -367,10 +368,10 @@ mod tests {
|
||||
|
||||
let pmc_net1 = create_mock_peer_manager_for_foreign_network("net1").await;
|
||||
connect_peer_manager(pmc_net1.clone(), pm_center.clone()).await;
|
||||
wait_route_appear(pma_net1.clone(), pmc_net1.my_peer_id())
|
||||
wait_route_appear(pma_net1.clone(), pmc_net1.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(pmb_net1.clone(), pmc_net1.my_peer_id())
|
||||
wait_route_appear(pmb_net1.clone(), pmc_net1.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(2, pmc_net1.list_routes().await.len());
|
||||
@@ -379,7 +380,7 @@ mod tests {
|
||||
let pmb_net2 = create_mock_peer_manager_for_foreign_network("net2").await;
|
||||
connect_peer_manager(pma_net2.clone(), pm_center.clone()).await;
|
||||
connect_peer_manager(pmb_net2.clone(), pm_center.clone()).await;
|
||||
wait_route_appear(pma_net2.clone(), pmb_net2.my_peer_id())
|
||||
wait_route_appear(pma_net2.clone(), pmb_net2.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(1, pma_net2.list_routes().await.len());
|
||||
|
||||
@@ -3,6 +3,7 @@ pub mod peer;
|
||||
pub mod peer_conn;
|
||||
pub mod peer_manager;
|
||||
pub mod peer_map;
|
||||
pub mod peer_ospf_route;
|
||||
pub mod peer_rip_route;
|
||||
pub mod peer_rpc;
|
||||
pub mod route_trait;
|
||||
|
||||
@@ -34,6 +34,7 @@ use super::{
|
||||
foreign_network_manager::ForeignNetworkManager,
|
||||
peer_conn::PeerConnId,
|
||||
peer_map::PeerMap,
|
||||
peer_ospf_route::PeerRoute,
|
||||
peer_rip_route::BasicRoute,
|
||||
peer_rpc::PeerRpcManager,
|
||||
route_trait::{ArcRoute, Route},
|
||||
@@ -43,7 +44,7 @@ use super::{
|
||||
struct RpcTransport {
|
||||
my_peer_id: PeerId,
|
||||
peers: Weak<PeerMap>,
|
||||
foreign_peers: Mutex<Option<Weak<PeerMap>>>,
|
||||
foreign_peers: Mutex<Option<Weak<ForeignNetworkClient>>>,
|
||||
|
||||
packet_recv: Mutex<UnboundedReceiver<Bytes>>,
|
||||
peer_rpc_tspt_sender: UnboundedSender<Bytes>,
|
||||
@@ -66,7 +67,7 @@ impl PeerRpcManagerTransport for RpcTransport {
|
||||
.ok_or(Error::Unknown)?;
|
||||
let peers = self.peers.upgrade().ok_or(Error::Unknown)?;
|
||||
|
||||
if foreign_peers.has_peer(dst_peer_id) {
|
||||
if foreign_peers.has_next_hop(dst_peer_id) {
|
||||
return foreign_peers.send_msg(msg, dst_peer_id).await;
|
||||
}
|
||||
|
||||
@@ -82,6 +83,18 @@ impl PeerRpcManagerTransport for RpcTransport {
|
||||
}
|
||||
}
|
||||
|
||||
pub enum RouteAlgoType {
|
||||
Rip,
|
||||
Ospf,
|
||||
None,
|
||||
}
|
||||
|
||||
enum RouteAlgoInst {
|
||||
Rip(Arc<BasicRoute>),
|
||||
Ospf(Arc<PeerRoute>),
|
||||
None,
|
||||
}
|
||||
|
||||
pub struct PeerManager {
|
||||
my_peer_id: PeerId,
|
||||
|
||||
@@ -100,7 +113,7 @@ pub struct PeerManager {
|
||||
peer_packet_process_pipeline: Arc<RwLock<Vec<BoxPeerPacketFilter>>>,
|
||||
nic_packet_process_pipeline: Arc<RwLock<Vec<BoxNicPacketFilter>>>,
|
||||
|
||||
basic_route: Arc<BasicRoute>,
|
||||
route_algo_inst: RouteAlgoInst,
|
||||
|
||||
foreign_network_manager: Arc<ForeignNetworkManager>,
|
||||
foreign_network_client: Arc<ForeignNetworkClient>,
|
||||
@@ -117,7 +130,11 @@ impl Debug for PeerManager {
|
||||
}
|
||||
|
||||
impl PeerManager {
|
||||
pub fn new(global_ctx: ArcGlobalCtx, nic_channel: mpsc::Sender<SinkItem>) -> Self {
|
||||
pub fn new(
|
||||
route_algo: RouteAlgoType,
|
||||
global_ctx: ArcGlobalCtx,
|
||||
nic_channel: mpsc::Sender<SinkItem>,
|
||||
) -> Self {
|
||||
let my_peer_id = rand::random();
|
||||
|
||||
let (packet_send, packet_recv) = mpsc::channel(100);
|
||||
@@ -138,7 +155,17 @@ impl PeerManager {
|
||||
});
|
||||
let peer_rpc_mgr = Arc::new(PeerRpcManager::new(rpc_tspt.clone()));
|
||||
|
||||
let basic_route = Arc::new(BasicRoute::new(my_peer_id, global_ctx.clone()));
|
||||
let route_algo_inst = match route_algo {
|
||||
RouteAlgoType::Rip => {
|
||||
RouteAlgoInst::Rip(Arc::new(BasicRoute::new(my_peer_id, global_ctx.clone())))
|
||||
}
|
||||
RouteAlgoType::Ospf => RouteAlgoInst::Ospf(PeerRoute::new(
|
||||
my_peer_id,
|
||||
global_ctx.clone(),
|
||||
peer_rpc_mgr.clone(),
|
||||
)),
|
||||
RouteAlgoType::None => RouteAlgoInst::None,
|
||||
};
|
||||
|
||||
let foreign_network_manager = Arc::new(ForeignNetworkManager::new(
|
||||
my_peer_id,
|
||||
@@ -170,7 +197,7 @@ impl PeerManager {
|
||||
peer_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||
nic_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||
|
||||
basic_route,
|
||||
route_algo_inst,
|
||||
|
||||
foreign_network_manager,
|
||||
foreign_network_client,
|
||||
@@ -404,8 +431,16 @@ impl PeerManager {
|
||||
self.peers.add_route(arc_route).await;
|
||||
}
|
||||
|
||||
pub fn get_route(&self) -> Box<dyn Route + Send + Sync + 'static> {
|
||||
match &self.route_algo_inst {
|
||||
RouteAlgoInst::Rip(route) => Box::new(route.clone()),
|
||||
RouteAlgoInst::Ospf(route) => Box::new(route.clone()),
|
||||
RouteAlgoInst::None => panic!("no route"),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_routes(&self) -> Vec<crate::rpc::Route> {
|
||||
self.basic_route.list_routes().await
|
||||
self.get_route().list_routes().await
|
||||
}
|
||||
|
||||
async fn run_nic_packet_process_pipeline(&self, mut data: BytesMut) -> BytesMut {
|
||||
@@ -487,21 +522,27 @@ impl PeerManager {
|
||||
.foreign_peers
|
||||
.lock()
|
||||
.await
|
||||
.replace(Arc::downgrade(&self.foreign_network_client.get_peer_map()));
|
||||
.replace(Arc::downgrade(&self.foreign_network_client));
|
||||
|
||||
self.foreign_network_manager.run().await;
|
||||
self.foreign_network_client.run().await;
|
||||
}
|
||||
|
||||
pub async fn run(&self) -> Result<(), Error> {
|
||||
self.add_route(self.basic_route.clone()).await;
|
||||
match &self.route_algo_inst {
|
||||
RouteAlgoInst::Ospf(route) => self.add_route(route.clone()).await,
|
||||
RouteAlgoInst::Rip(route) => self.add_route(route.clone()).await,
|
||||
RouteAlgoInst::None => {}
|
||||
};
|
||||
|
||||
self.init_packet_process_pipeline().await;
|
||||
self.start_peer_recv().await;
|
||||
self.peer_rpc_mgr.run();
|
||||
|
||||
self.start_peer_recv().await;
|
||||
self.run_clean_peer_without_conn_routine().await;
|
||||
|
||||
self.run_foriegn_network().await;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -530,7 +571,10 @@ impl PeerManager {
|
||||
}
|
||||
|
||||
pub fn get_basic_route(&self) -> Arc<BasicRoute> {
|
||||
self.basic_route.clone()
|
||||
match &self.route_algo_inst {
|
||||
RouteAlgoInst::Rip(route) => route.clone(),
|
||||
_ => panic!("not rip route"),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn get_foreign_network_manager(&self) -> Arc<ForeignNetworkManager> {
|
||||
@@ -560,10 +604,10 @@ mod tests {
|
||||
connect_peer_manager(peer_mgr_b.clone(), peer_mgr_c.clone()).await;
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_c.clone()).await;
|
||||
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
||||
@@ -119,7 +119,12 @@ impl PeerMap {
|
||||
}
|
||||
|
||||
let Some(gateway_peer_id) = gateway_peer_id else {
|
||||
log::error!("no gateway for dst_peer_id: {}", dst_peer_id);
|
||||
log::error!(
|
||||
"no gateway for dst_peer_id: {}, peers: {:?}, my_peer_id: {}",
|
||||
dst_peer_id,
|
||||
self.peer_map.iter().map(|v| *v.key()).collect::<Vec<_>>(),
|
||||
self.my_peer_id
|
||||
);
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -653,27 +653,39 @@ mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::{
|
||||
common::PeerId,
|
||||
connector::udp_hole_punch::tests::create_mock_peer_manager_with_mock_stun,
|
||||
common::{global_ctx::tests::get_mock_global_ctx, PeerId},
|
||||
connector::udp_hole_punch::tests::replace_stun_info_collector,
|
||||
peers::{
|
||||
peer_manager::PeerManager,
|
||||
peer_manager::{PeerManager, RouteAlgoType},
|
||||
peer_rip_route::Version,
|
||||
tests::{connect_peer_manager, wait_route_appear},
|
||||
},
|
||||
rpc::NatType,
|
||||
};
|
||||
|
||||
async fn create_mock_pmgr() -> Arc<PeerManager> {
|
||||
let (s, _r) = tokio::sync::mpsc::channel(1000);
|
||||
let peer_mgr = Arc::new(PeerManager::new(
|
||||
RouteAlgoType::Rip,
|
||||
get_mock_global_ctx(),
|
||||
s,
|
||||
));
|
||||
replace_stun_info_collector(peer_mgr.clone(), NatType::Unknown);
|
||||
peer_mgr.run().await.unwrap();
|
||||
peer_mgr
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_rip_route() {
|
||||
let peer_mgr_a = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
let peer_mgr_b = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
let peer_mgr_c = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
let peer_mgr_a = create_mock_pmgr().await;
|
||||
let peer_mgr_b = create_mock_pmgr().await;
|
||||
let peer_mgr_c = create_mock_pmgr().await;
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_b.clone()).await;
|
||||
connect_peer_manager(peer_mgr_b.clone(), peer_mgr_c.clone()).await;
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
||||
@@ -451,10 +451,10 @@ mod tests {
|
||||
let peer_mgr_c = create_mock_peer_manager().await;
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_b.clone()).await;
|
||||
connect_peer_manager(peer_mgr_b.clone(), peer_mgr_c.clone()).await;
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -503,7 +503,7 @@ mod tests {
|
||||
let peer_mgr_a = create_mock_peer_manager().await;
|
||||
let peer_mgr_b = create_mock_peer_manager().await;
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_b.clone()).await;
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.my_peer_id())
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
||||
@@ -7,11 +7,15 @@ use crate::{
|
||||
tunnels::ring_tunnel::create_ring_tunnel_pair,
|
||||
};
|
||||
|
||||
use super::peer_manager::PeerManager;
|
||||
use super::peer_manager::{PeerManager, RouteAlgoType};
|
||||
|
||||
pub async fn create_mock_peer_manager() -> Arc<PeerManager> {
|
||||
let (s, _r) = tokio::sync::mpsc::channel(1000);
|
||||
let peer_mgr = Arc::new(PeerManager::new(get_mock_global_ctx(), s));
|
||||
let peer_mgr = Arc::new(PeerManager::new(
|
||||
RouteAlgoType::Ospf,
|
||||
get_mock_global_ctx(),
|
||||
s,
|
||||
));
|
||||
peer_mgr.run().await.unwrap();
|
||||
peer_mgr
|
||||
}
|
||||
@@ -47,8 +51,12 @@ pub async fn wait_route_appear_with_cost(
|
||||
return Err(Error::NotFound);
|
||||
}
|
||||
|
||||
pub async fn wait_route_appear(peer_mgr: Arc<PeerManager>, node_id: PeerId) -> Result<(), Error> {
|
||||
wait_route_appear_with_cost(peer_mgr, node_id, None).await
|
||||
pub async fn wait_route_appear(
|
||||
peer_mgr: Arc<PeerManager>,
|
||||
target_peer: Arc<PeerManager>,
|
||||
) -> Result<(), Error> {
|
||||
wait_route_appear_with_cost(peer_mgr.clone(), target_peer.my_peer_id(), None).await?;
|
||||
wait_route_appear_with_cost(target_peer, peer_mgr.my_peer_id(), None).await
|
||||
}
|
||||
|
||||
pub async fn wait_for_condition<F, FRet>(mut condition: F, timeout: std::time::Duration) -> ()
|
||||
|
||||
Reference in New Issue
Block a user