Struct StreamActorManager

pub(crate) struct StreamActorManager {
    pub(super) env: StreamEnvironment,
    pub(super) streaming_metrics: Arc<StreamingMetrics>,
    pub(super) watermark_epoch: AtomicU64Ref,
    pub(super) await_tree_reg: Option<Registry>,
    pub(super) runtime: BackgroundShutdownRuntime,
}

Fields

env: StreamEnvironment

streaming_metrics: Arc<StreamingMetrics>

watermark_epoch: AtomicU64Ref

Watermark epoch number.

await_tree_reg: Option<Registry>

Manages the await-trees of all actors.

runtime: BackgroundShutdownRuntime

Runtime for the streaming actors.

Implementations

impl StreamActorManager

fn create_dispatcher( &self, env: StreamEnvironment, input: Executor, dispatchers: &[Dispatcher], actor_id: ActorId, fragment_id: FragmentId, shared_context: &Arc<SharedContext>, ) -> StreamResult<DispatchExecutor>

Create dispatchers with downstream information registered before

fn get_executor_id(actor_context: &ActorContext, node: &StreamNode) -> u64

fn get_executor_info(node: &StreamNode, executor_id: u64) -> ExecutorInfo

fn create_snapshot_backfill_input( &self, upstream_node: &StreamNode, actor_context: &ActorContextRef, shared_context: &Arc<SharedContext>, chunk_size: usize, ) -> StreamResult<MergeExecutorInput>

fn create_snapshot_backfill_node( &self, stream_node: &StreamNode, node: &StreamScanNode, actor_context: &ActorContextRef, vnode_bitmap: Option<Bitmap>, shared_context: &Arc<SharedContext>, env: StreamEnvironment, state_store: impl StateStore, ) -> StreamResult<Executor>

fn create_nodes_inner<'life0, 'life1, 'life2, 'life3, 'life_self, 'async_recursion>( &'life_self self, fragment_id: FragmentId, node: &'life0 StreamNode, env: StreamEnvironment, store: impl StateStore, actor_context: &'life1 ActorContextRef, vnode_bitmap: Option<Bitmap>, has_stateful: bool, subtasks: &'life2 mut Vec<SubtaskHandle>, shared_context: &'life3 Arc<SharedContext>, ) -> Pin<Box<dyn Future<Output = StreamResult<Executor>> + Send + 'async_recursion>>

where 'life0: 'async_recursion, 'life1: 'async_recursion, 'life2: 'async_recursion, 'life3: 'async_recursion, 'life_self: 'async_recursion,

Create a chain(tree) of nodes, with given store.

async fn create_nodes( &self, fragment_id: FragmentId, node: &StreamNode, env: StreamEnvironment, actor_context: &ActorContextRef, vnode_bitmap: Option<Bitmap>, shared_context: &Arc<SharedContext>, ) -> StreamResult<(Executor, Vec<SubtaskHandle>)>

Create a chain(tree) of nodes and return the head executor.

async fn create_actor( self: Arc<Self>, actor: StreamActor, shared_context: Arc<SharedContext>, related_subscriptions: Arc<HashMap<TableId, HashSet<u32>>>, ) -> StreamResult<Actor<DispatchExecutor>>

impl StreamActorManager

pub(super) fn spawn_actor( self: &Arc<Self>, actor: StreamActor, related_subscriptions: Arc<HashMap<TableId, HashSet<u32>>>, current_shared_context: Arc<SharedContext>, ) -> (JoinHandle<()>, Option<JoinHandle<()>>)