Struct StreamBundler 

pub struct StreamBundler<C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>>, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>>, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>>, BSS: CorrelatedStream<Singlet<Gf2_128>>, BTS: CorrelatedStream<Triple<Gf2_128>>, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>>, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>>, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>>, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>>, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>>, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>>> {
    pub basefield_dabit_stream: BFDS,
    pub basefield_singlet_stream: BFSS,
    pub basefield_triple_stream: BFTS,
    pub binary_singlet_stream: BSS,
    pub binary_triple_stream: BTS,
    pub mpc_field_dabit_stream: MDS,
    pub mpc_field_singlet_stream: MSS,
    pub mpc_field_triple_stream: MTS,
    pub scalar_dabit_stream: SDS,
    pub scalar_singlet_stream: SSS,
    pub scalar_triple_stream: STS,
    pub _c: PhantomData<C>,
}

Stream bundler, holding one stream per preprocessing type, to provide all preprocessing for a circuit via its streams.

Fields

basefield_dabit_stream: BFDS

basefield_singlet_stream: BFSS

basefield_triple_stream: BFTS

binary_singlet_stream: BSS

binary_triple_stream: BTS

mpc_field_dabit_stream: MDS

mpc_field_singlet_stream: MSS

mpc_field_triple_stream: MTS

scalar_dabit_stream: SDS

scalar_singlet_stream: SSS

scalar_triple_stream: STS

_c: PhantomData<C>

Implementations

impl<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS> StreamBundler<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS>

where C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>>, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>>, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>>, BSS: CorrelatedStream<Singlet<Gf2_128>>, BTS: CorrelatedStream<Triple<Gf2_128>>, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>>, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>>, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>>, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>>, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>>, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>>,

pub fn new( basefield_dabit_stream: BFDS, basefield_singlet_stream: BFSS, basefield_triple_stream: BFTS, binary_singlet_stream: BSS, binary_triple_stream: BTS, mpc_field_dabit_stream: MDS, mpc_field_singlet_stream: MSS, mpc_field_triple_stream: MTS, scalar_dabit_stream: SDS, scalar_singlet_stream: SSS, scalar_triple_stream: STS, ) -> Self

Creates a new bundler from the given streams.

impl<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS> StreamBundler<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS>

where C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>, Error = AbortError>, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>, Error = AbortError>, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>, Error = AbortError>, BSS: CorrelatedStream<Singlet<Gf2_128>, Error = AbortError>, BTS: CorrelatedStream<Triple<Gf2_128>, Error = AbortError>, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>, Error = AbortError>, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>, Error = AbortError>, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>, Error = AbortError>, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>, Error = AbortError>, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>, Error = AbortError>, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>, Error = AbortError>,

pub fn prefetch(&self, req: &CircuitPreprocessing) -> PrefetchHandle<AbortError>

Prefetches the given amount into every stream and returns a single PrefetchHandle that resolves once all of them complete (first error wins). The prefetches run concurrently in the background; the handle can be awaited or dropped.

pub async fn resync( &self, targets: &CircuitPreprocessing, ) -> Result<(), AbortError>

Advances every stream to its per-type target, realigning all parties on the same prefix.

All-or-nothing on rewinds: any target behind a stream’s current position is rejected before touching any stream. Otherwise all resyncs are dispatched together, every handle awaited, and the first error (if any) returned — no short-circuiting mid-fan-out.

impl<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS> StreamBundler<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS>

where C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>> + Buffer, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>> + Buffer, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>> + Buffer, BSS: CorrelatedStream<Singlet<Gf2_128>> + Buffer, BTS: CorrelatedStream<Triple<Gf2_128>> + Buffer, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>> + Buffer, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>> + Buffer, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>> + Buffer, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>> + Buffer, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>> + Buffer, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>> + Buffer,

pub fn buffer_configs(&self) -> [SharedBufferConfig; 11]

The shared buffer config of every stream. The returned array is a Buffer slice, so the whole bundle can be tuned in one call: bundler.buffer_configs().set_capacity(n) writes all streams, while .capacity() etc. read the first.

Trait Implementations

impl<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS> Bundler for StreamBundler<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS>

where C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>, Error = AbortError>, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>, Error = AbortError>, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>, Error = AbortError>, BSS: CorrelatedStream<Singlet<Gf2_128>, Error = AbortError>, BTS: CorrelatedStream<Triple<Gf2_128>, Error = AbortError>, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>, Error = AbortError>, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>, Error = AbortError>, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>, Error = AbortError>, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>, Error = AbortError>, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>, Error = AbortError>, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>, Error = AbortError>,

type Iterator = PreprocessingIterator<C>

A type to hold the futures to all requested correlations.

fn fetch( &mut self, req: &CircuitPreprocessing, ) -> Result<PreprocessingIterator<C>, BundlerError>

Fetches the requested correlations and returns a ready-to-use iterator.

fn fetch_for<Consumer>( &mut self, consumer: &Consumer, ) -> Result<Self::Iterator, BundlerError>

where Consumer: BundleConsumer<Iterator = Self::Iterator>,

Fetches the requested correlations and returns a ready-to-use iterator.

impl<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS> BundlerPositions for StreamBundler<C, BFDS, BFSS, BFTS, BSS, BTS, MDS, MSS, MTS, SDS, SSS, STS>

where C: MpcConfig, BFDS: CorrelatedStream<DaBit<BaseFieldOf<C>>, Error = AbortError>, BFSS: CorrelatedStream<Singlet<BaseFieldOf<C>>, Error = AbortError>, BFTS: CorrelatedStream<Triple<BaseFieldOf<C>>, Error = AbortError>, BSS: CorrelatedStream<Singlet<Gf2_128>, Error = AbortError>, BTS: CorrelatedStream<Triple<Gf2_128>, Error = AbortError>, MDS: CorrelatedStream<DaBit<MpcFieldOf<C>>, Error = AbortError>, MSS: CorrelatedStream<Singlet<MpcFieldOf<C>>, Error = AbortError>, MTS: CorrelatedStream<Triple<MpcFieldOf<C>>, Error = AbortError>, SDS: CorrelatedStream<DaBit<ScalarFieldOf<C>>, Error = AbortError>, SSS: CorrelatedStream<Singlet<ScalarFieldOf<C>>, Error = AbortError>, STS: CorrelatedStream<Triple<ScalarFieldOf<C>>, Error = AbortError>,

fn positions(&self) -> CircuitPreprocessing

The logical position (elements delivered) of every stream, per type. Stays in sync across parties; take the per-type maximum to agree on a resync target.

fn buffered(&self) -> CircuitPreprocessing

The number of already-generated elements currently buffered in every stream, per type — a runtime occupancy metric, unlike positions.

async fn resync(&self, targets: &CircuitPreprocessing) -> Result<(), AbortError>

Advances every stream to its per-type target, realigning all parties on the same prefix — the recovery primitive for a peer whose positions fell behind. See StreamBundler::resync.