Struct mz_compute::logging::ConsolidateBuffer

pub struct ConsolidateBuffer<'a, T, D, R: Semigroup, P>where
    P: Push<Bundle<T, (D, T, R)>> + 'a,
    T: Clone + Lattice + Ord + TimelyTimestamp + 'a,
    D: 'a + ExchangeData,{
    buffer: Vec<(D, T, R)>,
    output_handle: OutputHandle<'a, T, (D, T, R), P>,
    cap: Option<Capability<T>>,
    port: usize,
}

A buffer that consolidates updates

The buffer implements a wrapper around OutputHandle consolidating elements pushed to it. It is backed by a capacity-limited buffer, which means that compaction only occurs within the dimensions of the buffer, i.e. the number of unique keys is less than half of the buffer’s capacity.

A cap is retained whenever the current time changes to be able to flush on drop or when the time changes again.

The buffer is filled with updates until it reaches its capacity. At this point, the updates are consolidated to free up space. This process repeats until the consolidation recovered less than half of the buffer’s capacity, at which point the buffer will be shipped.

The buffer retains a capability to send data on flush. It will flush all data once dropped, if time changes, or if the buffer capacity is reached.

Fields

buffer: Vec<(D, T, R)>

output_handle: OutputHandle<'a, T, (D, T, R), P>

cap: Option<Capability<T>>

port: usize

Implementations

impl<'a, T, D: ExchangeData, R: Semigroup, P> ConsolidateBuffer<'a, T, D, R, P>where
    T: Clone + Lattice + Ord + TimelyTimestamp + 'a,
    P: Push<Bundle<T, (D, T, R)>> + 'a,

pub fn new(output_handle: OutputHandle<'a, T, (D, T, R), P>, port: usize) -> Self

Create a new ConsolidateBuffer, wrapping the provided session.

• output_handle: The output to send data to.
• ‘port’: The output port to retain capabilities for.

pub fn give(&mut self, cap: &CapabilityRef<'_, T>, data: (D, T, R))

Give an element to the buffer

pub fn flush(&mut self)

Flush the internal buffer to the underlying session

Trait Implementations

impl<'a, T, D, R: Semigroup, P> Drop for ConsolidateBuffer<'a, T, D, R, P>where
    P: Push<Bundle<T, (D, T, R)>> + 'a,
    T: Clone + Lattice + Ord + TimelyTimestamp + 'a,
    D: 'a + ExchangeData,

fn drop(&mut self)

Executes the destructor for this type.