Made type checking fully polymorphic

crates/aterm/src/aterm_binary_stream.rs

@@ -255,62 +255,13 @@ impl<W: Write> BinaryATermWriter<W> {
 
             if !self.terms.read().contains(&current_term) || is_output {
                 if write_ready {
-                    if is_int_term(&current_term) {
-                        let int_term = ATermIntRef::from(current_term.copy());
-                        if is_output {
-                            // If the integer is output, write the header and just an integer
-                            self.stream.write_bits(PacketType::ATermIntOutput as u64, PACKET_BITS)?;
-                            self.stream.write_integer(int_term.value() as u64)?;
-                        } else {
-                            let symbol_index = self.write_function_symbol(&int_term.get_head_symbol())?;
-
-                            self.stream.write_bits(PacketType::ATerm as u64, PACKET_BITS)?;
-                            self.stream
-                                .write_bits(symbol_index as u64, self.function_symbol_index_width())?;
-                            self.stream.write_integer(int_term.value() as u64)?;
-                        }
-                    } else {
-                        let symbol_index = self.write_function_symbol(&current_term.get_head_symbol())?;
-                        let packet_type = if is_output {
-                            PacketType::ATermOutput
-                        } else {
-                            PacketType::ATerm
-                        };
-
-                        self.stream.write_bits(packet_type as u64, PACKET_BITS)?;
-                        self.stream
-                            .write_bits(symbol_index as u64, self.function_symbol_index_width())?;
-
-                        for arg in current_term.arguments() {
-                            let index = self.terms.read().index(&arg).expect("Argument must already be written");
-                            self.stream.write_bits(*index as u64, self.term_index_width())?;
-                        }
-                    }
-
-                    if !is_output {
-                        let (_, inserted) = self.terms.write().insert(current_term.copy());
-                        assert!(inserted, "This term should have a new index assigned.");
-                        self.term_index_width = bits_for_value(self.terms.read().len());
-                    }
+                    self.write_term_body(&current_term, is_output)?;
 
                     // Done with this entry now that it has been written (and, if not the
                     // top-level output term, protected independently by `terms` above).
                     self.stack.write().pop_back();
                 } else {
-                    // Mark ready for its next visit, once its (not yet written) arguments below
-                    // it on the stack are done -- so leave it in place rather than popping it.
-                    if let Some(back) = self.stack.write().back_mut() {
-                        back.1 = true;
-                    }
-
-                    // Add arguments to stack for processing first. Two equal
-                    // arguments are both pushed here, but that does not write
-                    // the term twice.
-                    for arg in current_term.arguments() {
-                        if !self.terms.read().contains(&arg) {
-                            self.stack.write().push_back((arg.copy(), false));
-                        }
-                    }
+                    self.queue_arguments(&current_term);
                 }
             } else {
                 // This term was already written and as such should be skipped. This can happen
@@ -321,6 +272,69 @@ impl<W: Write> BinaryATermWriter<W> {
 
         Ok(())
     }
+
+    /// Writes the body of `current_term` to the stream: the head packet and,
+    /// for a non-integer term, the already-written argument indices. When
+    /// `is_output` the term is the top-level output term and is written down
+    /// as such without being added to the term table.
+    fn write_term_body(&mut self, current_term: &ATermRef<'_>, is_output: bool) -> Result<(), MercError> {
+        if is_int_term(current_term) {
+            let int_term = ATermIntRef::from(current_term.copy());
+            if is_output {
+                // If the integer is output, write the header and just an integer
+                self.stream.write_bits(PacketType::ATermIntOutput as u64, PACKET_BITS)?;
+                self.stream.write_integer(int_term.value() as u64)?;
+            } else {
+                let symbol_index = self.write_function_symbol(&int_term.get_head_symbol())?;
+
+                self.stream.write_bits(PacketType::ATerm as u64, PACKET_BITS)?;
+                self.stream
+                    .write_bits(symbol_index as u64, self.function_symbol_index_width())?;
+                self.stream.write_integer(int_term.value() as u64)?;
+            }
+        } else {
+            let symbol_index = self.write_function_symbol(&current_term.get_head_symbol())?;
+            let packet_type = if is_output {
+                PacketType::ATermOutput
+            } else {
+                PacketType::ATerm
+            };
+
+            self.stream.write_bits(packet_type as u64, PACKET_BITS)?;
+            self.stream
+                .write_bits(symbol_index as u64, self.function_symbol_index_width())?;
+
+            for arg in current_term.arguments() {
+                let index = self.terms.read().index(&arg).expect("Argument must already be written");
+                self.stream.write_bits(*index as u64, self.term_index_width())?;
+            }
+        }
+
+        if !is_output {
+            let (_, inserted) = self.terms.write().insert(current_term.copy());
+            assert!(inserted, "This term should have a new index assigned.");
+            self.term_index_width = bits_for_value(self.terms.read().len());
+        }
+
+        Ok(())
+    }
+
+    /// Marks `current_term` ready for its next visit — once its (not yet
+    /// written) arguments below it on the stack are done — by flipping its
+    /// `write_ready` flag in place, then queues those arguments for
+    /// processing first. Two equal arguments are both pushed here, but that
+    /// does not write the term twice.
+    fn queue_arguments(&mut self, current_term: &ATermRef<'_>) {
+        if let Some(back) = self.stack.write().back_mut() {
+            back.1 = true;
+        }
+
+        for arg in current_term.arguments() {
+            if !self.terms.read().contains(&arg) {
+                self.stack.write().push_back((arg.copy(), false));
+            }
+        }
+    }
 }
 
 impl<W: Write> ATermWrite for BinaryATermWriter<W> {

crates/aterm/src/storage/global_aterm_pool.rs

@@ -326,6 +326,39 @@ impl GlobalTermPool {
 
     /// Collects garbage terms.
     pub fn collect_garbage(&mut self) {
+        let mark_time = Instant::now();
+        self.mark_roots();
+        let mark_time_elapsed = mark_time.elapsed();
+        let collect_time = Instant::now();
+
+        let (removed_terms, removed_symbols) = self.sweep_terms_and_symbols();
+
+        debug!(
+            "Garbage collection: marking took {}ms, collection took {}ms, {} terms and {} symbols removed",
+            mark_time_elapsed.as_millis(),
+            collect_time.elapsed().as_millis(),
+            removed_terms,
+            removed_symbols
+        );
+
+        debug!("{}", self.metrics());
+
+        // Print information from the protection sets.
+        for pool in self.thread_pools.iter().flatten() {
+            // SAFETY: We have exclusive access to the global term pool, so no other thread can modify the protection sets.
+            let pool = unsafe { &mut *pool.get() };
+            debug!("{}", pool.metrics());
+        }
+
+        // Clear marking data structures
+        self.marked_terms.clear();
+        self.marked_symbols.clear();
+        self.stack.clear();
+    }
+
+    /// Marks the default symbols and every root in every protection set as reachable,
+    /// and reclaims protection sets of threads that have exited.
+    fn mark_roots(&mut self) {
         // Mark the default symbols
         // SAFETY: mark-set entries only live for the duration of this collection pass
         // (the sets are drained by the sweep below), and a marked symbol is by
@@ -342,8 +375,6 @@ impl GlobalTermPool {
             stack: &mut self.stack,
         };
 
-        let mark_time = Instant::now();
-
         // Loop through all protection sets and mark the terms.
         for pool in self.thread_pools.iter().flatten() {
             // SAFETY: We have exclusive access to the global term pool, so no other thread can modify the protection sets.
@@ -425,10 +456,11 @@ impl GlobalTermPool {
                 *slot = None;
             }
         }
+    }
 
-        let mark_time_elapsed = mark_time.elapsed();
-        let collect_time = Instant::now();
-
+    /// Removes every term and symbol that was not marked by [`Self::mark_roots`], and
+    /// returns how many of each were removed.
+    fn sweep_terms_and_symbols(&mut self) -> (usize, usize) {
         let num_of_terms = self.len();
         let num_of_symbols = self.symbol_pool.len();
 
@@ -459,27 +491,7 @@ impl GlobalTermPool {
             });
         }
 
-        debug!(
-            "Garbage collection: marking took {}ms, collection took {}ms, {} terms and {} symbols removed",
-            mark_time_elapsed.as_millis(),
-            collect_time.elapsed().as_millis(),
-            num_of_terms - self.len(),
-            num_of_symbols - self.symbol_pool.len()
-        );
-
-        debug!("{}", self.metrics());
-
-        // Print information from the protection sets.
-        for pool in self.thread_pools.iter().flatten() {
-            // SAFETY: We have exclusive access to the global term pool, so no other thread can modify the protection sets.
-            let pool = unsafe { &mut *pool.get() };
-            debug!("{}", pool.metrics());
-        }
-
-        // Clear marking data structures
-        self.marked_terms.clear();
-        self.marked_symbols.clear();
-        self.stack.clear();
+        (num_of_terms - self.len(), num_of_symbols - self.symbol_pool.len())
     }
 
     /// Returns the metrics of the term pool, can be formatted and written to output.

crates/explore/src/cpu_topology.rs

@@ -198,15 +198,7 @@ fn cluster_by_latency(latency_ns: &[f64], num_cores: usize, factor: f64) -> Vec<
         return (0..num_cores).map(|core| vec![core]).collect();
     }
 
