Struct StreamReaderCore 

pub struct StreamReaderCore<A: DestinationAlloc = SystemDestinations> { /* private fields */ }

Decodes the Hurray stream wire format from bytes a driver supplies.

The core owns the destination every payload byte lands in, so the transport’s write into it is the only copy (ADR-041 § 2), and that destination carries the alignment the buffer handle declares (§ 5). Position is kept across every short read, so a driver that stops — because its transport would block, or because its future was cancelled — can resume on the next call (§ 6).

Examples

Driving it over a slice, with no runtime anywhere:

use hurray_core::{
    BufferHandle, DeviceTag, ElementType, LayoutDescriptor, Shape, SyncMode,
    TensorDescriptor, MIN_BUFFER_ALIGNMENT,
};
use hurray_io::stream::{ReadStep, StreamReaderCore, StreamWriterCore, WriteStep};

// A 64-byte float32 tensor, encoded with the write core.
let handle =
    BufferHandle::new(64, MIN_BUFFER_ALIGNMENT, DeviceTag::Cpu, SyncMode::ProducerSynced)?;
let descriptor = TensorDescriptor::new(
    1, 0, ElementType::Float32, Shape::new(vec![4u64, 4])?, 0,
    LayoutDescriptor::RowMajor, vec![handle], None, None, None, None,
)?;
let payload: Vec<u8> = (0u8..64).collect();
// Bound, not a temporary: the operation borrows the payloads until the item is
// written, which is ADR-041 § 3's lifetime rule showing up in the type system.
let buffers: [&[u8]; 1] = [&payload];

let mut wire = Vec::new();
let mut writer = StreamWriterCore::new();
let mut op = writer.tensor(&descriptor, &buffers)?;
while let WriteStep::Output { .. } = op.step()? {
    let src = op.output_src();
    wire.extend_from_slice(src);
    let n = src.len();
    op.report_written(n)?;
}

// Read it back, handing the core 7 bytes at a time to exercise short reads.
let mut core = StreamReaderCore::new();
let mut rest = &wire[..];
let mut tensors = Vec::new();
loop {
    match core.step()? {
        ReadStep::Input { len } => {
            let n = len.min(7).min(rest.len());
            core.input_dst()[..n].copy_from_slice(&rest[..n]);
            rest = &rest[n..];
            core.report_filled(n)?;
        }
        ReadStep::Tensor => tensors.extend(core.take_tensor()),
        ReadStep::Finished => break,
    }
}

assert_eq!(tensors.len(), 1);
let tensor = &tensors[0];
assert_eq!(tensor.buffers[0].as_ref(), &payload[..]);
// The destination carried the declared alignment, so a SIMD load is safe on it.
assert_eq!(tensor.buffers[0].as_ptr() as usize % 64, 0);

Implementations

impl StreamReaderCore<SystemDestinations>

pub fn new() -> Self

Creates a core with default options and the system allocator.

Examples

See the type-level example.

pub fn with_options(options: StreamReaderOptions) -> Self

Creates a core with custom options.

max_descriptor_bytes, max_buffer_bytes and enforce_cross_machine_sync apply here; max_composite_depth does not, because this core does not assemble composites — it bounds CompositeAssembler instead.

Examples

use hurray_io::stream::{StreamReaderCore, StreamReaderOptions};

let core = StreamReaderCore::with_options(StreamReaderOptions {
    max_buffer_bytes: 1 << 20,
    ..Default::default()
});
assert!(!core.is_unusable());

impl<A: DestinationAlloc> StreamReaderCore<A>

pub fn with_destinations(options: StreamReaderOptions, destinations: A) -> Self

Creates a core whose read destinations come from destinations.

This is the seam ADR-041 left open: a binding may supply the allocation, so a read can land directly in memory its runtime owns, as long as the destination satisfies the alignment the handle declares.

Examples

use hurray_io::stream::{
    AlignedDestination, DestinationAlloc, DestinationRequest, StreamReaderCore,
    StreamReaderOptions, SystemDestinations,
};

/// Counts what the core asks for, then defers to the default allocator.
struct Counting {
    seen: Vec<(u64, u32)>,
}

impl DestinationAlloc for Counting {
    type Dest = AlignedDestination;

    fn allocate(
        &mut self,
        request: DestinationRequest,
    ) -> hurray_io::Result<Self::Dest> {
        self.seen.push((request.byte_size, request.alignment));
        SystemDestinations.allocate(request)
    }
}

let core = StreamReaderCore::with_destinations(
    StreamReaderOptions::default(),
    Counting { seen: Vec::new() },
);
assert!(!core.is_unusable());

pub fn step(&mut self) -> Result<ReadStep>

Advances the machine and returns what it needs next.

Never blocks and never waits: every transition here is decode, allocate or bookkeeping, which is what lets one core serve a blocking driver and an event loop alike (ADR-041 § 4).

Errors

• Error::InvalidHeader, Error::Core — the frame or descriptor is malformed
• Error::FrameTooLarge — a descriptor or buffer exceeds the configured limit
• Error::InvalidCrossMachineSyncMode — cross-machine mode and a buffer that is not ProducerSynced
• Error::UnallocatableBuffer — a destination the allocator cannot provide
• Error::CoreFailed, Error::CoreUnusable — a previous call ended this core

Examples

See the type-level example.

pub fn input_dst(&mut self) -> &mut [u8]

The destination for the outstanding ReadStep::Input request.

Its address does not move while the request stands, so a driver may hold a pointer into it across a suspension (ADR-041 § 6a). Empty when no request is outstanding.

Examples

See the type-level example.

pub fn report_filled(&mut self, n: usize) -> Result<()>

Reports that n bytes were filled into input_dst.

Deliberately the cheapest call in the API: it records a count and nothing else, so a driver can pair the transfer and this report inside one synchronous callback, where no cancellation can interleave (ADR-041 § 6a).

A short fill is normal — the core keeps its position and re-offers the remainder. n == 0 means the transport reached end of input: clean at an item boundary, an error anywhere else.

Errors

• Error::UnexpectedEof — end of input mid-item
• Error::InvalidProgressReport — more bytes reported than were offered, which also makes the core unusable: the position is lost

Examples

See the type-level example.

pub fn take_tensor(&mut self) -> Option<StreamTensor>

Takes the completed tensor announced by ReadStep::Tensor.

Returns None when no tensor is ready. The tensor is handed over only once every one of its buffers is full, so a partially filled destination is never observable (ADR-041 § 6).

A composite head arrives here like any other tensor, carrying no buffers; putting the tree back together is CompositeAssembler’s job, not the core’s.

Examples

See the type-level example.

pub fn mark_unusable(&mut self)

Marks the core unusable: a transport moved bytes whose count was lost.

The position is unrecoverable, so every later call reports it by name rather than letting the core diverge silently from the transport (ADR-041 § 6). A readiness- driven driver never needs this; one built on a transfer-and-resolve primitive such as sock_recv_into does, for the window between the transfer and the report.

Examples

use hurray_io::stream::StreamReaderCore;

let mut core = StreamReaderCore::new();
core.mark_unusable();
assert!(core.is_unusable());
assert!(core.step().is_err());

pub fn is_unusable(&self) -> bool

Whether the core has been marked unusable.

Examples

See mark_unusable.

Trait Implementations

impl Default for StreamReaderCore

fn default() -> Self

Returns the “default value” for a type.