Struct mz_storage::storage_state::Worker

pub struct Worker<'w, A: Allocate> {
    pub timely_worker: &'w mut Worker<A>,
    pub client_rx: Receiver<(Receiver<StorageCommand>, UnboundedSender<StorageResponse>, UnboundedSender<Thread>)>,
    pub storage_state: StorageState,
}

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.

NOTE: This is pub for testing.

client_rx: Receiver<(Receiver<StorageCommand>, UnboundedSender<StorageResponse>, UnboundedSender<Thread>)>

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

storage_state: StorageState

The state associated with collection ingress and egress.

Implementations

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

pub fn new( timely_worker: &'w mut TimelyWorker<A>, client_rx: Receiver<(Receiver<StorageCommand>, UnboundedSender<StorageResponse>, UnboundedSender<Thread>)>, metrics: StorageMetrics, now: NowFn, connection_context: ConnectionContext, instance_context: StorageInstanceContext, persist_clients: Arc<PersistClientCache>, txns_ctx: TxnsContext, tracing_handle: Arc<TracingHandle>, shared_rocksdb_write_buffer_manager: SharedWriteBufferManager, ) -> Self

Creates new Worker state from the given components.

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

pub fn run(&mut self)

Waits for client connections and runs them to completion.

fn run_client( &mut self, command_rx: Receiver<StorageCommand>, response_tx: UnboundedSender<StorageResponse>, )

Runs this (timely) storage worker until the given command_rx is disconnected.

See the module documentation for this workers responsibilities, how it communicates with the other workers and how commands flow from the controller and through the workers.

pub fn handle_async_worker_response( &self, async_response: AsyncStorageWorkerResponse<Timestamp>, )

Entry point for applying a response from the async storage worker.

pub fn handle_internal_storage_command( &mut self, internal_cmd: InternalStorageCommand, )

Entry point for applying an internal storage command.

pub fn report_frontier_progress( &mut self, response_tx: &UnboundedSender<StorageResponse>, )

Emit information about write frontier progress, along with information that should be made durable for this to be the case.

The write frontier progress is “conditional” in that it is not until the information is made durable that the data are emitted to downstream workers, and indeed they should not rely on the completeness of what they hear until the information is made durable.

Specifically, this sends information about new timestamp bindings created by dataflow workers, with the understanding if that if made durable (and ack’d back to the workers) the source will in fact progress with this write frontier.

pub fn report_storage_statistics( &mut self, response_tx: &UnboundedSender<StorageResponse>, )

Report source statistics back to the controller.

fn send_storage_response( &self, response_tx: &UnboundedSender<StorageResponse>, response: StorageResponse, )

Send a response to the coordinator.

fn reconcile( &mut self, command_rx: &Receiver<StorageCommand>, ) -> Result<(), RecvError>

Extract commands until InitializationComplete, and make the worker reflect those commands. If the worker can not be made to reflect the commands, return an error.