Struct LocalBarrierWorker

pub(super) struct LocalBarrierWorker {
    pub(super) state: ManagedBarrierState,
    await_epoch_completed_futures: FuturesOrdered<impl Future<Output = (PartialGraphId, Barrier, StreamResult<BarrierCompleteResult>)> + 'static>,
    control_stream_handle: ControlStreamHandle,
    pub(super) actor_manager: Arc<StreamActorManager>,
    pub(super) current_shared_context: Arc<SharedContext>,
    barrier_event_rx: UnboundedReceiver<LocalBarrierEvent>,
    actor_failure_rx: UnboundedReceiver<(ActorId, StreamError)>,
}

LocalBarrierWorker manages barrier control flow, used by local stream manager. Specifically, LocalBarrierWorker serve barrier injection from meta server, send the barriers to and collect them from all actors, and finally report the progress.

Fields

state: ManagedBarrierState

Current barrier collection state.

await_epoch_completed_futures: FuturesOrdered<impl Future<Output = (PartialGraphId, Barrier, StreamResult<BarrierCompleteResult>)> + 'static>

Futures will be finished in the order of epoch in ascending order.

control_stream_handle: ControlStreamHandle

actor_manager: Arc<StreamActorManager>

current_shared_context: Arc<SharedContext>

barrier_event_rx: UnboundedReceiver<LocalBarrierEvent>

actor_failure_rx: UnboundedReceiver<(ActorId, StreamError)>

Implementations

impl LocalBarrierWorker

pub(crate) fn update_create_mview_progress( &mut self, epoch: EpochPair, actor: ActorId, state: BackfillState, )

impl LocalBarrierWorker

pub(super) fn new( actor_manager: Arc<StreamActorManager>, initial_partial_graphs: Vec<InitialPartialGraph>, ) -> Self

fn to_debug_info(&self) -> LocalBarrierWorkerDebugInfo<'_>

async fn run(self, actor_op_rx: UnboundedReceiver<LocalActorOperation>)

fn handle_streaming_control_request( &mut self, request: StreamingControlStreamRequest, ) -> StreamResult<()>

fn handle_barrier_event( &mut self, event: LocalBarrierEvent, ) -> Result<(), (ActorId, StreamError)>

fn handle_actor_op(&mut self, actor_op: LocalActorOperation)

impl LocalBarrierWorker

fn complete_barrier( &mut self, partial_graph_id: PartialGraphId, prev_epoch: u64, )

fn on_epoch_completed( &mut self, partial_graph_id: PartialGraphId, epoch: u64, result: BarrierCompleteResult, )

fn send_barrier( &mut self, barrier: &Barrier, request: InjectBarrierRequest, ) -> StreamResult<()>

Broadcast a barrier to all senders. Save a receiver which will get notified when this barrier is finished, in managed mode.

Note that the error returned here is typically a StreamError::barrier_send, which is not the root cause of the failure. The caller should then call Self::try_find_root_failure to find the root cause.

fn remove_partial_graphs( &mut self, partial_graph_ids: impl Iterator<Item = PartialGraphId>, )

pub(super) fn add_partial_graph(&mut self, partial_graph_id: PartialGraphId)

pub(super) fn reset_state( &mut self, initial_partial_graphs: Vec<InitialPartialGraph>, )

Reset all internal states.

fn collect(&mut self, actor_id: ActorId, epoch: EpochPair)

When a crate::executor::StreamConsumer (typically crate::executor::DispatchExecutor) get a barrier, it should report and collect this barrier with its own actor_id using this function.

async fn notify_actor_failure( &mut self, actor_id: ActorId, err: StreamError, err_context: &'static str, )

When a actor exit unexpectedly, the error is reported using this function. The control stream will be reset and the meta service will then trigger recovery.

async fn notify_other_failure( &mut self, err: StreamError, message: impl Into<String>, )

When some other failure happens (like failed to send barrier), the error is reported using this function. The control stream will be reset and the meta service will then trigger recovery.

This is similar to Self::notify_actor_failure, but since there’s not always an actor failure, the given err will be used if there’s no root failure found.

async fn try_find_root_failure( &mut self, first_err: StreamError, ) -> ScoredStreamError

Collect actor errors for a while and find the one that might be the root cause.

Returns None if there’s no actor error received.

impl LocalBarrierWorker

pub fn spawn( env: StreamEnvironment, streaming_metrics: Arc<StreamingMetrics>, await_tree_reg: Option<Registry>, watermark_epoch: AtomicU64Ref, actor_op_rx: UnboundedReceiver<LocalActorOperation>, ) -> JoinHandle<()>

Create a LocalBarrierWorker with managed mode.

impl LocalBarrierWorker

pub(super) async fn reset( &mut self, initial_partial_graphs: Vec<InitialPartialGraph>, )

Force stop all actors on this worker, and then drop their resources.

impl LocalBarrierWorker

pub fn update_actor_info( &self, new_actor_infos: impl Iterator<Item = ActorInfo>, ) -> StreamResult<()>

This function could only be called once during the lifecycle of LocalStreamManager for now.