Test against a bad network

For Rust developers serving a plaza app over WebSockets who want to see how it behaves on a slow or lossy link before a player on hotel wifi finds out

You will serve a shared counter over WebSockets on localhost and then make the link to one client slow, lossy and jittery while the program runs, using the link conditioner built into plaza_session. The client times how long each change takes to come back and the server times its own latency probes, so you can see what each setting changes and what it leaves alone.

Before you start

You need Rust with Cargo. plaza comes from crates.io:

plaza = "0"

Set up the project#

Start a new binary crate.

cargo new badlink
cd badlink

plaza holds the controller, plaza_session serves it over actix-web WebSockets and plaza_ws is the client. plaza_client_utils has the RttEstimator the client measures with. Your Cargo.toml should end like this:

Cargo.toml
[dependencies]
actix-web = "4"
async-trait = "0.1"
plaza = "0"
plaza_client_utils = "0"
plaza_session = "0"
plaza_wire = "0"
plaza_ws = { version = "0", features = ["native", "pump"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

The server and its client share one process. The server runs on its own thread inside an actix system and listens on 127.0.0.1:5780. The client runs on the main thread and polls its socket every millisecond, the way a game polls once per frame. Nothing leaves your machine: the conditioner sits inside the session, on the server's side of each connection.

Write the counter#

This is the same counter as in the previous tutorial. Put it in src/counter.rs:

src/counter.rs
use async_trait::async_trait;
use plaza::agent::Agent;
use plaza::error::{SnapshotError, StateLogicError};
use plaza::snapshot::{SnapshotContext, SnapshotProvider};
use plaza::state_logic::{LogicInput, LogicOutput, SnapshotRequest, StateLogic};
use serde::{Deserialize, Serialize};

pub type UserId = u64;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CounterOp {
    Increment(i64),
    Value(i64),
}

#[derive(Debug, Default)]
pub struct Counter {
    value: i64,
    here: Vec<Agent<UserId>>,
}

pub struct CounterLogic;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterLogic {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentJoined { agent } => state.here.push(agent),
            LogicInput::AgentLeft { agent_id } => state.here.retain(|a| a.id() != Some(&agent_id)),
            LogicInput::AgentOps { ops, .. } => {
                for op in ops {
                    if let CounterOp::Increment(by) = op {
                        state.value += by;
                    }
                }
                return Ok(
                    LogicOutput::none().and_snapshot(SnapshotRequest::to(state.here.clone()))
                );
            }
            LogicInput::TimeStep { .. } => {}
        }
        Ok(LogicOutput::none())
    }
}

pub struct CounterSnapshotter;

#[async_trait]
impl SnapshotProvider<UserId, Counter, CounterOp> for CounterSnapshotter {
    async fn create_snapshot(
        &self,
        state: &Counter,
        _target: Option<&Agent<UserId>>,
        _context: Option<SnapshotContext>,
    ) -> Result<Option<CounterOp>, SnapshotError<UserId>> {
        Ok(Some(CounterOp::Value(state.value)))
    }
}

Every batch of ops asks for a fresh snapshot for everyone connected, so a client that sends Increment(1) gets a Value back. The time between the two is what a player feels as the delay between doing something and seeing it.

Replace src/main.rs with this:

src/main.rs
mod counter;

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_client_utils::RttEstimator;
use plaza_session::{ActixWsPlazaSession, SessionOptions};
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, UserId};

const ADDR: &str = "127.0.0.1:5780";
const URL: &str = "ws://127.0.0.1:5780/ws";

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

static NEXT_USER: AtomicU64 = AtomicU64::new(1);

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    let id = NEXT_USER.fetch_add(1, Ordering::Relaxed);
    session.handle_connection(&req, stream, Agent::new_human(id))
}

