foca 1.0.0

Gossip-based cluster membership discovery, based on SWIM
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

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
extern crate alloc;
use alloc::vec::Vec;

use crate::{member::Member, ProbeNumber};

pub(crate) struct Probe<T> {
    direct: Option<Member<T>>,
    indirect: Vec<T>,
    probe_number: ProbeNumber,

    direct_ack_ok: bool,
    indirect_ack_count: usize,

    reached_indirect_probe_stage: bool,
}

impl<T: Clone + PartialEq> Probe<T> {
    pub(crate) fn new(indirect: Vec<T>) -> Self {
        Self {
            indirect,
            direct: None,
            direct_ack_ok: false,
            indirect_ack_count: 0,
            reached_indirect_probe_stage: false,
            probe_number: ProbeNumber::default(),
        }
    }

    #[must_use]
    pub(crate) fn start(&mut self, target: Member<T>) -> ProbeNumber {
        self.clear();
        self.direct = Some(target);
        self.probe_number = self.probe_number.wrapping_add(1);
        self.probe_number
    }

    pub(crate) const fn probe_number(&self) -> ProbeNumber {
        self.probe_number
    }

    pub(crate) fn clear(&mut self) {
        self.direct = None;
        self.indirect.clear();
        self.direct_ack_ok = false;
        self.indirect_ack_count = 0;
        self.reached_indirect_probe_stage = false;
        // do NOT reset probe_number
    }

    pub(crate) fn mark_indirect_probe_stage_reached(&mut self) {
        self.reached_indirect_probe_stage = true;
    }

    pub(crate) const fn validate(&self) -> bool {
        // A probe that hasn't been started is
        // valid
        self.direct.is_none()
            // Otherwise it's only valid if the indirect
            // probing stage has been reached
            || self.reached_indirect_probe_stage
    }

    pub(crate) fn take_failed(&mut self) -> Option<Member<T>> {
        if !self.succeeded() {
            self.direct.take()
        } else {
            None
        }
    }

    #[cfg(any(feature = "tracing", test))]
    pub(crate) fn target(&self) -> Option<&T> {
        self.direct.as_ref().map(|probed| probed.id())
    }

    pub(crate) fn is_probing(&self, id: &T) -> bool {
        self.direct.as_ref().is_some_and(|probed| probed.id() == id)
    }

    pub(crate) const fn succeeded(&self) -> bool {
        self.direct_ack_ok || self.indirect_ack_count > 0
    }

    pub(crate) fn receive_ack(&mut self, from: &T, probeno: ProbeNumber) -> bool {
        if probeno == self.probe_number
            && self
                .direct
                .as_ref()
                .is_some_and(|direct| direct.id() == from)
        {
            self.direct_ack_ok = true;
            true
        } else {
            false
        }
    }

    pub(crate) fn expect_indirect_ack(&mut self, from: T) {
        debug_assert!(self
            .direct
            .as_ref()
            .is_some_and(|probed| probed.id() != &from));
        self.indirect.push(from);
    }

    pub(crate) fn receive_indirect_ack(&mut self, from: &T, probeno: ProbeNumber) -> bool {
        if self.probe_number != probeno {
            return false;
        }

        if let Some(position) = self.indirect.iter().position(|id| id == from) {
            self.indirect_ack_count += 1;
            // Ensure we can't double count the same candidate
            self.indirect.swap_remove(position);
            true
        } else {
            false
        }
    }
}