Struct mz_dataflow_types::client::replicated::ActiveReplication

pub struct ActiveReplication<T> {
    replicas: HashMap<ReplicaId, (UnboundedSender<ComputeCommand<T>>, UnboundedReceiverStream<Result<ComputeResponse<T>, Error>>)>,
    peeks: HashMap<Uuid, PendingPeek>,
    tails: HashMap<GlobalId, Antichain<T>>,
    uppers: HashMap<GlobalId, (Antichain<T>, HashMap<ReplicaId, MutableAntichain<T>>)>,
    history: ComputeCommandHistory<T>,
    last_command_count: usize,
    pending_response: VecDeque<ActiveReplicationResponse<T>>,
}

A client backed by multiple replicas.

Fields

replicas: HashMap<ReplicaId, (UnboundedSender<ComputeCommand<T>>, UnboundedReceiverStream<Result<ComputeResponse<T>, Error>>)>

Handles to the replicas themselves.

peeks: HashMap<Uuid, PendingPeek>

Outstanding peek identifiers, to guide responses (and which to suppress).

tails: HashMap<GlobalId, Antichain<T>>

Reported frontier of each in-progress tail.

uppers: HashMap<GlobalId, (Antichain<T>, HashMap<ReplicaId, MutableAntichain<T>>)>

Frontier information, both unioned across all replicas and from each individual replica.

history: ComputeCommandHistory<T>

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

last_command_count: usize

Most recent count of the volume of unpacked commands (e.g. dataflows in CreateDataflows).

pending_response: VecDeque<ActiveReplicationResponse<T>>

Responses that should be emitted on the next recv call.

This is introduced to produce peek cancelation responses eagerly, without awaiting a replica responding with the response itself, which allows us to compact away the peek in self.history.

Implementations

impl<T> ActiveReplication<T> where
    T: Timestamp, 

pub fn add_replica<C: ComputeClient<T> + 'static>(
    &mut self,
    id: ReplicaId,
    client: C
)

Introduce a new replica, and catch it up to the commands of other replicas.

It is not yet clear under which circumstances a replica can be removed.

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

pub fn remove_replica(&mut self, id: ReplicaId)

Remove a replica by its identifier.

fn hydrate_replica(&mut self, replica_id: ReplicaId)

Pipes a command stream at the indicated replica, introducing new dataflow identifiers.

Trait Implementations

impl<T: Debug> Debug for ActiveReplication<T>

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

Formats the value using the given formatter.

impl<T> Default for ActiveReplication<T>

fn default() -> Self

Returns the “default value” for a type.

impl<T> GenericClient<ComputeCommand<T>, ActiveReplicationResponse<T>> for ActiveReplication<T> where
    T: Timestamp + Lattice + Debug, 

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

The ADAPTER layer’s isolation from COMPUTE depends on the fact that this function is essentially non-blocking, i.e. the ADAPTER blindly awaits calls to this function. This lets the ADAPTER continue operating even in the face of unhealthy or absent replicas.

If this function every become blocking (e.g. making networking calls), the ADAPTER must amend its contract with COMPUTE.

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

Receives the next response from the dataflow server.

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

Returns an adapter that treats the client as a stream.