Struct mz_compute::logging::compute::DemuxHandler

struct DemuxHandler<'a, 'b, A: Allocate + 'static> {
    state: &'a mut DemuxState<A>,
    shared_state: &'a mut SharedLoggingState,
    output: &'a mut DemuxOutput<'b>,
    logging_interval_ms: u128,
    time: Duration,
}

Event handler of the demux operator.

Fields

state: &'a mut DemuxState<A>

State kept by the demux operator.

shared_state: &'a mut SharedLoggingState

State shared across log receivers.

output: &'a mut DemuxOutput<'b>

Demux output sessions.

logging_interval_ms: u128

The logging interval specifying the time granularity for the updates.

time: Duration

The current event time.

Implementations

impl<A: Allocate> DemuxHandler<'_, '_, A>

fn ts(&self) -> Timestamp

Return the timestamp associated with the current event, based on the event time and the logging interval.

fn handle(&mut self, event: ComputeEvent)

Handle the given compute event.

fn handle_export(&mut self, id: GlobalId, dataflow_id: usize)

fn handle_export_dropped(&mut self, id: GlobalId)

fn handle_dataflow_dropped(&mut self, id: usize)

fn handle_dataflow_shutdown(&mut self, id: usize)

fn handle_error_count(&mut self, export_id: GlobalId, diff: i64)

fn handle_hydration(&mut self, export_id: GlobalId)

fn handle_peek_install(&mut self, peek: Peek, peek_type: PeekType)

fn handle_peek_retire(&mut self, peek: Peek, peek_type: PeekType)

fn handle_frontier( &mut self, export_id: GlobalId, frontier: Timestamp, diff: i8, )

fn handle_import_frontier( &mut self, import_id: GlobalId, export_id: GlobalId, frontier: Timestamp, diff: i8, )

fn handle_arrangement_heap_size(&mut self, operator_id: usize, size: isize)

Update the allocation size for an arrangement.

fn handle_arrangement_heap_capacity( &mut self, operator_id: usize, capacity: isize, )

Update the allocation capacity for an arrangement.

fn handle_arrangement_heap_allocations( &mut self, operator_id: usize, count: isize, )

Update the allocation count for an arrangement.

fn handle_arrangement_heap_size_operator( &mut self, operator_id: usize, address: Rc<[usize]>, )

Indicate that a new arrangement exists, start maintaining the heap size state.

fn handle_arrangement_heap_size_operator_dropped(&mut self, operator_id: usize)

Indicate that an arrangement has been dropped and we can cleanup the heap size state.

fn handle_lir_mapping( &mut self, global_id: GlobalId, mapping: Box<[(LirId, LirMetadata)]>, )

Indicate that a new LIR operator exists; record the dataflow address it maps to.

fn handle_dataflow_global(&mut self, id: usize, global_id: GlobalId)