fix(interactive): Fix decode error in groot (#3340)

This pull request improves the error handling, logging, and debugging of
the IPC communication between the executor and the store modules in the
interactive engine.

interactive_engine/executor/assembly/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/check_format.sh

@@ -34,4 +34,5 @@ CURRENT=$(pwd)
 for dir in "${directories[@]}"; do
   cd "${CURRENT}/${dir}"
   cargo +nightly fmt -- --check
+  #cargo +nightly fmt
 done

interactive_engine/executor/common/dyn_type/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/engine/pegasus/benchmark/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/engine/pegasus/network/src/lib.rs

@@ -221,7 +221,12 @@ pub fn ipc_channel_recv<T: Codec + 'static>(
 }
 
 pub fn check_ipc_ready(local: u64, remotes: &[u64]) -> bool {
-    crate::state::check_connect(local, remotes)
-        && crate::send::check_remotes_send_ready(local, remotes)
-        && crate::receive::check_remotes_read_ready(local, remotes)
+    let f1 = crate::state::check_connect(local, remotes);
+    let f2 = crate::send::check_remotes_send_ready(local, remotes);
+    let f3 = crate::receive::check_remotes_read_ready(local, remotes);
+    let ret = f1 && f2 && f3;
+    if !ret {
+        warn!("IPC not ready {}, {}, {}", f1, f2, f3);
+    }
+    ret
 }

interactive_engine/executor/engine/pegasus/network/src/receive/mod.rs

