esp-p4-eth/examples/embassy_tcp_soak.rs

#![no_std]
#![no_main]

//! Long-duration TCP soak harness.
//!
//! Four parallel TCP echo listeners on ports 7780-7783, each backed by a
//! 4 KB RX + 4 KB TX socket buffer (2× the per-connection window of
//! `embassy_tcp_echo_irq`, 4× the parallelism). Per-port byte counters
//! plus full diagnostic snapshot every 60 seconds, hourly summary line.
//!
//! Designed for backbone reliability validation — leaks, ring-wedge,
//! cumulative cache-call drift, link-state churn — not for peak throughput.
//! The 192 KB on-chip SRAM does not fit four 32 KB sockets; if you need
//! BDP-saturating windows on a single connection, drop the listener count
//! to one and bump the buffer sizes locally.
//!
//! Run from host:
//!
//! ```bash
//! # Open four parallel iperf-style streams via socat (or use the soak
//! # driver script under `examples/dev/soak_driver.sh`):
//! for p in 7780 7781 7782 7783; do
//!     socat -u FILE:/dev/urandom TCP:192.168.0.50:$p &
//! done
//! ```
//!
//! Build:
//!   cargo build --no-default-features \
//!       --features p4-example,p4-time-driver-irq,embassy-net-tcp \
//!       --example embassy_tcp_soak

use core::fmt::{self, Write};
use core::hint::spin_loop;
use core::ptr::{read_volatile, write_volatile};
use core::sync::atomic::{AtomicU32, Ordering};

use embassy_executor::{Executor, Spawner};
use embassy_net::tcp::TcpSocket;
use embassy_net::{
    Config, IpListenEndpoint, Ipv4Address, Ipv4Cidr, Runner as NetRunner, Stack, StackResources,
    StaticConfigV4,
};
use embassy_time::{Duration, Timer};
use panic_halt as _;
use riscv_rt::entry;
use esp_p4_eth::{
    diag::{
        CACHE_INV_CALLS, CACHE_INV_TICKS, CACHE_WB_CALLS, CACHE_WB_TICKS, RX_ERROR_FRAMES_TOTAL,
        RX_FRAMES, RX_OVERSIZED_FRAMES_TOTAL, RX_RUNT_FRAMES_TOTAL, TX_FRAMES,
    },
    ethernet_task, new_from_static_resources as emac_new, release_waveshare_phy_reset, systimer,
    time_driver_irq, Device, StaticDmaResources,
};
use static_cell::{ConstStaticCell, StaticCell};

const UART0_BASE: usize = 0x500C_A000;
const UART_FIFO_REG: *mut u32 = UART0_BASE as *mut u32;
const UART_STATUS_REG: *const u32 = (UART0_BASE + 0x1C) as *const u32;
const UART_TXFIFO_CNT_SHIFT: u32 = 16;
const UART_TXFIFO_CNT_MASK: u32 = 0xFF << UART_TXFIFO_CNT_SHIFT;
const UART_TXFIFO_CAPACITY: u32 = 128;

const MAC_ADDR: [u8; 6] = [0xE8, 0xF6, 0x0A, 0xE0, 0x93, 0xF5];
const NET_SEED: u64 = 0x504F_414B_5350_4F4B; // "POAKSPOK"

const SELF_IP: Ipv4Address = Ipv4Address::new(192, 168, 0, 50);
const SELF_PREFIX: u8 = 24;
const GATEWAY: Ipv4Address = Ipv4Address::new(192, 168, 0, 1);

const SOAK_PORTS: [u16; 4] = [7780, 7781, 7782, 7783];
// 4 KB per direction × 4 listeners = 32 KB CPU-side buffers, the largest
// footprint that fits inside the 192 KB on-chip SRAM next to the existing
// driver code, ring storage, NetStackResources, and stack. Bump only after
// shrinking listener count or trimming the static_cell pools.
const TCP_BUF_SIZE: usize = 4 * 1024;
const STAT_PERIOD_S: u64 = 60;
const HOURLY_TICKS: u64 = 60;

static EXECUTOR: StaticCell<Executor> = StaticCell::new();
#[link_section = ".dma_bss"]
static DMA_RESOURCES: ConstStaticCell<StaticDmaResources> =
    ConstStaticCell::new(StaticDmaResources::new());
