subetha-cxc 0.1.3

//! End-to-end demonstration of the AdaptiveRing protocol-shape
//! morph + the PinnedRing handoff to native-primitive speed.
//!
//! What this proves:
//!  1. An AdaptiveRing starts in one shape (SPSC), runs at native
//!     speed via PinnedRing, observes a peer-count change, morphs
//!     to a new shape (MPSC, then MPMC), and the pinned handle
//!     correctly sees the morph via `is_still_valid()`.
//!  2. In-flight items survive each morph (zero lost, zero
//!     duplicated; integrity asserted at the end).
//!  3. The pinned-path hot loop runs at near-native primitive
//!     speed.
//!
//! Morph triggers in this example are manual (calling
//! `ring.morph_to(...)` when a peer count crosses a threshold).
//! The sidecar policy that does this automatically is a thin
//! layer on top of the same trigger.
//!
//! Run with:
//!     cargo run --release --example adaptive_ring_morph

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
use std::time::Instant;

use subetha_cxc::adaptive_ring::ADAPTIVE_SPSC_PAYLOAD_BYTES;
use subetha_cxc::{AdaptiveRing, RingShape};

const ITEMS_PER_STAGE: u64 = 100_000;
const RING_CAPACITY: usize = 1024;

fn main() {
    println!("=== AdaptiveRing morph + pinned-handoff E2E ===");
    println!();

    let ring = Arc::new(
        AdaptiveRing::create_anon(8, 8, RING_CAPACITY).expect("ring create"),
    );
    let produced = Arc::new(AtomicU64::new(0));
    let consumed = Arc::new(AtomicU64::new(0));
    let sum_produced = Arc::new(AtomicU64::new(0));
    let sum_consumed = Arc::new(AtomicU64::new(0));
    let stop_consumers = Arc::new(AtomicBool::new(false));

    let run_start = Instant::now();

    // SPSC: 1 producer + 1 consumer.
    println!("[stage SPSC, 1P/1C] starting...");
    let t0 = Instant::now();
    let producer_a = spawn_producer(
        ring.clone(), 0, ITEMS_PER_STAGE,
        produced.clone(), sum_produced.clone(),
    );
    let consumer_a = spawn_consumer(
        ring.clone(), 0, consumed.clone(),
        sum_consumed.clone(), stop_consumers.clone(),
    );
    producer_a.join().expect("producer A");
    wait_drain(&consumed, ITEMS_PER_STAGE);
    let stage1 = t0.elapsed();
    println!(
        "[stage SPSC, 1P/1C] {} items in {:?} = {:.2} M items/s",
        ITEMS_PER_STAGE, stage1,
        ITEMS_PER_STAGE as f64 / stage1.as_secs_f64() / 1e6,
    );
    println!("   shape = {:?}, pin_generation = {}",
             ring.current_shape(), ring.pin_generation());

    // Morph SPSC -> MPSC. Add a second producer.
    println!();
    println!("[morph SPSC -> MPSC] triggered by 2nd producer registration");
    ring.morph_to(RingShape::Mpsc).expect("morph to MPSC");
    println!("   shape = {:?}, pin_generation = {} (bumped)",
             ring.current_shape(), ring.pin_generation());

    let t0 = Instant::now();
    let producer_b = spawn_producer(
        ring.clone(), 0, ITEMS_PER_STAGE,
        produced.clone(), sum_produced.clone(),
    );
    let producer_c = spawn_producer(
        ring.clone(), 1, ITEMS_PER_STAGE,
        produced.clone(), sum_produced.clone(),
    );
    producer_b.join().expect("producer B");
    producer_c.join().expect("producer C");
    wait_drain(&consumed, 3 * ITEMS_PER_STAGE);
    let stage2 = t0.elapsed();
    println!(
        "[stage MPSC, 2P/1C] {} items in {:?} = {:.2} M items/s",
        2 * ITEMS_PER_STAGE, stage2,
        (2 * ITEMS_PER_STAGE) as f64 / stage2.as_secs_f64() / 1e6,
    );

    // Morph MPSC -> MPMC. 4P/4C.
    println!();
    println!("[morph MPSC -> MPMC] triggered by 4P/4C target shape");
    stop_consumers.store(true, Ordering::Release);
    consumer_a.join().expect("consumer A");
    stop_consumers.store(false, Ordering::Release);

    ring.morph_to(RingShape::Mpmc).expect("morph to MPMC");
    println!("   shape = {:?}, pin_generation = {} (bumped again)",
             ring.current_shape(), ring.pin_generation());

    let t0 = Instant::now();
    let mut consumers = Vec::new();
    for cid in 0..4 {
        consumers.push(spawn_consumer(
            ring.clone(), cid,
            consumed.clone(), sum_consumed.clone(), stop_consumers.clone(),
        ));
    }
    let mut producers = Vec::new();
    for pid in 0..4 {
        producers.push(spawn_producer(
            ring.clone(), pid, ITEMS_PER_STAGE,
            produced.clone(), sum_produced.clone(),
        ));
    }
    for p in producers { p.join().expect("producer"); }
    wait_drain(&consumed, 3 * ITEMS_PER_STAGE + 4 * ITEMS_PER_STAGE);
    let stage3 = t0.elapsed();
    println!(
        "[stage MPMC, 4P/4C] {} items in {:?} = {:.2} M items/s",
        4 * ITEMS_PER_STAGE, stage3,
        (4 * ITEMS_PER_STAGE) as f64 / stage3.as_secs_f64() / 1e6,
    );

    // Morph MPMC -> Vyukov. Demonstrates the morph across the family.
    println!();
    println!("[morph MPMC -> Vyukov] triggered by global-FIFO requirement");
    stop_consumers.store(true, Ordering::Release);
    for c in consumers { c.join().expect("consumer"); }
    stop_consumers.store(false, Ordering::Release);

    ring.morph_to(RingShape::Vyukov).expect("morph to Vyukov");
    println!("   shape = {:?}, pin_generation = {} (bumped)",
             ring.current_shape(), ring.pin_generation());

    let t0 = Instant::now();
    let consumer_e = spawn_consumer(
        ring.clone(), 0,
        consumed.clone(), sum_consumed.clone(), stop_consumers.clone(),
    );
    let producer_e = spawn_producer(
        ring.clone(), 0, ITEMS_PER_STAGE,
        produced.clone(), sum_produced.clone(),
    );
    producer_e.join().expect("producer E");
    let target_total = 3 * ITEMS_PER_STAGE + 4 * ITEMS_PER_STAGE + ITEMS_PER_STAGE;
    wait_drain(&consumed, target_total);
    let stage4 = t0.elapsed();
    println!(
        "[stage Vyukov, 1P/1C] {} items in {:?} = {:.2} M items/s",
        ITEMS_PER_STAGE, stage4,
        ITEMS_PER_STAGE as f64 / stage4.as_secs_f64() / 1e6,
    );

    stop_consumers.store(true, Ordering::Release);
    consumer_e.join().expect("consumer E");

    // Pinned-handoff demonstration: pin the current Vyukov shape,
    // run a native hot loop, observe is_still_valid() flips after
    // a manual morph.
    println!();
    println!("[pinned-handoff demo] pin Vyukov shape, run native hot loop");

    let pin = ring.pin_current_shape();
    println!(
        "   pinned at shape={:?}, generation={}, is_still_valid={}",
        pin.shape(), ring.pin_generation(), pin.is_still_valid(),
    );

    let payload = [0xEEu8; 56];
    let pin_pushes = 1_000u64;
    let mut pinned_pushed = 0u64;
    let mut pinned_popped = 0u64;
    let mut out = [0u8; ADAPTIVE_SPSC_PAYLOAD_BYTES];
    for _ in 0..pin_pushes {
        while pin.vyukov_try_push(&payload).is_err() {
            if pin.vyukov_try_pop(&mut out).is_ok() {
                pinned_popped += 1;
            }
        }
        pinned_pushed += 1;
        if pin.vyukov_try_pop(&mut out).is_ok() {
            pinned_popped += 1;
        }
    }
    while pin.vyukov_try_pop(&mut out).is_ok() {
        pinned_popped += 1;
    }
    println!(
        "   pinned hot loop: {} pushed, {} popped, is_still_valid={}",
        pinned_pushed, pinned_popped, pin.is_still_valid(),
    );

    println!();
    println!("[morph trigger] morph Vyukov -> SPSC; pinned handle must invalidate");
    let pin_before_morph_gen = ring.pin_generation();
    ring.morph_to(RingShape::Spsc).expect("morph back to SPSC");
    assert!(!pin.is_still_valid(), "morph must invalidate the prior pin");
    assert!(ring.pin_generation() > pin_before_morph_gen);
    let new_pin = ring.pin_current_shape();
    println!(
        "   new pin: shape={:?}, generation={}, is_still_valid={}",
        new_pin.shape(), ring.pin_generation(), new_pin.is_still_valid(),
    );

    let total_produced = produced.load(Ordering::Acquire);
    let total_consumed = consumed.load(Ordering::Acquire);
    let elapsed = run_start.elapsed();

    println!();
    println!("=== Result ===");
    println!("  total elapsed:              {:?}", elapsed);
    println!("  morphs:                     SPSC -> MPSC -> MPMC -> Vyukov -> SPSC");
    println!("  stages produced via app:    {total_produced}");
    println!("  stages consumed via app:    {total_consumed}");
    println!("  pinned hot loop pushed:     {pinned_pushed}");
    println!("  pinned hot loop popped:     {pinned_popped}");
    println!();
    assert_eq!(
        total_produced, total_consumed,
        "INTEGRITY FAIL: produced != consumed across morph lifecycle",
    );
    assert_eq!(
        pinned_pushed, pinned_popped,
        "INTEGRITY FAIL: pinned hot loop produced != consumed",
    );
    assert_eq!(
        sum_produced.load(Ordering::Acquire),
        sum_consumed.load(Ordering::Acquire),
        "INTEGRITY FAIL: per-item sums diverge across morph lifecycle",
    );
    println!("  integrity:                  PASS");
    println!("    sum_produced == sum_consumed: every item arrived exactly once");
    println!("    pinned hot loop balanced");
    println!("    morphs preserved in-flight items via the stale-backing walk");
}