fn serve(ready: mpsc::Sender<Arc<CounterSession>>) {
    actix_web::rt::System::new().block_on(async move {
        let options = SessionOptions::default().probe_schedule(
            8,
            Duration::from_millis(125),
            Duration::from_millis(250),
        );
        let session = CounterSession::with_options(JsonCodec, options);
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let data = web::Data::new(session.clone());
        let server = HttpServer::new(move || {
            App::new()
                .app_data(data.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(session).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
}

impl Client {
    fn connect() -> Self {
        let pump = FramePump::connect(URL, JsonCodec, 0).expect("connect");
        Client {
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => panic!("closed: {closed}"),
                Arrival::Opened | Arrival::Mismatch { .. } => {}
            }
        }
    }

    fn idle(&mut self, time: Duration) {
        let until = Instant::now() + time;
        while Instant::now() < until {
            self.poll();
            thread::sleep(Duration::from_millis(1));
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                return value;
            }
            thread::sleep(Duration::from_millis(1));
        }
        panic!("no value within 5s");
    }
}

fn floor_to(step: u64, ms: f32) -> u64 {
    ms as u64 / step * step
}

fn measure(label: &str, client: &mut Client, session: &CounterSession) {
    client.idle(Duration::from_millis(500));

    let mut ops = RttEstimator::default();
    let mut slowest = 0;
    for _ in 0..20 {
        let sent = Instant::now();
        client.send(CounterOp::Increment(1));
        client.next_value();
        let took = sent.elapsed().as_millis() as u64;
        ops.observe(took);
        slowest = slowest.max(took);
    }

    let manager = session.manager();
    let conn = manager.connections_of(&1)[0];
    let probe = manager.min_link_rtt(conn).expect("the session has probed");
    println!(
        "{label}: op round trip {} ms, probe round trip {} ms, slowest op {} ms",
        floor_to(10, ops.min_rtt().expect("twenty samples")),
        floor_to(10, probe.as_secs_f32() * 1000.0),
        floor_to(100, slowest as f32),
    );
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    let session = started.recv().expect("the server started");

    let mut client = Client::connect();
    client.next_value();

    measure("clean", &mut client, &session);
}

The session measures every connection itself. It sends a Ping frame down the connection, the client's FramePump answers with a Pong and the server times the round trip. By default it sends eight probes at 125 ms intervals and then one every five seconds. Setting a link profile clears the readings for that connection, so probe_schedule here drops the idle interval to 250 ms to get new readings soon after each change.

measure takes two readings. On the client it sends twenty increments one at a time and feeds how long each took into a fresh RttEstimator. On the server it reads min_link_rtt for the client's connection, the smallest probe round trip since the readings were last cleared. Both report the minimum, because jitter only ever adds delay and the smallest sample is the best estimate of the link itself. The program rounds the minimums down to tens of milliseconds and the slowest op down to hundreds, since the last few milliseconds depend on scheduling.

FramePump keeps an RttEstimator of its own for the probes it sends, which you can read as pump.timeline().rtt or pump.rtt_ms(). The client cannot tell when the server changes its link, so that estimator keeps its old minimum across a change. That is why measure starts a new one each time.

Run it:

$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms

On localhost both round trips take a millisecond or two.

Add delay while the program runs#

A LinkProfile has an up direction for what the client sends and a down direction for what the server sends. Each is a DirectionProfile with a delay, a jitter, a loss probability and what a loss means. LinkProfile::symmetric uses the same profile both ways. Change the plaza_session import to bring in the two types:

src/main.rs
use plaza_session::{ActixWsPlazaSession, DirectionProfile, LinkProfile, SessionOptions};

Then add this to the end of main:

src/main.rs
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile::delayed(Duration::from_millis(100))),
    );
    measure("delay", &mut client, &session);

The client is agent 1, the first connection the route numbered. set_agent_link_profile applies to every connection that agent holds and takes effect at once on a live connection. set_all_link_profiles does the same for every connection the session has, which is what a room-wide latency slider wants.

Run it:

$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms

A hundred milliseconds each way is a 200 ms round trip, which is what both measurements show. The op takes the same time as the probe because both cross the same conditioner: the increment waits 100 ms on the way up and the snapshot waits 100 ms on the way down. Everything on the connection goes through it, your ops, snapshots and the probes alike. Without rounding, the minimums land a few milliseconds above 200.

Add loss#

Add this to the end of main:

src/main.rs
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            delay: Duration::from_millis(100),
            loss: 0.05,
            ..DirectionProfile::default()
        }),
    );
    measure("loss", &mut client, &session);
    println!(
        "server: the link dropped {} frames",
        session.agent_link_dropped(&1)
    );

Each frame now has a 5% chance of being lost in each direction. What a loss means is the profile's delivery, which defaults to Delivery::Reliable.

Run it:

$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames

