fix(interactive): fix memory leak in pegasus (#3367)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes
#3366

interactive_engine/executor/engine/pegasus/common/src/rc.rs

@@ -97,7 +97,9 @@ impl<T: std::fmt::Debug> std::fmt::Debug for RcPointer<T> {
 impl<T: ?Sized> Drop for RcPointer<T> {
     fn drop(&mut self) {
         if self.count.fetch_sub(1, Ordering::SeqCst) == 1 {
-            unsafe { std::ptr::drop_in_place(self.ptr.as_ptr()) }
+            unsafe {
+                let _ = Box::<T>::from_raw(self.ptr.as_ptr());
+            }
         }
     }
 }

interactive_engine/executor/engine/pegasus/pegasus/src/errors/mod.rs

@@ -343,6 +343,7 @@ impl From<std::io::Error> for StartupError {
 #[derive(Debug)]
 pub enum CancelError {
     JobNotFoundError(u64),
+    CancelMapPoisonedError,
 }
 
 impl Display for CancelError {
@@ -351,6 +352,9 @@ impl Display for CancelError {
             CancelError::JobNotFoundError(e) => {
                 write!(f, "fail to find job, job id: {};", e)
             }
+            CancelError::CancelMapPoisonedError => {
+                write!(f, "JOB_CANCEL_MAP is poisoned!;")
+            }
         }
     }
 }

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

@@ -299,11 +299,23 @@ where
 }
 
 pub fn cancel_job(job_id: u64) -> Result<(), CancelError> {
-    let mut hook = JOB_CANCEL_MAP.write().expect("lock poisoned");
-    if let Some(cancel_hook) = hook.get_mut(&job_id) {
-        cancel_hook.store(true, Ordering::SeqCst);
+    if let Ok(mut hook) = JOB_CANCEL_MAP.write() {
+        if let Some(cancel_hook) = hook.get_mut(&job_id) {
+            cancel_hook.store(true, Ordering::SeqCst);
+        } else {
+            return Err(CancelError::JobNotFoundError(job_id));
+        }
+    } else {
+        return Err(CancelError::CancelMapPoisonedError);
+    }
+    Ok(())
+}
+
+pub fn remove_cancel_hook(job_id: u64) -> Result<(), CancelError> {
+    if let Ok(mut hook) = JOB_CANCEL_MAP.write() {
+        hook.remove(&job_id);
     } else {
-        return Err(CancelError::JobNotFoundError(job_id));
+        return Err(CancelError::CancelMapPoisonedError);
     }
     Ok(())
 }

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

@@ -132,6 +132,9 @@ impl<D: Data, T: Debug + Send + 'static> Worker<D, T> {
 
     fn release(&mut self) {
         self.peer_guard.fetch_sub(1, Ordering::SeqCst);
+        if !crate::remove_cancel_hook(self.conf.job_id).is_ok() {
+            error!("JOB_CANCEL_MAP is poisoned!");
+        }
     }
 }