void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-06-08 21:43:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Simplification of some code in controller.

Browse source
This commit is contained in:
Adam Ierymenko 2022-10-23 20:47:05 -04:00
parent bc5861c539
commit 88e613e043
No known key found for this signature in database
GPG key ID: C8877CF2D7A5D7F3
1 changed files with 26 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
controller/src/handler.rs
View file

@ -26,11 +26,8 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// ZeroTier VL2 network controller packet handler, answers VL2 netconf queries. /// ZeroTier VL2 network controller packet handler, answers VL2 netconf queries.
pub struct Handler { pub struct Handler {
inner: Arc<Inner>, self_ref: Weak<Self>,
} service: RwLock<Weak<VL1Service<dyn Database, Self, Self>>>,
struct Inner {
service: RwLock<Weak<VL1Service<dyn Database, Handler, Handler>>>,
reaper: Reaper, reaper: Reaper,
daemons: Mutex<Vec<tokio::task::JoinHandle<()>>>, // drop() aborts these daemons: Mutex<Vec<tokio::task::JoinHandle<()>>>, // drop() aborts these
runtime: tokio::runtime::Handle, runtime: tokio::runtime::Handle,
@ -43,19 +40,15 @@ impl Handler {
pub async fn new(database: Arc<dyn Database>, runtime: tokio::runtime::Handle) -> Result<Arc<Self>, Box<dyn Error>> { pub async fn new(database: Arc<dyn Database>, runtime: tokio::runtime::Handle) -> Result<Arc<Self>, Box<dyn Error>> {
if let Some(local_identity) = database.load_node_identity() { if let Some(local_identity) = database.load_node_identity() {
assert!(local_identity.secret.is_some()); assert!(local_identity.secret.is_some());
Ok(Arc::new_cyclic(|r| Self {
let inner = Arc::new(Inner { self_ref: r.clone(),
service: RwLock::new(Weak::default()), service: RwLock::new(Weak::default()),
reaper: Reaper::new(&runtime), reaper: Reaper::new(&runtime),
daemons: Mutex::new(Vec::with_capacity(1)), daemons: Mutex::new(Vec::with_capacity(1)),
runtime, runtime,
database: database.clone(), database: database.clone(),
local_identity, local_identity,
}); }))
let h = Arc::new(Self { inner: inner.clone() });
Ok(h)
} else { } else {
Err(Box::new(InvalidParameterError( Err(Box::new(InvalidParameterError(
"local controller's identity not readable by database", "local controller's identity not readable by database",
@ -69,8 +62,8 @@ impl Handler {
/// won't actually do anything. The reference the handler holds is weak to prevent /// won't actually do anything. The reference the handler holds is weak to prevent
/// a circular reference, so if the VL1Service is dropped this must be called again to /// a circular reference, so if the VL1Service is dropped this must be called again to
/// tell the controller handler about a new instance. /// tell the controller handler about a new instance.
pub fn set_service(&self, service: &Arc<VL1Service<dyn Database, Handler, Handler>>) { pub fn set_service(&self, service: &Arc<VL1Service<dyn Database, Self, Self>>) {
*self.inner.service.write().unwrap() = Arc::downgrade(service); *self.service.write().unwrap() = Arc::downgrade(service);
} }
/// Start a change watcher to respond to changes detected by the database. /// Start a change watcher to respond to changes detected by the database.
@ -79,20 +72,20 @@ impl Handler {
/// unnecessary async tasks. If the database being used does not support changes, this /// unnecessary async tasks. If the database being used does not support changes, this
/// does nothing. /// does nothing.
pub async fn start_change_watcher(&self) { pub async fn start_change_watcher(&self) {
if let Some(cw) = self.inner.database.changes().await.map(|mut ch| { if let Some(cw) = self.database.changes().await.map(|mut ch| {
let inner = self.inner.clone(); let self2 = self.self_ref.upgrade().unwrap();
self.inner.runtime.spawn(async move { self.runtime.spawn(async move {
loop { loop {
if let Ok(change) = ch.recv().await { if let Ok(change) = ch.recv().await {
inner.reaper.add( self2.reaper.add(
inner.runtime.spawn(inner.clone().handle_change_notification(change)), self2.runtime.spawn(self2.clone().handle_change_notification(change)),
Instant::now().checked_add(REQUEST_TIMEOUT).unwrap(), Instant::now().checked_add(REQUEST_TIMEOUT).unwrap(),
); );
} }
} }
}) })
}) { }) {
self.inner.daemons.lock().unwrap().push(cw); self.daemons.lock().unwrap().push(cw);
} }
} }
} }
@ -103,8 +96,8 @@ impl PathFilter for Handler {}
impl InnerProtocol for Handler { impl InnerProtocol for Handler {
fn handle_packet<HostSystemImpl: HostSystem + ?Sized>( fn handle_packet<HostSystemImpl: HostSystem + ?Sized>(
&self, &self,
_host_system: &HostSystemImpl, _: &HostSystemImpl,
_node: &Node, _: &Node,
source: &Arc<Peer>, source: &Arc<Peer>,
source_path: &Arc<Path>, source_path: &Arc<Path>,
source_hops: u8, source_hops: u8,
@ -157,19 +150,17 @@ impl InnerProtocol for Handler {
*/ */
// Launch handler as an async background task. // Launch handler as an async background task.
let (inner, peer, source_remote_endpoint) = (self.inner.clone(), source.clone(), source_path.endpoint.clone()); let (self2, peer, source_remote_endpoint) =
self.inner.reaper.add( (self.self_ref.upgrade().unwrap(), source.clone(), source_path.endpoint.clone());
self.inner.runtime.spawn(async move { self.reaper.add(
self.runtime.spawn(async move {
let node_id = peer.identity.address; let node_id = peer.identity.address;
let node_fingerprint = Blob::from(peer.identity.fingerprint); let node_fingerprint = Blob::from(peer.identity.fingerprint);
let now = ms_since_epoch(); let now = ms_since_epoch();
let result = match inner let result = match self2.handle_network_config_request(&peer.identity, network_id, now).await {
.handle_network_config_request::<HostSystemImpl>(&peer.identity, network_id, now)
.await
{
Result::Ok((result, Some(config))) => { Result::Ok((result, Some(config))) => {
inner.send_network_config(peer.as_ref(), &config, Some(message_id)); self2.send_network_config(peer.as_ref(), &config, Some(message_id));
result result
} }
Result::Ok((result, None)) => result, Result::Ok((result, None)) => result,
@ -179,13 +170,13 @@ impl InnerProtocol for Handler {
} }
}; };
let _ = inner let _ = self2
.database .database
.log_request(RequestLogItem { .log_request(RequestLogItem {
network_id, network_id,
node_id, node_id,
node_fingerprint, node_fingerprint,
controller_node_id: inner.local_identity.address, controller_node_id: self2.local_identity.address,
metadata: if meta_data.is_empty() { metadata: if meta_data.is_empty() {
Vec::new() Vec::new()
} else { } else {
@ -245,7 +236,7 @@ impl InnerProtocol for Handler {
} }
} }
impl Inner { impl Handler {
fn send_network_config( fn send_network_config(
&self, &self,
peer: &Peer, peer: &Peer,
@ -301,7 +292,7 @@ impl Inner {
async fn handle_change_notification(self: Arc<Self>, _change: Change) {} async fn handle_change_notification(self: Arc<Self>, _change: Change) {}
async fn handle_network_config_request<HostSystemImpl: HostSystem + ?Sized>( async fn handle_network_config_request(
self: &Arc<Self>, self: &Arc<Self>,
source_identity: &Identity, source_identity: &Identity,
network_id: NetworkId, network_id: NetworkId,
@ -424,7 +415,7 @@ impl Inner {
} }
} }
impl Drop for Inner { impl Drop for Handler {
fn drop(&mut self) { fn drop(&mut self) {
for h in self.daemons.lock().unwrap().drain(..) { for h in self.daemons.lock().unwrap().drain(..) {
h.abort(); h.abort();