fix(interactive): event message error

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

@@ -46,6 +46,7 @@ pub struct Worker<D: Data, T: Debug + Send + 'static> {
     sink: ResultSink<T>,
     resources: ResourceMap,
     keyed_resources: KeyedResources,
+    is_finished: bool,
     _ph: std::marker::PhantomData<D>,
 }
 
@@ -65,6 +66,7 @@ impl<D: Data, T: Debug + Send + 'static> Worker<D, T> {
             sink,
             resources: ResourceMap::default(),
             keyed_resources: KeyedResources::default(),
+            is_finished: false,
             _ph: std::marker::PhantomData,
         }
     }
@@ -131,7 +133,7 @@ impl<D: Data, T: Debug + Send + 'static> Worker<D, T> {
     }
 
     fn release(&mut self) {
-        if self.peer_guard.fetch_sub(1, Ordering::SeqCst) == 1 {
+        if self.peer_guard.load(Ordering::SeqCst) == 0 {
             pegasus_memory::alloc::remove_task(self.conf.job_id as usize);
         }
         if !crate::remove_cancel_hook(self.conf.job_id).is_ok() {
@@ -237,9 +239,18 @@ impl<D: Data, T: Debug + Send + 'static> Task for Worker<D, T> {
                         self.id.job_id,
                         self.conf.job_name,
                         self.start.elapsed().as_millis()
-                    )
+                    );
+                    self.is_finished = true;
+                    // if this is last worker, return Finished
+                    if self.peer_guard.fetch_sub(1, Ordering::SeqCst) == 1 {
+                        state
+                    } else {
+                        // if other workers are not finished, return NotReady until all workers finished
+                        TaskState::NotReady
+                    }
+                } else {
+                    state
                 }
-                state
             }
             Err(e) => {
                 error_worker!("job({}) execute error: {}", self.id.job_id, e);
@@ -255,23 +266,33 @@ impl<D: Data, T: Debug + Send + 'static> Task for Worker<D, T> {
             self.sink.set_cancel_hook(true);
             return TaskState::Finished;
         }
-
-        match self.task.check_ready() {
-            Ok(state) => {
-                if TaskState::Finished == state {
-                    info_worker!(
-                        "job({}) '{}' finished, used {:?};",
-                        self.id.job_id,
-                        self.conf.job_name,
-                        self.start.elapsed()
-                    );
+        if !self.is_finished {
+            match self.task.check_ready() {
+                Ok(state) => {
+                    {
+                        if TaskState::Finished == state {
+                            info_worker!(
+                                "job({}) '{}' finished, used {:?};",
+                                self.id.job_id,
+                                self.conf.job_name,
+                                self.start.elapsed()
+                            );
+                        }
+                    }
+                    state
+                }
+                Err(e) => {
+                    error_worker!("job({}) execute error: {}", self.id.job_id, e);
+                    self.sink.on_error(e);
+                    TaskState::Finished
                 }
-                state
             }
-            Err(e) => {
-                error_worker!("job({}) execute error: {}", self.id.job_id, e);
-                self.sink.on_error(e);
-                TaskState::Finished
+        } else {
+            // all workers are finished, return state Finished
+            if self.peer_guard.load(Ordering::SeqCst) == 0 {
+                return TaskState::Finished;
+            } else {
+                return TaskState::NotReady;
             }
         }
     }

interactive_engine/executor/engine/pegasus/pegasus/tests/limit_test.rs

@@ -18,7 +18,7 @@
 extern crate lazy_static;
 
 use pegasus::api::{
-    Collect, CorrelatedSubTask, HasAny, Iteration, Limit, Map, Merge, Sink, SortLimit, SortLimitBy,
+    Collect, CorrelatedSubTask, Count, HasAny, Iteration, Limit, Map, Merge, Sink, SortLimit, SortLimitBy,
 };
 use pegasus::JobConf;
 
@@ -35,6 +35,28 @@ lazy_static! {
     .collect();
 }
 
+// worker 1 finished before worker 0 start
+#[test]
+fn limit_filtermap_flatmap_count_test() {
+    let mut conf = JobConf::new("limit_filtermap_flatmap_count_test");
+    conf.set_workers(2);
+    let mut result = pegasus::run(conf, || {
+        |input, output| {
+            input
+                .input_from(1..50000000u32)?
+                .limit(1000000)?
+                .flat_map(|x| Ok(vec![x + 1].into_iter()))?
+                .count()?
+                .sink_into(output)
+        }
+    })
+    .expect("build job failure");
+
+    while let Some(v) = result.next() {
+        println!("get {}", v.unwrap());
+    }
+}
+
 // the most common case with early-stop
 #[test]
 fn flatmap_limit_test() {