mm1-node 0.7.23

An Erlang-style actor runtime for Rust.
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

use std::collections::{BTreeMap, HashMap, VecDeque};
use std::sync::Arc;

use mm1_address::address::Address;
use mm1_address::address_range::AddressRange;
use mm1_address::pool::{Lease, Pool};
use mm1_address::subnet::NetAddress;
use mm1_common::log;
use mm1_common::types::AnyError;
use mm1_core::envelope::Envelope;
use mm1_core::tap::MessageTap;
use mm1_core::tracing::TraceId;
use tokio::sync::{Notify, mpsc};

use crate::actor_key::ActorKey;
use crate::config::{Mm1NodeConfig, Valid};
use crate::registry::{MessageWithPermit, MessageWithoutPermit, WeakSubnetMailboxTx};
use crate::runtime::rt_api::RtApi;
use crate::runtime::sys_call;
use crate::runtime::sys_msg::SysMsg;

mod impl_context_api;

pub struct ActorContext {
    pub(crate) fork_address: Address,
    pub(crate) fork_lease:   Option<Lease>,
    pub(crate) ack_to:       Option<Address>,
    pub(crate) call:         sys_call::Tx,

    pub(crate) subnet_context: Arc<spin::lock_api::Mutex<SubnetContext>>,
}

#[derive(Default)]
pub(crate) struct ForkEntry {
    fork_notifiy:       Arc<Notify>,
    inbox_priority:     VecDeque<MessageWithoutPermit<Envelope>>,
    inbox_regular:      VecDeque<MessageWithPermit<Envelope>>,
    last_ping_received: Option<u64>,
}

pub(crate) struct SubnetContext {
    pub(crate) rt_api:    RtApi,
    pub(crate) rt_config: Arc<Valid<Mm1NodeConfig>>,

    pub(crate) actor_key:      ActorKey,
    pub(crate) subnet_pool:    Pool,
    pub(crate) subnet_address: NetAddress,

    pub(crate) subnet_mailbox_tx: WeakSubnetMailboxTx<(TraceId, SysMsg), Envelope>,

    pub(crate) subnet_notify: Arc<Notify>,
    pub(crate) rx_priority:   kanal::Receiver<MessageWithoutPermit<Envelope>>,
    pub(crate) rx_regular:    kanal::Receiver<MessageWithPermit<Envelope>>,

    pub(crate) fork_entries:  HashMap<Address, ForkEntry>,
    pub(crate) bound_subnets: BTreeMap<AddressRange, Address>,
    pub(crate) message_tap:   Arc<dyn MessageTap>,

    // TODO: send the NetAddress instead of Address here
    pub(crate) tx_actor_failure: mpsc::UnboundedSender<(Address, AnyError)>,
}

impl Drop for ActorContext {
    fn drop(&mut self) {
        let Self {
            fork_address,
            subnet_context,
            ..
        } = self;
        let mut subnet_context_locked = subnet_context
            .try_lock()
            .expect("could not lock subnet_context");
        let SubnetContext {
            rt_api,
            subnet_address,
            subnet_mailbox_tx,
            fork_entries,
            bound_subnets,
            ..
        } = &mut *subnet_context_locked;

        let registry = rt_api.registry();

        let mut unbound_subnets = vec![];
        for (address_range, bound_by) in bound_subnets.iter() {
            if bound_by == fork_address {
                unbound_subnets.push(*address_range);
            }
        }

        for address_range in unbound_subnets {
            bound_subnets.remove(&address_range);
            if !registry.unregister(address_range.into()) {
                log::error!(%address_range, "could not unregister bound subnet");
            }
        }
        if fork_entries.remove(fork_address).is_none() {
            log::error!(
                fork = %fork_address,
                "nothing actually removed from fork_entries"
            );
        }

        let fork_lease_opt = self.fork_lease.take();
        let subnet_system_tx_opt = subnet_mailbox_tx.tx_system.upgrade();

        if let Some((fork_lease, system_tx)) = fork_lease_opt.zip(subnet_system_tx_opt) {
            let fork_done = SysMsg::ForkDone(fork_lease);
            let _ = system_tx.send((TraceId::current(), fork_done));
        }

        if fork_entries.is_empty() {
            let _ = registry.unregister(*subnet_address);
        }
    }
}