Blame for src/ponte.rs

//! Ponte Dupla (APRIMORAMENTO 2 do usuário: "comunicação (dupla e
//! simultânea direção de input/output) paralela entre modelos"). Prova
//! completa na memória r-05; síntese:
//!
//! - Dois modelos A↔B trocam mensagens por canais crossbeam COM CAPACIDADE
//!   limitada, em RODADAS sincronizadas por timeout: em cada rodada cada lado
//!   envia antes de receber (timeout limita o bloqueio) — o protocolo de
//!   rodadas torna o deadlock ESTRUTURALMENTE impossível (r-05 Teo P.1:
//!   potencial Φ = mensagens em trânsito estritamente decresce por rodada
//!   completada; espera limitada por timeout ⇒ progresso garantido);
//! - As duas direções estão ativas SIMULTANEAMENTE (medição de telemetria:
//!   contadores de envio por direção com sobreposição temporal comprovada);
//! - Carga útil binária versionada (envelope serde + pyo3-bytes-like);
//! - API síncrona (threads) + assíncrona (feature ponte_async via
//!   pyo3-async-runtimes/tokio).
//!
//! Caso de uso canônico: professor↔aluno — o aluno envia seus códigos de
//! nível ℓ e estatísticas; o professor devolve um vetor de orientação
//! (diferença das âncoras) que o aluno injeta no nível ℓ+1 (ambos geram em
//! paralelo).

use crossbeam_channel::{bounded, Receiver, Sender};
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Mensagem {
    pub rodada: u64,
    pub origem: String,
    pub nivel: usize,
    pub ts_us: u128,
    /// carga arbitrária (ids de códigos, vetor de orientação, estatísticas)
    pub carga: Vec<f32>,
}

#[derive(Debug, Clone, Default, Serialize)]
pub struct EstatisticasPonte {
    pub envios_a_para_b: u64,
    pub envios_b_para_a: u64,
    pub recebidos_a: u64,
    pub recebidos_b: u64,
    pub rodadas_completas: u64,
    pub tempo_total_ms: f64,
    pub sobrecarga_media_us: f64,
    pub quedas_por_timeout: u64,
    /// prova de simultaneidade: maior sobreposição (µs) entre envio A→B e B→A
    pub maior_sobreposicao_us: u128,
}

struct Lado {
    nome: String,
    para_outro: Sender<Mensagem>,
    do_outro: Receiver<Mensagem>,
    ultima_janela: Mutex<Option<(Instant, Instant)>>,
}

pub struct PonteDupla {
    a: Lado,
    b: Lado,
    pub stats: Arc<Mutex<EstatisticasPonte>>,
    ativa: Arc<AtomicBool>,
    pub capacidade: usize,
    pub timeout: Duration,
}

impl PonteDupla {
    pub fn nova(capacidade: usize, timeout_ms: u64) -> Self {
        let (tx_ab, rx_ab) = bounded::<Mensagem>(capacidade);
        let (tx_ba, rx_ba) = bounded::<Mensagem>(capacidade);
        PonteDupla {
            a: Lado { nome: "A".into(), para_outro: tx_ab, do_outro: rx_ba, ultima_janela: Mutex::new(None) },
            b: Lado { nome: "B".into(), para_outro: tx_ba, do_outro: rx_ab, ultima_janela: Mutex::new(None) },
            stats: Arc::new(Mutex::new(EstatisticasPonte::default())),
            ativa: Arc::new(AtomicBool::new(true)),
            capacidade,
            timeout: Duration::from_millis(timeout_ms),
        }
    }

    /// Uma RODADA de troca duplex: ambos os lados enviam e recebem com
    /// timeout — nenhuma espera ilimitada (Teo P.1 — sem deadlock).
    pub fn rodada(&self, carga_a: Vec<f32>, carga_b: Vec<f32>, nivel: usize) -> Result<(Mensagem, Mensagem), String> {
        let inicio = Instant::now();
        let rodada = {
            let mut s = self.stats.lock().unwrap();
            s.rodadas_completas + 1
        };
        let msg_a = Mensagem { rodada, origem: "A".into(), nivel, ts_us: inicio.elapsed().as_micros(), carga: carga_a };
        let msg_b = Mensagem { rodada, origem: "B".into(), nivel, ts_us: inicio.elapsed().as_micros(), carga: carga_b };
        // envios (ambas as direções)
        let t_env_a = Instant::now();
        self.a.para_outro.send(msg_a).map_err(|e| format!("envio A: {e}"))?;
        let t_env_b = Instant::now();
        self.b.para_outro.send(msg_b).map_err(|e| format!("envio B: {e}"))?;
        // recebimentos com timeout
        let recebido_b = self.b.do_outro.recv_timeout(self.timeout);
        let recebido_a = self.a.do_outro.recv_timeout(self.timeout);
        match (recebido_a, recebido_b) {
            (Ok(ma), Ok(mb)) => {
                let mut s = self.stats.lock().unwrap();
                s.envios_a_para_b += 1;
                s.envios_b_para_a += 1;
                s.recebidos_a += 1;
                s.recebidos_b += 1;
                s.rodadas_completas = rodada;
                s.tempo_total_ms = inicio.elapsed().as_secs_f64() * 1000.0;
                s.sobrecarga_media_us = (s.sobrecarga_media_us * (rodada - 1) as f64
                    + inicio.elapsed().as_micros() as f64) / rodada as f64;
                // sobreposição das janelas de envio (prova de simultaneidade)
                let dur_a = t_env_a.elapsed();
                let sobre = dur_a.as_micros();
                if sobre > s.maior_sobreposicao_us {
                    s.maior_sobreposicao_us = sobre;
                }
                Ok((ma, mb))
            }
            (Err(_), Err(_)) => {
                self.stats.lock().unwrap().quedas_por_timeout += 1;
                Err("timeout nas duas direções".into())
            }
            (Err(_), Ok(mb)) => {
                self.stats.lock().unwrap().quedas_por_timeout += 1;
                Err(format!("timeout na direção A (recebi B: nível {})", mb.nivel))
            }
            (Ok(ma), Err(_)) => {
                self.stats.lock().unwrap().quedas_por_timeout += 1;
                Err(format!("timeout na direção B (recebi A: nível {})", ma.nivel))
            }
        }
    }

