Struct StreamWriterCore 

pub struct StreamWriterCore { /* private fields */ }

Encodes the Hurray stream wire format into bytes a driver writes.

A plain tensor or a whole composite is submitted as one operation; the core emits the descriptor, then each payload in turn, and says when the item is written. Validation happens on submission, so a torn or invalid item never reaches a sink.

Examples

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

let handle =
    BufferHandle::new(8, MIN_BUFFER_ALIGNMENT, DeviceTag::Cpu, SyncMode::ProducerSynced)?;
let descriptor = TensorDescriptor::new(
    1, 0, ElementType::Float64, Shape::new(vec![1u64])?, 0,
    LayoutDescriptor::RowMajor, vec![handle], None, None, None, None,
)?;
let payload = [1u8; 8];
// 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 sink = Vec::new();
let mut core = StreamWriterCore::new();
let mut op = core.tensor(&descriptor, &buffers)?;
loop {
    match op.step()? {
        WriteStep::Output { .. } => {
            // A sink that accepts 3 bytes at a time: a short write is normal.
            let src = op.output_src();
            let n = src.len().min(3);
            sink.extend_from_slice(&src[..n]);
            op.report_written(n)?;
        }
        WriteStep::ItemWritten => break,
    }
}
assert_eq!(sink.len(), descriptor.encode()?.len() + payload.len());

Implementations

impl StreamWriterCore

pub fn new() -> Self

Creates a core that accepts any valid SyncMode.

Examples

See the type-level example.

pub fn cross_machine() -> Self

Creates a core that rejects buffers whose sync_mode is not SyncMode::ProducerSynced, for a stream that crosses machine boundaries.

Examples

use hurray_io::stream::StreamWriterCore;

let core = StreamWriterCore::cross_machine();
assert!(!core.is_unusable());

pub fn tensor<'a>( &'a mut self, descriptor: &'a TensorDescriptor, buffers: &'a [&'a [u8]], ) -> Result<WriteOp<'a>>

Submits one tensor: its descriptor followed by one payload slice per buffer.

The payloads stay borrowed for the operation’s lifetime and are released when it reports WriteStep::ItemWritten — which is also the point a binding may drop the reference pinning them (ADR-041 § 3).

Errors

• Error::MultiBufferLengthMismatch — buffers.len() ≠ descriptor.buffers.len()
• Error::BufferSizeMismatch — a slice’s length ≠ its handle’s byte_size
• Error::InvalidCrossMachineSyncMode — cross-machine mode and a buffer that is not ProducerSynced
• Error::Core — the descriptor does not encode
• Error::CoreUnusable — a previous write lost its byte count, so this core and its sink are poisoned

Examples

See the type-level example.

pub fn composite<'a>( &'a mut self, head: &'a TensorDescriptor, members: &'a [CompositeNode<'a>], ) -> Result<WriteOp<'a>>

Submits a composite: its head, then every member’s descriptor and payloads, in order (ADR-027 § Binding).

The whole tree is validated before a byte is emitted, at every nesting level.

Errors

The errors of tensor for each member, plus Error::Core when the head is not a valid composite head or the group fails validation (member count, partition coverage, overlay ordering).

Examples

use hurray_core::{
    layout::{CompositeLayout, CompositionRule},
    BufferHandle, DeviceTag, ElementType, LayoutDescriptor, Shape, SyncMode,
    TensorDescriptor, MIN_BUFFER_ALIGNMENT,
};
use hurray_io::stream::{CompositeNode, StreamWriterCore, WriteStep};

let head = TensorDescriptor::new(
    1, 0, ElementType::Float32, Shape::new(vec![1u64])?, 0,
    LayoutDescriptor::Composite(CompositeLayout::new(CompositionRule::Group, 1)?),
    vec![], None, None, None, None,
)?;
let handle =
    BufferHandle::new(4, MIN_BUFFER_ALIGNMENT, DeviceTag::Cpu, SyncMode::ProducerSynced)?;
let member = TensorDescriptor::new(
    1, 0, ElementType::Float32, Shape::new(vec![1u64])?, 0,
    LayoutDescriptor::RowMajor, vec![handle], None, None, None, None,
)?;
let data = [7u8; 4];
let buffers: [&[u8]; 1] = [&data];
let members = [CompositeNode::Tensor { descriptor: &member, buffers: &buffers }];

let mut core = StreamWriterCore::new();
let mut op = core.composite(&head, &members)?;
let mut emitted = 0usize;
while let WriteStep::Output { len } = op.step()? {
    emitted += len;
    op.report_written(len)?;
}
assert_eq!(emitted, head.encode()?.len() + member.encode()?.len() + data.len());

pub fn is_unusable(&self) -> bool

Whether this core — and the sink behind it — has been poisoned.

Examples

See WriteOp::mark_unusable.

Trait Implementations

impl Default for StreamWriterCore

fn default() -> Self

Returns the “default value” for a type.