DistLink

DistLink: cross-node process monitors and bidirectional links (P5c).

Monitor protocol (two-step):

Local: call monitor(target_pid, timeout_ms) → sends MONITOR_REQ to

   the node that hosts target_pid, returns Ok(MonitorRef).
2. Remote node: receives MONITOR_REQ, calls march_dist_monitor_register
   in its C runtime.  On actor exit it sends MONITOR_FIRE back.
3. Local net-kernel loop: receives MONITOR_FIRE, delivers
   Down(ref, pid, reason) to the watcher's mailbox.

Node-down: when the peer connection drops (PeerDown from SwimDriver), the DistLink actor sweeps its pending_monitors table and fires Down(NodeDown) locally to all watchers of targets on that node.

Links build on monitors: both sides call monitor on each other. A Down message triggers EXIT propagation in the linked process (unless trapping).

Down reason: Normal | Killed | Crash(String) | NodeDown

MonitorRef is an opaque integer; demonitor(ref) cancels the monitor.

Types

typeDownReasonDownReason = Normal | Killed | Crash(String) | NodeDown#

typeMonitorRefMonitorRef = MonitorRef(Int)#

Opaque monitor reference returned by monitor/2.

typePendingMonitorPendingMonitor = {#

Internal: one entry in the pending monitor table.

typeMonitorTableMonitorTable = MonitorTable(Map(Int, PendingMonitor))#

Functions

fndecode_ackdecode_ack(bytes : List(Int)) : Result((Int, Int), String)#

Decode a MONITOR_ACK frame into (target_pid, watcher_pid). Returns Err for invalid MessagePack or any frame that is not exactly [12, target_pid, watcher_pid].

fndecode_firedecode_fire(bytes : List(Int)) : Result((Int, DownReason), String)#

Decode an inbound MONITOR_FIRE frame into (target_pid, reason). Returns Err for invalid MessagePack or a frame that is not [8, target_pid, reason_tag, reason_msg]; an unknown reason tag decodes as NodeDown.

fndecode_reqdecode_req(bytes : List(Int)) : Result((String, Int, Int), String)#

Decode an inbound MONITOR_REQ frame into (watcher_node, watcher_pid, target_pid). Returns Err for invalid MessagePack or a frame that is not [7, watcher_node, watcher_pid, target_pid].

fndown_msgdown_msg(m : PendingMonitor, reason : DownReason) : (MonitorRef, Int, DownReason)#

The (ref, target_pid, reason) Down tuple a watcher should receive for monitor [m]. Delivering it to the watcher's mailbox is the runtime glue's job.

fnempty_tableempty_table() : MonitorTable do MonitorTable(Map.empty()) end#

An empty pending-monitor table.

fnencode_ackencode_ack(target_pid : Int, watcher_pid : Int) : List(Int)#

Encode a MONITOR_ACK frame [12, target_pid, watcher_pid].

fnencode_fireencode_fire(target_pid : Int, reason : DownReason) : List(Int)#

Encode a MONITOR_FIRE frame (tag 8) reporting that [target_pid] exited with [reason]; resend_pending uses it to re-send unacked fires.

fnencode_reqencode_req(watcher_node : String, watcher_pid : Int, target_pid : Int) : List(Int)#

Encode a MONITOR_REQ frame (tag 7) asking the target's node to watch [target_pid] on behalf of [watcher_pid] at [watcher_node].

fnfind_by_target_pidfind_by_target_pid(tbl : MonitorTable, target_pid : Int) : Option(PendingMonitor)#

The first monitor whose target_pid is [target_pid], or None. O(n) scan. When several monitors (several watchers, or the same pid number on different nodes) match, which one is returned is unspecified: the target node is not compared.

fninsert_monitorinsert_monitor(tbl : MonitorTable, m : PendingMonitor) : MonitorTable#

Add [m] to the table keyed by its monitor ref, replacing any entry with the same ref.

fnmake_monitormake_monitor(ref : MonitorRef, target_node : String, target_pid : Int,#

Construct a PendingMonitor. Keeps record construction inside the module so callers avoid the bare-vs-qualified type name mismatch.

fnmonitors_on_nodemonitors_on_node(tbl : MonitorTable, node_id : String) : List(PendingMonitor)#

Every monitor whose target lives on [node_id], in unspecified order; used for the NodeDown sweep when that peer disconnects. O(n).

fnreason_of_tagreason_of_tag(tag : Int, msg : String) : DownReason#

Convert a wire reason tag to a DownReason: 0 Normal, 1 Killed, 2 Crash(msg), and any other tag (3, or anything unexpected, negative included) NodeDown. [msg] is ignored except for tag 2.

fnremove_monitorremove_monitor(tbl : MonitorTable, ref : MonitorRef) : MonitorTable#

Remove the entry for [ref]; the table is unchanged if it has none.

fnresend_pendingresend_pending(reg : PeerRegistry.Registry) : Int#

At-least-once delivery: resend every fired-but-unacked MONITOR_FIRE (from the runtime's pending-fire table) to the watcher node's CURRENT control connection, looked up in [reg], since the fd the fire was first written on may be gone. Returns how many frames were written; entries whose node is not in [reg], or whose write fails, wait for the next call. Entries stay pending until a MONITOR_ACK arrives. Call from a timer or after a reconnect.

fntag_acktag_ack() : Int do 12 end#

The wire tag of a MONITOR_ACK frame (12): the watcher's node has delivered the Down, so the firing node forgets the pending fire.

fntag_of_reasontag_of_reason(r : DownReason) : (Int, String)#

Convert a DownReason to its wire (tag, message) pair; the message is "" except for Crash(msg).