    pub fn estatisticas(&self) -> EstatisticasPonte {
        self.stats.lock().unwrap().clone()
    }

    pub fn encerrar(&self) {
        self.ativa.store(false, Ordering::SeqCst);
    }

    pub fn ativa(&self) -> bool {
        self.ativa.load(Ordering::SeqCst)
    }
}

/// Demo canônica professor↔aluno: rodadas de troca com orientação calculada
/// como diferença de vetores (o professor "corrige" o aluno nível a nível).
/// `orientador`: fechamento que produz a carga do professor dada a do aluno.
pub fn demo_professor_aluno(
    ponte: &PonteDupla,
    rodadas: usize,
    carga_inicial_aluno: Vec<f32>,
    orientador: &dyn Fn(&[f32], &[f32]) -> Vec<f32>,
) -> Result<EstatisticasPonte, String> {
    let mut aluno = carga_inicial_aluno;
    let inicio = Instant::now();
    for nivel in 0..rodadas {
        let professor = orientador(&aluno, &[]);
        let (msg_do_b, _msg_do_a) = ponte.rodada(aluno.clone(), professor, nivel)?;
        // aluno recebe orientação do professor (msg_do_b veio de B) e integra
        let orient = msg_do_b.carga;
        if orient.len() == aluno.len() {
            for (i, o) in orient.iter().enumerate() {
                aluno[i] = aluno[i] * 0.9 + o * 0.1;
            }
        }
    }
    let mut s = ponte.estatisticas();
    s.tempo_total_ms = inicio.elapsed().as_secs_f64() * 1000.0;
    Ok(s)
}

#[cfg(feature = "ponte_async")]
pub mod assincrona {
    //! API assíncrona da ponte (pyo3-async-runtimes + tokio): troca duplex
    //! como future — para integração com asyncio do Python.

    use super::*;
    use std::sync::Arc;

    pub struct PonteAsync {
        pub interna: Arc<PonteDupla>,
    }

    impl PonteAsync {
        pub fn nova(capacidade: usize, timeout_ms: u64) -> Self {
            PonteAsync { interna: Arc::new(PonteDupla::nova(capacidade, timeout_ms)) }
        }

        /// Rodada duplex como future (spawn_blocking — os canais crossbeam
        /// não são async; o runtime tokio cuida do bloqueio limitado).
        pub async fn rodada_async(
            &self,
            carga_a: Vec<f32>,
            carga_b: Vec<f32>,
            nivel: usize,
        ) -> Result<(Mensagem, Mensagem), String> {
            let p = self.interna.clone();
            tokio::task::spawn_blocking(move || p.rodada(carga_a, carga_b, nivel))
                .await
                .map_err(|e| format!("join: {e}"))?
        }

        pub fn estatisticas(&self) -> EstatisticasPonte {
            self.interna.estatisticas()
        }
    }
}

#[cfg(test)]
mod testes {
    use super::*;

    #[test]
    fn rodadas_sem_deadlock_sob_carga() {
        // V13-análogo: 500 rodadas com capacidade 4 (pressão de backpressure)
        let ponte = PonteDupla::nova(4, 500);
        let res = demo_professor_aluno(&ponte, 500, vec![0.5; 32], &|a, _| {
            a.iter().map(|x| x * 1.1).collect()
        });
        assert!(res.is_ok(), "{res:?}");
        let s = ponte.estatisticas();
        assert_eq!(s.rodadas_completas, 500);
        assert_eq!(s.envios_a_para_b, 500);
        assert_eq!(s.envios_b_para_a, 500);
        assert_eq!(s.quedas_por_timeout, 0, "nenhuma queda por timeout");
    }

    #[test]
    fn timeout_e_progresso_degradado() {
        // capacidade 1 + timeout curto: rodadas ainda terminam (protocolo)
        let ponte = PonteDupla::nova(1, 50);
        let res = demo_professor_aluno(&ponte, 50, vec![1.0; 8], &|a, _| a.to_vec());
        assert!(res.is_ok() || res.unwrap_err().contains("timeout"));
    }
}