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
//! Cross-node process-group propagation for the scheduler.
//!
//! [`SchedulerPgPropagation`] is the production [`PgPropagation`] backend
//! installed into the scheduler's `PgRegistry` once `SharedState` exists. It
//! replaces the no-op `NullPgPropagation`: every local `pg:join`/`pg:leave`
//! that changes membership is encoded as a `PG_UPDATE` control frame and
//! transmitted to every currently-connected node.
//!
//! ## `Arc`-cycle avoidance
//!
//! `PgRegistry` is a field of `SharedState`, but the propagation backend needs
//! `SharedState` (to reach the connection table and the local node name). Storing
//! an `Arc<SharedState>` here would create a cycle
//! (`SharedState -> PgRegistry -> propagation -> SharedState`) and leak the
//! scheduler forever. We therefore hold a [`Weak<SharedState>`] and upgrade it
//! per-broadcast; if the scheduler has been dropped the upgrade fails and the
//! broadcast is a no-op.
use std::sync::{Arc, Weak};
use crate::distribution::control::encode_pg_update_frame;
use crate::distribution::pg::{PgPropagation, PgUpdate};
use crate::distribution::sender::DistOutbound;
use super::SharedState;
/// Production [`PgPropagation`] backend: encodes membership updates as
/// `PG_UPDATE` control frames and sends them to every connected node.
pub(super) struct SchedulerPgPropagation {
/// Weak handle to the scheduler shared state. Weak (not `Arc`) to avoid the
/// `SharedState -> PgRegistry -> propagation -> SharedState` reference cycle.
pub(super) shared: Weak<SharedState>,
}
impl PgPropagation for SchedulerPgPropagation {
fn broadcast(&self, update: PgUpdate) {
// If the scheduler has been dropped, there is nothing to broadcast to.
let Some(shared) = self.shared.upgrade() else {
return;
};
// No bundle (distribution Disabled) or no sender (replay) ⇒ nothing to
// broadcast to; pg stays fully functional locally with zero propagation.
let Some(dist) = shared.distribution() else {
return;
};
let Some(sender) = dist.sender() else {
return;
};
// Encode the member as an external PID carrying our local node name so
// the receiver records a fully-attributed RemoteMember.
let local_node = shared.local_node.name;
let Ok(frame) = encode_pg_update_frame(update, local_node, &shared.atom_table) else {
// Encoding a pg control frame cannot fail for well-formed atoms/pids;
// a failure here means the update could not be serialised, so there is
// nothing safe to transmit. Drop it rather than panicking — a dropped
// membership update is self-correcting on the next join/leave or
// node-down purge, and `!`-style sends are infallible in BEAM.
return;
};
// Share the encoded frame across the fan-out so each connected node
// clones the `Arc` handle, not the bytes.
let frame: Arc<[u8]> = Arc::from(frame.into_boxed_slice());
// ENQUEUE to every currently-connected node and return immediately. The
// owned-runtime drain task performs the actual TCP I/O, so a slow or dead
// peer never blocks this scheduler worker thread. `connected_nodes()` is
// called with no PgState/propagation lock held (the registry drops its
// guards before invoking `broadcast`).
for node in dist.connections().connected_nodes() {
sender.enqueue(DistOutbound::ToNode {
node,
frame: Arc::clone(&frame),
});
}
}
}