Struct mz_compute_client::controller::instance::Instance

pub(super) struct Instance<T> {

    build_info: &'static BuildInfo,
    initialized: bool,
    replicas: BTreeMap<ReplicaId, ReplicaState<T>>,
    collections: BTreeMap<GlobalId, CollectionState<T>>,
    log_sources: BTreeMap<LogVariant, GlobalId>,
    peeks: BTreeMap<Uuid, PendingPeek<T>>,
    subscribes: BTreeMap<GlobalId, ActiveSubscribe<T>>,
    copy_tos: BTreeSet<GlobalId>,
    history: ComputeCommandHistory<UIntGauge, T>,
    response_tx: Sender<ComputeControllerResponse<T>>,
    introspection_tx: Sender<(IntrospectionType, Vec<(Row, Diff)>)>,
    envd_epoch: NonZeroI64,
    replica_epochs: BTreeMap<ReplicaId, u64>,
    metrics: InstanceMetrics,
    enable_aggressive_readhold_downgrades: bool,
}

The state we keep for a compute instance.

Fields

build_info: &'static BuildInfo

Build info for spawning replicas

initialized: bool

Whether instance initialization has been completed.

replicas: BTreeMap<ReplicaId, ReplicaState<T>>

The replicas of this compute instance.

collections: BTreeMap<GlobalId, CollectionState<T>>

Currently installed compute collections.

New entries are added for all collections exported from dataflows created through ActiveInstance::create_dataflow.

Entries are removed by Instance::cleanup_collections. See that method’s documentation about the conditions for removing collection state.

log_sources: BTreeMap<LogVariant, GlobalId>

IDs of log sources maintained by this compute instance.

peeks: BTreeMap<Uuid, PendingPeek<T>>

Currently outstanding peeks.

New entries are added for all peeks initiated through ActiveInstance::peek.

The entry for a peek is only removed once all replicas have responded to the peek. This is currently required to ensure all replicas have stopped reading from the peeked collection’s inputs before we allow them to compact. #16641 tracks changing this so we only have to wait for the first peek response.

subscribes: BTreeMap<GlobalId, ActiveSubscribe<T>>

Currently in-progress subscribes.

New entries are added for all subscribes exported from dataflows created through ActiveInstance::create_dataflow.

The entry for a subscribe is removed once at least one replica has reported the subscribe to have advanced to the empty frontier or to have been dropped, implying that no further updates will be emitted for this subscribe.

Note that subscribes are tracked both in collections and subscribes. collections keeps track of the subscribe’s upper and since frontiers and ensures appropriate read holds on the subscribe’s input. subscribes is only used to track which updates have been emitted, to decide if new ones should be emitted or suppressed.

copy_tos: BTreeSet<GlobalId>

Tracks all in-progress COPY TOs.

New entries are added for all s3 oneshot sinks (corresponding to a COPY TO) exported from dataflows created through ActiveInstance::create_dataflow.

The entry for a copy to is removed once at least one replica has finished or the exporting collection is dropped.

history: ComputeCommandHistory<UIntGauge, T>

The command history, used when introducing new replicas or restarting existing replicas.

response_tx: Sender<ComputeControllerResponse<T>>

Sender for responses to be delivered.

introspection_tx: Sender<(IntrospectionType, Vec<(Row, Diff)>)>

Sender for introspection updates to be recorded.

envd_epoch: NonZeroI64

A number that increases with each restart of environmentd.

replica_epochs: BTreeMap<ReplicaId, u64>

Numbers that increase with each restart of a replica.

metrics: InstanceMetrics

The registry the controller uses to report metrics.

enable_aggressive_readhold_downgrades: bool

Whether to aggressively downgrade read holds for sink dataflows.

This flag exists to derisk the rollout of the aggressive downgrading approach. TODO(teskje): Remove this after a couple weeks.

Implementations

impl<T: Timestamp> Instance<T>

pub fn collection( &self, id: GlobalId ) -> Result<&CollectionState<T>, CollectionMissing>

Acquire a handle to the collection state associated with id.

fn collection_mut( &mut self, id: GlobalId ) -> Result<&mut CollectionState<T>, CollectionMissing>

Acquire a mutable handle to the collection state associated with id.

pub fn expect_collection(&self, id: GlobalId) -> &CollectionState<T>

Acquire a handle to the collection state associated with id.

Panics

Panics if the identified collection does not exist.

fn expect_collection_mut(&mut self, id: GlobalId) -> &mut CollectionState<T>

Acquire a mutable handle to the collection state associated with id.

Panics

Panics if the identified collection does not exist.

pub fn collections_iter( &self ) -> impl Iterator<Item = (&GlobalId, &CollectionState<T>)>

fn add_collection( &mut self, id: GlobalId, as_of: Antichain<T>, storage_dependencies: Vec<GlobalId>, compute_dependencies: Vec<GlobalId>, write_only: bool )

Add a collection to the instance state.

Panics

Panics if a collection with the same ID exists already.