static NET_RESOURCES: ConstStaticCell<StackResources<6>> =
    ConstStaticCell::new(StackResources::<6>::new());

// One TCP socket buffer pair per listener. Plain RAM (CPU-side), not DMA.
static TCP_RX_BUF_0: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_TX_BUF_0: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_RX_BUF_1: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_TX_BUF_1: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_RX_BUF_2: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_TX_BUF_2: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_RX_BUF_3: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);
static TCP_TX_BUF_3: ConstStaticCell<[u8; TCP_BUF_SIZE]> = ConstStaticCell::new([0; TCP_BUF_SIZE]);

// Per-port observability. AtomicU32 (RISC-V 32-bit has no native AtomicU64)
// rolls over every ~4 GB; at saturating 100 Mbps line rate that is ~5.7 min,
// so the printed cumulative total is informational only. The per-tick delta
// (60 s × 12.5 MB/s = 750 MB max) always fits in u32 and the stat task uses
// `wrapping_sub` so the throughput line stays correct across rollovers.
static BYTES_ECHOED: [AtomicU32; 4] = [
    AtomicU32::new(0),
    AtomicU32::new(0),
    AtomicU32::new(0),
    AtomicU32::new(0),
];
static CONN_COUNT: [AtomicU32; 4] = [
    AtomicU32::new(0),
    AtomicU32::new(0),
    AtomicU32::new(0),
    AtomicU32::new(0),
];

struct Uart0;

impl Uart0 {
    fn write_byte(&mut self, byte: u8) {
        while txfifo_count() >= UART_TXFIFO_CAPACITY {
            spin_loop();
        }
        unsafe {
            write_volatile(UART_FIFO_REG, byte as u32);
        }
    }
}

impl Write for Uart0 {
    fn write_str(&mut self, s: &str) -> fmt::Result {
        for byte in s.bytes() {
            if byte == b'\n' {
                self.write_byte(b'\r');
            }
            self.write_byte(byte);
        }
        Ok(())
    }
}

fn txfifo_count() -> u32 {
    let status = unsafe { read_volatile(UART_STATUS_REG) };
    (status & UART_TXFIFO_CNT_MASK) >> UART_TXFIFO_CNT_SHIFT
}

#[embassy_executor::task]
async fn net_task(mut runner: NetRunner<'static, Device<'static>>) -> ! {
    runner.run().await
}

#[embassy_executor::task]
async fn dma_recovery_task() -> ! {
    loop {
        Timer::after(Duration::from_millis(50)).await;
        let dma_status = unsafe { read_volatile(0x5009_9014 as *const u32) };
        let rps = (dma_status >> 17) & 0b111;
        if rps == 0 || rps == 4 || dma_status & (1 << 7) != 0 {
            unsafe {
                write_volatile(0x5009_9014 as *mut u32, dma_status & 0x0001E7FF);
                write_volatile(0x5009_9008 as *mut u32, 1);
            }
        }
    }
}

/// Waits for link-up and applies the post-bringup MAC pokes once, then
/// returns. Listener tasks call this in parallel; idempotent register
/// writes mean the four serialized races are safe.
async fn finish_emac_bringup(stack: Stack<'static>, idx: usize, uart: &mut Uart0) {
    if idx == 0 {
        let t0 = systimer::now_us();
        while !stack.is_link_up() {
            Timer::after(Duration::from_millis(100)).await;
        }
        let _ = writeln!(
            uart,
            "soak: link up after {} ms",
            (systimer::now_us() - t0) / 1000
        );

        unsafe {
            write_volatile(0x5009_8004 as *mut u32, 0x80000011);
            let dma_op_mode = 0x5009_9018 as *mut u32;
            write_volatile(dma_op_mode, read_volatile(dma_op_mode) | (1 << 6));
            let mac_cfg = 0x5009_8000 as *mut u32;
            write_volatile(mac_cfg, read_volatile(mac_cfg) & !(1 << 25));
        }
        let _ = writeln!(uart, "soak: filter+FUF+ACS-clear applied");
    } else {
        while !stack.is_link_up() {
            Timer::after(Duration::from_millis(100)).await;
        }
    }
}

