beamr 0.4.5

A Rust runtime with the BEAM's execution model, targeting Gleam
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

//! Test-support helpers for scheduler integration tests.

use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};

use crate::atom::Atom;
use crate::module::Module;
use crate::namespace::NamespaceId;
use crate::process::Process;
use crate::process::heap::DEFAULT_HEAP_SIZE;
use crate::term::Term;

use super::{ProcessSlot, ScheduledProcess, Scheduler, lock_or_recover};

impl Scheduler {
    /// Enqueue an immediate atom message into a live process mailbox.
    #[must_use]
    pub fn enqueue_atom_message(&self, target_pid: u64, atom: Atom) -> bool {
        let delivered = self
            .with_process(target_pid, |process| {
                process.mailbox_mut().push_owned(Term::atom(atom));
                true
            })
            .unwrap_or(false);
        if delivered {
            self.wake_process(target_pid);
        }
        delivered
    }

    fn with_process<T>(&self, pid: u64, f: impl FnOnce(&mut Process) -> T) -> Option<T> {
        let entry = self.shared.process_bodies.get(&pid)?;
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(scheduled) => Some(f(&mut scheduled.0)),
            ProcessSlot::Executing(_) | ProcessSlot::Absent => None,
        }
    }

    /// Spawn an inert process without module code for host-side policy tests.
    pub fn spawn_test_process(&self, trap_exit: bool) -> u64 {
        let pid = self.shared.next_pid.fetch_add(1, Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(pid);
        let mut process = Process::new(pid, DEFAULT_HEAP_SIZE);
        process.set_trap_exit(trap_exit);
        self.shared.process_bodies.insert(
            pid,
            Mutex::new(ProcessSlot::Present(ScheduledProcess(process))),
        );
        pid
    }

    /// Spawn an inert process pinned to a module in a namespace for policy tests.
    pub fn spawn_test_process_in(&self, namespace: NamespaceId, module: Arc<Module>) -> u64 {
        let pid = self.shared.next_pid.fetch_add(1, Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(pid);
        let mut process = Process::new(pid, DEFAULT_HEAP_SIZE);
        process.set_namespace_id(namespace);
        process.set_current_module(module);
        self.shared.process_bodies.insert(
            pid,
            Mutex::new(ProcessSlot::Present(ScheduledProcess(process))),
        );
        pid
    }

    /// Spawn an inert process linked to a live parent for host-side policy tests.
    pub fn spawn_linked_test_process(
        &self,
        parent_pid: u64,
    ) -> Result<u64, crate::native::links::LinkError> {
        let child_pid = self.shared.next_pid.fetch_add(1, Ordering::Relaxed);
        // Scope the parent map guard so it is released before the insert below:
        // holding a DashMap shard guard across an insert into the same map
        // self-deadlocks when both pids hash to the same shard.
        let parent_group_leader = {
            let Some(parent_entry) = self.shared.process_bodies.get(&parent_pid) else {
                return Err(crate::native::links::LinkError::NoProc);
            };
            let mut parent_slot = lock_or_recover(&parent_entry);
            let ProcessSlot::Present(ScheduledProcess(parent)) = &mut *parent_slot else {
                return Err(crate::native::links::LinkError::NoProc);
            };
            parent.add_link(child_pid);
            parent.group_leader()
        };
        self.shared.process_table.spawn_with_pid(child_pid);
        let mut child = Process::new(child_pid, DEFAULT_HEAP_SIZE);
        child.set_group_leader(parent_group_leader);
        child.add_link(parent_pid);
        self.shared.process_bodies.insert(
            child_pid,
            Mutex::new(ProcessSlot::Present(ScheduledProcess(child))),
        );
        Ok(child_pid)
    }

    /// Return true when a term is queued in a live process mailbox.
    #[must_use]
    pub fn has_message(&self, target_pid: u64, expected: Term) -> Option<bool> {
        self.with_process(target_pid, |process| {
            process.mailbox_mut().drain_arrival();
            process
                .mailbox()
                .scan_iter()
                .any(|message| *message == expected)
        })
    }

    /// Return true when a trapped EXIT message from `source_pid` is queued.
    #[must_use]
    pub fn has_trapped_exit_message(&self, target_pid: u64, source_pid: u64) -> Option<bool> {
        self.with_process(target_pid, |process| {
            process.mailbox_mut().drain_arrival();
            process.mailbox().scan_iter().any(|message| {
                let Some(tuple) = crate::term::boxed::Tuple::new(*message) else {
                    return false;
                };
                tuple.arity() == 3
                    && tuple.get(0) == Some(Term::atom(Atom::EXIT))
                    && tuple.get(1) == Some(Term::pid(source_pid))
            })
        })
    }

    /// Return a live process's group leader term for scheduler tests.
    #[must_use]
    pub fn test_group_leader(&self, pid: u64) -> Option<Term> {
        self.with_process(pid, |process| process.group_leader())
    }

    /// Set a live process's group leader term for scheduler tests.
    pub fn set_test_group_leader(&self, pid: u64, group_leader: Term) -> bool {
        self.with_process(pid, |process| process.set_group_leader(group_leader))
            .is_some()
    }
}