fn remove_collection(&mut self, id: GlobalId)

fn add_replica_state( &mut self, id: ReplicaId, client: ReplicaClient<T>, config: ReplicaConfig )

fn remove_replica_state(&mut self, id: ReplicaId) -> Option<ReplicaState<T>>

fn deliver_response(&mut self, response: ComputeControllerResponse<T>)

Enqueue the given response for delivery to the controller clients.

fn deliver_introspection_updates( &mut self, type_: IntrospectionType, updates: Vec<(Row, Diff)> )

Enqueue the given introspection updates for recording.

pub fn activate<'a>( &'a mut self, storage_controller: &'a mut dyn StorageController<Timestamp = T> ) -> ActiveInstance<'a, T>

Acquire an ActiveInstance by providing a storage controller.

pub fn replica_exists(&self, id: ReplicaId) -> bool

Returns whether the identified replica exists.

pub fn replica_ids(&self) -> impl Iterator<Item = ReplicaId> + '_

Returns the ids of all replicas of this instance.

fn peeks_targeting( &self, replica_id: ReplicaId ) -> impl Iterator<Item = (Uuid, &PendingPeek<T>)>

Return the IDs of pending peeks targeting the specified replica.

fn subscribes_targeting( &self, replica_id: ReplicaId ) -> impl Iterator<Item = GlobalId> + '_

Return the IDs of in-progress subscribes targeting the specified replica.

fn refresh_state_metrics(&self)

Refresh the controller state metrics for this instance.

We could also do state metric updates directly in response to state changes, but that would mean littering the code with metric update calls. Encapsulating state metric maintenance in a single method is less noisy.

This method is invoked by ActiveComputeController::process, which we expect to be periodically called during normal operation.

fn report_dependency_updates(&mut self, id: GlobalId, diff: i64)

Report updates (inserts or retractions) to the identified collection’s dependencies.

Panics

Panics if the identified collection does not exist.

fn update_hydration_status( &mut self, id: GlobalId, replica_id: ReplicaId, frontier: &Antichain<T> )

Update the tracked hydration status for the given collection and replica according to an observed frontier update.

fn update_operator_hydration_status( &mut self, replica_id: ReplicaId, status: OperatorHydrationStatus )

Update the tracked hydration status for an operator according to a received status update.

fn cleanup_collections(&mut self)

Clean up collection state that is not needed anymore.

Three conditions need to be true before we can remove state for a collection:

1. A client must have explicitly dropped the collection. If that is not the case, clients can still reasonably assume that the controller knows about the collection and can answer queries about it.
2. There must be no outstanding read capabilities on the collection. As long as someone still holds read capabilities on a collection, we need to keep it around to be able to properly handle downgrading of said capabilities.
3. All replica write frontiers for the collection must have advanced to the empty frontier. Advancement to the empty frontier signals that replicas are done computing the collection and that they won’t send more ComputeResponses for it. As long as we might receive responses for a collection we want to keep it around to be able to validate and handle these responses.

pub fn collection_reverse_dependencies( &self, id: GlobalId ) -> impl Iterator<Item = &GlobalId>

List compute collections that depend on the given collection.

impl<T> Instance<T>where T: Timestamp + Lattice, ComputeGrpcClient: ComputeClient<T>,

pub fn new( build_info: &'static BuildInfo, arranged_logs: BTreeMap<LogVariant, GlobalId>, envd_epoch: NonZeroI64, metrics: InstanceMetrics, response_tx: Sender<ComputeControllerResponse<T>>, introspection_tx: Sender<(IntrospectionType, Vec<(Row, Diff)>)>, enable_aggressive_readhold_downgrades: bool ) -> Self

pub fn update_configuration(&mut self, config_params: ComputeParameters)

Update instance configuration.

pub fn initialization_complete(&mut self)

Marks the end of any initialization commands.

Intended to be called by Controller, rather than by other code. Calling this method repeatedly has no effect.

pub fn drop(self)

Drop this compute instance.

Panics

Panics if the compute instance still has active replicas. Panics if the compute instance still has collections installed.

pub fn send(&mut self, cmd: ComputeCommand<T>)

Sends a command to all replicas of this instance.

pub async fn recv( &mut self ) -> Result<(ReplicaId, ComputeResponse<T>), ReplicaId>

Receives the next response from any replica of this instance.

Returns Err if receiving from a replica has failed, to signal that it is in need of rehydration.

This method is cancellation safe.

fn register_replica_heartbeat(&mut self, replica_id: ReplicaId)

Register a heartbeat from the given replica.

Panics

Panics if the specified replica does not exist.

pub fn set_subscribe_target_replica( &mut self, id: GlobalId, target_replica: ReplicaId ) -> Result<(), SubscribeTargetError>

Assign a target replica to the identified subscribe.

If a subscribe has a target replica assigned, only subscribe responses sent by that replica are considered.

Trait Implementations

impl<T: Debug> Debug for Instance<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.