Method take_for_message

nodedb-cluster/src/swim/dissemination/queue.rs:70–101

Return up to `max` least-disseminated updates for a single outgoing message. Increments each returned entry's `sent_count` and drops entries whose new count has reached `lambda_log_n`. `lambda_log_n` is `ceil(lambda * log2(cluster_size + 1))` — computed by the caller because it depends on the current membership size and the [`super::super::config::SwimConfig`] `fanout_lambda` knob.

(&self, max: usize, lambda_log_n: u32)

    /// membership size and the [`super::super::config::SwimConfig`]
    /// `fanout_lambda` knob.
    pub fn take_for_message(&self, max: usize, lambda_log_n: u32) -> Vec<MemberUpdate> {
        if max == 0 {
            return Vec::new();
        }
        let mut guard = self.inner.lock().expect("dissemination lock poisoned");

        // Sort a snapshot of keys by (sent_count, node_id) ascending;
        // we can't use BinaryHeap directly without cloning because the
        // values need to be mutated in place after the decision.
        let mut keys: Vec<NodeId> = guard.keys().cloned().collect();
        keys.sort_by(|a, b| {
            let pa = &guard[a];
            let pb = &guard[b];
            pa.sent_count
                .cmp(&pb.sent_count)
                .then_with(|| a.as_str().cmp(b.as_str()))
                .then_with(|| pa.update.incarnation.cmp(&pb.update.incarnation))
        });
        keys.truncate(max);

        let mut out = Vec::with_capacity(keys.len());
        for k in keys {
            if let Some(pending) = guard.get_mut(&k) {
                pending.record_sent();
                out.push(pending.update.clone());
                if pending.sent_count >= lambda_log_n {
                    guard.remove(&k);
                }
            }
        }
        out
    }

    /// Compute `ceil(lambda * log2(cluster_size + 1))`. Exposed so the
    /// runner can pass the result straight into [`take_for_message`].