:erlang.link/1

Processes are linked and when either sides exits, it propagates an exit
to the other side of the link.

Author: KronicDeth

liblumen_alloc/src/erts/process.rs

@@ -19,7 +19,7 @@ use core::sync::atomic::{AtomicU16, AtomicU64, AtomicUsize, Ordering};
 
 use ::alloc::sync::Arc;
 
-use hashbrown::HashMap;
+use hashbrown::{HashMap, HashSet};
 use intrusive_collections::{LinkedList, UnsafeRef};
 
 use liblumen_core::locks::{Mutex, MutexGuard, RwLock, SpinLock};
@@ -100,6 +100,9 @@ pub struct ProcessControlBlock {
     code_stack: Mutex<code::stack::Stack>,
     pub status: RwLock<Status>,
     pub registered_name: RwLock<Option<Atom>>,
+    /// Pids of processes that are linked to this process and need to be exited when this process
+    /// exits
+    pub linked_pid_set: Mutex<HashSet<Pid>>,
     pub mailbox: Mutex<RefCell<Mailbox>>,
     // process heap, cache line aligned to avoid false sharing with rest of struct
     heap: Mutex<ProcessHeap>,
@@ -140,6 +143,7 @@ impl ProcessControlBlock {
             run_reductions: Default::default(),
             total_reductions: Default::default(),
             registered_name: Default::default(),
+            linked_pid_set: Default::default(),
         }
     }
 
@@ -242,6 +246,8 @@ impl ProcessControlBlock {
         heap.virtual_alloc(bin)
     }
 
+    // Stack
+
     /// Frees stack space occupied by the last term on the stack,
     /// adjusting the stack pointer accordingly.
     ///
@@ -290,6 +296,23 @@ impl ProcessControlBlock {
         self.heap.lock().stack_used()
     }
 
+    // Links
+
+    pub fn link(&self, other: &ProcessControlBlock) {
+        // link in order so that locks are always taken in the same order to prevent deadlocks
+        if self.pid < other.pid {
+            let mut self_pid_set = self.linked_pid_set.lock();
+            let mut other_pid_set = other.linked_pid_set.lock();
+
+            self_pid_set.insert(other.pid);
+            other_pid_set.insert(self.pid);
+        } else {
+            other.link(self)
+        }
+    }
+
+    // Pid
+
     pub fn pid(&self) -> Pid {
         self.pid
     }

lumen_runtime/src/otp/erlang/link_1.rs

@@ -0,0 +1,88 @@
+// wasm32 proptest cannot be compiled at the same time as non-wasm32 proptest, so disable tests that
+// use proptest completely for wasm32
+//
+// See https://github.com/rust-lang/cargo/issues/4866
+#[cfg(all(not(target_arch = "wasm32"), test))]
+mod test;
+
+use std::sync::Arc;
+
+use liblumen_alloc::erts::exception;
+use liblumen_alloc::erts::exception::system::Alloc;
+use liblumen_alloc::erts::process::code::stack::frame::{Frame, Placement};
+use liblumen_alloc::erts::process::code::{self, result_from_exception};
+use liblumen_alloc::erts::process::ProcessControlBlock;
+use liblumen_alloc::erts::term::{atom_unchecked, Atom, Term, TypedTerm};
+use liblumen_alloc::{badarg, error, ModuleFunctionArity};
+
+use crate::registry::pid_to_process;
+
+pub fn place_frame_with_arguments(
+    process: &ProcessControlBlock,
+    placement: Placement,
+    pid_or_port: Term,
+) -> Result<(), Alloc> {
+    process.stack_push(pid_or_port)?;
+    process.place_frame(frame(), placement);
+
+    Ok(())
+}
+
+// Private
+
+fn code(arc_process: &Arc<ProcessControlBlock>) -> code::Result {
+    arc_process.reduce();
+
+    let pid_or_port = arc_process.stack_pop().unwrap();
+
+    match native(arc_process, pid_or_port) {
+        Ok(true_term) => {
+            arc_process.return_from_call(true_term)?;
+
+            ProcessControlBlock::call_code(arc_process)
+        }
+        Err(exception) => result_from_exception(arc_process, exception),
+    }
+}
+
+fn frame() -> Frame {
+    Frame::new(module_function_arity(), code)
+}
+
+fn function() -> Atom {
+    Atom::try_from_str("link").unwrap()
+}
+
+fn module_function_arity() -> Arc<ModuleFunctionArity> {
+    Arc::new(ModuleFunctionArity {
+        module: super::module(),
+        function: function(),
+        arity: 1,
+    })
+}
+
+fn native(process: &ProcessControlBlock, pid_or_port: Term) -> exception::Result {
+    match pid_or_port.to_typed_term().unwrap() {
+        TypedTerm::Pid(pid) => {
+            if pid == process.pid() {
+                Ok(true.into())
+            } else {
+                match pid_to_process(&pid) {
+                    Some(pid_arc_process) => {
+                        process.link(&pid_arc_process);
+
+                        Ok(true.into())
+                    }
+                    None => Err(error!(atom_unchecked("noproc")).into()),
+                }
+            }
+        }
+        TypedTerm::Port(_) => unimplemented!(),
+        TypedTerm::Boxed(boxed) => match boxed.to_typed_term().unwrap() {
+            TypedTerm::ExternalPid(_) => unimplemented!(),
+            TypedTerm::ExternalPort(_) => unimplemented!(),
+            _ => Err(badarg!().into()),
+        },
+        _ => Err(badarg!().into()),
+    }
+}

