Struct ControlStreamManager 

pub(super) struct ControlStreamManager {
    workers: HashMap<WorkerId, (WorkerNode, WorkerNodeState)>,
    pub env: MetaSrvEnv,
}

Fields

workers: HashMap<WorkerId, (WorkerNode, WorkerNodeState)>

env: MetaSrvEnv

Implementations

impl ControlStreamManager

pub(super) fn new(env: MetaSrvEnv) -> Self

pub(super) fn host_addr(&self, worker_id: WorkerId) -> HostAddress

pub(super) async fn add_worker( &mut self, node: WorkerNode, inflight_infos: impl Iterator<Item = (DatabaseId, impl Iterator<Item = TableId>)>, term_id: String, context: &impl GlobalBarrierWorkerContext, )

pub(super) fn remove_worker(&mut self, node: WorkerNode)

fn retry_connect( node: WorkerNode, term_id: String, context: Arc<impl GlobalBarrierWorkerContext>, ) -> BoxFuture<'static, StreamingControlHandle>

pub(super) async fn recover( env: MetaSrvEnv, nodes: &HashMap<WorkerId, WorkerNode>, term_id: &str, context: Arc<impl GlobalBarrierWorkerContext>, ) -> Self

pub(super) fn clear(&mut self)

Clear all nodes and response streams in the manager.

impl ControlStreamManager

fn poll_next_event<'a>( this_opt: &mut Option<&'a mut Self>, cx: &mut Context<'_>, term_id: &str, context: &Arc<impl GlobalBarrierWorkerContext>, poll_reconnect: bool, ) -> Poll<(WorkerId, WorkerNodeEvent<'a>)>

pub(super) async fn next_event<'a>( &'a mut self, term_id: &str, context: &Arc<impl GlobalBarrierWorkerContext>, ) -> (WorkerId, WorkerNodeEvent<'a>)

pub(super) async fn next_response( &mut self, term_id: &str, context: &Arc<impl GlobalBarrierWorkerContext>, ) -> (WorkerId, MetaResult<Response>)

fn collect_init_partial_graph( initial_inflight_infos: impl Iterator<Item = (DatabaseId, impl Iterator<Item = TableId>)>, ) -> impl Iterator<Item = PbCreatePartialGraphRequest>

impl ControlStreamManager

pub(super) fn inject_database_initial_barrier( &mut self, database_id: DatabaseId, jobs: HashMap<TableId, HashMap<FragmentId, InflightFragmentInfo>>, state_table_committed_epochs: &mut HashMap<TableId, u64>, state_table_log_epochs: &mut HashMap<TableId, Vec<(Vec<u64>, u64)>>, edges: &mut FragmentEdgeBuildResult, stream_actors: &HashMap<ActorId, StreamActor>, source_splits: &mut HashMap<ActorId, Vec<SplitImpl>>, background_jobs: &mut HashMap<TableId, String>, mv_depended_subscriptions: &mut HashMap<TableId, HashMap<SubscriptionId, u64>>, is_paused: bool, hummock_version_stats: &HummockVersionStats, cdc_table_snapshot_split_assignment: &mut CdcTableSnapshotSplitAssignmentWithGeneration, ) -> MetaResult<DatabaseInitialBarrierCollector>

Extract information from the loaded runtime barrier worker snapshot info, and inject the initial barrier.

pub(super) fn inject_command_ctx_barrier( &mut self, database_id: DatabaseId, command: Option<&Command>, barrier_info: &BarrierInfo, is_paused: bool, pre_applied_graph_info: &InflightDatabaseInfo, applied_graph_info: &InflightDatabaseInfo, edges: &mut Option<FragmentEdgeBuildResult>, ) -> MetaResult<HashMap<WorkerId, bool>>

fn connected_workers( &self, ) -> impl Iterator<Item = (WorkerId, &ControlStreamNode)> + '_

pub(super) fn inject_barrier<'a>( &mut self, database_id: DatabaseId, creating_table_id: Option<TableId>, mutation: Option<Mutation>, barrier_info: &BarrierInfo, pre_applied_graph_info: impl IntoIterator<Item = &InflightFragmentInfo>, applied_graph_info: impl IntoIterator<Item = &'a InflightFragmentInfo> + 'a, new_actors: Option<StreamJobActorsToCreate>, ) -> MetaResult<HashMap<WorkerId, bool>>

pub(super) fn add_partial_graph( &mut self, database_id: DatabaseId, creating_job_id: Option<TableId>, )

pub(super) fn remove_partial_graph( &mut self, database_id: DatabaseId, creating_job_ids: Vec<TableId>, )

pub(super) fn reset_database( &mut self, database_id: DatabaseId, reset_request_id: u32, ) -> HashSet<WorkerId>