Skip to main content

nuvec_io/
safetensors.rs

1//! The safetensors format: an 8-byte little-endian header size, a JSON header mapping
2//! tensor names to their dtype, shape and byte range, then the tensors' bytes.
3//!
4//! Loading maps the file without a copy (see [`MappedFile`]): every tensor is a view of the
5//! mapped bytes, read from disk (or the page cache) when first used. Files whose tensors are
6//! not aligned to their element sizes (MLX writes headers without padding) are not mapped:
7//! their tensors are read once, with parallel positioned reads, into one buffer per file
8//! where each starts at a multiple of 256 bytes ([`MappedFile::read_ranges`]).
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::path::Path;
12
13use nuvec_core::{Array, DType, Error, MappedFile, Result};
14
15use crate::json::{self, Json};
16
17fn invalid(msg: impl Into<String>) -> Error {
18    Error::InvalidArgument(format!("safetensors: {}", msg.into()))
19}
20
21/// The tensors and metadata of a safetensors file.
22#[derive(Debug, Clone, Default)]
23pub struct SafeTensors {
24    /// The tensors by name.
25    pub tensors: BTreeMap<String, Array>,
26    /// The `__metadata__` strings.
27    pub metadata: BTreeMap<String, String>,
28}
29
30/// safetensors' dtype names, with MLX's `C64` for complex64 and fp8 tensors read as bytes.
31fn dtype_of(name: &str) -> Option<DType> {
32    Some(match name {
33        "BOOL" => DType::Bool,
34        "U8" => DType::U8,
35        "U16" => DType::U16,
36        "U32" => DType::U32,
37        "U64" => DType::U64,
38        "I8" => DType::I8,
39        "I16" => DType::I16,
40        "I32" => DType::I32,
41        "I64" => DType::I64,
42        "F16" => DType::F16,
43        "BF16" => DType::BF16,
44        "F32" => DType::F32,
45        "F64" => DType::F64,
46        "C64" => DType::Complex64,
47        // MLX's reading of fp8 tensors: bytes (`to_fp8`/`from_fp8`, quantization scales);
48        // `F8_E5M2` has no reading (MLX rejects it too).
49        "F8_E4M3" | "F8_E8M0" => DType::U8,
50        _ => return None,
51    })
52}
53
54fn name_of(dtype: DType) -> &'static str {
55    match dtype {
56        DType::Bool => "BOOL",
57        DType::U8 => "U8",
58        DType::U16 => "U16",
59        DType::U32 => "U32",
60        DType::U64 => "U64",
61        DType::I8 => "I8",
62        DType::I16 => "I16",
63        DType::I32 => "I32",
64        DType::I64 => "I64",
65        DType::F16 => "F16",
66        DType::BF16 => "BF16",
67        DType::F32 => "F32",
68        DType::F64 => "F64",
69        _ => "C64",
70    }
71}
72
73fn usize_of(v: &Json, what: &str) -> Result<usize> {
74    match v {
75        Json::Int(n) if *n >= 0 => Ok(*n as usize),
76        _ => Err(invalid(format!("{what} must be a non-negative integer"))),
77    }
78}
79
80/// The reference implementation's limit on the header size.
81const MAX_HEADER: usize = 100_000_000;
82
83/// Where tensors read from a misaligned file start in their buffer: multiples of 256 bytes
84/// (decode at 0.99 → 1.00× mlx-lm on Gemma 4 E4B; 1 KiB and 16 KiB pages measured the same).
85const TENSOR_ALIGN: usize = 256;
86
87/// A parsed and validated header: the metadata and the tensors' byte ranges (sorted, tiling
88/// the data section from `start` to the end of the file).
89struct Header {
90    metadata: BTreeMap<String, String>,
91    /// `(begin, end, name, dtype, shape)`, relative to the data section.
92    ranges: Vec<(usize, usize, String, DType, Vec<usize>)>,
93    /// Where the data section starts (8 + the header's length).
94    start: usize,
95}
96
97impl Header {
98    /// Whether every tensor is aligned to its element size in the file (else the loader
99    /// reads the tensors instead of mapping the file).
100    fn aligned(&self) -> bool {
101        self.ranges
102            .iter()
103            .all(|(begin, _, _, dtype, _)| (self.start + begin).is_multiple_of(dtype.size()))
104    }
105
106    /// Where the tensors `keep` accepts go in one buffer read from the file: in file order,
107    /// each continuing the previous kept tensor's file range if it follows it in the file
108    /// and lands at a multiple of [`TENSOR_ALIGN`], else starting a new range at the next
109    /// one. (MLX's tensors are separate allocations; at the file's own offsets, a
110    /// quantized product ran up to 4% slower, depending on where its weights fell relative
111    /// to its scales.)
112    fn plan(&self, keep: &dyn Fn(&str) -> bool) -> ReadPlan {
113        let mut plan = ReadPlan::default();
114        // The file end of the current range, if any.
115        let mut file_end = None;
116        for (index, (begin, end, name, _, _)) in self.ranges.iter().enumerate() {
117            if !keep(name) {
118                continue;
119            }
120            let at = (self.start + begin) as u64;
121            let n = end - begin;
122            let offset = if file_end == Some(at) && plan.len.is_multiple_of(TENSOR_ALIGN) {
123                let last = plan.ranges.last_mut().expect("an open range");
124                last.2 += n;
125                plan.len
126            } else {
127                let offset = plan.len.next_multiple_of(TENSOR_ALIGN);
128                plan.ranges.push((at, offset, n));
129                offset
130            };
131            plan.tensors.push((index, offset));
132            plan.len = offset + n;
133            file_end = Some(at + n as u64);
134        }
135        plan
136    }
137}
138
139/// The reads loading a misaligned file makes ([`Header::plan`]).
140#[derive(Default)]
141struct ReadPlan {
142    /// `(file offset, buffer offset, length)`, increasing and disjoint in the buffer.
143    ranges: Vec<(u64, usize, usize)>,
144    /// `(index into Header::ranges, buffer offset)` of each kept tensor.
145    tensors: Vec<(usize, usize)>,
146    /// The buffer's length.
147    len: usize,
148}
149
150/// The data section's start from the 8-byte header size, for a file of `file_len` bytes.
151fn header_end(size: [u8; 8], file_len: usize) -> Result<usize> {
152    usize::try_from(u64::from_le_bytes(size))
153        .ok()
154        .filter(|&n| n <= MAX_HEADER)
155        .and_then(|n| n.checked_add(8))
156        .filter(|&end| end <= file_len)
157        .ok_or_else(|| invalid("header larger than the file or 100 MB"))
158}
159
160/// Parses and validates the JSON header (`json`, the bytes after the size) of a file of
161/// `file_len` bytes whose data starts at `start`.
162fn parse_header(json_bytes: &[u8], start: usize, file_len: usize) -> Result<Header> {
163    let Json::Object(entries) = json::parse(json_bytes)? else {
164        return Err(invalid("the header must be a JSON object"));
165    };
166    let mut metadata = BTreeMap::new();
167    let mut ranges = Vec::new();
168    let mut names = BTreeSet::new();
169    for (name, entry) in entries {
170        if name == "__metadata__" {
171            match entry {
172                Json::Null => {}
173                Json::Object(fields) => {
174                    for (k, v) in fields {
175                        let Json::Str(s) = v else {
176                            return Err(invalid("metadata values must be strings"));
177                        };
178                        metadata.insert(k, s);
179                    }
180                }
181                _ => return Err(invalid("`__metadata__` must be an object")),
182            }
183            continue;
184        }
185        // A repeated name would tile the data and then silently replace the first tensor.
186        if !names.insert(name.clone()) {
187            return Err(invalid(format!("duplicate tensor `{name}`")));
188        }
189        let Json::Object(fields) = entry else {
190            return Err(invalid(format!("entry `{name}` must be an object")));
191        };
192        let field = |key: &str| fields.iter().find(|(k, _)| k == key).map(|(_, v)| v);
193        let dtype = match field("dtype") {
194            Some(Json::Str(d)) => {
195                dtype_of(d).ok_or_else(|| invalid(format!("`{name}`: unsupported dtype {d}")))?
196            }
197            _ => return Err(invalid(format!("`{name}`: missing dtype"))),
198        };
199        let shape = match field("shape") {
200            Some(Json::Array(dims)) => dims
201                .iter()
202                .map(|d| usize_of(d, "a dimension"))
203                .collect::<Result<Vec<_>>>()?,
204            _ => return Err(invalid(format!("`{name}`: missing shape"))),
205        };
206        let (begin, end) = match field("data_offsets") {
207            Some(Json::Array(o)) if o.len() == 2 => {
208                (usize_of(&o[0], "an offset")?, usize_of(&o[1], "an offset")?)
209            }
210            _ => return Err(invalid(format!("`{name}`: missing data_offsets"))),
211        };
212        let expected = shape
213            .iter()
214            .try_fold(dtype.size(), |n, &d| n.checked_mul(d))
215            .ok_or_else(|| invalid(format!("`{name}`: size overflows")))?;
216        if end.checked_sub(begin) != Some(expected) {
217            return Err(invalid(format!(
218                "`{name}`: data_offsets [{begin}, {end}] do not hold {expected} bytes"
219            )));
220        }
221        ranges.push((begin, end, name, dtype, shape));
222    }
223    // The ranges must tile the data exactly.
224    ranges.sort_by_key(|r| (r.0, r.1));
225    let mut at = 0;
226    for (begin, end, name, ..) in &ranges {
227        if *begin != at {
228            return Err(invalid(format!(
229                "`{name}`: data_offsets overlap or leave a gap"
230            )));
231        }
232        at = *end;
233    }
234    if start.checked_add(at) != Some(file_len) {
235        return Err(invalid(
236            "the tensors do not cover the data to the end of the file",
237        ));
238    }
239    Ok(Header {
240        metadata,
241        ranges,
242        start,
243    })
244}
245
246/// Reads and validates a file's header with ordinary reads (nothing is mapped), returning
247/// it and the file's length.
248fn read_header(path: &Path) -> Result<(Header, usize)> {
249    use std::io::Read;
250    let io = |e: std::io::Error| Error::Io(format!("{}: {e}", path.display()));
251    let mut file = std::fs::File::open(path).map_err(io)?;
252    let file_len = usize::try_from(file.metadata().map_err(io)?.len())
253        .map_err(|_| Error::Io(format!("{}: too large", path.display())))?;
254    let mut size = [0u8; 8];
255    file.read_exact(&mut size).map_err(|_| {
256        invalid(format!(
257            "{}: file shorter than its header size",
258            path.display()
259        ))
260    })?;
261    let start = header_end(size, file_len).map_err(|e| named(path, e))?;
262    let mut json_bytes = vec![0u8; start - 8];
263    file.read_exact(&mut json_bytes).map_err(io)?;
264    let header = parse_header(&json_bytes, start, file_len).map_err(|e| named(path, e))?;
265    Ok((header, file_len))
266}
267
268/// `e` with the file's name in front of its message.
269fn named(path: &Path, e: Error) -> Error {
270    match e {
271        Error::InvalidArgument(m) => Error::InvalidArgument(format!("{}: {m}", path.display())),
272        Error::Io(m) => Error::Io(format!("{}: {m}", path.display())),
273        e => e,
274    }
275}
276
277/// What loading safetensors checkpoints holds in the GPU allocator, in bytes (see
278/// [`safetensors_footprint`]). Compare with
279/// `nuvec_gpu::MemoryStats`'s `mapped`, `active` and `peak_with_mapped`.
280#[non_exhaustive]
281#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
282pub struct Footprint {
283    /// Kept mapped: the page-rounded whole files (header included) of aligned files.
284    pub mapped: u64,
285    /// Read into allocator memory: the size-class-rounded tensors of files whose tensors are
286    /// not aligned (as MLX writes them; mlx-community checkpoints). Such files are not
287    /// mapped.
288    pub copied: u64,
289    /// The peak of `mapped + copied` while loading: `mapped + copied` (no file is mapped
290    /// and copied at once).
291    pub transient: u64,
292}
293
294impl Footprint {
295    /// The footprint of one file of `file_len` bytes with this header, loading the tensors
296    /// `keep` accepts.
297    fn of(header: &Header, file_len: usize, keep: &dyn Fn(&str) -> bool) -> Self {
298        if header.aligned() {
299            let mapping = nuvec_core::MappedFile::mapping_size(file_len as u64);
300            Footprint {
301                mapped: mapping,
302                copied: 0,
303                transient: mapping,
304            }
305        } else {
306            let copied = nuvec_core::MappedFile::copy_size(header.plan(keep).len as u64);
307            Footprint {
308                mapped: 0,
309                copied,
310                transient: copied,
311            }
312        }
313    }
314}
315
316/// What loading a safetensors file, or every `*.safetensors` file of a directory, will
317/// hold, from their headers only (ordinary reads; nothing is mapped, copied or wired):
318/// the bytes kept mapped, the bytes copied, and the transient peak of the load.
319///
320/// After [`load_safetensors`] (file by file for a directory) the GPU allocator's `mapped`
321/// grows by `mapped`, its `active` by `copied`, and `peak_with_mapped` rises by at most
322/// `transient`, which is `mapped + copied`: misaligned files are read, not mapped.
323/// [`load_safetensors_filtered`] copies at most `copied`.
324///
325/// ```
326/// # fn main() -> nuvec_core::Result<()> {
327/// # if cfg!(miri) { return Ok(()); }
328/// use nuvec_core::array;
329///
330/// let path = std::env::temp_dir().join("nuvec-footprint-doc.safetensors");
331/// nuvec_io::save_safetensors(&path, &[("w", &array![1.0, 2.0])], &[])?;
332/// let fp = nuvec_io::safetensors_footprint(&path)?;
333/// assert_eq!(fp.copied, 0); // nuvec writes aligned files: mapped, not copied
334/// assert!(fp.mapped >= std::fs::metadata(&path).unwrap().len());
335/// # Ok(())
336/// # }
337/// ```
338pub fn safetensors_footprint(path: impl AsRef<Path>) -> Result<Footprint> {
339    safetensors_footprint_filtered(path, |_| true)
340}
341
342/// What [`load_safetensors_filtered`] with `keep` will hold, as [`safetensors_footprint`]
343/// (a file, or every `*.safetensors` file of a directory): `copied` counts only the kept
344/// tensors of misaligned files; aligned files are mapped whole whatever is kept.
345///
346/// ```
347/// # fn main() -> nuvec_core::Result<()> {
348/// # if cfg!(miri) { return Ok(()); }
349/// use nuvec_core::array;
350///
351/// let path = std::env::temp_dir().join("nuvec-footprint-filtered-doc.safetensors");
352/// nuvec_io::save_safetensors(&path, &[("a", &array![1.0]), ("b", &array![2.0])], &[])?;
353/// let all = nuvec_io::safetensors_footprint(&path)?;
354/// let some = nuvec_io::safetensors_footprint_filtered(&path, |name| name == "b")?;
355/// assert_eq!(some, all); // aligned: mapped whole
356/// # Ok(())
357/// # }
358/// ```
359pub fn safetensors_footprint_filtered(
360    path: impl AsRef<Path>,
361    keep: impl Fn(&str) -> bool,
362) -> Result<Footprint> {
363    let path = path.as_ref();
364    let files = if path.is_dir() {
365        let mut files: Vec<_> = std::fs::read_dir(path)
366            .map_err(|e| Error::Io(format!("{}: {e}", path.display())))?
367            .filter_map(|e| e.ok().map(|e| e.path()))
368            .filter(|p| p.extension().is_some_and(|x| x == "safetensors"))
369            .collect();
370        files.sort();
371        files
372    } else {
373        vec![path.to_path_buf()]
374    };
375    let mut total = Footprint::default();
376    for file in &files {
377        let (header, len) = read_header(file)?;
378        let one = Footprint::of(&header, len, &keep);
379        total.mapped += one.mapped;
380        total.copied += one.copied;
381    }
382    total.transient = total.mapped + total.copied;
383    Ok(total)
384}
385
386/// Loads a safetensors file. The tensors are views of the memory-mapped file or, if they
387/// are not aligned to their element sizes (MLX writes headers without padding), of one
388/// buffer they are read into with parallel positioned reads (the file is then not mapped).
389///
390/// Validation follows the reference implementation: a header of at most 100 MB, an
391/// optional (or `null`) `__metadata__` of strings, and tensors whose byte ranges, sorted
392/// by offset, tile the data from its start to the end of the file.
393///
394/// Before mapping or reading, the file's load transient ([`safetensors_footprint`]) is
395/// checked against the GPU's memory limit: a load that would pass it fails with an error
396/// naming the file, with nothing mapped or allocated.
397pub fn load_safetensors(path: impl AsRef<Path>) -> Result<SafeTensors> {
398    load_safetensors_filtered(path, |_| true)
399}
400
401/// [`load_safetensors`], returning only the tensors whose names `keep` accepts. In files
402/// that are read rather than mapped (misaligned ones, as MLX writes them), the others are
403/// not read at all: loading a language model from a multimodal checkpoint, say, skips the
404/// vision weights.
405///
406/// ```
407/// # fn main() -> nuvec_core::Result<()> {
408/// # if cfg!(miri) { return Ok(()); }
409/// use nuvec_core::array;
410///
411/// let path = std::env::temp_dir().join("nuvec-filtered-doc.safetensors");
412/// nuvec_io::save_safetensors(&path, &[("a", &array![1.0]), ("b", &array![2.0])], &[])?;
413/// let loaded = nuvec_io::load_safetensors_filtered(&path, |name| name == "b")?;
414/// assert_eq!(loaded.tensors.keys().collect::<Vec<_>>(), ["b"]);
415/// # Ok(())
416/// # }
417/// ```
418pub fn load_safetensors_filtered(
419    path: impl AsRef<Path>,
420    keep: impl Fn(&str) -> bool,
421) -> Result<SafeTensors> {
422    let path = path.as_ref();
423    let (header, len) = read_header(path)?;
424    let aligned = header.aligned();
425    let plan = (!aligned).then(|| header.plan(&keep));
426    let need = match &plan {
427        Some(plan) => nuvec_core::MappedFile::copy_size(plan.len as u64),
428        None => Footprint::of(&header, len, &|_| true).transient,
429    };
430    if let Some(room) = nuvec_core::MappedFile::headroom()
431        && need > room
432    {
433        return Err(Error::Io(format!(
434            "{}: out of memory: loading needs {need} bytes (mapped or read), {room} \
435             are left under the memory limit",
436            path.display()
437        )));
438    }
439    let oom = |e| match e {
440        Error::OutOfMemory(n) => Error::Io(format!(
441            "{}: out of memory: {n} bytes pass the memory limit",
442            path.display()
443        )),
444        e => e,
445    };
446    let mut out = SafeTensors {
447        metadata: header.metadata,
448        ..SafeTensors::default()
449    };
450    match plan {
451        // Read (not mapped): a mapping besides the copy would hold the checkpoint twice.
452        Some(plan) => {
453            let data = MappedFile::read_ranges(path, &plan.ranges, plan.len).map_err(oom)?;
454            let mut ranges: Vec<_> = header.ranges.into_iter().map(Some).collect();
455            for (index, offset) in plan.tensors {
456                let (_, _, name, dtype, shape) = ranges[index].take().expect("kept once");
457                out.tensors.insert(name, data.array(offset, dtype, &shape)?);
458            }
459        }
460        None => {
461            let file = MappedFile::open(path).map_err(oom)?;
462            // The file is the one whose header was read; it must not change while it is
463            // mapped (see `MappedFile`), so a different length is an error.
464            if file.len() != len {
465                return Err(Error::Io(format!(
466                    "{}: the file changed while it was loaded",
467                    path.display()
468                )));
469            }
470            for (begin, _, name, dtype, shape) in header.ranges {
471                if keep(&name) {
472                    let tensor = file.array(header.start + begin, dtype, &shape)?;
473                    out.tensors.insert(name, tensor);
474                }
475            }
476        }
477    }
478    Ok(out)
479}
480
481/// The array's elements as little-endian bytes, row-major.
482pub(crate) fn array_bytes(a: &Array) -> Result<Vec<u8>> {
483    macro_rules! le {
484        ($t:ty) => {
485            a.to_vec::<$t>()?
486                .iter()
487                .flat_map(|v| v.to_le_bytes())
488                .collect()
489        };
490    }
491    Ok(match a.dtype() {
492        DType::Bool => a.to_vec::<bool>()?.iter().map(|&b| u8::from(b)).collect(),
493        DType::U8 => a.to_vec::<u8>()?,
494        DType::U16 => le!(u16),
495        DType::U32 => le!(u32),
496        DType::U64 => le!(u64),
497        DType::I8 => le!(i8),
498        DType::I16 => le!(i16),
499        DType::I32 => le!(i32),
500        DType::I64 => le!(i64),
501        DType::F16 => le!(nuvec_core::f16),
502        DType::BF16 => le!(nuvec_core::bf16),
503        DType::F32 => le!(f32),
504        DType::F64 => le!(f64),
505        _ => a
506            .to_vec::<nuvec_core::Complex64>()?
507            .iter()
508            .flat_map(|z| [z.re.to_le_bytes(), z.im.to_le_bytes()].concat())
509            .collect(),
510    })
511}
512
513/// The file's bytes for `tensors` (evaluating them) and `metadata`.
514fn encode(tensors: &[(&str, &Array)], metadata: &[(&str, &str)]) -> Result<Vec<u8>> {
515    let mut sorted: Vec<&(&str, &Array)> = tensors.iter().collect();
516    sorted.sort_by(|(ln, la), (rn, ra)| ra.dtype().size().cmp(&la.dtype().size()).then(ln.cmp(rn)));
517    let mut header = Vec::new();
518    if !metadata.is_empty() {
519        let map: BTreeMap<String, String> = metadata
520            .iter()
521            .map(|(k, v)| (k.to_string(), v.to_string()))
522            .collect();
523        header.push(format!("\"__metadata__\":{}", json::object(&map)));
524    }
525    let mut data = Vec::new();
526    for (name, a) in sorted {
527        let bytes = array_bytes(a)?;
528        let shape: Vec<String> = a.shape().iter().map(i32::to_string).collect();
529        header.push(format!(
530            "{}:{{\"dtype\":\"{}\",\"shape\":[{}],\"data_offsets\":[{},{}]}}",
531            json::quote(name),
532            name_of(a.dtype()),
533            shape.join(","),
534            data.len(),
535            data.len() + bytes.len()
536        ));
537        data.extend(bytes);
538    }
539    let mut header = format!("{{{}}}", header.join(",")).into_bytes();
540    header.resize(header.len().next_multiple_of(8), b' ');
541    let mut out = Vec::with_capacity(8 + header.len() + data.len());
542    out.extend((header.len() as u64).to_le_bytes());
543    out.extend(header);
544    out.extend(data);
545    Ok(out)
546}
547
548/// Writes `bytes` to `path` atomically: to `path.tmp`, renamed over `path` on success.
549fn write_atomically(path: &Path, bytes: &[u8]) -> Result<()> {
550    let io = |e: std::io::Error| Error::Io(format!("{}: {e}", path.display()));
551    let mut tmp = path.as_os_str().to_owned();
552    tmp.push(".tmp");
553    let tmp = std::path::PathBuf::from(tmp);
554    if let Err(e) = std::fs::write(&tmp, bytes) {
555        let _ = std::fs::remove_file(&tmp);
556        return Err(io(e));
557    }
558    std::fs::rename(&tmp, path).map_err(|e| {
559        let _ = std::fs::remove_file(&tmp);
560        io(e)
561    })
562}
563
564/// Saves `tensors` (evaluating them) and `metadata` as a safetensors file, written to
565/// `path.tmp` and renamed into place.
566///
567/// As the reference implementation, tensors are laid out by decreasing element size (then
568/// by name) after a header padded to 8 bytes, so every tensor is aligned to its element
569/// size and loads without a copy.
570pub fn save_safetensors(
571    path: impl AsRef<Path>,
572    tensors: &[(&str, &Array)],
573    metadata: &[(&str, &str)],
574) -> Result<()> {
575    write_atomically(path.as_ref(), &encode(tensors, metadata)?)
576}
577
578/// A background safetensors write ([`save_safetensors_async`]). [`SaveHandle::wait`]
579/// returns its result; dropping the handle waits for the write to finish.
580#[must_use = "wait() returns the write's result; dropping the handle only waits"]
581#[derive(Debug)]
582pub struct SaveHandle {
583    thread: Option<std::thread::JoinHandle<Result<()>>>,
584}
585
586impl SaveHandle {
587    /// Waits for the write and returns its result.
588    pub fn wait(mut self) -> Result<()> {
589        self.join()
590    }
591
592    fn join(&mut self) -> Result<()> {
593        match self.thread.take() {
594            Some(t) => t
595                .join()
596                .unwrap_or_else(|_| Err(Error::Io("the safetensors writer panicked".into()))),
597            None => Ok(()),
598        }
599    }
600}
601
602impl Drop for SaveHandle {
603    fn drop(&mut self) {
604        let _ = self.join();
605    }
606}
607
608/// [`save_safetensors`] in the background, for snapshots taken while work continues (a
609/// KV cache or recurrent state copied with `copy`, say): the arrays are submitted for
610/// evaluation ([`nuvec_core::async_eval`]) and returned at once with a handle; a
611/// background thread waits for them and writes `path.tmp`, renamed into place on success.
612/// The file is byte-identical to [`save_safetensors`]'s. The handle holds the arrays until
613/// the write is done, so an in-place update of one of them copies instead of writing into
614/// memory being saved.
615///
616/// ```
617/// # fn main() -> nuvec_core::Result<()> {
618/// # if cfg!(miri) { return Ok(()); }
619/// use nuvec_core::array;
620///
621/// let path = std::env::temp_dir().join("nuvec-async-save-doc.safetensors");
622/// let state = array![1.0, 2.0, 3.0];
623/// let handle = nuvec_io::save_safetensors_async(&path, &[("state", &state.copy())], &[]);
624/// // ... more work here ...
625/// handle.wait()?;
626/// assert_eq!(nuvec_io::load_safetensors(&path)?.tensors["state"].to_vec::<f32>()?, [1.0, 2.0, 3.0]);
627/// # Ok(())
628/// # }
629/// ```
630pub fn save_safetensors_async(
631    path: impl AsRef<Path>,
632    tensors: &[(&str, &Array)],
633    metadata: &[(&str, &str)],
634) -> SaveHandle {
635    let path = path.as_ref().to_path_buf();
636    let tensors: Vec<(String, Array)> = tensors
637        .iter()
638        .map(|(n, a)| ((*n).to_string(), (*a).clone()))
639        .collect();
640    let metadata: Vec<(String, String)> = metadata
641        .iter()
642        .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
643        .collect();
644    let submitted = nuvec_core::async_eval(&tensors.iter().map(|(_, a)| a).collect::<Vec<_>>());
645    let thread = std::thread::spawn(move || {
646        submitted?;
647        let tensors: Vec<(&str, &Array)> = tensors.iter().map(|(n, a)| (n.as_str(), a)).collect();
648        let metadata: Vec<(&str, &str)> = metadata
649            .iter()
650            .map(|(k, v)| (k.as_str(), v.as_str()))
651            .collect();
652        write_atomically(&path, &encode(&tensors, &metadata)?)
653    });
654    SaveHandle {
655        thread: Some(thread),
656    }
657}
658
659#[cfg(test)]
660mod tests {
661    use super::*;
662
663    #[test]
664    fn read_tensors_start_at_multiples_of_tensor_align() {
665        // As MLX writes them: an unpadded header (data at an odd offset), tensors back to
666        // back. `a` (300 bytes) ends off the alignment; `b` follows it; `skip` is rejected;
667        // `c` (256 bytes) is followed by `d`, which then lands aligned and continues `c`.
668        let json = br#"{"a":{"dtype":"U32","shape":[75],"data_offsets":[0,300]},
669            "b":{"dtype":"BF16","shape":[5],"data_offsets":[300,310]},
670            "skip":{"dtype":"U8","shape":[2],"data_offsets":[310,312]},
671            "c":{"dtype":"U32","shape":[64],"data_offsets":[312,568]},
672            "d":{"dtype":"U32","shape":[1],"data_offsets":[568,572]}}"#;
673        let start = 8 + json.len();
674        let header = parse_header(json, start, start + 572).unwrap();
675        let plan = header.plan(&|name| name != "skip");
676        let offsets: Vec<usize> = plan.tensors.iter().map(|&(_, at)| at).collect();
677        assert_eq!(offsets, [0, 512, 768, 1024]);
678        assert!(offsets.iter().all(|o| o.is_multiple_of(TENSOR_ALIGN)));
679        let s = start as u64;
680        assert_eq!(
681            plan.ranges,
682            [(s, 0, 300), (s + 300, 512, 10), (s + 312, 768, 260)]
683        );
684        assert_eq!(plan.len, 1028);
685    }
686}