lumen_runtime/src/otp/erlang/link_1/test.rs

@@ -0,0 +1,35 @@
+mod with_local_pid;
+
+use proptest::prop_assert_eq;
+use proptest::strategy::Strategy;
+use proptest::test_runner::{Config, TestRunner};
+
+use liblumen_alloc::badarg;
+use liblumen_alloc::erts::process::ProcessControlBlock;
+
+use crate::otp::erlang::link_1::native;
+use crate::scheduler::{with_process, with_process_arc};
+use crate::test::strategy;
+
+#[test]
+fn without_pid_or_port_errors_badarg() {
+    with_process_arc(|arc_process| {
+        TestRunner::new(Config::with_source_file(file!()))
+            .run(
+                &strategy::term(arc_process.clone())
+                    .prop_filter("Cannot be pid or port", |pid_or_port| {
+                        !(pid_or_port.is_pid() || pid_or_port.is_port())
+                    }),
+                |pid_or_port| {
+                    prop_assert_eq!(native(&arc_process, pid_or_port), Err(badarg!().into()));
+
+                    Ok(())
+                },
+            )
+            .unwrap();
+    });
+}
+
+fn link_count(process: &ProcessControlBlock) -> usize {
+    process.linked_pid_set.lock().len()
+}

lumen_runtime/src/otp/erlang/link_1/test/with_local_pid.rs

@@ -0,0 +1,127 @@
+use super::*;
+
+use liblumen_alloc::error;
+use liblumen_alloc::erts::process::code::stack::frame::Placement;
+use liblumen_alloc::erts::term::{atom_unchecked, next_pid};
+
+use crate::otp::erlang;
+use crate::process;
+use crate::scheduler::Scheduler;
+
+#[test]
+fn with_self_returns_true_but_does_not_create_link() {
+    with_process(|process| {
+        let link_count_before = link_count(process);
+
+        assert_eq!(native(process, process.pid_term()), Ok(true.into()));
+
+        assert_eq!(link_count(process), link_count_before);
+    });
+}
+
+#[test]
+fn with_non_existent_pid_errors_noproc() {
+    with_process(|process| {
+        let link_count_before = link_count(process);
+
+        assert_eq!(
+            native(process, next_pid()),
+            Err(error!(atom_unchecked("noproc")).into())
+        );
+
+        assert_eq!(link_count(process), link_count_before);
+    });
+}
+
+#[test]
+fn with_existing_unlinked_pid_links_to_process() {
+    with_process(|process| {
+        let other_process = process::test(process);
+
+        let process_link_count_before = link_count(process);
+        let other_process_link_count_before = link_count(process);
+
+        assert_eq!(native(process, other_process.pid_term()), Ok(true.into()));
+
+        assert_eq!(link_count(process), process_link_count_before + 1);
+        assert_eq!(
+            link_count(&other_process),
+            other_process_link_count_before + 1
+        );
+    });
+}
+
+#[test]
+fn with_existing_linked_pid_returns_true() {
+    with_process(|process| {
+        let other_process = process::test(process);
+
+        process.link(&other_process);
+
+        let process_link_count_before = link_count(process);
+        let other_process_link_count_before = link_count(process);
+
+        assert_eq!(native(process, other_process.pid_term()), Ok(true.into()));
+
+        assert_eq!(link_count(process), process_link_count_before);
+        assert_eq!(link_count(&other_process), other_process_link_count_before);
+    });
+}
+
+#[test]
+fn when_a_linked_process_exits_the_process_exits_too() {
+    with_process(|process| {
+        let other_arc_process = process::test(process);
+
+        assert_eq!(
+            native(process, other_arc_process.pid_term()),
+            Ok(true.into())
+        );
+
+        assert!(Scheduler::current().run_through(&other_arc_process));
+
+        assert!(!other_arc_process.is_exiting());
+        assert!(!process.is_exiting());
+
+        erlang::exit_1::place_frame_with_arguments(
+            &other_arc_process,
+            Placement::Replace,
+            atom_unchecked("normal"),
+        )
+        .unwrap();
+
+        assert!(Scheduler::current().run_through(&other_arc_process));
+
+        assert!(other_arc_process.is_exiting());
+        assert!(process.is_exiting())
+    });
+}
+
+#[test]
+fn when_the_process_exits_linked_processes_exit_too() {
+    with_process_arc(|arc_process| {
+        let other_arc_process = process::test(&arc_process);
+
+        assert_eq!(
+            native(&arc_process, other_arc_process.pid_term()),
+            Ok(true.into())
+        );
+
+        assert!(Scheduler::current().run_through(&other_arc_process));
+
+        assert!(!other_arc_process.is_exiting());
+        assert!(!arc_process.is_exiting());
+
+        erlang::exit_1::place_frame_with_arguments(
+            &arc_process,
+            Placement::Replace,
+            atom_unchecked("normal"),
+        )
+        .unwrap();
+
+        assert!(Scheduler::current().run_through(&arc_process));
+
+        assert!(other_arc_process.is_exiting());
+        assert!(arc_process.is_exiting())
+    });
+}

