better synchronization

better deadlock detection to also track wait/sync
2026-08-24 11:58:04 +02:00
parent d634484df2
commit 336f0b506e
20 changed files with 1123 additions and 329 deletions
@@ -886,11 +886,21 @@ pub fn recv(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus>
    Ok(InstructionStatus::Receiving(data))
 }

+#[inline(never)]
+pub fn isa_wait(functor: usize) -> bool {
+    (wait as *const () as usize) == functor
+}
+
 #[inline(never)]
 pub fn wait(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
    Ok(InstructionStatus::Waiting(data))
 }

+#[inline(never)]
+pub fn isa_sync(functor: usize) -> bool {
+    (sync as *const () as usize) == functor
+}
+
 #[inline(never)]
 pub fn sync(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
    Ok(InstructionStatus::Sync(data))
@@ -14,7 +14,7 @@ use crate::{
    cpu::CPU,
    instruction_set::{
        Instruction, InstructionStatus, Instructions,
-        isa::{NAMES, functor_to_name, isa_recv, isa_send},
+        isa::{NAMES, functor_to_name, isa_recv, isa_send, isa_sync, isa_wait},
    },
    memory_manager::type_traits::TryToUsize,
    send_recv::{SendRecv, handle_send_recv},
@@ -104,7 +104,13 @@ struct DeadlockInfo {
    states: String,
 }

-type SyncEvents = Vec<[i32; 32]>;
+#[derive(Debug, Clone, Default)]
+struct SyncEvent {
+    count: i32,
+    sources: HashMap<i32, i32>,
+}
+
+type SyncEvents = Vec<[SyncEvent; 32]>;

 fn print_status(core_instructions: &[CoreInstructions]) {
    let mut tot_instructions = 0;
@@ -182,7 +188,9 @@ impl<'a> Executable<'a> {
        } = self;
        let mut cpu_progressed = 0;
        let max_core = cpu.num_core();
-        let mut sync_events: SyncEvents = vec![[0; 32]; max_core];
+        let mut sync_events: SyncEvents = (0..max_core)
+            .map(|_| std::array::from_fn(|_| SyncEvent::default()))
+            .collect();
        let mut cpu_index = 0;
        let mut now = SystemTime::now();

@@ -222,9 +230,11 @@ impl<'a> Executable<'a> {
                }
                if (now.elapsed().unwrap() > Duration::from_secs(5)) {
                    print_status(cores_instructions);
-                    if let Some(deadlock) = detect_deadlock(cores_instructions) {
+                    if let Some(deadlock) =
+                        detect_deadlock(cores_instructions, &sync_events, batch_size)
+                    {
                        bail!(
-                            "Deadlock cycle detected: {} [{}]",
+                            "Communication deadlock detected: {} [{}]",
                            deadlock.cycle,
                            deadlock.states
                        );
@@ -255,9 +265,9 @@ impl<'a> Executable<'a> {
        }
        print_status(cores_instructions);

-        if let Some(deadlock) = detect_deadlock(cores_instructions) {
+        if let Some(deadlock) = detect_deadlock(cores_instructions, &sync_events, batch_size) {
            bail!(
-                "Deadlock cycle detected: {} [{}]",
+                "Communication deadlock detected: {} [{}]",
                deadlock.cycle,
                deadlock.states
            );
@@ -316,18 +326,23 @@ fn store_input(cpu: &mut CPU, input: &[u8], input_regions: &[(usize, usize)]) ->
    Ok(())
 }

-fn detect_deadlock(cores_instructions: &[CoreInstructions]) -> Option<DeadlockInfo> {
+fn detect_deadlock(
+    cores_instructions: &[CoreInstructions],
+    events: &SyncEvents,
+    batch_size: u32,
+) -> Option<DeadlockInfo> {
    #[derive(Debug, PartialEq, Eq)]
    enum CoreState {
        SendingTo(i32, i32),
        ReceivingFrom(i32, i32),
+        WaitingEvent(i32, i32, i32),
        Working,
        Halted,
    }

    let mut states = HashMap::new();

-    for core_inst in cores_instructions.iter() {
+    for (core, core_inst) in cores_instructions.iter().enumerate() {
        if core_inst.program_counter >= core_inst.instructions.len() {
            continue;
        }
@@ -344,94 +359,191 @@ fn detect_deadlock(cores_instructions: &[CoreInstructions]) -> Option<DeadlockIn
            );
        } else if isa_send(functor_address) {
            states.insert(this_core, CoreState::SendingTo(target_core, data.imm_len()));
+        } else if isa_wait(functor_address) {
+            let register = data.offset_select();
+            states.insert(
+                this_core,
+                CoreState::WaitingEvent(
+                    register,
+                    data.offset_value(),
+                    events[core][register as usize].count,
+                ),
+            );
        } else {
            states.insert(this_core, CoreState::Working);
        }
    }

-    let mut wait_for = HashMap::new();
+    let event_sources = |target: i32, register: i32| {
+        let mut sources = Vec::new();
+        let mut signal_count = 0usize;
+        let event = &events[target as usize][register as usize];
+        for core_inst in cores_instructions {
+            let matches = |instruction: &&Instruction| {
+                isa_sync(instruction.functor as usize)
+                    && instruction.data.get_core_immcore().1 == target
+                    && instruction.data.offset_select() == register
+            };
+            let remaining = core_inst.instructions[core_inst.program_counter..]
+                .iter()
+                .filter(matches)
+                .count();
+            let per_iteration = core_inst.instructions.iter().filter(matches).count();
+            if per_iteration == 0 {
+                continue;
+            }
+            let future_iterations = batch_size.saturating_sub(core_inst.current_iteration + 1);
+            let source = core_inst.instructions.iter().find(|instruction| {
+                isa_sync(instruction.functor as usize)
+                    && instruction.data.get_core_immcore().1 == target
+                    && instruction.data.offset_select() == register
+            });
+            let source = source.unwrap().data.get_core_immcore().0;
+            let contributed = event.sources.get(&source).copied().unwrap_or(0) as usize;
+            let needed = per_iteration.saturating_sub(contributed);
+            let count = (remaining + per_iteration * future_iterations as usize).min(needed);
+            if count != 0 {
+                sources.push(source);
+                signal_count += count;
+            }
+        }
+        sources.sort_unstable();
+        sources.dedup();
+        (sources, signal_count)
+    };
+
+    let format_state = |core: &i32| {
+        let position = cores_instructions.get(*core as usize);
+        let location = position.map_or_else(
+            || format!("core {}", core - 1),
+            |instructions| {
+                format!(
+                    "core {} iteration {} pc {}",
+                    core - 1,
+                    instructions.current_iteration,
+                    instructions.program_counter
+                )
+            },
+        );
+        match states.get(core).unwrap_or(&CoreState::Halted) {
+            CoreState::SendingTo(target, size) => {
+                format!("{location} send {}B -> {}", size, target - 1)
+            }
+            CoreState::ReceivingFrom(source, size) => {
+                format!("{location} recv {}B <- {}", size, source - 1)
+            }
+            CoreState::WaitingEvent(register, expected, observed) => {
+                format!("{location} wait event {register} == {expected} (observed {observed})")
+            }
+            CoreState::Working => format!("{location} working"),
+            CoreState::Halted => format!("{location} halted"),
+        }
+    };
+
+    let mut wait_for: HashMap<i32, Vec<i32>> = HashMap::new();

    for (&core_id, state) in states.iter() {
        match state {
            CoreState::SendingTo(target_core, size) => {
                let target_state = states.get(target_core).unwrap_or(&CoreState::Halted);
                if target_state != &CoreState::ReceivingFrom(core_id, *size) {
-                    wait_for.insert(core_id, *target_core);
+                    wait_for.insert(core_id, vec![*target_core]);
                }
            }
            CoreState::ReceivingFrom(target_core, size) => {
                let target_state = states.get(target_core).unwrap_or(&CoreState::Halted);
                if target_state != &CoreState::SendingTo(core_id, *size) {
-                    wait_for.insert(core_id, *target_core);
+                    wait_for.insert(core_id, vec![*target_core]);
                }
            }
+            CoreState::WaitingEvent(register, expected, observed) => {
+                if observed > expected {
+                    return Some(DeadlockInfo {
+                        cycle: format!(
+                            "core {} WAIT event {} overshot exact value {} with {}",
+                            core_id - 1,
+                            register,
+                            expected,
+                            observed
+                        ),
+                        states: format_state(&core_id),
+                    });
+                }
+                if observed == expected {
+                    continue;
+                }
+                let (sources, remaining_signals) = event_sources(core_id, *register);
+                if *observed as usize + remaining_signals < *expected as usize {
+                    return Some(DeadlockInfo {
+                        cycle: format!(
+                            "core {} WAIT event {} needs {} but only {} signal(s) can arrive",
+                            core_id - 1,
+                            register,
+                            expected,
+                            *observed as usize + remaining_signals
+                        ),
+                        states: format_state(&core_id),
+                    });
+                }
+                wait_for.insert(core_id, sources);
+            }
            CoreState::Working | CoreState::Halted => {}
        }
    }

-    let mut visited = HashSet::new();
-
-    for &start_core in wait_for.keys() {
-        if visited.contains(&start_core) {
-            continue;
+    fn find_cycle(
+        core: i32,
+        wait_for: &HashMap<i32, Vec<i32>>,
+        path: &mut Vec<i32>,
+        positions: &mut HashMap<i32, usize>,
+        visited: &mut HashSet<i32>,
+    ) -> Option<Vec<i32>> {
+        if let Some(position) = positions.get(&core) {
+            return Some(path[*position..].to_vec());
        }
+        if !visited.insert(core) {
+            return None;
+        }
+        positions.insert(core, path.len());
+        path.push(core);
+        if let Some(targets) = wait_for.get(&core) {
+            for target in targets {
+                if let Some(cycle) = find_cycle(*target, wait_for, path, positions, visited) {
+                    return Some(cycle);
+                }
+            }
+        }
+        path.pop();
+        positions.remove(&core);
+        None
+    }

+    let mut visited = HashSet::new();
+    for start_core in wait_for.keys() {
        let mut path = Vec::new();
-        let mut current_core = start_core;
-        let mut in_path = HashSet::new();
-
-        while let Some(&waiting_for) = wait_for.get(&current_core) {
-            path.push(current_core);
-            in_path.insert(current_core);
-            visited.insert(current_core);
-
-            // Found a closed loop!
-            if in_path.contains(&waiting_for) {
-                let cycle_start = path.iter().position(|&c| c == waiting_for).unwrap();
-                let cycle = &path[cycle_start..];
-                let format_core = |core: &i32| (core - 1).to_string();
-
-                let cycle_str = cycle
-                    .iter()
-                    .map(format_core)
-                    .collect::<Vec<_>>()
-                    .join(" -> ");
-
-                let cycle = cycle
-                    .iter()
-                    .copied()
-                    .chain(std::iter::once(waiting_for))
-                    .collect::<Vec<_>>();
-                let cycle_msg = format!("{} -> {}", cycle_str, waiting_for - 1);
-                let states_msg = cycle
-                    .iter()
-                    .filter_map(|core| {
-                        states.get(core).map(|state| match state {
-                            CoreState::SendingTo(target, size) => {
-                                format!("core {} send {}B -> {}", core - 1, size, target - 1)
-                            }
-                            CoreState::ReceivingFrom(source, size) => {
-                                format!("core {} recv {}B <- {}", core - 1, size, source - 1)
-                            }
-                            CoreState::Working => format!("core {} working", core - 1),
-                            CoreState::Halted => format!("core {} halted", core - 1),
-                        })
-                    })
-                    .collect::<Vec<_>>()
-                    .join(", ");
-
-                return Some(DeadlockInfo {
-                    cycle: cycle_msg,
-                    states: states_msg,
-                });
-            }
-
-            // Hit a known branch that didn't result in a cycle
-            if visited.contains(&waiting_for) {
-                break;
-            }
-
-            current_core = waiting_for;
+        let mut positions = HashMap::new();
+        if let Some(cycle) = find_cycle(
+            *start_core,
+            &wait_for,
+            &mut path,
+            &mut positions,
+            &mut visited,
+        ) {
+            let cycle_msg = cycle
+                .iter()
+                .chain(std::iter::once(&cycle[0]))
+                .map(|core| (core - 1).to_string())
+                .collect::<Vec<_>>()
+                .join(" -> ");
+            let states_msg = cycle
+                .iter()
+                .map(&format_state)
+                .collect::<Vec<_>>()
+                .join(", ");
+            return Some(DeadlockInfo {
+                cycle: cycle_msg,
+                states: states_msg,
+            });
        }
    }
    None
@@ -446,7 +558,9 @@ fn handle_wait_sync(
        InstructionStatus::Sync(data) => {
            let (source, target) = data.get_core_immcore();
            let register = data.offset_select() as usize;
-            events[target as usize][register] += 1;
+            let event = &mut events[target as usize][register];
+            event.count += 1;
+            *event.sources.entry(source).or_default() += 1;
            core_instructions[source as usize].program_counter += 1;
            true
        }
@@ -454,8 +568,8 @@ fn handle_wait_sync(
            let core = data.core_indx() as usize;
            let register = data.offset_select() as usize;
            let value = data.offset_value();
-            if events[core][register] >= value {
-                events[core][register] -= value;
+            if events[core][register].count == value {
+                events[core][register] = SyncEvent::default();
                core_instructions[core].program_counter += 1;
                true
            } else {
@@ -465,3 +579,66 @@ fn handle_wait_sync(
        _ => false,
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::instruction_set::{
+        InstructionsBuilder,
+        instruction_data::InstructionDataBuilder,
+        isa::{sync, wait},
+    };
+
+    fn wait_then_sync(core: i32, target: i32) -> CoreInstructions {
+        let mut instructions = InstructionsBuilder::new();
+        let mut data = InstructionDataBuilder::new();
+        data.set_core_indx(core).fix_core_indx();
+        instructions.make_inst(wait, data.set_offset_select_value(0, 2).build());
+        instructions.make_inst(
+            sync,
+            data.set_imm_core(target)
+                .set_offset_select_value(1, 0)
+                .build(),
+        );
+        CoreInstructions::from(instructions.build())
+    }
+
+    fn sync_then_wait(core: i32, target: i32) -> CoreInstructions {
+        let mut instructions = InstructionsBuilder::new();
+        let mut data = InstructionDataBuilder::new();
+        data.set_core_indx(core).fix_core_indx();
+        instructions.make_inst(
+            sync,
+            data.set_imm_core(target)
+                .set_offset_select_value(0, 0)
+                .build(),
+        );
+        instructions.make_inst(wait, data.set_offset_select_value(1, 1).build());
+        CoreInstructions::from(instructions.build())
+    }
+
+    #[test]
+    fn contributed_sync_source_is_not_a_wait_dependency() {
+        let mut writer = wait_then_sync(1, 2);
+        writer.current_iteration = 1;
+        let mut contributed_reader = sync_then_wait(2, 1);
+        contributed_reader.current_iteration = 1;
+        contributed_reader.program_counter = 1;
+        let pending_reader = sync_then_wait(3, 1);
+        let cores = vec![
+            CoreInstructions::empty(),
+            writer,
+            contributed_reader,
+            pending_reader,
+        ];
+        let mut events: SyncEvents = (0..cores.len())
+            .map(|_| std::array::from_fn(|_| SyncEvent::default()))
+            .collect();
+        events[1][0].count = 1;
+        events[1][0].sources.insert(2, 1);
+
+        assert!(detect_deadlock(&cores, &events, 3).is_none());
+        events[1][0].sources.clear();
+        assert!(detect_deadlock(&cores, &events, 3).is_some());
+    }
+}
@@ -297,7 +297,7 @@ fn multiple_send_recv_test() {
 }

 #[test]
-fn sync_wait_tokens_test() {
+fn sync_wait_exact_count_resets_test() {
    let cpu = common::empty_cpu(2);
    let mut cores = CoreInstructionsBuilder::new(2);
    let mut instructions = InstructionsBuilder::new();
@@ -313,14 +313,68 @@ fn sync_wait_tokens_test() {
    cores.set_core(1, instructions.build());

    data.set_core_indx(2).fix_core_indx();
-    for _ in 0..2 {
-        instructions.make_inst(wait, data.set_offset_select_value(0, 1).build());
-    }
+    instructions.make_inst(wait, data.set_offset_select_value(0, 2).build());
    cores.set_core(2, instructions.build());

    Executable::new(cpu, cores.build()).execute().unwrap();
 }

+#[test]
+fn sync_wait_rejects_overshoot() {
+    let cpu = common::empty_cpu(3);
+    let mut cores = CoreInstructionsBuilder::new(3);
+    let mut instructions = InstructionsBuilder::new();
+    let mut data = InstructionDataBuilder::new();
+
+    data.set_core_indx(1).fix_core_indx();
+    instructions.make_inst(
+        sync,
+        data.set_imm_core(3).set_offset_select_value(0, 0).build(),
+    );
+    cores.set_core(1, instructions.build());
+
+    data.set_core_indx(2).fix_core_indx();
+    instructions.make_inst(
+        sync,
+        data.set_imm_core(3).set_offset_select_value(0, 0).build(),
+    );
+    cores.set_core(2, instructions.build());
+
+    data.set_core_indx(3).fix_core_indx();
+    instructions.make_inst(wait, data.set_offset_select_value(0, 1).build());
+    cores.set_core(3, instructions.build());
+
+    let error = Executable::new(cpu, cores.build()).execute().unwrap_err();
+    assert!(error.to_string().contains("overshot exact value"));
+}
+
+#[test]
+fn sync_wait_deadlock_cycle_is_reported() {
+    let cpu = common::empty_cpu(2);
+    let mut cores = CoreInstructionsBuilder::new(2);
+    let mut instructions = InstructionsBuilder::new();
+    let mut data = InstructionDataBuilder::new();
+
+    data.set_core_indx(1).fix_core_indx();
+    instructions.make_inst(wait, data.set_offset_select_value(0, 1).build());
+    instructions.make_inst(
+        sync,
+        data.set_imm_core(2).set_offset_select_value(0, 0).build(),
+    );
+    cores.set_core(1, instructions.build());
+
+    data.set_core_indx(2).fix_core_indx();
+    instructions.make_inst(wait, data.set_offset_select_value(0, 1).build());
+    instructions.make_inst(
+        sync,
+        data.set_imm_core(1).set_offset_select_value(0, 0).build(),
+    );
+    cores.set_core(2, instructions.build());
+
+    let error = Executable::new(cpu, cores.build()).execute().unwrap_err();
+    assert!(error.to_string().contains("wait event"));
+}
+
 #[test]
 fn blocked_transfers_do_not_starve_sync_producer() {
    let cpu = common::empty_cpu(4);