Struct LocalBarrierWorker

pub(super) struct LocalBarrierWorker {
    pub(super) state: ManagedBarrierState,
    await_epoch_completed_futures: HashMap<DatabaseId, FuturesOrdered<impl Future<Output = (PartialGraphId, Barrier, StreamResult<BarrierCompleteResult>)> + 'static>>,
    control_stream_handle: ControlStreamHandle,
    pub(super) actor_manager: Arc<StreamActorManager>,
}

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: HashMap<DatabaseId, 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>

Implementations

impl LocalBarrierWorker

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

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

pub(crate) fn get_or_insert_database_shared_context<'a>( current_shared_context: &'a mut HashMap<DatabaseId, Arc<SharedContext>>, database_id: DatabaseId, actor_manager: &StreamActorManager, ) -> &'a Arc<SharedContext>

async fn next_completed_epoch( futures: &mut HashMap<DatabaseId, FuturesOrdered<impl Future<Output = (PartialGraphId, Barrier, StreamResult<BarrierCompleteResult>)> + 'static>>, ) -> (DatabaseId, PartialGraphId, Barrier, StreamResult<BarrierCompleteResult>)

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

fn handle_streaming_control_request( &mut self, request: Request, ) -> Result<(), (DatabaseId, StreamError)>

fn handle_actor_op(&mut self, actor_op: LocalActorOperation)

impl LocalBarrierWorker

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

fn on_epoch_completed( &mut self, database_id: DatabaseId, 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 try_find_root_failure to find the root cause.

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

fn add_partial_graph( &mut self, database_id: DatabaseId, partial_graph_id: PartialGraphId, )

fn reset_database(&mut self, req: ResetDatabaseRequest)

fn ack_database_reset( &mut self, database_id: DatabaseId, reset_output: Option<ResetDatabaseOutput>, reset_request_id: u32, )

fn on_database_failure( &mut self, database_id: DatabaseId, failed_actor: Option<ActorId>, 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 responded with a message to notify about the error, and the global barrier worker will later reset and rerun the database.

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, init_request: InitRequest)

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

impl LocalBarrierWorker

pub fn update_actor_info( &mut self, database_id: DatabaseId, new_actor_infos: impl Iterator<Item = ActorInfo>, )

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