Struct mz_compute::server::Worker

struct Worker<'w, A: Allocate> {
    timely_worker: &'w mut Worker<A>,
    client_rx: Receiver<(Receiver<ComputeCommand>, UnboundedSender<ComputeResponse>, UnboundedSender<SyncActivator>)>,
    compute_state: Option<ComputeState>,
    metrics: ComputeMetrics,
    persist_clients: Arc<PersistClientCache>,
    txns_ctx: TxnsContext,
    tracing_handle: Arc<TracingHandle>,
    context: ComputeInstanceContext,
}

State maintained for each worker thread.

Much of this state can be viewed as local variables for the worker thread, holding state that persists across function calls.

Fields

timely_worker: &'w mut Worker<A>

The underlying Timely worker.

client_rx: Receiver<(Receiver<ComputeCommand>, UnboundedSender<ComputeResponse>, UnboundedSender<SyncActivator>)>

The channel over which communication handles for newly connected clients are delivered.

compute_state: Option<ComputeState>

metrics: ComputeMetrics

Compute metrics.

persist_clients: Arc<PersistClientCache>

A process-global cache of (blob_uri, consensus_uri) -> PersistClient. This is intentionally shared between workers

txns_ctx: TxnsContext

Context necessary for rendering txn-wal operators.

tracing_handle: Arc<TracingHandle>

A process-global handle to tracing configuration.

context: ComputeInstanceContext

Implementations

impl<'w, A: Allocate + 'static> Worker<'w, A>

pub fn run(&mut self)

Waits for client connections and runs them to completion.

fn split_command<T: Timestamp>( command: ComputeCommand<T>, parts: usize, ) -> Vec<ComputeCommand<T>>

fn setup_channel_and_run_client( &mut self, command_rx: Receiver<ComputeCommand>, response_tx: UnboundedSender<ComputeResponse>, activator_tx: UnboundedSender<SyncActivator>, )

fn run_client( &mut self, command_rx: CommandReceiverQueue, response_tx: ResponseSender, )

Draws commands from a single client until disconnected.

fn handle_command( &mut self, response_tx: &mut ResponseSender, cmd: ComputeCommand, )

fn activate_compute<'a>( &'a mut self, response_tx: &'a mut ResponseSender, ) -> Option<ActiveComputeState<'a, A>>

fn reconcile( &mut self, command_rx: &CommandReceiverQueue, response_tx: &mut ResponseSender, ) -> Result<(), RecvError>

Extract commands until InitializationComplete, and make the worker reflect those commands.

This method is meant to be a function of the commands received thus far (as recorded in the compute state command history) and the new commands from command_rx. It should not be a function of other characteristics, like whether the worker has managed to respond to a peek or not. Some effort goes in to narrowing our view to only the existing commands we can be sure are live at all other workers.

The methodology here is to drain command_rx until an InitializationComplete, at which point the prior commands are “reconciled” in. Reconciliation takes each goal dataflow and looks for an existing “compatible” dataflow (per compatible()) it can repurpose, with some additional tests to be sure that we can cut over from one to the other (no additional compaction, no tails/sinks). With any connections established, old orphaned dataflows are allow to compact away, and any new dataflows are created from scratch. “Kept” dataflows are allowed to compact up to any new as_of.

Some additional tidying happens, cleaning up pending peeks, reported frontiers, and creating a new subscribe response buffer. We will need to be vigilant with future modifications to ComputeState to line up changes there with clean resets here.