Repository: risingwavelabs/risingwave
File: src/stream/src/executor/mod.rs (Rust)

Code before:
    pub fn is_empty(&self) -> bool {
        self.blocked.is_empty() && self.active.is_empty()
    }
}

// Explanation of why we need `DispatchBarrierBuffer`:
//
// When we need to create or replace an upstream fragment for the current fragment, the `Merge` operator must
// add some new upstream actor inputs. However, the `Merge` operator may still have old upstreams. We must wait
// for these old upstreams to completely process their barriers and align before we can safely update the
// `upstream-input-set`.
//
// Meanwhile, the creation of a new upstream actor can only succeed after the channel to the downstream `Merge`
// operator has been established. This creates a potential dependency chain: [new_actor_creation ->
// downstream_merge_update -> old_actor_processing]
//
// To address this, we split the application of a barrier's `Mutation` into two steps:
// 1. Parse the `Mutation`. If there is an addition on the upstream-set, establish a channel with the upstream
//    and cache it.
// 2. When the upstream barrier actually arrives, apply the cached upstream changes to the upstream-set
//
// Additionally, since receiving a barrier from current upstream input and from the `barrier_rx` are
// asynchronous, we cannot determine which will arrive first. Therefore, when a barrier is received from an
// upstream: if a cached mutation is present, we apply it. Otherwise, we must fetch a new barrier from
// `barrier_rx`.
pub(crate) struct DispatchBarrierBuffer {
    buffer: VecDeque<(Barrier, Option<Vec<BoxedActorInput>>)>,
    barrier_rx: mpsc::UnboundedReceiver<Barrier>,
    recv_state: BarrierReceiverState,
    curr_upstream_fragment_id: FragmentId,
    actor_id: ActorId,
    // read-only context for building new inputs
    build_input_ctx: Arc<BuildInputContext>,
}

struct BuildInputContext {
    pub actor_id: ActorId,
    pub local_barrier_manager: LocalBarrierManager,
    pub metrics: Arc<StreamingMetrics>,
    pub fragment_id: FragmentId,
}

type BoxedNewInputsFuture =
    Pin<Box<dyn Future<Output = StreamExecutorResult<Vec<BoxedActorInput>>> + Send>>;

enum BarrierReceiverState {
    ReceivingBarrier,
    CreatingNewInput(Barrier, BoxedNewInputsFuture),
}

impl DispatchBarrierBuffer {

Review comment (suggestion):
The `continuously_fetch_barrier_rx()` branch does not track elapsed time before returning the error. This means blocking time is not recorded when errors occur from the barrier receiver, leading to incomplete metrics. Consider adding `metrics.actor_input_buffer_blocking_duration_ns.inc_by(start_time.elapsed().as_nanos() as u64);` before line 1829.
```suggestion
                e = self.continuously_fetch_barrier_rx() => {
                    metrics
                        .actor_input_buffer_blocking_duration_ns
                        .inc_by(start_time.elapsed().as_nanos() as u64);
```

Code after:
        }
    }

    pub async fn await_next_message(
        &mut self,
        stream: &mut (impl Stream<Item = StreamExecutorResult<DispatcherMessage>> + Unpin),
        metrics: &ActorInputMetrics,
    ) -> StreamExecutorResult<DispatcherMessage> {
        let mut start_time = Instant::now();
        let interval_duration = Duration::from_secs(15);
        let mut interval =
            tokio::time::interval_at(start_time + interval_duration, interval_duration);

        loop {
            tokio::select! {
                biased;
                msg = stream.try_next() => {
                    metrics
                        .actor_input_buffer_blocking_duration_ns
                        .inc_by(start_time.elapsed().as_nanos() as u64);
                    return msg?.ok_or_else(
                        || StreamExecutorError::channel_closed("upstream executor closed unexpectedly")
                    );
                }

                e = self.continuously_fetch_barrier_rx() => {
                    return Err(e);
                }

                _ = interval.tick() => {
                    start_time = Instant::now();
                    metrics.actor_input_buffer_blocking_duration_ns.inc_by(interval_duration.as_nanos() as u64);
                    continue;
                }
            }
        }
    }

    pub async fn pop_barrier_with_inputs(
        &mut self,
        barrier: DispatcherBarrier,
    ) -> StreamExecutorResult<(Barrier, Option<Vec<BoxedActorInput>>)> {
        while self.buffer.is_empty() {
            self.try_fetch_barrier_rx(false).await?;
        }
        let (mut recv_barrier, inputs) = self.buffer.pop_front().unwrap();
        apply_dispatcher_barrier(&mut recv_barrier, barrier);

        Ok((recv_barrier, inputs))
    }