course/hse-2026
1
Fork 1
Code Issues Pull Requests Actions Packages Activity

Add HW 4

Browse Source
This commit is contained in:
osukhoroslov
2026-10-03 11:42:32 +03:00
parent 6008cff9f6
commit 209c66683d
15 changed files with 5061 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homework/04-broadcast/tests/src/common.rs
+36
View File
@@ -0,0 +1,36 @@
use serde::Serialize;
use sugars::boxed;
use anysystem::python::PyProcessFactory;
use anysystem::System;
#[derive(Copy, Clone)]
pub struct TestConfig<'a> {
pub proc_factory: &'a PyProcessFactory,
pub proc_count: u64,
pub seed: u64,
pub monkeys: u32,
pub debug: bool,
}
#[derive(Serialize)]
pub struct BroadcastMessage<'a> {
pub text: &'a str,
}
pub fn build_system(config: &TestConfig) -> System {
let mut sys = System::new(config.seed);
let mut proc_names = Vec::new();
for n in 0..config.proc_count {
proc_names.push(format!("{n}"));
}
for proc_name in &proc_names {
let proc = config
.proc_factory
.build((proc_name, proc_names.clone()), config.seed);
// process and node on which it runs have the same name
sys.add_node(proc_name);
sys.add_process(proc_name, boxed!(proc), proc_name);
}
sys
}
homework/04-broadcast/tests/src/main.rs
+219
View File
@@ -0,0 +1,219 @@
mod common;
mod optimization;
mod runner;
mod shared_state;
mod tests;
mod tests_mc;
use anysystem::test::TestResult;
use clap::Parser;
use indexmap::IndexMap;
use std::collections::HashSet;
use std::path::PathBuf;
/// Broadcast Homework Tests
#[derive(Parser, Debug)]
#[clap(about, long_about = None)]
struct Args {
/// Path to Python file with solution
#[clap(long = "impl", short = 'i', default_value = "solution/broadcast.py")]
solution_path: String,
/// Test to run (optional)
#[clap(long = "test", short)]
test: Option<String>,
/// Print execution trace
#[clap(long, short)]
debug: bool,
/// Random seed used in tests
#[clap(long, short, default_value = "2023")]
seed: u64,
/// Number of processes
#[clap(long, short, default_value = "5")]
proc_count: u64,
/// Number of chaos monkey runs
#[clap(long, short, default_value = "10")]
monkeys: u32,
/// Do not run model checking tests
#[clap(long)]
disable_mc_tests: bool,
/// Optional scalable solution (defaults to broadcast_scalable.py beside --impl)
#[clap(long)]
optimized_impl: Option<String>,
/// Repetitions per OPTIMIZATION scenario and system size
#[clap(long, default_value = "10", value_parser = clap::value_parser!(u32).range(1..=100))]
optimization_runs: u32,
/// Shared wall-clock budget in seconds for both implementations in OPTIMIZATION
#[clap(long, default_value = "90", value_parser = clap::value_parser!(u64).range(1..=90))]
optimization_timeout: u64,
#[clap(long, hide = true)]
optimization_worker: bool,
/// Wall-clock limit in seconds per ordinary test
#[clap(long, default_value = "15", value_parser = clap::value_parser!(u64).range(1..=300))]
test_timeout: u64,
/// Wall-clock limit in seconds for SCALABILITY (up to 50 processes)
#[clap(long, default_value = "60", value_parser = clap::value_parser!(u64).range(1..=300))]
scalability_timeout: u64,
/// Wall-clock limit in seconds for all CHAOS MONKEY runs
#[clap(long, default_value = "30", value_parser = clap::value_parser!(u64).range(1..=300))]
chaos_timeout: u64,
/// External MC test limit (NORMAL DELIVERY is capped at 110s, CAUSAL ORDER at 60s)
#[clap(long, default_value = "210", value_parser = clap::value_parser!(u64).range(1..=210))]
mc_timeout: u64,
/// Total wall-clock budget in seconds for correctness tests
#[clap(long, default_value = "300", value_parser = clap::value_parser!(u64).range(1..=300))]
suite_timeout: u64,
/// Continue after failures for local diagnostics; resource limits and crashes still stop
#[clap(long)]
keep_going: bool,
#[clap(long, hide = true, requires = "test-result")]
test_worker: Option<String>,
#[clap(long, hide = true)]
test_result: Option<PathBuf>,
}
fn main() {
let args = Args::parse();
if args.test_worker.is_some() {
runner::worker(&args);
return;
}
if args.optimization_worker {
optimization::worker(&args);
return;
}
if args.test.as_deref() == Some("OPTIMIZATION") {
optimization::run(&args);
return;
}
if !shared_state::validate_or_report(&args.solution_path, &["BroadcastProcess"]) {
return;
}
let results = runner::run(&args);
if args.test.is_none() {
let score = score(results);
optimization::run_if_present(&args, score);
println!("SCORE: {score}\n");
}
}
fn score(results: IndexMap<String, TestResult>) -> f32 {
let mut violated = HashSet::new();
let mut causal_incomplete = false;
for (name, result) in results {
if let Err(e) = result {
if e.starts_with("Violated") {
for prop in e.replace("Violated ", "").split(", ") {
violated.insert(prop.to_string());
}
} else if runner::causal_mc_incomplete(&name, &e) {
// Keep partial credit separate from a discovered violation.
causal_incomplete = true;
} else {
return 0.;
}
}
}
let violated: HashSet<&str> = violated.iter().map(|s| s.as_str()).collect();
let base_props = HashSet::from([
"NO DUPLICATION",
"NO CREATION",
"VALIDITY",
"UNIFORM AGREEMENT",
]);
if violated.is_disjoint(&base_props) {
if !violated.is_empty() {
5.
} else if causal_incomplete {
6.
} else {
7.
}
} else {
0.
}
}
#[cfg(test)]
mod score_tests {
use super::score;
use indexmap::IndexMap;
#[test]
fn incomplete_causal_mc_gets_partial_credit() {
for error in [
"time limit of 55s exceeded",
"test time limit of 60s exceeded",
"suite time limit of 300s exceeded",
"causal model-checking state limit of 100000 exceeded",
] {
let mut results = IndexMap::new();
results.insert("MODEL CHECKING CAUSAL ORDER".to_string(), Err(error.into()));
assert_eq!(score(results), 6., "{error}");
}
}
#[test]
fn discovered_violation_takes_precedence_over_incomplete_search() {
for (error, expected) in [
("Violated CAUSAL ORDER", 5.),
("Violated VALIDITY", 0.),
("Violated VALIDITY, CAUSAL ORDER", 0.),
] {
for incomplete_first in [false, true] {
let mut entries = vec![
("CAUSAL ORDER", error),
("MODEL CHECKING CAUSAL ORDER", "time limit of 55s exceeded"),
];
if incomplete_first {
entries.reverse();
}
let results = entries
.into_iter()
.map(|(name, error)| (name.to_string(), Err(error.to_string())))
.collect();
assert_eq!(score(results), expected, "{error}");
}
}
}
#[test]
fn other_failures_do_not_get_partial_credit() {
for (name, error) in [
("NORMAL", "test time limit of 15s exceeded"),
(
"MODEL CHECKING NORMAL DELIVERY",
"time limit of 100s exceeded",
),
(
"MODEL CHECKING CAUSAL ORDER",
"test worker exited with exit code: 1",
),
(
"MODEL CHECKING CAUSAL ORDER",
"nothing left to do to reach the goal",
),
] {
let mut results = IndexMap::new();
results.insert(name.to_string(), Err(error.to_string()));
assert_eq!(score(results), 0., "{name}: {error}");
}
}
}
homework/04-broadcast/tests/src/optimization.rs
+516
View File
@@ -0,0 +1,516 @@
//! Bounded measurements, deliberately separate from correctness tests and SCORE.
use std::cell::Cell;
use std::collections::{BTreeMap, BTreeSet};
use std::io::{Read, Write};
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::rc::Rc;
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use std::thread;
use std::time::{Duration, Instant};
use anysystem::logger::LogEntry;
use anysystem::python::PyProcessFactory;
use anysystem::{Context, Message, Process, System};
use serde::{Deserialize, Serialize};
use crate::common::{BroadcastMessage, TestConfig};
use crate::{shared_state, Args};
const SIZES: [u64; 4] = [5, 10, 20, 50];
const SCENARIOS: [&str; 3] = ["NORMAL", "CRASHES", "DELAYED"];
const MESSAGES: usize = 10;
const HORIZON: f64 = 109.; // 100 model-time units after the last SEND at t=9.
const EVENT_LIMIT: u64 = 100_000;
const OUTPUT_LIMIT: u64 = 2 * 1024 * 1024;
const RESULT_PREFIX: &str = "OPTIMIZATION RESULT ";
// Coverage counts first valid DELIVERs to correct processes, including the
// sender. Missing deliveries at the horizon do not prove a liveness violation.
// Latencies use model time and only those deliveries. Network counters include
// service traffic; duplicate/unexpected/malformed outputs count all processes.
#[derive(Debug, Deserialize, Serialize)]
struct Measurement {
scenario: String,
n: u64,
seed: u64,
status: String,
observed_until: f64,
sent: usize,
messages: u64,
bytes: u64,
max_sent: u64,
delivered: usize,
expected: usize,
latency_mean: Option<f64>,
latency_max: Option<f64>,
duplicates: usize,
unexpected: usize,
malformed: usize,
}
// A native test driver schedules external actions independently of the student's
// traffic. It has no network traffic and is excluded from all measurements.
#[derive(Clone)]
struct Workload {
action: Rc<Cell<usize>>,
crashes: bool,
}
impl Process for Workload {
fn on_local_message(&mut self, _: Message, ctx: &mut Context) -> Result<(), String> {
for i in 1..MESSAGES {
ctx.set_timer(&i.to_string(), i as f64);
}
if self.crashes {
ctx.set_timer("10", 4.5);
}
ctx.set_timer("11", HORIZON);
Ok(())
}
fn on_message(&mut self, _: Message, _: String, _: &mut Context) -> Result<(), String> {
Ok(())
}
fn on_timer(&mut self, timer: String, _: &mut Context) -> Result<(), String> {
self.action.set(timer.parse().unwrap());
Ok(())
}
}
fn send(sys: &mut System, index: usize) {
let text = format!("{index:02}:{}", "x".repeat(61)); // fixed 64-byte payload
sys.send_local_message(
"0",
Message::json("SEND", &BroadcastMessage { text: &text }),
);
}
fn measure(config: &TestConfig, scenario: &str) -> Measurement {
let mut sys = System::new(config.seed);
if scenario == "DELAYED" {
sys.network().set_delays(1., 10.);
} else {
sys.network().set_delays(1., 3.);
}
let processes: Vec<String> = (0..config.proc_count).map(|i| i.to_string()).collect();
for proc in &processes {
let implementation = config
.proc_factory
.build((proc, processes.clone()), config.seed);
sys.add_node(proc);
sys.add_process(proc, Box::new(implementation), proc);
}
let action = Rc::new(Cell::new(0));
sys.add_node("_workload");
sys.add_process(
"_workload",
Box::new(Workload {
action: action.clone(),
crashes: scenario == "CRASHES",
}),
"_workload",
);
sys.send_local_message("_workload", Message::new("START", "{}"));
send(&mut sys, 0);
let mut status = "event limit";
for _ in 0..EVENT_LIMIT {
if !sys.step() {
status = "queue exhausted";
break;
}
match action.replace(0) {
0 => {}
index @ 1..=9 => send(&mut sys, index),
10 => {
sys.crash_node("1");
sys.crash_node("2");
}
11 => {
status = "horizon";
break;
}
_ => unreachable!(),
}
}
summarize(&sys, config, scenario, status)
}
fn summarize(sys: &System, config: &TestConfig, scenario: &str, status: &str) -> Measurement {
let mut latencies = Vec::new();
let mut duplicates = 0;
let mut unexpected = 0;
let mut malformed = 0;
let mut network_sends = BTreeMap::<String, u64>::new();
let mut sent = BTreeMap::new();
let mut seen = BTreeSet::new();
// Trace order matters: a guessed DELIVER before SEND is unexpected even
// when the two events have the same model timestamp.
for entry in sys.logger().trace() {
match entry {
LogEntry::MessageSent {
src_proc, dst_proc, ..
} if src_proc != dst_proc => {
*network_sends.entry(src_proc.clone()).or_default() += 1;
}
LogEntry::LocalMessageReceived {
time, proc, msg, ..
} if proc == "0" && msg.tip == "SEND" => {
let data: serde_json::Value = serde_json::from_str(&msg.data).unwrap();
sent.insert(data["text"].as_str().unwrap().to_string(), *time);
}
LogEntry::LocalMessageSent {
time, proc, msg, ..
} => {
let data = serde_json::from_str::<serde_json::Value>(&msg.data).ok();
let text = data.as_ref().and_then(|v| v["text"].as_str());
if msg.tip != "DELIVER" || text.is_none() {
malformed += 1;
continue;
}
let text = text.unwrap();
if !seen.insert((proc.clone(), text.to_string())) {
duplicates += 1;
continue;
}
if let Some(sent_at) = sent.get(text) {
if scenario != "CRASHES" || (proc != "1" && proc != "2") {
latencies.push(time - sent_at);
}
} else {
unexpected += 1;
}
}
_ => {}
}
}
let correct_count = config.proc_count - if scenario == "CRASHES" { 2 } else { 0 };
let network = sys.network();
Measurement {
scenario: scenario.to_string(),
n: config.proc_count,
seed: config.seed,
status: status.to_string(),
observed_until: sys.time(),
sent: sent.len(),
messages: network.network_message_count(),
bytes: network.traffic(),
max_sent: network_sends.into_values().max().unwrap_or(0),
delivered: latencies.len(),
expected: MESSAGES * correct_count as usize,
latency_mean: (!latencies.is_empty())
.then(|| latencies.iter().sum::<f64>() / latencies.len() as f64),
latency_max: latencies.into_iter().reduce(f64::max),
duplicates,
unexpected,
malformed,
}
}
pub fn worker(args: &Args) {
// This process may panic, hang in Python, or print SCORE from the validator.
// The supervisor captures its output and never forwards it verbatim.
if !shared_state::validate_or_report(&args.solution_path, &["BroadcastProcess"]) {
return;
}
let factory = PyProcessFactory::new(&args.solution_path, "BroadcastProcess");
let mut rows = Vec::new();
'measurements: for scenario in SCENARIOS {
for n in SIZES {
for run in 0..args.optimization_runs {
let config = TestConfig {
proc_factory: &factory,
proc_count: n,
seed: args.seed.wrapping_add(run as u64),
monkeys: 0,
debug: false,
};
let result = measure(&config, scenario);
let incomplete = result.status == "event limit";
rows.push(result);
if incomplete {
break 'measurements;
}
}
}
}
drop(factory);
// Emit one final record only after all simulator callbacks and cleanup.
// Earlier stdout belongs to the solution and is diagnostic text, even if it
// resembles this record. This framing is not a sandbox for hostile Python.
println!("\n{RESULT_PREFIX}{}", serde_json::to_string(&rows).unwrap());
std::io::stdout().flush().unwrap();
}
struct WorkerResult {
rows: Vec<Measurement>,
problem: Option<String>,
diagnostics: Vec<String>,
}
fn read_bounded(
reader: impl Read + Send + 'static,
exceeded: Arc<AtomicBool>,
) -> thread::JoinHandle<Vec<u8>> {
thread::spawn(move || {
let mut bytes = Vec::new();
let _ = reader.take(OUTPUT_LIMIT + 1).read_to_end(&mut bytes);
if bytes.len() as u64 > OUTPUT_LIMIT {
exceeded.store(true, Ordering::Relaxed);
}
bytes
})
}
fn run_worker(args: &Args, path: &str, deadline: Instant) -> std::io::Result<WorkerResult> {
if Instant::now() >= deadline {
return Ok(WorkerResult {
rows: Vec::new(),
problem: Some("shared wall-clock budget exhausted".to_string()),
diagnostics: Vec::new(),
});
}
let mut child = Command::new(std::env::current_exe()?)
.args([
"--optimization-worker",
"-i",
path,
"-s",
&args.seed.to_string(),
"--optimization-runs",
&args.optimization_runs.to_string(),
])
.env("PYTHONUNBUFFERED", "1")
// The validator also initializes Python, so set the seed before spawn.
.env("PYTHONHASHSEED", args.seed.to_string())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let exceeded = Arc::new(AtomicBool::new(false));
let out = read_bounded(child.stdout.take().unwrap(), exceeded.clone());
let err = read_bounded(child.stderr.take().unwrap(), exceeded.clone());
let mut problem = None;
loop {
match child.try_wait() {
Ok(Some(status)) => {
if !status.success() {
problem = Some(format!("worker exited with {status}"));
}
break;
}
Ok(None) => {}
Err(error) => {
problem = Some(error.to_string());
let _ = child.kill();
let _ = child.wait();
break;
}
}
if exceeded.load(Ordering::Relaxed) {
problem = Some("output limit".to_string());
let _ = child.kill();
let _ = child.wait();
break;
}
if Instant::now() >= deadline {
problem = Some(format!(
"shared wall-clock limit ({}s)",
args.optimization_timeout
));
let _ = child.kill();
let _ = child.wait();
break;
}
thread::sleep(Duration::from_millis(10));
}
let stdout = out.join().unwrap_or_default();
let stderr = err.join().unwrap_or_default();
if stdout.len() as u64 > OUTPUT_LIMIT || stderr.len() as u64 > OUTPUT_LIMIT {
problem = Some("output limit".to_string());
}
let output = String::from_utf8_lossy(&stdout);
let (diagnostic_output, result_line) = output
.strip_suffix('\n')
.unwrap_or(&output)
.rsplit_once('\n')
.unwrap_or(("", output.as_ref()));
let mut rows = Vec::new();
if problem.is_none() {
match parse_measurements(result_line.trim_end_matches('\r'), args) {
Ok(measurements) => rows = measurements,
Err(error) => problem = Some(error),
}
}
let mut diagnostics = Vec::new();
for line in diagnostic_output.lines() {
if !line.is_empty() && diagnostics.len() < 4 {
diagnostics.push(line.chars().take(240).collect());
}
}
// A missing or malformed final record is also useful diagnostic output.
if rows.is_empty() && !result_line.is_empty() && diagnostics.len() < 4 {
diagnostics.push(result_line.chars().take(240).collect());
}
for line in String::from_utf8_lossy(&stderr).lines().take(4) {
diagnostics.push(line.chars().take(240).collect());
}
Ok(WorkerResult {
rows,
problem,
diagnostics,
})
}
fn parse_measurements(line: &str, args: &Args) -> Result<Vec<Measurement>, String> {
let json = line
.strip_prefix(RESULT_PREFIX)
.ok_or("worker did not return measurements")?;
let rows: Vec<Measurement> =
serde_json::from_str(json).map_err(|e| format!("invalid measurements: {e}"))?;
let expected = SCENARIOS.iter().flat_map(|scenario| {
SIZES.iter().flat_map(move |n| {
(0..args.optimization_runs)
.map(move |run| (*scenario, *n, args.seed.wrapping_add(run as u64)))
})
});
let total = SCENARIOS.len() * SIZES.len() * args.optimization_runs as usize;
if rows.is_empty()
|| rows.len() > total
|| rows.iter().zip(expected).any(|(row, (scenario, n, seed))| {
row.scenario != scenario || row.n != n || row.seed != seed
})
|| rows[..rows.len() - 1]
.iter()
.any(|row| row.status == "event limit")
|| (rows.len() < total && rows.last().unwrap().status != "event limit")
{
return Err("worker returned an incomplete or unexpected measurement sequence".to_string());
}
Ok(rows)
}
fn optimized_path(args: &Args) -> PathBuf {
args.optimized_impl
.as_ref()
.map(PathBuf::from)
.unwrap_or_else(|| {
PathBuf::from(&args.solution_path).with_file_name("broadcast_scalable.py")
})
}
pub fn run_if_present(args: &Args, score: f32) {
if args.optimized_impl.is_some() || optimized_path(args).exists() {
if score == 7. {
run(args);
} else {
println!("SKIPPED: OPTIMIZATION (requires main SCORE: 7)");
}
}
}
fn cell(row: Option<&Measurement>) -> String {
let Some(r) = row else {
return "NOT MEASURED".to_string();
};
let latency = |value: Option<f64>| {
value
.map(|v| format!("{v:.3}"))
.unwrap_or_else(|| "n/a".to_string())
};
format!(
"{} {} {} {}/{} {} {} {} {} {} [{}; t={:.3}; sent={}]",
r.messages,
r.bytes,
r.max_sent,
r.delivered,
r.expected,
latency(r.latency_mean),
latency(r.latency_max),
r.duplicates,
r.unexpected,
r.malformed,
if r.status == "event limit" {
"INCOMPLETE: event limit"
} else {
&r.status
},
r.observed_until,
r.sent
)
}
pub fn run(args: &Args) {
println!("\n--- OPTIMIZATION ---\n");
let optimized = optimized_path(args);
if !optimized.is_file() {
println!(
"SKIPPED: scalable implementation not found: {}",
optimized.display()
);
return;
}
// Both implementations receive the same external workload: ten unique
// 64-byte messages from process 0 at t=0..9, observed until HORIZON.
// Run indices start at zero; internal traffic depends on the implementation.
// Measurements have no performance threshold or automatic optimization score.
// Event/time/output limits mark incomplete measurements without changing SCORE.
println!("Workload: N={SIZES:?}, messages={MESSAGES} x 64B, sender=0, send_t=0..9, horizon={HORIZON}");
println!("Delays: NORMAL/CRASHES=[1,3], DELAYED=[1,10]; CRASHES: [1,2] at t=4.5");
println!("Runs={}, base_seed={}; limits: {EVENT_LIMIT} events/case, {}s/comparison, {OUTPUT_LIMIT} bytes/stream", args.optimization_runs, args.seed, args.optimization_timeout);
let deadline = Instant::now() + Duration::from_secs(args.optimization_timeout);
let mut results = Vec::new();
for (label, path) in [
("BASE", args.solution_path.as_str()),
("SCALABLE", optimized.to_str().unwrap_or("")),
] {
if label == "SCALABLE"
&& results.first().is_some_and(|r: &WorkerResult| {
r.problem.is_some() || r.rows.iter().any(|row| row.status == "event limit")
})
{
println!("SKIPPED: SCALABLE (BASE measurement incomplete)");
results.push(WorkerResult {
rows: Vec::new(),
problem: None,
diagnostics: Vec::new(),
});
continue;
}
println!("Measuring {label}: {path}");
let result = run_worker(args, path, deadline).unwrap_or_else(|e| WorkerResult {
rows: Vec::new(),
problem: Some(e.to_string()),
diagnostics: Vec::new(),
});
if let Some(problem) = &result.problem {
println!("{label}: INCOMPLETE: {problem}");
}
for line in &result.diagnostics {
println!("{label} diagnostic: {line}");
}
results.push(result);
}
println!("scenario N seed | BASE: messages bytes max_sent delivered/expected latency_mean latency_max dup unexpected malformed | SCALABLE: same columns");
for scenario in SCENARIOS {
for n in SIZES {
for run in 0..args.optimization_runs {
let seed = args.seed.wrapping_add(run as u64);
let find = |side: usize| {
results[side]
.rows
.iter()
.find(|r| r.scenario == scenario && r.n == n && r.seed == seed)
};
println!(
"{scenario} {n} {seed} | {} | {}",
cell(find(0)),
cell(find(1))
);
}
}
}
}
homework/04-broadcast/tests/src/runner.rs
+336
View File
@@ -0,0 +1,336 @@
//! Run each correctness test in a killable process. Student callbacks can block
//! inside Python, so an in-simulator deadline alone cannot enforce wall time.
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use std::process::Command;
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anysystem::python::PyProcessFactory;
use anysystem::test::TestResult;
use env_logger::Builder;
use indexmap::IndexMap;
use log::LevelFilter;
use crate::common::TestConfig;
use crate::tests::*;
use crate::tests_mc::*;
use crate::Args;
struct Test {
name: &'static str,
run: fn(&TestConfig) -> TestResult,
}
const TESTS: &[Test] = &[
Test {
name: "NORMAL",
run: test_normal,
},
Test {
name: "SENDER CRASH",
run: test_sender_crash,
},
Test {
name: "SENDER CRASH 2",
run: test_sender_crash2,
},
Test {
name: "TWO CRASHES",
run: test_two_crashes,
},
Test {
name: "TWO CRASHES 2",
run: test_two_crashes2,
},
Test {
name: "TWO CRASHES AFTER DELIVERY",
run: test_two_crashes_after_delivery,
},
Test {
name: "CAUSAL ORDER",
run: test_causal_order,
},
Test {
name: "QUORUM",
run: test_quorum,
},
Test {
name: "CHAOS MONKEY",
run: test_chaos_monkey,
},
Test {
name: "SCALABILITY",
run: test_scalability,
},
Test {
name: "MODEL CHECKING NORMAL DELIVERY",
run: test_mc_normal_delivery,
},
Test {
name: "MODEL CHECKING SENDER CRASH",
run: test_mc_sender_crash,
},
Test {
name: "MODEL CHECKING CAUSAL ORDER",
run: test_mc_causal_order,
},
];
pub fn worker(args: &Args) {
let test = TESTS
.iter()
.find(|test| Some(test.name) == args.test_worker.as_deref())
.unwrap();
if args.debug {
Builder::new()
.filter(Some("anysystem"), LevelFilter::Debug)
.format(|buf, record| writeln!(buf, "{}", record.args()))
.init();
}
let factory = PyProcessFactory::new(&args.solution_path, "BroadcastProcess");
let config = TestConfig {
proc_factory: &factory,
proc_count: if test.name.starts_with("MODEL CHECKING") {
3
} else {
args.proc_count
},
seed: args.seed,
monkeys: args.monkeys,
debug: args.debug,
};
let result = (test.run)(&config);
// Keep the structured result separate from student prints and debug traces.
fs::write(
args.test_result.as_ref().unwrap(),
serde_json::to_vec(&result).unwrap(),
)
.unwrap();
}
struct ResultFile(PathBuf);
impl ResultFile {
fn new() -> std::io::Result<Self> {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let path = std::env::temp_dir().join(format!(
"broadcast-test-{}-{nonce}.json",
std::process::id()
));
OpenOptions::new()
.write(true)
.create_new(true)
.open(&path)?;
Ok(Self(path))
}
}
impl Drop for ResultFile {
fn drop(&mut self) {
let _ = fs::remove_file(&self.0);
}
}
fn run_one(args: &Args, test: &Test, suite_deadline: Instant) -> TestResult {
let seconds = if test.name == "CHAOS MONKEY" {
args.chaos_timeout
} else if test.name == "SCALABILITY" {
args.scalability_timeout
} else if test.name.starts_with("MODEL CHECKING") {
// The process deadline also catches a callback hanging inside the MC.
args.mc_timeout
.min(if test.name == "MODEL CHECKING NORMAL DELIVERY" {
110
} else if test.name == "MODEL CHECKING CAUSAL ORDER" {
60
} else {
210
})
} else {
args.test_timeout
};
let test_deadline = Instant::now() + Duration::from_secs(seconds);
let deadline = test_deadline.min(suite_deadline);
let timeout = if suite_deadline <= test_deadline {
format!("suite time limit of {}s exceeded", args.suite_timeout)
} else {
format!("test time limit of {seconds}s exceeded")
};
if Instant::now() >= deadline {
return Err(timeout);
}
let result_file = ResultFile::new().map_err(|e| format!("test worker error: {e}"))?;
let mut command = Command::new(std::env::current_exe().map_err(|e| e.to_string())?);
command
.args([
"--test-worker",
test.name,
"-i",
&args.solution_path,
"-s",
&args.seed.to_string(),
"-p",
&args.proc_count.to_string(),
"-m",
&args.monkeys.to_string(),
"--test-result",
])
.arg(&result_file.0)
.env("PYTHONUNBUFFERED", "1")
// Python must see the seed before interpreter initialization.
.env("PYTHONHASHSEED", args.seed.to_string());
if args.debug {
command.arg("-d");
}
let mut child = command
.spawn()
.map_err(|e| format!("test worker error: {e}"))?;
loop {
match child.try_wait() {
Ok(Some(status)) => {
if !status.success() {
return Err(format!("test worker exited with {status}"));
}
return serde_json::from_slice(
&fs::read(&result_file.0).map_err(|e| format!("test worker result: {e}"))?,
)
.map_err(|e| format!("test worker result: {e}"))?;
}
Ok(None) => {}
Err(error) => {
let _ = child.kill();
let _ = child.wait();
return Err(format!("test worker error: {error}"));
}
}
if Instant::now() >= deadline {
let _ = child.kill();
let _ = child.wait();
return Err(timeout);
}
thread::sleep(Duration::from_millis(10));
}
}
fn must_stop(result: &TestResult, keep_going: bool) -> bool {
let Err(error) = result else {
return false;
};
// Resource exhaustion and process failures always stop, including in local
// diagnostic mode. A pure causal violation must not hide a later base error.
if error.contains("time limit")
|| error.starts_with("causal model-checking ")
|| error.starts_with("test worker")
{
return true;
}
if keep_going {
return false;
}
!error.strip_prefix("Violated ").is_some_and(|properties| {
properties
.split(", ")
.all(|property| property == "CAUSAL ORDER")
})
}
pub(crate) fn causal_mc_incomplete(name: &str, error: &str) -> bool {
name == "MODEL CHECKING CAUSAL ORDER"
&& (error.contains("time limit") || error.starts_with("causal model-checking "))
}
pub fn run(args: &Args) -> IndexMap<String, TestResult> {
let tests: Vec<_> = TESTS
.iter()
.filter(|test| {
if let Some(name) = &args.test {
test.name == name
} else {
!args.disable_mc_tests || !test.name.starts_with("MODEL CHECKING")
}
})
.collect();
let deadline = Instant::now() + Duration::from_secs(args.suite_timeout);
let mut results = IndexMap::new();
let mut stopped = None;
let mut passed = 0;
let mut causal_violated = false;
for test in &tests {
if let Some(failed_test) = stopped {
println!("SKIPPED: {} (stopped after {failed_test})", test.name);
continue;
}
if test.name == "MODEL CHECKING CAUSAL ORDER" && causal_violated && !args.keep_going {
println!("SKIPPED: {} (CAUSAL ORDER already violated)", test.name);
continue;
}
println!("\n--- {} ---\n", test.name);
std::io::stdout().flush().unwrap();
let result = run_one(args, test, deadline);
match &result {
Ok(_) => {
println!("\nPASSED\n");
passed += 1;
}
Err(error) => {
causal_violated |= error.strip_prefix("Violated ").is_some_and(|properties| {
properties
.split(", ")
.any(|property| property == "CAUSAL ORDER")
});
println!("\nFAILED: {error}");
if causal_mc_incomplete(test.name, error) {
println!("INCOMPLETE: causal model checking did not finish within its resource budget");
}
println!();
}
}
if must_stop(&result, args.keep_going) {
stopped = Some(test.name);
}
results.insert(test.name.to_string(), result);
}
if let Some(name) = &args.test {
if tests.is_empty() {
println!("FAILED: unknown test {name}");
}
} else {
println!("-------------------------------------------------------------------------------");
println!("\nPassed {passed} from {} tests\n", tests.len());
if results.len() != tests.len() {
println!("Skipped {} tests\n", tests.len() - results.len());
}
if results.values().any(Result::is_err) {
println!("Failed tests:");
for (name, result) in &results {
if let Err(error) = result {
println!("- {name}: {error}");
}
}
println!();
}
}
results
}
#[cfg(test)]
mod tests {
use super::must_stop;
#[test]
fn mc_timeout_always_stops_but_causal_violation_does_not() {
assert!(must_stop(&Err("time limit of 100s exceeded".into()), false));
assert!(must_stop(&Err("time limit of 100s exceeded".into()), true));
assert!(!must_stop(&Err("Violated CAUSAL ORDER".into()), false));
assert!(must_stop(
&Err("Violated VALIDITY, CAUSAL ORDER".into()),
false
));
assert!(!must_stop(&Err("Violated VALIDITY".into()), true));
}
}
homework/04-broadcast/tests/src/shared_state.rs
+78
View File
@@ -0,0 +1,78 @@
use std::ffi::CString;
use pyo3::prelude::*;
use pyo3::types::{PyList, PyModule};
const VALIDATOR_CODE: &str = include_str!("../shared_state_validator.py");
fn validator_module<'py>(py: Python<'py>) -> PyResult<Bound<'py, PyModule>> {
let code = CString::new(VALIDATOR_CODE).expect("validator source contains a null byte");
PyModule::from_code(
py,
&code,
c"shared_state_validator.py",
c"shared_state_validator",
)
}
#[cfg(test)]
fn validate_source(
source: &str,
filename: &str,
class_names: &[&str],
) -> Result<Vec<String>, String> {
Python::attach(|py| {
let module = validator_module(py).map_err(|error| error.to_string())?;
let class_names = PyList::new(py, class_names).map_err(|error| error.to_string())?;
module
.getattr("validate_source")
.and_then(|function| function.call1((source, filename, class_names)))
.and_then(|result| result.extract::<Vec<String>>())
.map_err(|error| error.to_string())
})
}
fn validate_solution(path: &str, class_names: &[&str]) -> Result<Vec<String>, String> {
Python::attach(|py| {
let module = validator_module(py).map_err(|error| error.to_string())?;
let class_names = PyList::new(py, class_names).map_err(|error| error.to_string())?;
module
.getattr("validate_solution")
.and_then(|function| function.call1((path, class_names)))
.and_then(|result| result.extract::<Vec<String>>())
.map_err(|error| error.to_string())
})
}
pub fn validate_or_report(path: &str, class_names: &[&str]) -> bool {
match validate_solution(path, class_names) {
Ok(violations) if violations.is_empty() => true,
Ok(violations) => {
println!("ERROR: shared state between AnySystem processes is forbidden");
for violation in violations {
println!("{violation}");
}
println!("\nSCORE: 0\n");
false
}
Err(error) => panic!("failed to validate solution for shared state: {error}"),
}
}
#[cfg(test)]
mod tests {
use super::validate_source;
#[test]
fn embeds_validator_and_rejects_shared_state() {
let violations = validate_source(
"class ProcessImpl:\n shared = {}\n",
"solution.py",
&["ProcessImpl"],
)
.unwrap();
assert_eq!(violations.len(), 1);
assert!(violations[0].contains("ProcessImpl.shared"));
}
}
homework/04-broadcast/tests/src/tests.rs
+495
View File
@@ -0,0 +1,495 @@
use std::collections::{BTreeMap, BTreeSet, HashSet};
use rand::prelude::*;
use rand_pcg::Pcg64;
use serde_json::Value;
use anysystem::test::TestResult;
use anysystem::{Message, ProcessEvent, System};
use crate::common::{build_system, BroadcastMessage, TestConfig};
pub fn test_normal(config: &TestConfig) -> TestResult {
let mut sys = build_system(config);
let text = "0:Hello";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_sender_crash(config: &TestConfig) -> TestResult {
println!("Process 0 will crash after another process receives a network message.");
let mut sys = build_system(config);
let text = "0:Hello";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
// run until the message is received by one other process
let mut received = false;
while !received {
if !sys.step() {
return Err("no events left before another process received the message".to_string());
}
for n in 1..config.proc_count {
if sys.received_message_count(&n.to_string()) == 1 {
received = true;
break;
}
}
}
// crash source node
sys.crash_node("0");
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_sender_crash2(config: &TestConfig) -> TestResult {
println!("Process 0 is isolated and will crash; its messages cannot reach other processes.");
let mut sys = build_system(config);
let text = "0:Hello";
// let the message to be received only by the sender itself by disconnecting it
sys.network().disconnect_node("0");
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.step();
sys.crash_node("0");
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_two_crashes(config: &TestConfig) -> TestResult {
if config.proc_count < 5 {
return Err(
"There are 2 crashes in this test, so at least 5 processes are required.".to_string(),
);
};
let mut sys = build_system(config);
let survivors: Vec<String> = (2..config.proc_count).map(|n| n.to_string()).collect();
println!(
"Processes [0, 1] are isolated from [{}] and will crash; links within each group remain enabled.",
survivors.join(", ")
);
sys.network().make_partition(
&["0", "1"],
&survivors.iter().map(String::as_str).collect::<Vec<_>>(),
);
let text = "0:Hello";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.steps(config.proc_count.pow(2));
sys.crash_node("0");
sys.crash_node("1");
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_two_crashes2(config: &TestConfig) -> TestResult {
if config.proc_count < 5 {
return Err(
"There are 2 crashes in this test, so at least 5 processes are required.".to_string(),
);
};
let mut sys = build_system(config);
println!(
"Outgoing messages from processes [1, 2] are dropped; both processes will later crash."
);
// Model messages that remain undelivered until their senders crash.
sys.network().drop_outgoing("1");
sys.network().drop_outgoing("2");
let text = "0:Hello";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.steps(config.proc_count.pow(2));
sys.crash_node("1");
sys.crash_node("2");
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_two_crashes_after_delivery(config: &TestConfig) -> TestResult {
if config.proc_count < 5 {
return Err(
"There are 2 crashes in this test, so at least 5 processes are required.".to_string(),
);
}
println!("Process 0 and the first other process to deliver to its local user will crash immediately after that delivery.");
let mut sys = build_system(config);
let text = "0:Hello";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
// Crash the sender and the first other process to deliver to its user.
// Uniform agreement must still hold even though that recipient has crashed.
while sys.step() {
let recipient = (1..config.proc_count)
.map(|n| n.to_string())
.find(|proc| !sys.local_outbox(proc).is_empty());
if let Some(recipient) = recipient {
println!("Process {recipient} delivered to its local user; crashing processes [0, {recipient}].");
sys.crash_node("0");
sys.crash_node(&recipient);
break;
}
}
sys.step_until_no_events();
check(&sys, config, HashSet::from([text.to_string()]))
}
pub fn test_quorum(config: &TestConfig) -> TestResult {
for node_count in [7, 10] {
let mut run_config = *config;
run_config.proc_count = node_count;
let failures = (node_count - 1) / 2;
let text = "0:Hello";
println!("- N={node_count}: {failures} isolated processes crash");
let mut sys = build_system(&run_config);
let minority: Vec<String> = (0..failures).map(|n| n.to_string()).collect();
let majority: Vec<String> = (failures..node_count).map(|n| n.to_string()).collect();
println!(
"Processes [{}] are isolated from [{}] and will crash; links within each group remain enabled.",
minority.join(", "),
majority.join(", ")
);
sys.network().make_partition(
&minority.iter().map(String::as_str).collect::<Vec<_>>(),
&majority.iter().map(String::as_str).collect::<Vec<_>>(),
);
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.step_until_no_events();
for n in 0..failures {
sys.crash_node(&n.to_string());
}
sys.step_until_no_events();
check(&sys, &run_config, HashSet::from([text.to_string()]))?;
println!("- N={node_count}: {failures} processes crash before broadcast");
println!(
"Processes [{}] will crash; the remaining majority must deliver to their local users.",
((node_count - failures)..node_count)
.map(|n| n.to_string())
.collect::<Vec<_>>()
.join(", ")
);
let mut sys = build_system(&run_config);
// The surviving majority, including the sender, must complete delivery.
for n in (node_count - failures)..node_count {
sys.crash_node(&n.to_string());
}
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.step_until_no_events();
check(&sys, &run_config, HashSet::from([text.to_string()]))?;
}
Ok(true)
}
pub fn test_causal_order(config: &TestConfig) -> TestResult {
let mut sys = build_system(config);
sys.network().set_delays(100., 200.);
let texts = ["0:Hello", "1:How?", "0:Fine!"];
sys.send_local_message(
"0",
Message::json("SEND", &BroadcastMessage { text: texts[0] }),
);
while sys.event_log("1").is_empty() {
if !sys.step() {
return Err("no events left before the second SEND".to_string());
}
}
sys.network().set_delays(10., 20.);
sys.send_local_message(
"1",
Message::json("SEND", &BroadcastMessage { text: texts[1] }),
);
while sys.event_log("0").len() < 3 {
if !sys.step() {
return Err("no events left before the third SEND".to_string());
}
}
sys.network().set_delay(1.);
sys.send_local_message(
"0",
Message::json("SEND", &BroadcastMessage { text: texts[2] }),
);
sys.step_until_no_events();
let sent_messages = HashSet::from_iter(texts.into_iter().map(String::from));
check(&sys, config, sent_messages)
}
pub fn test_chaos_monkey(config: &TestConfig) -> TestResult {
if config.proc_count < 5 {
return Err(
"There are 2 crashes in this test, so at least 5 processes are required.".to_string(),
);
};
let mut rand = Pcg64::seed_from_u64(config.seed);
println!("Only outgoing messages from processes that will later crash may be dropped.");
for i in 1..=config.monkeys {
// Preserve the random sequence: this draw was previously printed as an unused seed.
let _ = rand.next_u64();
let mut sys = build_system(config);
let victim1 = format!("{}", rand.gen_range(0..config.proc_count));
let mut victim2 = format!("{}", rand.gen_range(0..config.proc_count));
while victim2 == victim1 {
victim2 = format!("{}", rand.gen_range(0..config.proc_count));
}
println!(
"- Run {i} (base seed: {}; processes that will crash: [{victim1}, {victim2}])",
config.seed
);
let mut sent_messages = HashSet::new();
for i in 0..10 {
let sender = format!("{}", rand.gen_range(0..config.proc_count));
let text = format!("{sender}:{i}");
sent_messages.insert(text.clone());
if i % 2 == 0 {
sys.network().set_delays(10., 20.);
} else {
sys.network().set_delays(1., 2.);
}
for j in 0..8 {
if rand.gen_range(0.0..1.0) > 0.3 {
sys.network().drop_outgoing(&victim1);
} else {
sys.network().pass_outgoing(&victim1);
}
if rand.gen_range(0.0..1.0) > 0.3 {
sys.network().drop_outgoing(&victim2);
} else {
sys.network().pass_outgoing(&victim2);
}
if j == 0 {
sys.send_local_message(
&sender,
Message::json("SEND", &BroadcastMessage { text: &text }),
);
} else {
sys.step();
}
}
}
sys.crash_node(&victim1);
sys.crash_node(&victim2);
sys.step_until_no_events();
check(&sys, config, sent_messages)?;
}
Ok(true)
}
pub fn test_scalability(config: &TestConfig) -> TestResult {
let sys_sizes = [
config.proc_count,
config.proc_count * 2,
config.proc_count * 4,
config.proc_count * 10,
];
let mut msg_counts = Vec::new();
for node_count in sys_sizes {
let mut run_config = *config;
run_config.proc_count = node_count;
let mut sys = build_system(&run_config);
let text = "0:Hello!";
sys.send_local_message("0", Message::json("SEND", &BroadcastMessage { text }));
sys.step_until_no_events();
check(&sys, &run_config, HashSet::from([text.to_string()]))?;
msg_counts.push(sys.network().network_message_count());
}
println!("\nMessage count:");
for i in 0..sys_sizes.len() {
let baseline = sys_sizes[i] * (sys_sizes[i] - 1);
println!(
"- N={}: {} (baseline {})",
sys_sizes[i], msg_counts[i], baseline
);
}
Ok(true)
}
fn check(sys: &System, config: &TestConfig, all_sent_messages: HashSet<String>) -> TestResult {
let mut sent = BTreeMap::new();
let mut delivered = BTreeMap::new();
let mut all_delivered = BTreeSet::new();
let mut histories = BTreeMap::new();
let mut debug_histories = BTreeMap::new();
for proc in sys.process_names() {
let mut history = Vec::new();
let mut debug_history = Vec::new();
let mut sent_msgs = Vec::new();
let mut delivered_msgs = Vec::new();
for e in sys.event_log(&proc) {
match e.event {
ProcessEvent::LocalMessageReceived { msg: m } => {
let data: Value = serde_json::from_str(&m.data).unwrap();
let message = data["text"].as_str().unwrap().to_string();
sent_msgs.push(message.clone());
if config.debug {
debug_history.push(format!("SEND {message:?}"));
}
history.push(message);
}
ProcessEvent::LocalMessageSent { msg: m } => {
let data: Value = serde_json::from_str(&m.data).unwrap();
let message = data["text"].as_str().unwrap().to_string();
delivered_msgs.push(message.clone());
all_delivered.insert(message.clone());
if config.debug {
debug_history.push(format!("DELIVER {message:?}"));
}
history.push(message);
}
_ => {}
}
}
sent.insert(proc.clone(), sent_msgs);
delivered.insert(proc.clone(), delivered_msgs);
debug_histories.insert(proc.clone(), debug_history);
histories.insert(proc, history);
}
if config.debug {
println!(
"Messages sent across network: {}",
sys.network().network_message_count()
);
println!("Process histories:");
for proc in sys.process_names() {
println!(
"- [{}] {}",
proc,
debug_histories.get(&proc).unwrap().join(", ")
);
}
}
// NO DUPLICATION
let mut no_duplication = true;
for (proc, delivered_msgs) in &delivered {
let mut uniq = HashSet::new();
for msg in delivered_msgs {
if uniq.contains(msg) {
println!("NO DUPLICATION: process {proc} delivered message {msg:?} to its local user more than once.");
no_duplication = false;
};
uniq.insert(msg);
}
}
// NO CREATION
let mut no_creation = true;
for (proc, delivered_msgs) in &delivered {
for msg in delivered_msgs {
if !all_sent_messages.contains(msg) {
println!("NO CREATION: process {proc} delivered message {msg:?} to its local user, but no user sent it.");
no_creation = false;
}
}
}
// VALIDITY
let mut validity = true;
for (proc, sent_msgs) in &sent {
if sys.proc_node_is_crashed(proc) {
continue;
}
let delivered_msgs = delivered.get(proc).unwrap();
for msg in sent_msgs {
if !delivered_msgs.contains(msg) {
println!("VALIDITY: correct process {proc} received SEND {msg:?} from its local user, but did not deliver it back with DELIVER.");
validity = false;
}
}
}
// UNIFORM AGREEMENT
let mut uniform_agreement = true;
for msg in all_delivered.iter() {
let mut delivered_by = Vec::new();
let mut missing = Vec::new();
for (proc, delivered_msgs) in &delivered {
if delivered_msgs.contains(msg) {
delivered_by.push(if sys.proc_node_is_crashed(proc) {
format!("{proc} (later crashed)")
} else {
proc.clone()
});
} else if !sys.proc_node_is_crashed(proc) {
missing.push(proc.clone());
}
}
if !missing.is_empty() {
println!("UNIFORM AGREEMENT violated for message {msg:?}:");
println!(
" Delivered to local users by processes: {}.",
delivered_by.join(", ")
);
println!(
" Not delivered by correct processes: {}.",
missing.join(", ")
);
uniform_agreement = false;
}
}
// CAUSAL ORDER
let mut causal_order = true;
for (src, sent_msgs) in &sent {
for msg in sent_msgs.iter() {
if !all_delivered.contains(msg) {
continue;
}
// build sender past for send message event
let mut src_past = HashSet::new();
for e in histories.get(src).unwrap() {
if e != msg {
src_past.insert(e.clone());
} else {
break;
}
}
// check that other correct processes have delivered all past events before delivering the message
for (dst, delivered_msgs) in &delivered {
if sys.proc_node_is_crashed(dst) {
continue;
}
let mut dst_past = HashSet::new();
for e in delivered_msgs {
if e != msg {
dst_past.insert(e.clone());
} else {
break;
}
}
if !dst_past.is_superset(&src_past) {
let mut missing = src_past
.difference(&dst_past)
.cloned()
.collect::<Vec<String>>();
missing.sort();
println!(
"CAUSAL ORDER: process {} did not deliver messages {:?} before message {:?} from process {}.",
dst,
missing,
msg,
src
);
causal_order = false;
}
}
}
}
if no_duplication & no_creation & validity & uniform_agreement & causal_order {
Ok(true)
} else {
let mut violated = Vec::new();
if !no_duplication {
violated.push("NO DUPLICATION")
}
if !no_creation {
violated.push("NO CREATION")
}
if !validity {
violated.push("VALIDITY")
}
if !uniform_agreement {
violated.push("UNIFORM AGREEMENT")
}
if !causal_order {
violated.push("CAUSAL ORDER")
}
Err(format!("Violated {}", violated.join(", ")))
}
}
homework/04-broadcast/tests/src/tests_mc.rs
+289
View File
@@ -0,0 +1,289 @@
use std::cell::Cell;
use std::collections::HashSet;
use std::rc::Rc;
use std::time::Duration;
use serde_json::Value;
use sugars::boxed;
use anysystem::logger::LogEntry;
use anysystem::test::TestResult;
use anysystem::Message;
use anysystem::mc::predicates::{collects, goals, invariants, prunes};
use anysystem::mc::strategies::Bfs;
use anysystem::mc::strategy::{GoalFn, InvariantFn, PruneFn, StrategyConfig};
use anysystem::mc::{McState, ModelChecker};
use crate::common::{build_system, BroadcastMessage, TestConfig};
// The causal test explores two broadcasts on three processes. The state budget
// is checked as an invariant, so exhausting it cannot silently pass a pruned search.
const CAUSAL_MC_STATE_LIMIT: usize = 100_000;
pub fn test_mc_normal_delivery(config: &TestConfig) -> TestResult {
let mut sys = build_system(config);
let proc_names = sys.process_names();
let text = "0:Hello";
sys.send_local_message(
proc_names[0].as_str(),
Message::json("SEND", &BroadcastMessage { text }),
);
let goal = goals::all_goals(
proc_names
.iter()
.map(|name| goals::got_n_local_messages(name, name, 1))
.collect::<Vec<GoalFn>>(),
);
let strategy_config = StrategyConfig::default()
.goal(goal)
.prune(prunes::any_prune(vec![
prunes::state_depth(10),
mc_prune_proc_permutations(&proc_names[1..]),
// Prune states with more than 2 messages received from any process
mc_prune_msg_per_proc_limit(&proc_names, 2),
]))
.invariant(invariants::all_invariants(vec![
mc_invariant(proc_names.clone(), text.to_string()),
invariants::time_limit(Duration::from_secs(100)),
]));
let mut mc = ModelChecker::new(&sys);
let res = mc.run::<Bfs>(strategy_config);
if let Err(err) = res {
err.print_trace();
Err(err.message())
} else {
Ok(true)
}
}
pub fn test_mc_sender_crash(config: &TestConfig) -> TestResult {
let mut sys = build_system(config);
let proc_names = sys.process_names();
let text = "0:Hello";
sys.send_local_message(
proc_names[0].as_str(),
Message::json("SEND", &BroadcastMessage { text }),
);
let goal = goals::all_goals(
proc_names
.iter()
.map(|name| goals::got_n_local_messages(name, name, 1))
.collect::<Vec<GoalFn>>(),
);
let strategy_config = StrategyConfig::default()
.prune(prunes::any_prune(vec![
prunes::state_depth(4),
mc_prune_proc_permutations(&proc_names[1..]),
]))
.goal(goal)
.invariant(invariants::all_invariants(vec![
mc_invariant(proc_names.clone(), text.to_string()),
invariants::time_limit(Duration::from_secs(100)),
]))
.collect(collects::any_collect(
proc_names[1..]
.iter()
.map(|proc| collects::got_n_local_messages(proc, proc, 1))
.collect(),
));
let mut mc = ModelChecker::new(&sys);
let res = mc.run::<Bfs>(strategy_config);
let intermediate_states = res
.map_err(|err| {
err.print_trace();
err.message()
})?
.collected_states;
if intermediate_states.is_empty() {
return Err("no states collected after first stage".to_string());
}
// Crash first node in the list
let left_proc_names = proc_names[1..].to_vec();
let goal = goals::all_goals(
left_proc_names
.iter()
.map(|name| goals::got_n_local_messages(name, name, 1))
.collect::<Vec<GoalFn>>(),
);
let strategy_config = StrategyConfig::default()
.goal(goal)
.invariant(invariants::all_invariants(vec![
mc_invariant(left_proc_names, text.to_string()),
invariants::time_limit(Duration::from_secs(100)),
]))
.prune(prunes::any_prune(vec![
prunes::state_depth(6),
mc_prune_proc_permutations(&proc_names[1..]),
// Prune states with more than 4 messages received from any process
mc_prune_msg_per_proc_limit(&proc_names, 4),
]));
let res = mc.run_from_states_with_change::<Bfs>(strategy_config, intermediate_states, |sys| {
sys.crash_node(proc_names[0].clone());
});
if let Err(err) = res {
err.print_trace();
Err(err.message())
} else {
Ok(true)
}
}
// Send "a" from P2, then send "b" from P1 after P1 delivers "a",
// which gives the causal order "a" -> "b", then check that all processes
// deliver both messages and deliver "a" before "b".
pub fn test_mc_causal_order(config: &TestConfig) -> TestResult {
let mut sys = build_system(config);
sys.send_local_message("2", Message::json("SEND", &BroadcastMessage { text: "a" }));
let proc_names = sys.process_names();
let report_error = |err: anysystem::mc::McError| {
err.print_trace();
err.message()
};
let checked = Rc::new(Cell::new(0));
let make_strategy = |texts: Vec<&'static str>| {
let goal_procs = proc_names.clone();
let invariant_procs = proc_names.clone();
let checked = checked.clone();
StrategyConfig::default()
// Stop this branch when everyone has delivered the messages
// sent in this stage. If events end before that, the test fails.
.goal(boxed!(move |state: &McState| {
goal_procs
.iter()
.all(|proc| {
texts
.iter()
.all(|text| mc_delivery_index(state, proc, text).is_some())
})
.then(|| "all messages have been delivered by every process".to_string())
}))
// Since P1 sends "b" after it delivers "a",
// every process must also deliver "a" before "b".
.invariant(invariants::all_invariants(vec![
boxed!(move |state: &McState| {
for proc in &invariant_procs {
if let Some(b) = mc_delivery_index(state, proc, "b") {
if !mc_delivery_index(state, proc, "a").is_some_and(|a| a < b) {
return Err("Violated CAUSAL ORDER".to_string());
}
}
}
Ok(())
}),
boxed!(move |_: &McState| {
let next = checked.get() + 1;
checked.set(next);
if next > CAUSAL_MC_STATE_LIMIT {
Err(format!(
"causal model-checking state limit of {CAUSAL_MC_STATE_LIMIT} exceeded"
))
} else {
Ok(())
}
}),
invariants::time_limit(Duration::from_secs(55)),
]))
};
let mut mc = ModelChecker::new(&sys);
// Explore all schedules of "a" before P1 sends "b". The first stage
// checks delivery of "a" on every branch and retains every state where
// P1 can send "b" after delivering "a".
let states = mc
.run::<Bfs>(make_strategy(vec!["a"]).collect(boxed!(|state: &McState| {
mc_delivery_index(state, "1", "a").is_some()
})))
.map_err(&report_error)?
.collected_states;
if states.is_empty() {
return Err("No state with delivery of a on process 1 was reached".to_string());
}
// Explore every continuation after sending "b" from each collected state.
// Resource exhaustion is reported as incomplete rather than pruning paths.
mc.run_from_states_with_change::<Bfs>(make_strategy(vec!["a", "b"]), states, |sys| {
sys.send_local_message(
"1",
"1",
Message::json("SEND", &BroadcastMessage { text: "b" }),
);
})
.map(|_| true)
.map_err(report_error)
}
fn mc_delivery_index(state: &McState, proc: &str, text: &str) -> Option<usize> {
state.node_states[proc].proc_states[proc]
.local_outbox
.iter()
.position(|msg| {
msg.tip == "DELIVER"
&& serde_json::from_str::<Value>(&msg.data).unwrap()["text"] == text
})
}
fn mc_prune_proc_permutations(equivalent_procs: &[String]) -> PruneFn {
let equivalent_procs = equivalent_procs.to_vec();
boxed!(move |state| {
let proc_names = HashSet::<String>::from_iter(equivalent_procs.clone().into_iter());
let mut used_proc_names = HashSet::<String>::new();
let mut waiting_for_proc = 0;
for entry in &state.trace {
match entry {
LogEntry::McMessageReceived { src: proc, .. }
| LogEntry::McTimerFired { proc, .. } => {
if used_proc_names.contains(proc) || !proc_names.contains(proc) {
continue;
}
if equivalent_procs[waiting_for_proc] != *proc {
return Some(
"state is the same as another state with renumerated processes"
.to_owned(),
);
}
used_proc_names.insert(proc.clone());
waiting_for_proc += 1;
}
_ => {}
}
}
None
})
}
fn mc_prune_msg_per_proc_limit(proc_names: &[String], limit: usize) -> PruneFn {
prunes::events_limit_per_proc(
|entry: &LogEntry, proc: &String| match entry {
LogEntry::McMessageReceived { src, .. } => src == proc,
_ => false,
},
proc_names.to_owned(),
limit,
)
}
fn mc_invariant(proc_names: Vec<String>, sent_message: String) -> InvariantFn {
boxed!(move |state: &McState| {
for name in &proc_names {
let outbox = &state.node_states[name].proc_states[name].local_outbox;
let mut message_data = HashSet::new();
for message in outbox {
let data: Value = serde_json::from_str(&message.data).unwrap();
let message = data["text"].as_str().unwrap().to_string();
if message_data.contains(&message) {
return Err("Violated NO DUPLICATION".to_owned());
}
message_data.insert(message.clone());
if message != sent_message {
return Err("Violated NO CREATION".to_owned());
}
}
}
Ok(())
})
}