#[embassy_executor::task(pool_size = 4)]
async fn tcp_echo_task(
    stack: Stack<'static>,
    idx: usize,
    port: u16,
    rx_buf: &'static mut [u8],
    tx_buf: &'static mut [u8],
) -> ! {
    let mut uart = Uart0;
    finish_emac_bringup(stack, idx, &mut uart).await;

    let mut socket = TcpSocket::new(stack, rx_buf, tx_buf);
    socket.set_timeout(Some(Duration::from_secs(120)));
    socket.set_keep_alive(Some(Duration::from_secs(30)));

    let _ = writeln!(uart, "soak[{}]: listening on :{}", idx, port);

    let mut buf = [0u8; 2048];
    loop {
        match socket
            .accept(IpListenEndpoint { addr: None, port })
            .await
        {
            Ok(()) => {
                CONN_COUNT[idx].fetch_add(1, Ordering::Relaxed);
            }
            Err(e) => {
                let _ = writeln!(uart, "soak[{}]: accept err {:?}", idx, e);
                socket.abort();
                let _ = socket.flush().await;
                continue;
            }
        }

        loop {
            match socket.read(&mut buf).await {
                Ok(0) => break,
                Ok(n) => {
                    // Drain the slice across potentially-partial writes so the
                    // soak run cannot lose bytes when the TX window is small.
                    let mut wrote = 0usize;
                    let mut write_err = false;
                    while wrote < n {
                        match socket.write(&buf[wrote..n]).await {
                            Ok(0) => {
                                write_err = true;
                                break;
                            }
                            Ok(w) => wrote += w,
                            Err(e) => {
                                let _ = writeln!(uart, "soak[{}]: write err {:?}", idx, e);
                                write_err = true;
                                break;
                            }
                        }
                    }
                    if write_err {
                        break;
                    }
                    BYTES_ECHOED[idx].fetch_add(n as u32, Ordering::Relaxed);
                }
                Err(e) => {
                    let _ = writeln!(uart, "soak[{}]: read err {:?}", idx, e);
                    break;
                }
            }
        }
        socket.close();
        let _ = socket.flush().await;
        socket.abort();
        let _ = socket.flush().await;
    }
}

#[embassy_executor::task]
async fn stat_task() -> ! {
    let mut uart = Uart0;
    let mut tick = 0u64;
    let mut prev_bytes = [0u32; 4];
    let mut prev_rx = 0u32;
    let mut prev_tx = 0u32;

    loop {
        Timer::after(Duration::from_secs(STAT_PERIOD_S)).await;
        tick += 1;
        let uptime_s = tick * STAT_PERIOD_S;

        let bytes: [u32; 4] = [
            BYTES_ECHOED[0].load(Ordering::Relaxed),
            BYTES_ECHOED[1].load(Ordering::Relaxed),
            BYTES_ECHOED[2].load(Ordering::Relaxed),
            BYTES_ECHOED[3].load(Ordering::Relaxed),
        ];
        let conns: [u32; 4] = [
            CONN_COUNT[0].load(Ordering::Relaxed),
            CONN_COUNT[1].load(Ordering::Relaxed),
            CONN_COUNT[2].load(Ordering::Relaxed),
            CONN_COUNT[3].load(Ordering::Relaxed),
        ];
        let rx_total = RX_FRAMES.load(Ordering::Relaxed);
        let tx_total = TX_FRAMES.load(Ordering::Relaxed);

        // wrapping_sub keeps deltas correct across u32 rollover (every ~4 GB
        // / ~6 min at line rate).
        let delta_bytes: u64 = bytes
            .iter()
            .zip(prev_bytes.iter())
            .map(|(b, p)| b.wrapping_sub(*p) as u64)
            .sum();
        let throughput_bps = delta_bytes * 8 / STAT_PERIOD_S;

        let _ = writeln!(
            uart,
            "[t={}m] thrpt={} kbps  rx_frm={} (+{})  tx_frm={} (+{})",
            uptime_s / 60,
            throughput_bps / 1000,
            rx_total,
            rx_total.wrapping_sub(prev_rx),
            tx_total,
            tx_total.wrapping_sub(prev_tx),
        );
        for i in 0..4 {
            let _ = writeln!(
                uart,
                "  port {}: bytes={} (+{}) conns={}",
                SOAK_PORTS[i],
                bytes[i],
                bytes[i].wrapping_sub(prev_bytes[i]),
                conns[i],
            );
        }
        let runt = RX_RUNT_FRAMES_TOTAL.load(Ordering::Relaxed);
        let oversized = RX_OVERSIZED_FRAMES_TOTAL.load(Ordering::Relaxed);
        let err = RX_ERROR_FRAMES_TOTAL.load(Ordering::Relaxed);
        if runt != 0 || oversized != 0 || err != 0 {
            let _ = writeln!(
                uart,
                "  ERRORS: runt={} oversized={} err={}",
                runt, oversized, err
            );
        }

        if tick % HOURLY_TICKS == 0 {
            let inv_calls = CACHE_INV_CALLS.load(Ordering::Relaxed);
            let inv_ticks = CACHE_INV_TICKS.load(Ordering::Relaxed);
            let wb_calls = CACHE_WB_CALLS.load(Ordering::Relaxed);
            let wb_ticks = CACHE_WB_TICKS.load(Ordering::Relaxed);
            let _ = writeln!(
                uart,
                "=== HOURLY @ t={}h: rx={} tx={} cache_inv={}/{} cache_wb={}/{} ===",
                tick / HOURLY_TICKS,
                rx_total,
                tx_total,
                inv_calls,
                inv_ticks,
                wb_calls,
                wb_ticks,
            );
        }

        prev_bytes = bytes;
        prev_rx = rx_total;
        prev_tx = tx_total;
    }
}