@@ -67,6 +67,7 @@ pub fn check_remotes_read_ready(local: u64, remotes: &[u64]) -> bool {
     for id in remotes.iter() {
         if *id != local {
             if !lock.contains_key(&(local, *id)) {
+                warn!("remote {} is not ready.", *id);
                 return false;
             }
         }

interactive_engine/executor/engine/pegasus/network/src/receive/net_rx.rs

@@ -80,7 +80,7 @@ impl Inbox {
                 self.buffer.push(msg);
             } else {
                 if let Err(_) = unsafe { (*tx).send(msg) } {
-                    error!("Inbox#push: send data failure;");
+                    error!("Channel {}, Inbox#push: send data failure;", self.channel_id);
                 }
             }
         }

interactive_engine/executor/engine/pegasus/network/src/send/mod.rs

@@ -143,6 +143,7 @@ pub fn check_remotes_send_ready(local: u64, remotes: &[u64]) -> bool {
     for id in remotes {
         if *id != local {
             if !lock.contains_key(&(local, *id)) {
+                warn!("remote {} is not ready.", *id);
                 return false;
             }
         }
@@ -216,6 +217,7 @@ pub(crate) fn start_net_sender(
             .name(format!("net-sender-{}", remote.id))
             .spawn(move || {
                 busy_send(&mut net_tx, is_block, timeout, local_id, remote.id, recv_poisoned);
+                error!("Connection to server {} lost", remote.id);
                 disconnected.store(true, Ordering::SeqCst);
                 net_tx
                     .take_writer()
@@ -233,6 +235,7 @@ pub(crate) fn start_net_sender(
             .name(format!("net-sender-{}", remote.id))
             .spawn(move || {
                 busy_send(&mut net_tx, is_block, timeout, local_id, remote.id, recv_poisoned);
+                error!("Connection to server {} lost", remote.id);
                 disconnected.store(true, Ordering::SeqCst);
                 net_tx
                     .take_writer()

interactive_engine/executor/engine/pegasus/network/src/state.rs

@@ -80,15 +80,21 @@ pub fn is_connected(local_id: u64, remote_id: u64) -> bool {
 
 pub fn check_connect(local: u64, remotes: &[u64]) -> bool {
     let states = CONNECTION_STATES.read().expect("lock poisoned");
+    let mut connect_status = true;
+    let mut disconnected_servers = vec![];
     for id in remotes {
         if *id != local
             && !states
                 .get(&(local, *id))
                 .map(|s| s.is_connected())
                 .unwrap_or(false)
         {
-            return false;
+            connect_status = false;
+            disconnected_servers.push(*id);
         }
     }
-    true
+    if !connect_status {
+        error!("Servers {:?} are not connected", disconnected_servers);
+    }
+    connect_status
 }

interactive_engine/executor/engine/pegasus/network/src/transport/block/mod.rs

@@ -116,7 +116,7 @@ pub fn connect<A: ToSocketAddrs>(
     debug!("connect to server {:?};", addr);
     let hb_sec = params.get_hb_interval_sec();
     super::setup_connection(local_id, hb_sec, &mut conn)?;
-    debug!("setup connection to {:?} success;", addr);
+    info!("setup connection to {:?} success;", addr);
     if let Some((id, hb_sec)) = super::check_connection(&mut conn)? {
         if id == remote_id {
             info!("connect server {} on {:?} success;", remote_id, addr);

interactive_engine/executor/engine/pegasus/pegasus/src/lib.rs

@@ -134,7 +134,7 @@ pub fn wait_servers_ready(server_conf: &ServerConf) {
         };
         if !remotes.is_empty() {
             while !pegasus_network::check_ipc_ready(local, &remotes) {
-                std::thread::sleep(std::time::Duration::from_millis(100));
+                std::thread::sleep(std::time::Duration::from_millis(1000));
                 info!("waiting remote servers connect ...");
             }
         }

interactive_engine/executor/engine/pegasus/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/ir/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/executor/store/groot/src/db/graph/codec.rs

@@ -186,23 +186,15 @@ impl Decoder {
     fn decode_var_len_property_at<'a>(
         &self, reader: &UnsafeBytesReader<'a>, idx: usize,
     ) -> Option<ValueRef<'a>> {
-        let info = &self.src.props[idx];
-        let offset = self.src.offsets[idx];
-        let end_off = bytes_to_len(reader.read_bytes(offset, 3));
-        let mut start_off = 0;
+        let end_off = bytes_to_len(reader.read_bytes(self.src.offsets[idx], 3));
+        let mut start_off = 0; // idx == self.src.fixed_len_prop_count
         if idx > self.src.fixed_len_prop_count {
-            let offset = self.src.offsets[idx - 1];
-            start_off = bytes_to_len(reader.read_bytes(offset, 3));
+            start_off = bytes_to_len(reader.read_bytes(self.src.offsets[idx - 1], 3));
         }
         let len = end_off - start_off;
-        if len > bytes_to_len(reader.read_bytes(*self.src.offsets.last().unwrap(), 3)) {
-            let msg = format!("fatal error! This codec cannot decode the bytes");
-            let err = gen_graph_err!(GraphErrorCode::DecodeError, msg);
-            error!("{:?}", err);
-            return None;
-        }
         let start_off = start_off + self.src.var_len_prop_start_offset;
         let bytes = reader.read_bytes(start_off, len);
+        let info = &self.src.props[idx];
         let ret = ValueRef::new(info.r#type, bytes);
         Some(ret)
     }

interactive_engine/executor/store/rustfmt.toml

@@ -5,5 +5,5 @@ unstable_features = true
 chain_width = 48
 max_width = 108
 use_small_heuristics = "Max"
-fn_args_layout = "Compressed"
+fn_params_layout = "Compressed"
 group_imports = "StdExternalCrate"

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogService.java

@@ -134,12 +134,29 @@ private AdminClient getAdmin() {
         if (this.adminClient == null) {
             synchronized (this) {
                 if (this.adminClient == null) {
-                    Map<String, Object> adminConfig = new HashMap<>();
-                    adminConfig.put("bootstrap.servers", this.servers);
-                    this.adminClient = AdminClient.create(adminConfig);
+                    try {
+                        this.adminClient = createAdminWithRetry();
+                    } catch (InterruptedException e) {
+                        logger.error("Create Kafka Client interrupted");
+                    }
                 }
             }
         }
         return this.adminClient;
     }
+
+    private AdminClient createAdminWithRetry() throws InterruptedException {
+        Map<String, Object> adminConfig = new HashMap<>();
+        adminConfig.put("bootstrap.servers", this.servers);
+
+        for (int i = 0; i < 10; ++i) {
+            try {
+                return AdminClient.create(adminConfig);
+            } catch (Exception e) {
+                logger.warn("Error creating Kafka AdminClient", e);
+                Thread.sleep(5000);
+            }
+        }
+        throw new RuntimeException("Create Kafka Client failed");
+    }
 }