flutter and macos work

2022-06-10 17:07:10 -04:00
parent 86567ea78d
commit 5931f1254f
33 changed files with 312 additions and 96 deletions
--- a/veilid-core/src/network_manager/mod.rs
+++ b/veilid-core/src/network_manager/mod.rs
@@ -233,6 +233,7 @@ impl NetworkManager {
        self.inner.lock().relay_node.clone()
    }

+    #[instrument(level = "debug", skip_all, err)]
    pub async fn init(&self, update_callback: UpdateCallback) -> Result<(), String> {
        let routing_table = RoutingTable::new(self.clone());
        routing_table.init().await?;
@@ -240,6 +241,8 @@ impl NetworkManager {
        self.inner.lock().update_callback = Some(update_callback);
        Ok(())
    }
+
+    #[instrument(level = "debug", skip_all)]
    pub async fn terminate(&self) {
        let routing_table = {
            let mut inner = self.inner.lock();
@@ -251,6 +254,7 @@ impl NetworkManager {
        self.inner.lock().update_callback = None;
    }

+    #[instrument(level = "debug", skip_all, err)]
    pub async fn internal_startup(&self) -> Result<(), String> {
        trace!("NetworkManager::internal_startup begin");
        if self.inner.lock().components.is_some() {
@@ -281,6 +285,7 @@ impl NetworkManager {
        Ok(())
    }

+    #[instrument(level = "debug", skip_all, err)]
    pub async fn startup(&self) -> Result<(), String> {
        if let Err(e) = self.internal_startup().await {
            self.shutdown().await;
@@ -292,6 +297,7 @@ impl NetworkManager {
        Ok(())
    }

+    #[instrument(level = "debug", skip_all)]
    pub async fn shutdown(&self) {
        trace!("NetworkManager::shutdown begin");

@@ -339,6 +345,7 @@ impl NetworkManager {
        }
    }

+    #[instrument(level = "trace", skip(self), ret)]
    pub fn check_client_whitelist(&self, client: DHTKey) -> bool {
        let mut inner = self.inner.lock();

@@ -351,6 +358,7 @@ impl NetworkManager {
        }
    }

+    #[instrument(level = "trace", skip(self))]
    pub fn purge_client_whitelist(&self) {
        let timeout_ms = self.config.get().network.client_whitelist_timeout_ms;
        let mut inner = self.inner.lock();
@@ -366,6 +374,15 @@ impl NetworkManager {
        }
    }

+    #[instrument(level = "debug", skip_all, err)]
+    async fn restart_net(&self, net: Network) -> Result<(), String> {
+        net.shutdown().await;
+        self.send_network_update();
+        net.startup().await?;
+        self.send_network_update();
+        Ok(())
+    }
+
    pub async fn tick(&self) -> Result<(), String> {
        let (routing_table, net, receipt_manager) = {
            let inner = self.inner.lock();
@@ -380,10 +397,7 @@ impl NetworkManager {
        // If the network needs to be reset, do it
        // if things can't restart, then we fail out of the attachment manager
        if net.needs_restart() {
-            net.shutdown().await;
-            self.send_network_update();
-            net.startup().await?;
-            self.send_network_update();
+            self.restart_net(net.clone()).await?;
        }

        // Run the rolling transfers task
@@ -448,6 +462,7 @@ impl NetworkManager {
    }

    // Generates a multi-shot/normal receipt
+    #[instrument(level = "trace", skip(self, extra_data, callback), err)]
    pub fn generate_receipt<D: AsRef<[u8]>>(
        &self,
        expiration_us: u64,
@@ -473,6 +488,7 @@ impl NetworkManager {
    }

    // Generates a single-shot/normal receipt
+    #[instrument(level = "trace", skip(self, extra_data), err)]
    pub fn generate_single_shot_receipt<D: AsRef<[u8]>>(
        &self,
        expiration_us: u64,
@@ -498,6 +514,7 @@ impl NetworkManager {
    }

    // Process a received out-of-band receipt
+    #[instrument(level = "trace", skip(self, receipt_data), err)]
    pub async fn handle_out_of_band_receipt<R: AsRef<[u8]>>(
        &self,
        receipt_data: R,
@@ -511,6 +528,7 @@ impl NetworkManager {
    }

    // Process a received in-band receipt
+    #[instrument(level = "trace", skip(self, receipt_data), err)]
    pub async fn handle_in_band_receipt<R: AsRef<[u8]>>(
        &self,
        receipt_data: R,
@@ -527,6 +545,7 @@ impl NetworkManager {
    }

    // Process a received signal
+    #[instrument(level = "trace", skip(self), err)]
    pub async fn handle_signal(&self, signal_info: SignalInfo) -> Result<(), String> {
        match signal_info {
            SignalInfo::ReverseConnect { receipt, peer_info } => {
@@ -588,6 +607,7 @@ impl NetworkManager {
    }

    // Builds an envelope for sending over the network
+    #[instrument(level = "trace", skip(self, body), err)]
    fn build_envelope<B: AsRef<[u8]>>(
        &self,
        dest_node_id: DHTKey,
@@ -614,6 +634,7 @@ impl NetworkManager {
    // node_ref is the direct destination to which the envelope will be sent
    // If 'node_id' is specified, it can be different than node_ref.node_id()
    // which will cause the envelope to be relayed
+    #[instrument(level = "trace", skip(self, body), ret, err)]
    pub async fn send_envelope<B: AsRef<[u8]>>(
        &self,
        node_ref: NodeRef,
@@ -665,6 +686,7 @@ impl NetworkManager {
    }

    // Called by the RPC handler when we want to issue an direct receipt
+    #[instrument(level = "trace", skip(self, rcpt_data), err)]
    pub async fn send_out_of_band_receipt(
        &self,
        dial_info: DialInfo,
@@ -683,6 +705,7 @@ impl NetworkManager {
    }

    // Figure out how to reach a node
+    #[instrument(level = "trace", skip(self), ret, err)]
    fn get_contact_method(&self, mut target_node_ref: NodeRef) -> Result<ContactMethod, String> {
        let routing_table = self.routing_table();

@@ -802,6 +825,7 @@ impl NetworkManager {

    // Send a reverse connection signal and wait for the return receipt over it
    // Then send the data across the new connection
+    #[instrument(level = "trace", skip(self, data), err)]
    pub async fn do_reverse_connect(
        &self,
        relay_nr: NodeRef,
@@ -872,6 +896,7 @@ impl NetworkManager {

    // Send a hole punch signal and do a negotiating ping and wait for the return receipt
    // Then send the data across the new connection
+    #[instrument(level = "trace", skip(self, data), err)]
    pub async fn do_hole_punch(
        &self,
        relay_nr: NodeRef,
@@ -1036,6 +1061,7 @@ impl NetworkManager {
    // Called when a packet potentially containing an RPC envelope is received by a low-level
    // network protocol handler. Processes the envelope, authenticates and decrypts the RPC message
    // and passes it to the RPC handler
+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    async fn on_recv_envelope(
        &self,
        data: &[u8],
@@ -1175,6 +1201,7 @@ impl NetworkManager {
    }

    // Keep relays assigned and accessible
+    #[instrument(level = "trace", skip(self), err)]
    async fn relay_management_task_routine(self, _last_ts: u64, cur_ts: u64) -> Result<(), String> {
        // log_net!("--- network manager relay_management task");

@@ -1227,6 +1254,7 @@ impl NetworkManager {
    }

    // Compute transfer statistics for the low level network
+    #[instrument(level = "trace", skip(self), err)]
    async fn rolling_transfers_task_routine(self, last_ts: u64, cur_ts: u64) -> Result<(), String> {
        // log_net!("--- network manager rolling_transfers task");
        {
--- a/veilid-core/src/network_manager/native/mod.rs
+++ b/veilid-core/src/network_manager/native/mod.rs
@@ -267,6 +267,7 @@ impl Network {
    // This creates a short-lived connection in the case of connection-oriented protocols
    // for the purpose of sending this one message.
    // This bypasses the connection table as it is not a 'node to node' connection.
+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_unbound_to_dial_info(
        &self,
        dial_info: DialInfo,
@@ -300,6 +301,7 @@ impl Network {
        res
    }

+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_to_existing_connection(
        &self,
        descriptor: ConnectionDescriptor,
@@ -357,6 +359,7 @@ impl Network {
    }

    // Send data directly to a dial info, possibly without knowing which node it is going to
+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_to_dial_info(
        &self,
        dial_info: DialInfo,
@@ -404,9 +407,8 @@ impl Network {
        self.inner.lock().protocol_config
    }

+    #[instrument(level = "debug", err, skip_all)]
    pub async fn startup(&self) -> Result<(), String> {
-        trace!("startup network");
-
        // initialize interfaces
        let mut interfaces = NetworkInterfaces::new();
        interfaces.refresh().await?;
@@ -492,10 +494,12 @@ impl Network {
        self.inner.lock().network_started
    }

+    #[instrument(level = "debug", skip_all)]
    pub fn restart_network(&self) {
        self.inner.lock().network_needs_restart = true;
    }

+    #[instrument(level = "debug", skip_all)]
    pub async fn shutdown(&self) {
        info!("stopping network");

@@ -524,6 +528,7 @@ impl Network {
        inner.network_class
    }

+    #[instrument(level = "debug", skip_all)]
    pub fn reset_network_class(&self) {
        let mut inner = self.inner.lock();
        inner.network_class = None;
--- a/veilid-core/src/network_manager/native/network_class_discovery.rs
+++ b/veilid-core/src/network_manager/native/network_class_discovery.rs
@@ -51,30 +51,30 @@ impl DiscoveryContext {
    // Pick the best network class we have seen so far
    pub fn set_detected_network_class(&self, network_class: NetworkClass) {
        let mut inner = self.inner.lock();
-        log_net!( debug
-            "=== set_detected_network_class {:?} {:?}: {:?} ===",
-            inner.protocol_type,
-            inner.address_type,
-            network_class
+        debug!(target: "net",
+            protocol_type=?inner.protocol_type,
+            address_type=?inner.address_type,
+            ?network_class,
+            "set_detected_network_class"
        );
-
        inner.detected_network_class = Some(network_class);
    }

    pub fn set_detected_public_dial_info(&self, dial_info: DialInfo, class: DialInfoClass) {
        let mut inner = self.inner.lock();
-        log_net!( debug
-            "=== set_detected_public_dial_info {:?} {:?}: {} {:?} ===",
-            inner.protocol_type,
-            inner.address_type,
-            dial_info,
-            class
+        debug!(target: "net",
+            protocol_type=?inner.protocol_type,
+            address_type=?inner.address_type,
+            ?dial_info,
+            ?class,
+            "set_detected_public_dial_info"
        );
        inner.detected_public_dial_info = Some(DetectedPublicDialInfo { dial_info, class });
    }

    // Ask for a public address check from a particular noderef
    // This is done over the normal port using RPC
+    #[instrument(level = "trace", skip(self), ret)]
    async fn request_public_address(&self, node_ref: NodeRef) -> Option<SocketAddress> {
        let rpc = self.routing_table.rpc_processor();
        rpc.rpc_call_status(node_ref.clone())
@@ -93,6 +93,7 @@ impl DiscoveryContext {

    // find fast peers with a particular address type, and ask them to tell us what our external address is
    // This is done over the normal port using RPC
+    #[instrument(level = "trace", skip(self), ret)]
    async fn discover_external_address(
        &self,
        protocol_type: ProtocolType,
@@ -122,6 +123,7 @@ impl DiscoveryContext {
    }

    // This pulls the already-detected local interface dial info from the routing table
+    #[instrument(level = "trace", skip(self), ret)]
    fn get_local_addresses(
        &self,
        protocol_type: ProtocolType,
@@ -143,6 +145,7 @@ impl DiscoveryContext {
            .collect()
    }

+    #[instrument(level = "trace", skip(self), ret)]
    async fn validate_dial_info(
        &self,
        node_ref: NodeRef,
@@ -165,6 +168,7 @@ impl DiscoveryContext {
            .unwrap_or(false)
    }

+    #[instrument(level = "trace", skip(self), ret)]
    async fn try_port_mapping(&self) -> Option<DialInfo> {
        //xxx
        None
@@ -189,6 +193,7 @@ impl DiscoveryContext {
    ///////
    // Per-protocol discovery routines

+    #[instrument(level = "trace", skip(self))]
    pub fn protocol_begin(&self, protocol_type: ProtocolType, address_type: AddressType) {
        // Get our interface addresses
        let intf_addrs = self.get_local_addresses(protocol_type, address_type);
@@ -203,6 +208,7 @@ impl DiscoveryContext {
    }

    // Get our first node's view of our external IP address via normal RPC
+    #[instrument(level = "trace", skip(self), ret)]
    pub async fn protocol_get_external_address_1(&self) -> bool {
        let (protocol_type, address_type) = {
            let inner = self.inner.lock();
@@ -234,6 +240,7 @@ impl DiscoveryContext {
    }

    // If we know we are not behind NAT, check our firewall status
+    #[instrument(level = "trace", skip(self), err)]
    pub async fn protocol_process_no_nat(&self) -> Result<(), String> {
        let (node_1, external_1_dial_info) = {
            let inner = self.inner.lock();
@@ -264,6 +271,7 @@ impl DiscoveryContext {
    }

    // If we know we are behind NAT check what kind
+    #[instrument(level = "trace", skip(self), ret, err)]
    pub async fn protocol_process_nat(&self) -> Result<bool, String> {
        let (node_1, external_1_dial_info, external_1_address, protocol_type, address_type) = {
            let inner = self.inner.lock();
@@ -353,6 +361,7 @@ impl DiscoveryContext {
 }

 impl Network {
+    #[instrument(level = "trace", skip(self, context), err)]
    pub async fn update_ipv4_protocol_dialinfo(
        &self,
        context: &DiscoveryContext,
@@ -414,6 +423,7 @@ impl Network {
        Ok(())
    }

+    #[instrument(level = "trace", skip(self, context), err)]
    pub async fn update_ipv6_protocol_dialinfo(
        &self,
        context: &DiscoveryContext,
@@ -454,9 +464,8 @@ impl Network {
        Ok(())
    }

+    #[instrument(level = "trace", skip(self), err)]
    pub async fn update_network_class_task_routine(self, _l: u64, _t: u64) -> Result<(), String> {
-        log_net!("--- updating network class");
-
        // Ensure we aren't trying to update this without clearing it first
        let old_network_class = self.inner.lock().network_class;
        assert_eq!(old_network_class, None);
--- a/veilid-core/src/network_manager/wasm/mod.rs
+++ b/veilid-core/src/network_manager/wasm/mod.rs
@@ -51,6 +51,7 @@ impl Network {

    /////////////////////////////////////////////////////////////////

+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_unbound_to_dial_info(
        &self,
        dial_info: DialInfo,
@@ -79,6 +80,7 @@ impl Network {
        res
    }

+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_to_existing_connection(
        &self,
        descriptor: ConnectionDescriptor,
@@ -115,6 +117,7 @@ impl Network {
        }
    }

+    #[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
    pub async fn send_data_to_dial_info(
        &self,
        dial_info: DialInfo,