shadow_rs/core/
sim_config.rs

Help
1//! Code for processing parsed Shadow configurations.
2//!
3//! This involves loading and verifying network graphs, converting options to types/formats that are
4//! easier to use in Shadow, verifying that paths exist, etc.
5
6use std::collections::hash_map::Entry;
7use std::collections::{BTreeMap, HashMap, HashSet};
8use std::ffi::{OsStr, OsString};
9use std::hash::{Hash, Hasher};
10use std::path::PathBuf;
11use std::sync::RwLock;
12use std::time::Duration;
13
14use anyhow::Context;
15use once_cell::sync::Lazy;
16use rand::{Rng, SeedableRng};
17use rand_xoshiro::Xoshiro256PlusPlus;
18use shadow_shim_helper_rs::simulation_time::SimulationTime;
19
20use crate::core::configuration::{
21    ConfigOptions, EnvName, Flatten, HostOptions, LogLevel, ProcessArgs, ProcessFinalState,
22    ProcessOptions, QDiscMode, parse_string_as_args,
23};
24use crate::network::graph::{IpAssignment, NetworkGraph, RoutingInfo, load_network_graph};
25use crate::utility::units::{self, Unit};
26use crate::utility::{tilde_expansion, verify_plugin_path};
27
28/// The simulation configuration after processing the configuration options and network graph.
29pub struct SimConfig {
30    // deterministic source of randomness for the simulation
31    pub random: Xoshiro256PlusPlus,
32
33    // map of ip addresses to graph nodes
34    pub ip_assignment: IpAssignment<u32>,
35
36    // routing information for paths between graph nodes
37    pub routing_info: RoutingInfo<u32>,
38
39    // bandwidths of hosts at ip addresses
40    pub host_bandwidths: HashMap<std::net::IpAddr, Bandwidth>,
41
42    // a list of hosts and their processes
43    pub hosts: Vec<HostInfo>,
44}
45
46impl SimConfig {
47    pub fn new(config: &ConfigOptions, hosts_to_debug: &HashSet<String>) -> anyhow::Result<Self> {
48        // Xoshiro256PlusPlus is not ideal when a seed with many zeros is used, but
49        // 'seed_from_u64()' uses SplitMix64 to derive the actual seed, so we are okay here
50        let seed = config.general.seed.unwrap();
51        let mut random = Xoshiro256PlusPlus::seed_from_u64(seed.into());
52
53        // this should be the same for all hosts
54        let randomness_for_seed_calc = random.random();
55
56        // build the host list
57        let mut hosts = vec![];
58        for (name, host_options) in &config.hosts {
59            let new_host = build_host(
60                config,
61                host_options,
62                name,
63                randomness_for_seed_calc,
64                hosts_to_debug,
65            )
66            .with_context(|| format!("Failed to configure host '{name}'"))?;
67            hosts.push(new_host);
68        }
69        if hosts.is_empty() {
70            return Err(anyhow::anyhow!(
71                "The configuration did not contain any hosts"
72            ));
73        }
74
75        // load and parse the network graph
76        let graph: String = load_network_graph(config.network.graph.as_ref().unwrap())
77            .map_err(|e| anyhow::anyhow!(e))
78            .context("Failed to load the network graph")?;
79        let graph = NetworkGraph::parse(&graph)
80            .map_err(|e| anyhow::anyhow!(e))
81            .context("Failed to parse the network graph")?;
82
83        // check that each node ID is valid
84        for host in &hosts {
85            if graph.node_id_to_index(host.network_node_id).is_none() {
86                return Err(anyhow::anyhow!(
87                    "The network node id {} for host '{}' does not exist",
88                    host.network_node_id,
89                    host.name
90                ));
91            }
92        }
93
94        // assign a bandwidth to every host
95        for host in &mut hosts {
96            let node_index = graph.node_id_to_index(host.network_node_id).unwrap();
97            let node = graph.graph().node_weight(*node_index).unwrap();
98
99            let graph_bw_down_bits = node
100                .bandwidth_down
101                .map(|x| x.convert(units::SiPrefixUpper::Base).unwrap().value());
102            let graph_bw_up_bits = node
103                .bandwidth_up
104                .map(|x| x.convert(units::SiPrefixUpper::Base).unwrap().value());
105
106            host.bandwidth_down_bits = host.bandwidth_down_bits.or(graph_bw_down_bits);
107            host.bandwidth_up_bits = host.bandwidth_up_bits.or(graph_bw_up_bits);
108
109            // check if no bandwidth was provided in the host options or graph node
110            if host.bandwidth_down_bits.is_none() {
111                return Err(anyhow::anyhow!(
112                    "No downstream bandwidth provided for host '{}'",
113                    host.name
114                ));
115            }
116            if host.bandwidth_up_bits.is_none() {
117                return Err(anyhow::anyhow!(
118                    "No upstream bandwidth provided for host '{}'",
119                    host.name
120                ));
121            }
122        }
123
124        // check if any hosts in 'hosts_to_debug' don't exist
125        for hostname in hosts_to_debug {
126            if !hosts.iter().any(|y| &y.name == hostname) {
127                return Err(anyhow::anyhow!(
128                    "The host to debug '{hostname}' doesn't exist"
129                ));
130            }
131        }
132
133        // assign IP addresses to hosts and graph nodes
134        let ip_assignment = assign_ips(&mut hosts)?;
135
136        // generate routing info between every pair of in-use nodes
137        let routing_info = generate_routing_info(
138            &graph,
139            &ip_assignment.get_nodes(),
140            config.network.use_shortest_path.unwrap(),
141        )?;
142
143        // get all host bandwidths
144        let host_bandwidths = hosts
145            .iter()
146            .map(|host| {
147                // we made sure above that every host has a bandwidth set
148                let bw = Bandwidth {
149                    up_bytes: host.bandwidth_up_bits.unwrap() / 8,
150                    down_bytes: host.bandwidth_down_bits.unwrap() / 8,
151                };
152
153                (host.ip_addr.unwrap(), bw)
154            })
155            .collect();
156
157        Ok(Self {
158            random,
159            ip_assignment,
160            routing_info,
161            host_bandwidths,
162            hosts,
163        })
164    }
165}
166
167#[derive(Clone)]
168pub struct HostInfo {
169    pub name: String,
170    pub processes: Vec<ProcessInfo>,
171    pub seed: u64,
172    pub network_node_id: u32,
173    pub pause_for_debugging: bool,
174    pub cpu_threshold: Option<SimulationTime>,
175    pub cpu_precision: Option<SimulationTime>,
176    pub bandwidth_down_bits: Option<u64>,
177    pub bandwidth_up_bits: Option<u64>,
178    pub ip_addr: Option<std::net::IpAddr>,
179    pub log_level: Option<LogLevel>,
180    pub pcap_config: Option<PcapConfig>,
181    pub send_buf_size: u64,
182    pub recv_buf_size: u64,
183    pub autotune_send_buf: bool,
184    pub autotune_recv_buf: bool,
185    pub qdisc: QDiscMode,
186}
187
188#[derive(Clone)]
189pub struct ProcessInfo {
190    pub plugin: PathBuf,
191    pub start_time: SimulationTime,
192    pub shutdown_time: Option<SimulationTime>,
193    pub shutdown_signal: nix::sys::signal::Signal,
194    pub args: Vec<OsString>,
195    pub env: BTreeMap<EnvName, String>,
196    pub expected_final_state: ProcessFinalState,
197}
198
199#[derive(Debug, Clone)]
200pub struct Bandwidth {
201    pub up_bytes: u64,
202    pub down_bytes: u64,
203}
204
205#[derive(Debug, Clone, Copy)]
206pub struct PcapConfig {
207    pub capture_size: u64,
208}
209
210/// For a host entry in the configuration options, build `HostInfo` object.
211fn build_host(
212    config: &ConfigOptions,
213    host: &HostOptions,
214    hostname: &str,
215    randomness_for_seed_calc: u64,
216    hosts_to_debug: &HashSet<String>,
217) -> anyhow::Result<HostInfo> {
218    let hostname = hostname.to_string();
219
220    // hostname hash is used as part of the host's seed
221    let hostname_hash = {
222        let mut hasher = std::hash::DefaultHasher::new();
223        hostname.hash(&mut hasher);
224        hasher.finish()
225    };
226
227    let pause_for_debugging = hosts_to_debug.contains(&hostname);
228
229    let processes: Vec<_> = host
230        .processes
231        .iter()
232        .map(|proc| {
233            build_process(proc, config)
234                .with_context(|| format!("Failed to configure process '{}'", proc.path.display()))
235        })
236        .collect::<anyhow::Result<_>>()?;
237
238    Ok(HostInfo {
239        name: hostname,
240        processes,
241
242        seed: randomness_for_seed_calc ^ hostname_hash,
243        network_node_id: host.network_node_id,
244        pause_for_debugging,
245
246        cpu_threshold: None,
247        cpu_precision: Some(SimulationTime::from_nanos(200)),
248
249        bandwidth_down_bits: host
250            .bandwidth_down
251            .map(|x| x.convert(units::SiPrefixUpper::Base).unwrap().value()),
252        bandwidth_up_bits: host
253            .bandwidth_down
254            .map(|x| x.convert(units::SiPrefixUpper::Base).unwrap().value()),
255
256        ip_addr: host.ip_addr.map(|x| x.into()),
257        log_level: host.host_options.log_level.flatten(),
258        pcap_config: host
259            .host_options
260            .pcap_enabled
261            .unwrap()
262            .then_some(PcapConfig {
263                capture_size: host
264                    .host_options
265                    .pcap_capture_size
266                    .unwrap()
267                    .convert(units::SiPrefixUpper::Base)
268                    .unwrap()
269                    .value(),
270            }),
271
272        // some options come from the config options and not the host options
273        send_buf_size: config
274            .experimental
275            .socket_send_buffer
276            .unwrap()
277            .convert(units::SiPrefixUpper::Base)
278            .unwrap()
279            .value(),
280        recv_buf_size: config
281            .experimental
282            .socket_recv_buffer
283            .unwrap()
284            .convert(units::SiPrefixUpper::Base)
285            .unwrap()
286            .value(),
287        autotune_send_buf: config.experimental.socket_send_autotune.unwrap(),
288        autotune_recv_buf: config.experimental.socket_recv_autotune.unwrap(),
289        qdisc: config.experimental.interface_qdisc.unwrap(),
290    })
291}
292
293/// For a process entry in the configuration options, build a `ProcessInfo` object.
294fn build_process(proc: &ProcessOptions, config: &ConfigOptions) -> anyhow::Result<ProcessInfo> {
295    let start_time = Duration::from(proc.start_time).try_into().unwrap();
296    let shutdown_time = proc
297        .shutdown_time
298        .map(|x| Duration::from(x).try_into().unwrap());
299    let shutdown_signal = *proc.shutdown_signal;
300    let sim_stop_time =
301        SimulationTime::try_from(Duration::from(config.general.stop_time.unwrap())).unwrap();
302
303    if start_time >= sim_stop_time {
304        return Err(anyhow::anyhow!(
305            "Process start time '{}' must be earlier than the simulation stop time '{}'",
306            proc.start_time,
307            config.general.stop_time.unwrap(),
308        ));
309    }
310
311    if let Some(shutdown_time) = shutdown_time {
312        if start_time >= shutdown_time {
313            return Err(anyhow::anyhow!(
314                "Process start time '{}' must be earlier than its shutdown_time time '{}'",
315                proc.start_time,
316                proc.shutdown_time.unwrap(),
317            ));
318        }
319        if shutdown_time >= sim_stop_time {
320            return Err(anyhow::anyhow!(
321                "Process shutdown_time '{}' must be earlier than the simulation stop time '{}'",
322                proc.shutdown_time.unwrap(),
323                config.general.stop_time.unwrap(),
324            ));
325        }
326    }
327
328    let mut args = match &proc.args {
329        ProcessArgs::List(x) => x.iter().map(|y| OsStr::new(y).to_os_string()).collect(),
330        ProcessArgs::Str(x) => parse_string_as_args(OsStr::new(&x.trim()))
331            .map_err(|e| anyhow::anyhow!(e))
332            .with_context(|| format!("Failed to parse arguments: {x}"))?,
333    };
334
335    let expanded_path = tilde_expansion(proc.path.to_str().unwrap());
336
337    // a cache so we don't resolve the same path multiple times
338    static RESOLVED_PATHS: Lazy<RwLock<HashMap<PathBuf, PathBuf>>> =
339        Lazy::new(|| RwLock::new(HashMap::new()));
340
341    let canonical_path = RESOLVED_PATHS.read().unwrap().get(&proc.path).cloned();
342    let canonical_path = match canonical_path {
343        Some(x) => x,
344        None => {
345            match RESOLVED_PATHS.write().unwrap().entry(proc.path.clone()) {
346                Entry::Occupied(entry) => entry.get().clone(),
347                Entry::Vacant(entry) => {
348                    // We currently use `which::which`, which searches the `PATH` similarly to a
349                    // shell.
350                    let canonical_path = which::which(&expanded_path)
351                        .map_err(anyhow::Error::from)
352                        // `which` returns an absolute path, but it may still contain
353                        // symbolic links, .., etc.
354                        .and_then(|p| Ok(p.canonicalize()?))
355                        .with_context(|| {
356                            format!("Failed to resolve plugin path '{expanded_path:?}'")
357                        })?;
358
359                    verify_plugin_path(&canonical_path).with_context(|| {
360                        format!("Failed to verify plugin path '{canonical_path:?}'")
361                    })?;
362                    log::info!("Resolved binary path {:?} to {canonical_path:?}", proc.path);
363
364                    entry.insert(canonical_path).clone()
365                }
366            }
367        }
368    };
369
370    // set argv[0] as the user-provided expanded string, not the canonicalized version
371    args.insert(0, expanded_path.into());
372
373    Ok(ProcessInfo {
374        plugin: canonical_path,
375        start_time,
376        shutdown_time,
377        shutdown_signal,
378        args,
379        env: proc.environment.clone(),
380        expected_final_state: proc.expected_final_state,
381    })
382}
383
384/// Generate an IP assignment map using hosts' configured IP addresses and graph node IDs. For hosts
385/// without IP addresses, they will be assigned an arbitrary IP address.
386fn assign_ips(hosts: &mut [HostInfo]) -> anyhow::Result<IpAssignment<u32>> {
387    let mut ip_assignment = IpAssignment::new();
388
389    // first register hosts that have a specific IP address
390    for host in hosts.iter().filter(|x| x.ip_addr.is_some()) {
391        let ip = host.ip_addr.unwrap();
392        let hostname = &host.name;
393        let node_id = host.network_node_id;
394        ip_assignment.assign_ip(node_id, ip).with_context(|| {
395            format!("Failed to assign IP address {ip} for host '{hostname}' to node '{node_id}'")
396        })?;
397    }
398
399    // then register remaining hosts
400    for host in hosts.iter_mut().filter(|x| x.ip_addr.is_none()) {
401        let ip = ip_assignment.assign(host.network_node_id);
402        // assign the new IP to the host
403        host.ip_addr = Some(ip);
404    }
405
406    Ok(ip_assignment)
407}
408
409/// Generate a map containing routing information (latency, packet loss, etc) for each pair of
410/// nodes.
411fn generate_routing_info(
412    graph: &NetworkGraph,
413    nodes: &std::collections::HashSet<u32>,
414    use_shortest_paths: bool,
415) -> anyhow::Result<RoutingInfo<u32>> {
416    // convert gml node IDs to petgraph indexes
417    let nodes: Vec<_> = nodes
418        .iter()
419        .map(|x| *graph.node_id_to_index(*x).unwrap())
420        .collect();
421
422    // helper to convert petgraph indexes back to gml node IDs
423    let to_ids = |((src, dst), path)| {
424        let src = graph.node_index_to_id(src).unwrap();
425        let dst = graph.node_index_to_id(dst).unwrap();
426        ((src, dst), path)
427    };
428
429    let paths = if use_shortest_paths {
430        graph
431            .compute_shortest_paths(&nodes[..])
432            .map_err(|e| anyhow::anyhow!(e))
433            .context("Failed to compute shortest paths between graph nodes")?
434            .into_iter()
435            .map(to_ids)
436            .collect()
437    } else {
438        graph
439            .get_direct_paths(&nodes[..])
440            .map_err(|e| anyhow::anyhow!(e))
441            .context("Failed to get the direct paths between graph nodes")?
442            .into_iter()
443            .map(to_ids)
444            .collect()
445    };
446
447    Ok(RoutingInfo::new(paths))
448}
shadow_rs/core/sim_config.rs

shadow_rs/core/
sim_config.rs