-    let mut min_latency = f64::INFINITY;
-    for i in 0..num_cores {
-        for j in 0..num_cores {
-            if i != j {
-                min_latency = min_latency.min(latency_ns[i * num_cores + j]);
-            }
-        }
-    }
-    let threshold = min_latency * factor;
+    let threshold = minimum_off_diagonal_latency(latency_ns, num_cores) * factor;
 
     let mut visited = vec![false; num_cores];
     let mut clusters = Vec::new();
@@ -215,21 +207,7 @@ fn cluster_by_latency(latency_ns: &[f64], num_cores: usize, factor: f64) -> Vec<
             continue;
         }
 
-        let mut component = Vec::new();
-        let mut queue = VecDeque::new();
-        queue.push_back(start);
-        visited[start] = true;
-
-        while let Some(node) = queue.pop_front() {
-            component.push(node);
-            for neighbor in 0..num_cores {
-                if !visited[neighbor] && latency_ns[node * num_cores + neighbor] <= threshold {
-                    visited[neighbor] = true;
-                    queue.push_back(neighbor);
-                }
-            }
-        }
-
+        let mut component = connected_component(start, latency_ns, num_cores, threshold, &mut visited);
         component.sort_unstable();
         clusters.push(component);
     }
@@ -238,6 +216,46 @@ fn cluster_by_latency(latency_ns: &[f64], num_cores: usize, factor: f64) -> Vec<
     clusters
 }
 
