Struct Worker

struct Worker<'w, A: Allocate> {
    timely_worker: &'w mut Worker<A>,
    command_rx: CommandReceiver,
    response_tx: ResponseSender,
    compute_state: Option<ComputeState>,
    metrics: WorkerMetrics,
    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.

command_rx: CommandReceiver

The channel over which commands are received.

response_tx: ResponseSender

The channel over which responses are sent.

compute_state: Option<ComputeState>

metrics: WorkerMetrics

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)

Runs a compute worker.

fn set_nonce(&mut self, nonce: Uuid)

fn run_client(&mut self) -> Result<Infallible, NonceChange>

Handles commands for a client connection, returns when the nonce changes.

fn handle_pending_commands(&mut self) -> Result<(), NonceChange>

fn handle_command(&mut self, cmd: ComputeCommand)

fn activate_compute(&mut self) -> Option<ActiveComputeState<'_, A>>

fn recv_command(&mut self) -> Result<ComputeCommand, NonceChange>

Receive the next compute command.

This method blocks if no command is currently available, but takes care to step the Timely worker while doing so.

fn reconcile(&mut self) -> Result<(), NonceChange>

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.