refactor: implement local barrier manager as event handler

src/compute/src/rpc/service/stream_service.rs

@@ -165,7 +165,7 @@ impl StreamService for StreamServiceImpl {
         }
 
         self.mgr
-            .send_barrier(&barrier, req.actor_ids_to_send, req.actor_ids_to_collect)
+            .send_barrier(barrier, req.actor_ids_to_send, req.actor_ids_to_collect)
             .await?;
 
         Ok(Response::new(InjectBarrierResponse {

src/stream/src/executor/actor.rs

@@ -219,7 +219,7 @@ where
             .into()));
 
             // Collect barriers to local barrier manager
-            self.context.lock_barrier_manager().collect(id, &barrier);
+            self.context.barrier_manager().collect(id, &barrier);
 
             // Then stop this actor if asked
             if barrier.is_stop(id) {

src/stream/src/executor/chain.rs

@@ -125,7 +125,6 @@ impl Executor for ChainExecutor {
 #[cfg(test)]
 mod test {
     use std::default::Default;
-    use std::sync::Arc;
 
     use futures::StreamExt;
     use risingwave_common::array::stream_chunk::StreamChunkTestExt;
@@ -144,8 +143,7 @@ mod test {
     #[tokio::test]
     async fn test_basic() {
         let barrier_manager = LocalBarrierManager::for_test();
-        let progress =
-            CreateMviewProgress::for_test(Arc::new(parking_lot::Mutex::new(barrier_manager)));
+        let progress = CreateMviewProgress::for_test(barrier_manager);
         let actor_id = progress.actor_id();
 
         let schema = Schema::new(vec![Field::unnamed(DataType::Int64)]);

src/stream/src/executor/values.rs

@@ -155,7 +155,6 @@ impl Executor for ValuesExecutor {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
 
     use futures::StreamExt;
     use risingwave_common::array::{
@@ -174,8 +173,7 @@ mod tests {
     #[tokio::test]
     async fn test_values() {
         let barrier_manager = LocalBarrierManager::for_test();
-        let progress =
-            CreateMviewProgress::for_test(Arc::new(parking_lot::Mutex::new(barrier_manager)));
+        let progress = CreateMviewProgress::for_test(barrier_manager);
         let actor_id = progress.actor_id();
         let (tx, barrier_receiver) = unbounded_channel();
         let value = StructValue::new(vec![Some(1.into()), Some(2.into()), Some(3.into())]);

src/stream/src/from_proto/barrier_recv.rs

@@ -37,7 +37,7 @@ impl ExecutorBuilder for BarrierRecvExecutorBuilder {
         let (sender, barrier_receiver) = unbounded_channel();
         stream
             .context
-            .lock_barrier_manager()
+            .barrier_manager()
             .register_sender(params.actor_context.id, sender);
 
         Ok(BarrierRecvExecutor::new(params.actor_context, params.info, barrier_receiver).boxed())

src/stream/src/from_proto/now.rs

@@ -36,7 +36,7 @@ impl ExecutorBuilder for NowExecutorBuilder {
         let (sender, barrier_receiver) = unbounded_channel();
         stream
             .context
-            .lock_barrier_manager()
+            .barrier_manager()
             .register_sender(params.actor_context.id, sender);
 
         let state_table =

src/stream/src/from_proto/source/trad_source.rs

@@ -44,7 +44,7 @@ impl ExecutorBuilder for SourceExecutorBuilder {
         let (sender, barrier_receiver) = unbounded_channel();
         stream
             .context
-            .lock_barrier_manager()
+            .barrier_manager()
             .register_sender(params.actor_context.id, sender);
         let system_params = params.env.system_params_manager_ref().get_params();
 

src/stream/src/from_proto/values.rs

@@ -39,7 +39,7 @@ impl ExecutorBuilder for ValuesExecutorBuilder {
         let (sender, barrier_receiver) = unbounded_channel();
         stream
             .context
-            .lock_barrier_manager()
+            .barrier_manager()
             .register_sender(params.actor_context.id, sender);
         let progress = stream
             .context