Struct mz_storage_client::controller::StorageControllerState

pub struct StorageControllerState<T: Timestamp + Lattice + Codec64 + TimestampManipulation> {

    now: NowFn,
    envd_epoch: NonZeroI64,
    pub(crate) collections: BTreeMap<GlobalId, CollectionState<T>>,
    pub(crate) exports: BTreeMap<GlobalId, ExportState<T>>,
    pub(crate) stash: Stash,
    pub(crate) persist_write_handles: PersistWriteWorker<T>,
    persist_read_handles: PersistReadWorker<T>,
    stashed_response: Option<StorageResponse<T>>,
    pending_source_drops: Vec<GlobalId>,
    pending_sink_drops: Vec<GlobalId>,
    pending_compaction_commands: Vec<(GlobalId, Antichain<T>, Option<StorageInstanceId>)>,
    pub(crate) collection_manager: CollectionManager,
    pub(crate) introspection_ids: BTreeMap<IntrospectionType, GlobalId>,
    introspection_tokens: BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>,
    source_statistics: Arc<Mutex<BTreeMap<GlobalId, BTreeMap<usize, SourceStatisticsUpdate>>>>,
    sink_statistics: Arc<Mutex<BTreeMap<GlobalId, BTreeMap<usize, SinkStatisticsUpdate>>>>,
    clients: BTreeMap<StorageInstanceId, RehydratingStorageClient<T>>,
    initialized: bool,
    config: StorageParameters,
}

Controller state maintained for each storage instance.

Fields

now: NowFn

A function that returns the current time.

envd_epoch: NonZeroI64

The fencing token for this instance of the controller.

collections: BTreeMap<GlobalId, CollectionState<T>>

Collections maintained by the storage controller.

This collection only grows, although individual collections may be rendered unusable. This is to prevent the re-binding of identifiers to other descriptions.

exports: BTreeMap<GlobalId, ExportState<T>>

stash: Stash

persist_write_handles: PersistWriteWorker<T>

Write handle for persist shards.

persist_read_handles: PersistReadWorker<T>

Read handles for persist shards.

These handles are on the other end of a Tokio task, so that work can be done asynchronously without blocking the storage controller.

stashed_response: Option<StorageResponse<T>>

pending_source_drops: Vec<GlobalId>

IDs of sources that were dropped whose statuses should be updated during the next call to StorageController::process.

pending_sink_drops: Vec<GlobalId>

IDs of sinks that were dropped whose statuses should be updated during the next call to StorageController::process.

pending_compaction_commands: Vec<(GlobalId, Antichain<T>, Option<StorageInstanceId>)>

Compaction commands to send during the next call to StorageController::process.

collection_manager: CollectionManager

Interface for managed collections

introspection_ids: BTreeMap<IntrospectionType, GlobalId>

Tracks which collection is responsible for which IntrospectionType.

introspection_tokens: BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>

Tokens for tasks that drive updating introspection collections. Dropping this will make sure that any tasks (or other resources) will stop when needed.

source_statistics: Arc<Mutex<BTreeMap<GlobalId, BTreeMap<usize, SourceStatisticsUpdate>>>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s.

sink_statistics: Arc<Mutex<BTreeMap<GlobalId, BTreeMap<usize, SinkStatisticsUpdate>>>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s.

clients: BTreeMap<StorageInstanceId, RehydratingStorageClient<T>>

Clients for all known storage instances.

initialized: bool

Set to true once initialization_complete has been called.

config: StorageParameters

Storage configuration to apply to newly provisioned instances.

Implementations

impl<T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> StorageControllerState<T>

pub(crate) async fn new(
    postgres_url: String,
    tx: UnboundedSender<StorageResponse<T>>,
    now: NowFn,
    factory: &StashFactory,
    envd_epoch: NonZeroI64
) -> Self

Trait Implementations

impl<T: Debug + Timestamp + Lattice + Codec64 + TimestampManipulation> Debug for StorageControllerState<T>

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

Formats the value using the given formatter.