fn spawn_producer(
    ring: Arc<AdaptiveRing>,
    producer_id: usize,
    items: u64,
    produced: Arc<AtomicU64>,
    sum_produced: Arc<AtomicU64>,
) -> thread::JoinHandle<()> {
    thread::spawn(move || {
        let mut buf = [0u8; 56];
        for i in 0..items {
            buf[..8].copy_from_slice(&i.to_le_bytes());
            buf[8] = producer_id as u8;
            while ring.try_send(producer_id, &buf).is_err() {
                std::hint::spin_loop();
            }
            produced.fetch_add(1, Ordering::Relaxed);
            sum_produced.fetch_add(i, Ordering::Relaxed);
        }
    })
}

fn spawn_consumer(
    ring: Arc<AdaptiveRing>,
    consumer_id: usize,
    consumed: Arc<AtomicU64>,
    sum_consumed: Arc<AtomicU64>,
    stop: Arc<AtomicBool>,
) -> thread::JoinHandle<()> {
    thread::spawn(move || {
        let mut out = [0u8; ADAPTIVE_SPSC_PAYLOAD_BYTES];
        while !stop.load(Ordering::Acquire) {
            if ring.try_recv(consumer_id, &mut out).is_ok() {
                let v = u64::from_le_bytes(out[..8].try_into().unwrap());
                consumed.fetch_add(1, Ordering::Relaxed);
                sum_consumed.fetch_add(v, Ordering::Relaxed);
            } else {
                std::hint::spin_loop();
            }
        }
        while ring.try_recv(consumer_id, &mut out).is_ok() {
            let v = u64::from_le_bytes(out[..8].try_into().unwrap());
            consumed.fetch_add(1, Ordering::Relaxed);
            sum_consumed.fetch_add(v, Ordering::Relaxed);
        }
    })
}

fn wait_drain(consumed: &AtomicU64, target: u64) {
    while consumed.load(Ordering::Acquire) < target {
        thread::sleep(std::time::Duration::from_millis(1));
    }
}