lumen_runtime/src/process.rs

@@ -7,6 +7,7 @@ use hashbrown::HashMap;
 
 use liblumen_core::locks::RwLockWriteGuard;
 
+use liblumen_alloc::erts::exception::runtime;
 use liblumen_alloc::erts::exception::system::Alloc;
 use liblumen_alloc::erts::process::code::stack::frame::{Frame, Placement};
 use liblumen_alloc::erts::process::code::Code;
@@ -19,6 +20,62 @@ use crate::code;
 use crate::otp::erlang::apply_3;
 use crate::registry::*;
 use crate::scheduler::Scheduler;
+use crate::system;
+#[cfg(test)]
+use crate::test;
+
+fn is_expected_exit_reason(reason: Term) -> bool {
+    match reason.to_typed_term().unwrap() {
+        TypedTerm::Atom(atom) => match atom.name() {
+            "normal" | "shutdown" => true,
+            _ => false,
+        },
+        TypedTerm::Boxed(boxed) => match boxed.to_typed_term().unwrap() {
+            TypedTerm::Tuple(tuple) => {
+                tuple.len() == 2 && {
+                    match tuple[0].to_typed_term().unwrap() {
+                        TypedTerm::Atom(atom) => atom.name() == "shutdown",
+                        _ => false,
+                    }
+                }
+            }
+            _ => false,
+        },
+        _ => false,
+    }
+}
+
+pub fn log_exit(process: &ProcessControlBlock, exception: &runtime::Exception) {
+    match exception.class {
+        runtime::Class::Exit => {
+            let reason = exception.reason;
+
+            if !is_expected_exit_reason(reason) {
+                system::io::puts(&format!(
+                    "** (EXIT from {}) exited with reason: {}",
+                    process, reason
+                ));
+            }
+        }
+        runtime::Class::Error { .. } => system::io::puts(&format!(
+            "** (EXIT from {}) exited with reason: an exception was raised: {}\n{}",
+            process,
+            exception.reason,
+            process.stacktrace()
+        )),
+        _ => unimplemented!("{:?}", exception),
+    }
+}
+
+pub fn propagate_exit(process: &ProcessControlBlock) {
+    for linked_pid in process.linked_pid_set.lock().iter() {
+        if let Some(linked_pid_arc_process) = pid_to_process(linked_pid) {
+            // only tell the linked process to exit.  When it is run by its scheduler, it will
+            // go through propagating its own exit.
+            linked_pid_arc_process.exit()
+        }
+    }
+}
 
 pub fn register_in(
     arc_process_control_block: Arc<ProcessControlBlock>,