manycastr/worker/
config.rs

1use crate::custom_module::manycastr::controller_client::ControllerClient;
2use crate::custom_module::manycastr::instruction::InstructionType;
3use crate::custom_module::manycastr::{Address, Origin};
4use local_ip_address::{local_ip, local_ipv6};
5use log::warn;
6use std::sync::Arc;
7use std::sync::atomic::AtomicBool;
8use tonic::transport::Channel;
9
10/// The worker that is run at the anycast PoPs and performs measurements as instructed by the orchestrator.
11/// The worker is responsible for establishing a connection with the orchestrator, receiving tasks, and performing measurements.
12pub struct Worker {
13    /// gRPC client to communicate with the orchestrator
14    pub(crate) grpc_client: ControllerClient<Channel>,
15    /// Hostname of the worker
16    pub(crate) hostname: String,
17    /// Whether a measurement is currently active on this worker
18    pub(crate) is_busy: Arc<AtomicBool>,
19    /// Instructions senders to the outbound probing threads, paired with their origin ID
20    pub(crate) outbound_txs: Vec<(u32, tokio::sync::mpsc::Sender<InstructionType>)>,
21    /// Join handles of the outbound probing threads, awaited on graceful end before closing inbound
22    pub(crate) outbound_handles: Vec<std::thread::JoinHandle<()>>,
23    /// Atomic boolean to signal the inbound thread to immediately stop listening for packets
24    pub(crate) abort_inbound: Arc<AtomicBool>,
25}
26
27/// Takes a list of origins, replaces any unicast placeholder addresses with the local
28/// address of the placeholder's IP version, and returns the modified list of origins.
29///
30/// Drops unicast origins when no local unicast address of that version can be found.
31///
32/// # Arguments
33/// * `origins` - A vector of Origin structs to be modified.
34///
35/// # Returns
36/// * A vector of Origin structs with unicast placeholders replaced by local addresses.
37pub fn set_unicast_origins(origins: Vec<Origin>) -> Vec<Origin> {
38    // Resolve the local addresses once (a version is looked up only when an origin needs it)
39    let mut local_v4: Option<Option<Address>> = None;
40    let mut local_v6: Option<Option<Address>> = None;
41
42    origins
43        .into_iter()
44        .filter_map(|mut o| {
45            if o.is_unicast() {
46                let is_ipv6 = o.src.expect("no src").is_v6();
47                let src_addr = if is_ipv6 {
48                    *local_v6.get_or_insert_with(|| local_ipv6().ok().map(Address::from))
49                } else {
50                    *local_v4.get_or_insert_with(|| local_ip().ok().map(Address::from))
51                };
52                match src_addr {
53                    Some(addr) => o.src = Some(addr),
54                    None => {
55                        warn!(
56                            "[Worker] No local {} address available; skipping unicast origin {}",
57                            if is_ipv6 { "IPv6" } else { "IPv4" },
58                            o.origin_id
59                        );
60                        return None;
61                    }
62                }
63            }
64            Some(o)
65        })
66        .collect()
67}
manycastr/worker/config.rs

manycastr/worker/
config.rs