TCP and WebSockets are reliable streams, so a lost segment is retransmitted and never reaches the application as a missing message. Delivery::Reliable models that. A lost frame arrives RETRANSMIT_PENALTY (200 ms, TCP's minimum retransmission timeout) late and everything behind it on the connection waits, so the link dropped nothing. The fastest op still took 200 ms. The slowest took about 400, the delay plus one retransmission. Nothing went missing: measure waits for each increment's snapshot and all twenty came back.

The jitter and loss draws are seeded from the connection id instead of the clock, so the same program on the same machine loses the same frames from run to run. That is why the slowest op here came out the same in every run. With different traffic the loss lands on different frames.

Add jitter#

Jitter is the part of the delay that changes from frame to frame. A delay round trip is a single number, but under jitter every op takes a different time. The program prints a property that holds whatever the draws are instead of a number. Add this function above main:

src/main.rs
fn jittery(client: &mut Client) {
    client.idle(Duration::from_millis(500));

    let mut inside = 0;
    for _ in 0..20 {
        let sent = Instant::now();
        client.send(CounterOp::Increment(1));
        client.next_value();
        let took = sent.elapsed();
        if took >= Duration::from_millis(200) && took < Duration::from_millis(300) {
            inside += 1;
        }
    }
    println!("jitter: {inside} of 20 op round trips took 200 to 300 ms");
}

Then add this to the end of main:

src/main.rs
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            delay: Duration::from_millis(100),
            jitter: Duration::from_millis(40),
            ..DirectionProfile::default()
        }),
    );
    jittery(&mut client);

Run it:

$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames
jitter: 20 of 20 op round trips took 200 to 300 ms

Each frame gets the delay plus an extra draw between zero and the jitter, so each direction takes 100 to 140 ms and an op round trip falls between 200 and 280 ms plus a few for scheduling. The conditioner keeps frames in order, as a stream does. A frame that drew a short delay still waits behind one that drew a long one, so a jitter spike arrives as a stall and then a burst. The individual round trips differ from run to run, but every one stays inside that range.

Drop frames for real#

Delivery::Datagram makes a lost frame disappear, which is what a UDP-style transport does. Over a WebSocket it simulates a datagram link, which is useful for testing how your client recovers. The client now needs to remember the last value it saw, since some increments and some snapshots will never arrive. Add a field to Client:

src/main.rs
struct Client {
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
    last: i64,
}

Set it to last: 0, in Client::connect and record each value in Client::poll:

src/main.rs
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                            self.last = value;
                        }

Add Delivery to the plaza_session import:

src/main.rs
use plaza_session::{ActixWsPlazaSession, Delivery, DirectionProfile, LinkProfile, SessionOptions};

Then add this to the end of main:

src/main.rs
    session.set_agent_link_profile(
        &1,
        LinkProfile::symmetric(DirectionProfile {
            loss: 0.2,
            delivery: Delivery::Datagram,
            ..DirectionProfile::default()
        }),
    );
    client.idle(Duration::from_millis(500));
    let before = client.last;
    for _ in 0..50 {
        client.send(CounterOp::Increment(1));
    }
    client.idle(Duration::from_secs(1));
    println!(
        "datagram: sent 50 increments, the client saw all 50: {}",
        client.last - before == 50
    );
    println!(
        "server: the link dropped frames: {}",
        session.agent_link_dropped(&1) > 0
    );

    session.set_agent_link_profile(&1, LinkProfile::default());
    measure("passthrough", &mut client, &session);

The client fires fifty increments without waiting for each answer, because under this profile an answer may never come. The last three lines set the link back to passthrough with LinkProfile::default() and measure it again.

Run it:

$ cargo run
clean: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms
delay: op round trip 200 ms, probe round trip 200 ms, slowest op 200 ms
loss: op round trip 200 ms, probe round trip 200 ms, slowest op 400 ms
server: the link dropped 0 frames
jitter: 20 of 20 op round trips took 200 to 300 ms
datagram: sent 50 increments, the client saw all 50: false
server: the link dropped frames: true
passthrough: op round trip 0 ms, probe round trip 0 ms, slowest op 0 ms

With 20% loss each way, some increments never reach the controller and some snapshots never reach the client. No frame kind is exempt: probes and the Hello can be lost too. agent_link_dropped counts what the link threw away. Your application cannot count that itself, because a frame the link lost never reached your code. How many go missing depends on how the frames interleave, so the program prints only that some did.

A passthrough profile costs nothing: no queue, no allocation and no deadline arithmetic. After it the connection measures as clean as it did at the start.

Where to go next#

Profiles are plain data, so a debug panel only has to send new ones. LinkPublisher wraps a closure such as move |profile| session.set_all_link_profiles(profile) and skips a setting that did not change, which suits a slider that fires on every frame. For tests with no session in the middle, plaza_client_utils has LatencyLink behind its net-sim feature, a deterministic latency, jitter and loss queue that keeps order by default, as the conditioner does.

Guide chapter 31, Faking a bad network explains the conditioner and the two measurement planes. Its lab points at the playground examples, whose sliders drive the same profiles from a browser.

The plaza_session usage guide covers impairing a link, the probe schedule and the counters a session keeps. The plaza_ws usage guide covers reading the link from the pump.

The full API references are on docs.rs: plaza_session, plaza_ws and plaza_client_utils.