Struct mz_compute_client::controller::sequential_hydration::SequentialHydration

pub(super) struct SequentialHydration<T> {
    command_tx: UnboundedSender<ComputeCommand<T>>,
    response_rx: UnboundedReceiver<Result<ComputeResponse<T>, Error>>,
    dyncfg: Arc<ConfigSet>,
    metrics: ReplicaMetrics,
    collections: BTreeMap<GlobalId, Collection<T>>,
    hydration_queue: VecDeque<GlobalId>,
    hydration_token: Arc<()>,
    _forwarder_task: AbortOnDropHandle<()>,
}

A client enforcing sequential dataflow hydration.

Fields

command_tx: UnboundedSender<ComputeCommand<T>>

A sender for commands to the wrapped client.

response_rx: UnboundedReceiver<Result<ComputeResponse<T>, Error>>

A receiver for responses from the wrapped client.

dyncfg: Arc<ConfigSet>

Dynamic system configuration.

metrics: ReplicaMetrics

Tracked metrics.

collections: BTreeMap<GlobalId, Collection<T>>

Tracked collections.

Entries are inserted in response to observed CreateDataflow commands. Entries are removed in response to Frontiers commands that report collection hydration, or in response to AllowCompaction commands that specify the empty frontier.

hydration_queue: VecDeque<GlobalId>

A queue of scheduled collections that are awaiting hydration.

hydration_token: Arc<()>

A token held by hydrating collections.

Useful to efficiently determine how many collections are currently in the process of hydration, and thus how much capacity is available.

_forwarder_task: AbortOnDropHandle<()>

Handle to the forwarder task, to abort it when SequentialHydration is dropped.

Implementations

impl<T> SequentialHydration<T>

where T: ComputeControllerTimestamp,

pub fn new<C>( client: C, dyncfg: Arc<ConfigSet>, metrics: ReplicaMetrics, ) -> Self

where C: ComputeClient<T> + 'static,

Create a new SequentialHydration client.

fn hydration_count(&self) -> usize

Return the number of hydrating collections.

fn absorb_command(&mut self, cmd: ComputeCommand<T>) -> Result<(), Error>

Absorb a command and send resulting commands to the wrapped client.

fn observe_response(&mut self, resp: &ComputeResponse<T>) -> Result<(), Error>

Observe a response and send resulting commands to the wrapped client.

fn hydrate_collections(&mut self) -> Result<(), Error>

Allow hydration based on the available capacity.

Trait Implementations

impl<T: Debug> Debug for SequentialHydration<T>

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

Formats the value using the given formatter.

impl<T> GenericClient<ComputeCommand<T>, ComputeResponse<T>> for SequentialHydration<T>

where T: ComputeControllerTimestamp,

fn recv<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<ComputeResponse<T>>, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Cancel safety

This method is cancel safe. If recv is used as the event in a tokio::select! statement and some other branch completes first, it is guaranteed that no messages were received by this client.

fn send<'life0, 'async_trait>( &'life0 mut self, cmd: ComputeCommand<T>, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Sends a command to the dataflow server.

fn as_stream<'a>( &'a mut self, ) -> Pin<Box<dyn Stream<Item = Result<R, Error>> + Send + 'a>>

where R: Send + 'a,

Returns an adapter that treats the client as a stream.