3.3 Kahn 程序網路：Fluxa 的時間模型#

void fir(int *x, int *y) {
#pragma HLS PIPELINE
    for (int i = 0; i < N; i++) {
        acc += x[i] * COEF[i];
    }
}

uN  ≡ int<0..=2^N-1>          iN ≡ int<-2^(N-1)..=2^(N-1)-1>

(T-ADD)  int<a..=b> + int<c..=d>  :  int<a+c..=b+d>
(T-SUB)  e1 - e2                    :  int<a-d..=b-c>
(T-MUL)  e1 * e2                    :  int<min P..=max P>,  P={ac,ad,bc,bd}
(T-SHL)  e << k                    :  int<a·2^k..=b·2^k>
(T-WRAP) e1 +% e2                  :  同 e1 的型別（明確的環繞）

(T-FADD) fixed<I1,F1> + fixed<I2,F2> : fixed<max(I1,I2)+1, max(F1,F2)>
(T-FMUL) fixed<I1,F1> * fixed<I2,F2> : fixed<I1+I2, F1+F2>

task Fir(x: in chan<i16>, y: out chan<i16>) {
    let mut hist: [i16; TAPS - 1] @reg = [0; TAPS - 1];

    pipeline(ii = 1) for s in x {
        let win: [i16; TAPS] = concat([s], hist);
        let acc = (0..TAPS).unrolled()
                           .map(|k| win[k] * COEF[k])
                           .tree_sum();
        hist = win[0..TAPS - 1];
        y.send(acc.to::<i16>(round = nearest, ovf = sat));
    }
}

/// 線性所有權轉移：端點必須由 from 擁有，轉給 to
pub fn transfer(&self, v: &Value, from: usize, to: usize, line: u32) -> R<()> {
    match v {
        Value::Tx(ch) | Value::Rx(ch) => {
            let owner = if is_tx { &mut w.chans[*ch].tx_owner } else { &mut w.chans[*ch].rx_owner };
            if *owner != from {
                return msg(line, format!("{} endpoint of channel `{}` was already moved (channel endpoints are linear)", ...));
            }
            *owner = to; Ok(())
        }
        ...
    }
}

/// 結束任務時關閉其仍擁有的 Tx 端點（等同硬體的 end-of-stream）
pub fn close_owned_tx(&self, task: usize, v: &Value) { ... }

// 執行期也拒絕非阻塞操作
(Value::Rx(_), "try_recv") | (Value::Rx(_), "empty") | (Value::Rx(_), "len") => {
    return self.err(format!(
        "`{}` is non-deterministic and is only allowed in `stage` code (task-level programs must use blocking recv)", name));
}

task Pe(a_in: in chan<i8>, b_in: in chan<i8>, a_out: out chan<i8>, b_out: out chan<i8>,
        k: u16, c_out: out chan<i32>) {
    carry acc: i32 = 0;
    pipeline(ii = 1) for _ in 0..k {
        let a = a_in.recv();
        let b = b_in.recv();
        acc += a * b;
        a_out.send(a);
        b_out.send(b);
    }
    c_out.send(acc);
}

pub struct Var {
    pub val: Value,
    pub ty: Option<RtType>,
    /// `@reg`：迭代內讀到舊值，寫入在迭代結束提交
    pub reg: bool,
    pub pending: Option<Value>,
}

fn commit_regs(&mut self) {
    for sc in self.scopes.iter_mut() {
        for v in sc.values_mut() { if let Some(p) = v.pending.take() { v.val = p; } }
    }
}

carry acc: i32 = 0;

// fluxa/synth/src/ty.rs 的 mul()
// 寬度照值域推導，但 carry 變數是「使用者指定」的型別，
// 所以合成時要檢查累加的結果是否放得下

pipeline(ii = 1) for s in x {
    ...
}

PipeOp::Recv(d, c) => {
    if !sends.is_empty() {
        return Err(format!("`{}` is received after a send in the same iteration", ...));
    }
    if !seen.insert(c.clone()) {
        return Err(format!("`{}` is received twice in an iteration (one token a clock): use ii = 2", ...));
    }
}
PipeOp::Send(c, v) => {
    if sends.iter().any(|(x,_)| *x == c) {
        return Err(format!("`{}` is sent twice in an iteration (one token a clock): use ii = 2", ...));
    }
}

// 若流水化形式不可行但 ii > 1 被要求，降階會回滾
// （記錄 mark = (blocks.len(), temps.len(), n)）並退回一般的 FSM 迴圈
// 若 ii == 1，錯誤就變成：
Err(e) => Err(format!("line {}: error[E0412]: pipeline(ii = {}) cannot be met: {}", line, ii, e));

// 每個 task 一條執行緒，8 MB 堆疊
std::thread::Builder::new().stack_size(8 * 1024 * 1024)

w.blocked += 1;
let my_epoch = w.epoch;
if w.blocked >= w.live {
    w.deadlock = true;
    // 產生報告，每個被阻塞的 task 一行
    w.deadlock_report = Some(format!("deadlock: all {} live task(s) are blocked\n{}", w.live, lines.join("\n")));
    return Err(RtErr::Deadlock);
}

task A: recv(x)      // 阻塞，blocked = 1
task B: send(x)      // 這個 send 喚醒 A

struct YieldOnce(bool);
impl Future for YieldOnce {
    fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
        if self.0 { Poll::Ready(()) } else { self.0 = true; Poll::Pending }
    }
}

pub fn run_coop<'a, T>(&self, mut main: Pin<Box<dyn Future<Output = R<T>> + 'a>>) -> R<T> {
    loop {
        let epoch = self.lock().epoch;
        if let Poll::Ready(r) = main.as_mut().poll(&mut cx) { return r; }
        // 輪詢每個 task 一次；新 spawn 的 task 加入佇列
        let mut w = self.lock();
        if w.epoch == epoch {
            /* 整整一輪都沒有任何進展 → 死結 */
        }
    }
}

pub enum Value {
    Unit, Bool(bool), Int(i128), Float(f64),
    Fixed(i128, u32),   // (raw, fraction bits)；值 = raw / 2^frac
    Str(String), Array(Vec<Value>), Tuple(Vec<Value>),
    Range(i128, i128),  // 半開 [lo,hi)
    Tx(usize), Rx(usize)   // 通道端點 = 通道索引
}
pub const MAX_WIDTH: u32 = 120;

// 窄化固定點值（有小數位）會被拒絕：
// "fractional bits would be lost"

ExprKind::Method(recv, "recv") => {
    return match self.sh.recv(self.task, *ch, line).await? {
        Some(v) => Ok(v),
        None => self.err("recv() on a closed, empty channel"),
    };
}

/// 一個 boxed future：打破非同步直譯器的遞迴（eval -> ... -> eval）
pub type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

pub fn eval<'a>(&'a mut self, e: &'a Expr) -> BoxFut<'a, R<Value>> { Box::pin(self.eval_impl(e)) }
fn exec_body<'a>(&'a mut self, b: &'a Block) -> BoxFut<'a, R<Value>> { Box::pin(self.exec_body_impl(b)) }

fn tree_sum(mut items: Vec<Value>) -> Result<Value, String> {
    while items.len() > 1 {
        let mut next = Vec::with_capacity(items.len() / 2 + 1);
        for ch in items.chunks(2) {
            if ch.len() == 2 { next.push(binop(BinOp::Add, &ch[0], &ch[1])?); }
            else { next.push(ch[0].clone()); }
        }
        items = next;
    }
    Ok(items.remove(0))
}