+/// Returns the smallest observed off-diagonal one-way latency.
+fn minimum_off_diagonal_latency(latency_ns: &[f64], num_cores: usize) -> f64 {
+    let mut min_latency = f64::INFINITY;
+    for i in 0..num_cores {
+        for j in 0..num_cores {
+            if i != j {
+                min_latency = min_latency.min(latency_ns[i * num_cores + j]);
+            }
+        }
+    }
+    min_latency
+}
+
+/// Single-linkage flood fill: returns every core reachable from `start` through
+/// edges of latency at most `threshold`, marking the reached cores as visited.
+fn connected_component(
+    start: usize,
+    latency_ns: &[f64],
+    num_cores: usize,
+    threshold: f64,
+    visited: &mut [bool],
+) -> Vec<usize> {
+    let mut component = Vec::new();
+    let mut queue = VecDeque::new();
+    queue.push_back(start);
+    visited[start] = true;
+
+    while let Some(node) = queue.pop_front() {
+        component.push(node);
+        for neighbor in 0..num_cores {
+            if !visited[neighbor] && latency_ns[node * num_cores + neighbor] <= threshold {
+                visited[neighbor] = true;
+                queue.push_back(neighbor);
+            }
+        }
+    }
+
+    component
+}
+
 /// Measures the row-major one-way latency matrix over `cores`, sequentially pair by pair.
 ///
 /// Pairs are measured one at a time because concurrently running pairs would perturb each

crates/reduction/src/weak_bisimulation.rs

@@ -406,8 +406,9 @@ fn compute_weak_acts_inner<L: LTS>(
             let [marked_s, marked_t] = marked
                 .get_disjoint_mut([*transition.from, *t])
                 .expect("The indices are disjoint");
-            for (i, number) in marked_s.as_raw_mut_slice().iter_mut().enumerate() {
-                *number |= marked_t.as_raw_slice()[i];
+            let marked_t_raw = marked_t.as_raw_slice();
+            for (s, t) in marked_s.as_raw_mut_slice().iter_mut().zip(marked_t_raw) {
+                *s |= *t;
             }
         }
     }