1use 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#[derive(Debug, Clone, Default)]
23pub struct SafeTensors {
24 pub tensors: BTreeMap<String, Array>,
26 pub metadata: BTreeMap<String, String>,
28}
29
30fn 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 "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
80const MAX_HEADER: usize = 100_000_000;
82
83const TENSOR_ALIGN: usize = 256;
86
87struct Header {
90 metadata: BTreeMap<String, String>,
91 ranges: Vec<(usize, usize, String, DType, Vec<usize>)>,
93 start: usize,
95}
96
97impl Header {
98 fn aligned(&self) -> bool {
101 self.ranges
102 .iter()
103 .all(|(begin, _, _, dtype, _)| (self.start + begin).is_multiple_of(dtype.size()))
104 }
105
106 fn plan(&self, keep: &dyn Fn(&str) -> bool) -> ReadPlan {
113 let mut plan = ReadPlan::default();
114 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#[derive(Default)]
141struct ReadPlan {
142 ranges: Vec<(u64, usize, usize)>,
144 tensors: Vec<(usize, usize)>,
146 len: usize,
148}
149
150fn 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
160fn 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 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 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
246fn 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
268fn 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#[non_exhaustive]
281#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
282pub struct Footprint {
283 pub mapped: u64,
285 pub copied: u64,
289 pub transient: u64,
292}
293
294impl Footprint {
295 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
316pub fn safetensors_footprint(path: impl AsRef<Path>) -> Result<Footprint> {
339 safetensors_footprint_filtered(path, |_| true)
340}
341
342pub 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
386pub fn load_safetensors(path: impl AsRef<Path>) -> Result<SafeTensors> {
398 load_safetensors_filtered(path, |_| true)
399}
400
401pub 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 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 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
481pub(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
513fn 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
548fn 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
564pub 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#[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 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
608pub 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 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}