Execute one superstep. Returns (local_delta, outbound_contributions).
(
&mut self,
damping: f64,
global_n: usize,
local_edge_iter: &dyn Fn(u32) -> Vec<u32>,
)
| 62 | |
| 63 | /// Execute one superstep. Returns (local_delta, outbound_contributions). |
| 64 | pub fn superstep( |
| 65 | &mut self, |
| 66 | damping: f64, |
| 67 | global_n: usize, |
| 68 | local_edge_iter: &dyn Fn(u32) -> Vec<u32>, |
| 69 | ) -> (f64, HashMap<u16, Vec<(String, f64)>>) { |
| 70 | let n = global_n as f64; |
| 71 | let teleport = (1.0 - damping) / n; |
| 72 | |
| 73 | let dangling_sum: f64 = self |
| 74 | .rank |
| 75 | .iter() |
| 76 | .enumerate() |
| 77 | .filter(|(i, _)| self.is_dangling[*i]) |
| 78 | .map(|(_, r)| r) |
| 79 | .sum(); |
| 80 | |
| 81 | let base = teleport + damping * dangling_sum / n; |
| 82 | |
| 83 | for r in self.next_rank.iter_mut() { |
| 84 | *r = base; |
| 85 | } |
| 86 | |
| 87 | let mut outbound: HashMap<u16, Vec<(String, f64)>> = HashMap::new(); |
| 88 | for u in 0..self.vertex_count { |
| 89 | let deg = self.out_degrees[u]; |
| 90 | if deg == 0 { |
| 91 | continue; |
| 92 | } |
| 93 | let contrib = damping * self.rank[u] / deg as f64; |
| 94 | |
| 95 | // Scatter to local edges. |
| 96 | for dst in local_edge_iter(u as u32) { |
| 97 | self.next_rank[dst as usize] += contrib; |
| 98 | } |
| 99 | |
| 100 | // Scatter to boundary edges (cross-shard). |
| 101 | if let Some(boundary) = self.boundary_edges.get(&(u as u32)) { |
| 102 | for (dst_name, target_shard) in boundary { |
| 103 | outbound |
| 104 | .entry(*target_shard) |
| 105 | .or_default() |
| 106 | .push((dst_name.clone(), contrib)); |
| 107 | } |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | let delta: f64 = self |
| 112 | .rank |
| 113 | .iter() |
| 114 | .zip(self.next_rank.iter()) |
| 115 | .map(|(old, new)| (old - new).abs()) |
| 116 | .sum(); |
| 117 | |
| 118 | std::mem::swap(&mut self.rank, &mut self.next_rank); |
| 119 | self.incoming_contributions.clear(); |
| 120 | |
| 121 | (delta, outbound) |