Merge pull request #568 from dora-rs/ignore-quicker-pending-drop-token

Ignore-quicker-pending-drop-token
dora-rs · Jul 3, 2024 · be1618a · be1618a
2 parents 320ef26 + 1f06034
commit be1618a
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 4 deletions.
diff --git a/apis/rust/node/src/event_stream/thread.rs b/apis/rust/node/src/event_stream/thread.rs
@@ -220,14 +220,14 @@ fn report_remaining_drop_tokens(
 
         let mut still_pending = Vec::new();
         for (token, rx, since, _) in pending_drop_tokens.drain(..) {
-            match rx.recv_timeout(Duration::from_millis(100)) {
+            match rx.recv_timeout(Duration::from_millis(50)) {
                 Ok(()) => return Err(eyre!("Node API should not send anything on ACK channel")),
                 Err(flume::RecvTimeoutError::Disconnected) => {
                     // the event was dropped -> add the drop token to the list
                     drop_tokens.push(token);
                 }
                 Err(flume::RecvTimeoutError::Timeout) => {
-                    let duration = Duration::from_secs(30);
+                    let duration = Duration::from_millis(200);
                     if since.elapsed() > duration {
                         tracing::warn!(
                             "timeout: node finished, but token {token:?} was still not \

diff --git a/apis/rust/node/src/node/mod.rs b/apis/rust/node/src/node/mod.rs
@@ -401,7 +401,7 @@ impl Drop for DoraNode {
                 );
             }
 
-            match self.drop_stream.recv_timeout(Duration::from_secs(10)) {
+            match self.drop_stream.recv_timeout(Duration::from_millis(500)) {
                 Ok(token) => {
                     self.sent_out_shared_memory.remove(&token);
                 }

diff --git a/binaries/daemon/src/lib.rs b/binaries/daemon/src/lib.rs
@@ -1567,7 +1567,7 @@ impl RunningDataflow {
         let running_nodes = self.running_nodes.clone();
         let grace_duration_kills = self.grace_duration_kills.clone();
         tokio::spawn(async move {
-            let duration = grace_duration.unwrap_or(Duration::from_millis(500));
+            let duration = grace_duration.unwrap_or(Duration::from_millis(2000));
             tokio::time::sleep(duration).await;
             let mut system = sysinfo::System::new();
             system.refresh_processes();