Function stream 

pub fn stream<'env, D: FnMut(&'env mut [Hash]), T>(
    mode: Mode<'_>,
    done: D,
    body: impl for<'s> FnOnce(&'s mut Stream<'env, D>) -> T,
) -> T

Hash a stream of messages on this crate’s threads, in place: body submits messages that live at least as long as the call (Stream::submit), each with the place its digest goes, in the program’s memory (a batch’s, the place for all its digests). The threads write the digests there, and done receives each place back, in submission order, once its digests are in it: from a submission (when the stream is full), collect and finish, on this thread. body calls finish last (a scope that ends with places not handed back panics). A message of any length; this crate’s threads read it where it is, and stream returns only after they are done, also when body panics.

The program calls crate::initialize_multithreaded first. A stream holds up to 1024 messages (a message over 64 KiB counts once per 64 KiB, a batch once per 64 KiB of messages) in flight; when it is full, a submission waits for room, handing places back and hashing a waiting task itself. A message stays unhashed in a task still open until the task fills or the program calls collect or finish, so a program that needs a digest soon calls one of them. State that done and body share goes in a Cell or RefCell.

Each message’s place comes back alone, about 1-2 ns a message: for many short messages of one length, Stream::submit_batch hands back one place per batch and hashes them side by side (64-byte messages one at a time ran 4-8% slower than in the stream’s earlier design, which handed back runs of digests from its own memory; batches of them, 5-13% faster; i7-12700K).

Streams, then other calls. While a stream is open, this crate’s threads keep looking for its next message for a few microseconds before they sleep, which makes streams of messages 3-5% faster. A program that streams and then soon makes a one-shot multithreaded call pays for it: on an i7-12700K under Linux, crate::hash_many_multithreaded on 16384 messages right after streams of 16-64 KiB messages ran about 5% slower. A program that keeps streaming does not pay it; one that alternates can stream its other work too (Stream::submit_batch for a batch, a stream per round with finish between rounds that depend on each other).

use blake3_servil::{Hash, Mode};
let messages: Vec<Vec<u8>> = (0..1000u32).map(|i| i.to_le_bytes().repeat(16)).collect();
let mut digests = vec![Hash::from_bytes([0; 32]); messages.len()];
let mut handed_back = 0;
blake3_servil::stream(Mode::Hash, |done: &mut [Hash]| handed_back += done.len(), |s| {
    for (message, digest) in messages.iter().zip(&mut digests) {
        s.submit(message, digest);
    }
    s.finish();
});
assert_eq!(handed_back, 1000);
assert_eq!(digests[7], blake3_servil::hash(&messages[7]));