Struct PartitionedComputeState

pub struct PartitionedComputeState<T> {
    parts: usize,
    max_result_size: u64,
    frontiers: BTreeMap<GlobalId, TrackedFrontiers<T>>,
    peek_responses: BTreeMap<Uuid, BTreeMap<usize, PeekResponse>>,
    copy_to_responses: BTreeMap<GlobalId, BTreeMap<usize, CopyToResponse>>,
    pending_subscribes: BTreeMap<GlobalId, PendingSubscribe<T>>,
}

Maintained state for partitioned compute clients.

This helper type unifies the responses of multiple partitioned workers in order to present as a single worker:

• It emits Frontiers responses reporting the minimum/meet of frontiers reported by the individual workers.
• It emits PeekResponses and SubscribeResponses reporting the union of the responses received from the workers.

In the compute communication stack, this client is instantiated several times:

• One instance on the controller side, dispatching between cluster processes.
• One instance in each cluster process, dispatching between timely worker threads.

Note that because compute commands, except CreateTimely and UpdateConfiguration, are only sent to the first process, the cluster-side instances of PartitionedComputeState are not guaranteed to see all compute commands. Or more specifically: The instance running inside process 0 sees all commands, whereas the instances running inside the other processes only see CreateTimely and UpdateConfiguration. The PartitionedComputeState implementation must be able to cope with this limited visiblity. It does so by performing most of its state management based on observed compute responses rather than commands.

Fields

parts: usize

Number of partitions the state machine represents.

max_result_size: u64

The maximum result size this state machine can return.

This is updated upon receiving ComputeCommand::UpdateConfigurations.

frontiers: BTreeMap<GlobalId, TrackedFrontiers<T>>

Tracked frontiers for indexes and sinks.

Frontier tracking for a collection is initialized when the first Frontiers response for that collection is received. Frontier tracking is ceased when all shards have reported advancement to the empty frontier for all frontier kinds.

The compute protocol requires that shards always emit Frontiers responses reporting empty frontiers for all frontier kinds when a collection is dropped. It further requires that no further Frontier responses are emitted for a collection after the empty frontiers were reported. These properties ensure that a) we always cease frontier tracking for collections that have been dropped and b) frontier tracking for a collection is not re-initialized after it was ceased.

peek_responses: BTreeMap<Uuid, BTreeMap<usize, PeekResponse>>

Pending responses for a peek; returnable once all are available.

Tracking of responses for a peek is initialized when the first PeekResponse for that peek is received. Once all shards have provided a PeekResponse, a unified peek response is emitted and the peek tracking state is dropped again.

The compute protocol requires that exactly one response is emitted for each peek. This property ensures that a) we can eventually drop the tracking state maintained for a peek and b) we won’t re-initialize tracking for a peek we have already served.

copy_to_responses: BTreeMap<GlobalId, BTreeMap<usize, CopyToResponse>>

Pending responses for a copy to; returnable once all are available.

Tracking of responses for a COPY TO is initialized when the first CopyResponse for that command is received. Once all shards have provided a CopyResponse, a unified copy response is emitted and the copy_to tracking state is dropped again.

The compute protocol requires that exactly one response is emitted for each COPY TO command. This property ensures that a) we can eventually drop the tracking state maintained for a copy and b) we won’t re-initialize tracking for a copy we have already served.

pending_subscribes: BTreeMap<GlobalId, PendingSubscribe<T>>

Tracks in-progress SUBSCRIBEs, and the stashed rows we are holding back until their timestamps are complete.

The updates may be Err if any of the batches have reported an error, in which case the subscribe is permanently borked.

Tracking of a subscribe is initialized when the first SubscribeResponse for that subscribe is received. Once all shards have emitted an “end-of-subscribe” response the subscribe tracking state is dropped again.

The compute protocol requires that for a subscribe that shuts down an end-of-subscribe response is emitted:

• Either a Batch response reporting advancement to the empty frontier…
• … or a DroppedAt response reporting that the subscribe was dropped before completing.

The compute protocol further requires that no further SubscribeResponses are emitted for a subscribe after an end-of-subscribe was reported.

These two properties ensure that a) once a subscribe has shut down, we can eventually drop the tracking state maintained for it and b) we won’t re-initialize tracking for a subscribe we have already dropped.

Implementations

impl<T> PartitionedComputeState<T>

where T: ComputeControllerTimestamp,

fn reset(&mut self)

pub fn observe_command(&mut self, command: &ComputeCommand<T>)

Observes commands that move past, and prepares state for responses.

Trait Implementations

impl<T: Debug> Debug for PartitionedComputeState<T>

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

Formats the value using the given formatter.

impl<T> PartitionedState<ComputeCommand<T>, ComputeResponse<T>> for PartitionedComputeState<T>

where T: ComputeControllerTimestamp,

fn split_command( &mut self, command: ComputeCommand<T>, ) -> Vec<Option<ComputeCommand<T>>>

Splits a command into multiple partitions.

fn absorb_response( &mut self, shard_id: usize, message: ComputeResponse<T>, ) -> Option<Result<ComputeResponse<T>, Error>>

Absorbs a response from a single partition.