#[entry]
fn main() -> ! {
    let mut uart = Uart0;
    let _ = writeln!(uart, "\n=== embassy_tcp_soak alive ===");

    time_driver_irq::init();
    let _ = writeln!(uart, "main: time_driver_irq::init() done");

    release_waveshare_phy_reset();
    for _ in 0..5_000_000 {
        spin_loop();
    }
    let _ = writeln!(uart, "main: PHY reset deasserted, waited 5M cycles");

    let executor = EXECUTOR.init(Executor::new());
    executor.run(init);
}

fn init(spawner: Spawner) {
    let mut uart = Uart0;
    let _ = writeln!(uart, "init: enter");
    let dma_resources = DMA_RESOURCES.take();
    let _ = writeln!(uart, "init: DMA_RESOURCES.take() ok");
    let (device, emac_runner, eth) = emac_new(MAC_ADDR, dma_resources);
    let _ = writeln!(uart, "init: emac_new returned");

    let config = Config::ipv4_static(StaticConfigV4 {
        address: Ipv4Cidr::new(SELF_IP, SELF_PREFIX),
        gateway: Some(GATEWAY),
        dns_servers: heapless::Vec::new(),
    });
    let resources = NET_RESOURCES.take();
    let _ = writeln!(uart, "init: NET_RESOURCES.take() ok");
    let (stack, net_runner) = embassy_net::new(device, config, resources, NET_SEED);
    let _ = writeln!(uart, "init: embassy_net::new returned");

    spawner.spawn(ethernet_task(emac_runner, eth)).unwrap();
    spawner.spawn(net_task(net_runner)).unwrap();
    spawner.spawn(dma_recovery_task()).unwrap();

    spawner
        .spawn(tcp_echo_task(
            stack,
            0,
            SOAK_PORTS[0],
            TCP_RX_BUF_0.take(),
            TCP_TX_BUF_0.take(),
        ))
        .unwrap();
    spawner
        .spawn(tcp_echo_task(
            stack,
            1,
            SOAK_PORTS[1],
            TCP_RX_BUF_1.take(),
            TCP_TX_BUF_1.take(),
        ))
        .unwrap();
    spawner
        .spawn(tcp_echo_task(
            stack,
            2,
            SOAK_PORTS[2],
            TCP_RX_BUF_2.take(),
            TCP_TX_BUF_2.take(),
        ))
        .unwrap();
    spawner
        .spawn(tcp_echo_task(
            stack,
            3,
            SOAK_PORTS[3],
            TCP_RX_BUF_3.take(),
            TCP_TX_BUF_3.take(),
        ))
        .unwrap();

    spawner.spawn(stat_task()).unwrap();
    let _ = writeln!(uart, "init: 4 listeners + stat_task spawned");
}