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>,
    pub(super) term_id: String,
}

LocalBarrierWorker manages barrier control flow. Specifically, LocalBarrierWorker serves barrier injection from meta server, sends the barriers to and collects them from all actors, and finally reports the progress.

Runs event loop in Self::run. Handles events sent by crate::task::LocalStreamManager.

See crate::task for architecture overview.

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>

term_id: String

Implementations

impl LocalBarrierWorker

pub(super) fn new( actor_manager: Arc<StreamActorManager>, term_id: String, ) -> Self

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

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.

async fn reset(&mut self, init_request: InitRequest)

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

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.