1#![allow(clippy::useless_conversion)]
5use std::sync::{Arc, OnceLock};
23
24use bytes::{Bytes, BytesMut};
25use pyo3::create_exception;
26use pyo3::exceptions::{PyIOError, PyNotImplementedError, PyRuntimeError, PyValueError};
27use pyo3::prelude::*;
28use pyo3::types::{PyBytes, PyDict, PyList};
29use s4_codec_rs::index::{FrameIndex, IndexError};
30use s4_codec_rs::multipart::{write_frame, FrameError, FrameHeader, FrameIter};
31use s4_codec_rs::{cpu_gzip, cpu_zstd, cpu_zstd_dict, ChunkManifest, Codec, CodecError, CodecKind};
32use tokio::runtime::{Builder, Runtime};
33
34create_exception!(s4_codec, S4Error, PyValueError);
55create_exception!(s4_codec, S4CrcMismatchError, S4Error);
56create_exception!(s4_codec, S4SizeMismatchError, S4Error);
57create_exception!(s4_codec, S4CodecMismatchError, S4Error);
58create_exception!(s4_codec, S4UnregisteredCodecError, S4Error);
59create_exception!(s4_codec, S4ManifestSizeExceedsLimitError, S4Error);
60create_exception!(s4_codec, S4ManifestSizeMismatchError, S4Error);
61create_exception!(s4_codec, S4BackendError, PyRuntimeError);
62create_exception!(s4_codec, S4IoError, PyIOError);
63create_exception!(s4_codec, S4FrameError, S4Error);
69create_exception!(s4_codec, S4IndexError, S4Error);
70
71fn runtime() -> &'static Runtime {
72 static RT: OnceLock<Runtime> = OnceLock::new();
73 RT.get_or_init(|| {
74 Builder::new_multi_thread()
75 .enable_all()
76 .thread_name("s4-codec-py")
77 .build()
78 .expect("failed to start tokio runtime for s4_codec python binding")
79 })
80}
81
82fn codec_err_to_py(e: CodecError) -> PyErr {
83 use s4_codec_rs::CodecError::*;
84 match e {
85 SizeMismatch { expected, got } => {
86 S4SizeMismatchError::new_err(format!("size mismatch: expected {expected}, got {got}"))
87 }
88 CrcMismatch { expected, got } => S4CrcMismatchError::new_err(format!(
89 "crc32c mismatch: expected {expected:#010x}, got {got:#010x}"
90 )),
91 CodecMismatch { expected, got } => S4CodecMismatchError::new_err(format!(
92 "codec mismatch: expected {expected:?}, got {got:?}"
93 )),
94 UnregisteredCodec(k) => {
95 S4UnregisteredCodecError::new_err(format!("codec {k:?} not registered"))
96 }
97 ManifestSizeExceedsLimit { requested, limit } => S4ManifestSizeExceedsLimitError::new_err(
98 format!("manifest claims {requested} bytes but limit is {limit}"),
99 ),
100 ManifestSizeMismatch { manifest, actual } => S4ManifestSizeMismatchError::new_err(format!(
101 "manifest claims {manifest} bytes but body is {actual}"
102 )),
103 Backend(msg) => S4BackendError::new_err(format!("backend: {msg}")),
104 Io(e) => S4IoError::new_err(format!("io: {e}")),
105 TruncatedStream { expected, got } => S4Error::new_err(format!(
106 "stream truncated: expected {expected} input bytes, got {got}"
107 )),
108 OverlengthStream { expected, got } => S4Error::new_err(format!(
114 "stream over-length: expected {expected} input bytes, got at least {got}"
115 )),
116 Join(e) => S4BackendError::new_err(format!("backend (worker join): {e}")),
120 other => S4Error::new_err(format!("codec error: {other}")),
126 }
127}
128
129fn manifest_from_parts(
130 kind: CodecKind,
131 payload_len: u64,
132 original_size: u64,
133 crc32c: u32,
134) -> ChunkManifest {
135 ChunkManifest {
136 codec: kind,
137 original_size,
138 compressed_size: payload_len,
139 crc32c,
140 }
141}
142
143fn block_on<F, T>(py: Python<'_>, fut: F) -> T
147where
148 F: std::future::Future<Output = T> + Send,
149 T: Send,
150{
151 py.detach(|| runtime().block_on(fut))
152}
153
154#[pyclass(name = "CpuZstd", module = "s4_codec")]
157struct PyCpuZstd {
158 inner: cpu_zstd::CpuZstd,
159}
160
161#[pymethods]
162impl PyCpuZstd {
163 #[new]
164 #[pyo3(signature = (level = 3))]
165 fn new(level: i32) -> Self {
166 Self {
167 inner: cpu_zstd::CpuZstd::new(level),
168 }
169 }
170
171 fn compress<'py>(
175 &self,
176 py: Python<'py>,
177 data: &Bound<'py, PyBytes>,
178 ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
179 let input = Bytes::copy_from_slice(data.as_bytes());
180 let codec = self.inner.clone();
181 let (out, manifest) =
182 block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
183 Ok((
184 PyBytes::new(py, &out),
185 manifest.original_size,
186 manifest.crc32c,
187 ))
188 }
189
190 fn decompress<'py>(
193 &self,
194 py: Python<'py>,
195 data: &Bound<'py, PyBytes>,
196 original_size: u64,
197 crc32c: u32,
198 ) -> PyResult<Bound<'py, PyBytes>> {
199 let input = Bytes::copy_from_slice(data.as_bytes());
200 let manifest = manifest_from_parts(
201 CodecKind::CpuZstd,
202 input.len() as u64,
203 original_size,
204 crc32c,
205 );
206 let codec = self.inner.clone();
207 let out = block_on(py, async move { codec.decompress(input, &manifest).await })
208 .map_err(codec_err_to_py)?;
209 Ok(PyBytes::new(py, &out))
210 }
211
212 fn __repr__(&self) -> String {
213 format!("CpuZstd(level={})", cpu_zstd::CpuZstd::DEFAULT_LEVEL)
214 }
215}
216
217#[pyclass(name = "CpuGzip", module = "s4_codec")]
221struct PyCpuGzip {
222 inner: cpu_gzip::CpuGzip,
223}
224
225#[pymethods]
226impl PyCpuGzip {
227 #[new]
228 #[pyo3(signature = (level = 6))]
229 fn new(level: u32) -> Self {
230 Self {
231 inner: cpu_gzip::CpuGzip::new(level),
232 }
233 }
234
235 fn compress<'py>(
236 &self,
237 py: Python<'py>,
238 data: &Bound<'py, PyBytes>,
239 ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
240 let input = Bytes::copy_from_slice(data.as_bytes());
241 let codec = self.inner.clone();
242 let (out, manifest) =
243 block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
244 Ok((
245 PyBytes::new(py, &out),
246 manifest.original_size,
247 manifest.crc32c,
248 ))
249 }
250
251 fn decompress<'py>(
252 &self,
253 py: Python<'py>,
254 data: &Bound<'py, PyBytes>,
255 original_size: u64,
256 crc32c: u32,
257 ) -> PyResult<Bound<'py, PyBytes>> {
258 let input = Bytes::copy_from_slice(data.as_bytes());
259 let manifest = manifest_from_parts(
260 CodecKind::CpuGzip,
261 input.len() as u64,
262 original_size,
263 crc32c,
264 );
265 let codec = self.inner.clone();
266 let out = block_on(py, async move { codec.decompress(input, &manifest).await })
267 .map_err(codec_err_to_py)?;
268 Ok(PyBytes::new(py, &out))
269 }
270
271 fn __repr__(&self) -> String {
272 format!("CpuGzip(level={})", cpu_gzip::CpuGzip::DEFAULT_LEVEL)
273 }
274}
275
276#[pyclass(name = "CpuZstdDict", module = "s4_codec")]
281struct PyCpuZstdDict {
282 inner: cpu_zstd_dict::CpuZstdDict,
283 level: i32,
284}
285
286#[pymethods]
287impl PyCpuZstdDict {
288 #[new]
289 #[pyo3(signature = (dict_bytes, level = 3))]
290 fn new(dict_bytes: &Bound<'_, PyBytes>, level: i32) -> PyResult<Self> {
291 let dict: Arc<[u8]> = Arc::from(dict_bytes.as_bytes().to_vec().into_boxed_slice());
292 let inner = cpu_zstd_dict::CpuZstdDict::new(dict, level).map_err(codec_err_to_py)?;
293 Ok(Self {
294 inner,
295 level: level.clamp(1, 22),
296 })
297 }
298
299 fn compress<'py>(
303 &self,
304 py: Python<'py>,
305 data: &Bound<'py, PyBytes>,
306 ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
307 let input = Bytes::copy_from_slice(data.as_bytes());
308 let codec = self.inner.clone();
309 let (out, manifest) =
310 block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
311 Ok((
312 PyBytes::new(py, &out),
313 manifest.original_size,
314 manifest.crc32c,
315 ))
316 }
317
318 fn decompress<'py>(
322 &self,
323 py: Python<'py>,
324 data: &Bound<'py, PyBytes>,
325 original_size: u64,
326 crc32c: u32,
327 ) -> PyResult<Bound<'py, PyBytes>> {
328 let input = Bytes::copy_from_slice(data.as_bytes());
329 let manifest = manifest_from_parts(
330 CodecKind::CpuZstdDict,
331 input.len() as u64,
332 original_size,
333 crc32c,
334 );
335 let codec = self.inner.clone();
336 let out = block_on(py, async move { codec.decompress(input, &manifest).await })
337 .map_err(codec_err_to_py)?;
338 Ok(PyBytes::new(py, &out))
339 }
340
341 fn __repr__(&self) -> String {
342 format!(
343 "CpuZstdDict(dict_len={}, level={})",
344 self.inner.dict().len(),
345 self.level
346 )
347 }
348}
349
350#[pyfunction]
353fn gpu_available() -> bool {
354 s4_codec_rs::nvcomp::is_gpu_available()
355}
356
357fn frame_err_to_py(e: FrameError) -> PyErr {
358 S4FrameError::new_err(format!("frame error: {e}"))
359}
360
361fn index_err_to_py(e: IndexError) -> PyErr {
362 S4IndexError::new_err(format!("index error: {e}"))
363}
364
365fn frame_header_dict<'py>(py: Python<'py>, h: &FrameHeader) -> PyResult<Bound<'py, PyDict>> {
369 let d = PyDict::new(py);
370 d.set_item("codec", h.codec.as_str())?;
371 d.set_item("original_size", h.original_size)?;
372 d.set_item("compressed_size", h.compressed_size)?;
373 d.set_item("crc32c", h.crc32c)?;
374 Ok(d)
375}
376
377#[pyfunction]
386fn read_frame<'py>(
387 py: Python<'py>,
388 data: &Bound<'py, PyBytes>,
389) -> PyResult<(Bound<'py, PyDict>, Bound<'py, PyBytes>, Bound<'py, PyBytes>)> {
390 let input = Bytes::copy_from_slice(data.as_bytes());
391 let (header, payload, rest) =
392 s4_codec_rs::multipart::read_frame(input).map_err(frame_err_to_py)?;
393 Ok((
394 frame_header_dict(py, &header)?,
395 PyBytes::new(py, &payload),
396 PyBytes::new(py, &rest),
397 ))
398}
399
400#[pyfunction]
405fn frame_iter<'py>(py: Python<'py>, data: &Bound<'py, PyBytes>) -> PyResult<Bound<'py, PyList>> {
406 let input = Bytes::copy_from_slice(data.as_bytes());
407 let out = PyList::empty(py);
408 for item in FrameIter::new(input) {
409 let (header, payload) = item.map_err(frame_err_to_py)?;
410 out.append((frame_header_dict(py, &header)?, PyBytes::new(py, &payload)))?;
411 }
412 Ok(out)
413}
414
415#[pyfunction]
436fn decode_index<'py>(py: Python<'py>, data: &Bound<'py, PyBytes>) -> PyResult<Bound<'py, PyDict>> {
437 let input = Bytes::copy_from_slice(data.as_bytes());
438 let idx: FrameIndex = s4_codec_rs::index::decode_index(input).map_err(index_err_to_py)?;
439 let d = PyDict::new(py);
440 d.set_item("total_padded_size", idx.total_padded_size)?;
441 d.set_item("total_original_size", idx.total_original_size())?;
442 d.set_item("source_etag", idx.source_etag.as_deref())?;
443 d.set_item("source_compressed_size", idx.source_compressed_size)?;
444 let entries = PyList::empty(py);
445 for e in &idx.entries {
446 let ed = PyDict::new(py);
447 ed.set_item("original_offset", e.original_offset)?;
448 ed.set_item("original_size", e.original_size)?;
449 ed.set_item("compressed_offset", e.compressed_offset)?;
450 ed.set_item("compressed_size", e.compressed_size)?;
451 entries.append(ed)?;
452 }
453 d.set_item("entries", entries)?;
454 match idx.sse_v3 {
455 Some(sse) => {
456 let sd = PyDict::new(py);
457 sd.set_item("enc_chunk_size", sse.enc_chunk_size)?;
458 sd.set_item("enc_chunk_count", sse.enc_chunk_count)?;
459 sd.set_item("enc_key_id", sse.enc_key_id)?;
460 sd.set_item("enc_salt", PyBytes::new(py, &sse.enc_salt))?;
461 sd.set_item("enc_plaintext_len", sse.enc_plaintext_len)?;
462 sd.set_item("enc_header_bytes", sse.enc_header_bytes)?;
463 d.set_item("sse", sd)?;
464 }
465 None => d.set_item("sse", py.None())?,
466 }
467 Ok(d)
468}
469
470#[pyfunction]
474#[pyo3(name = "crc32c")]
475fn crc32c_py(data: &Bound<'_, PyBytes>) -> u32 {
476 ::crc32c::crc32c(data.as_bytes())
477}
478
479const DEFAULT_S4F2_CHUNK_SIZE: usize = 4 * 1024 * 1024;
494
495fn pick_chunk_size_impl(content_length: u64) -> usize {
503 if content_length <= 1024 * 1024 {
504 1024 * 1024
505 } else if content_length <= 100 * 1024 * 1024 {
506 DEFAULT_S4F2_CHUNK_SIZE
507 } else {
508 16 * 1024 * 1024
509 }
510}
511
512#[pyfunction]
516fn pick_chunk_size(content_length: u64) -> usize {
517 pick_chunk_size_impl(content_length)
518}
519
520fn encode_framed_zstd(
527 py: Python<'_>,
528 input: Bytes,
529 level: i32,
530) -> PyResult<(Bytes, ChunkManifest)> {
531 let chunk_size = pick_chunk_size_impl(input.len() as u64);
532 let codec = cpu_zstd::CpuZstd::new(level);
533 let res: Result<(Bytes, ChunkManifest), CodecError> = block_on(py, async move {
534 let mut framed = BytesMut::new();
535 let mut rolling_crc: u32 = 0;
536 let total = input.len() as u64;
537 let mut off = 0usize;
538 while off < input.len() {
539 let end = usize::min(off + chunk_size, input.len());
540 let chunk = input.slice(off..end);
541 let chunk_crc = ::crc32c::crc32c(&chunk);
542 rolling_crc = ::crc32c::crc32c_append(rolling_crc, &chunk);
543 let chunk_len = chunk.len() as u64;
544 let (compressed, _manifest) = codec.compress(chunk).await?;
545 let header = FrameHeader {
546 codec: CodecKind::CpuZstd,
547 original_size: chunk_len,
548 compressed_size: compressed.len() as u64,
549 crc32c: chunk_crc,
550 };
551 write_frame(&mut framed, header, &compressed);
552 off = end;
553 }
554 let body = framed.freeze();
555 let compressed_size = body.len() as u64;
556 Ok((
557 body,
558 ChunkManifest {
559 codec: CodecKind::CpuZstd,
560 original_size: total,
561 compressed_size,
562 crc32c: rolling_crc,
563 },
564 ))
565 });
566 res.map_err(codec_err_to_py)
567}
568
569#[pyfunction]
606#[pyo3(signature = (data, codec = "cpu-zstd", level = 3))]
607fn encode_s4_object<'py>(
608 py: Python<'py>,
609 data: &Bound<'py, PyBytes>,
610 codec: &str,
611 level: i32,
612) -> PyResult<Bound<'py, PyDict>> {
613 let input = Bytes::copy_from_slice(data.as_bytes());
614 let (body, manifest, framed) = match codec {
615 "cpu-zstd" => {
616 let (body, manifest) = encode_framed_zstd(py, input, level)?;
617 (body, manifest, true)
618 }
619 "passthrough" => {
623 let crc = py.detach(|| ::crc32c::crc32c(&input));
627 let len = input.len() as u64;
628 let manifest = ChunkManifest {
629 codec: CodecKind::Passthrough,
630 original_size: len,
631 compressed_size: len,
632 crc32c: crc,
633 };
634 (input, manifest, false)
635 }
636 other => {
637 return Err(PyNotImplementedError::new_err(format!(
638 "encode_s4_object supports codecs 'cpu-zstd' and 'passthrough' only; \
639 {other:?} (cpu-gzip / cpu-zstd-dict / nvcomp-* / dietgpu-ans) must be \
640 written through the S4 gateway, which owns those codec paths"
641 )));
642 }
643 };
644 let sidecar = if framed {
648 let idx = s4_codec_rs::index::build_index_from_body(&body).map_err(frame_err_to_py)?;
649 if idx.entries.len() > 1 {
650 Some(s4_codec_rs::index::encode_index(&idx))
651 } else {
652 None
653 }
654 } else {
655 None
656 };
657 let out = PyDict::new(py);
658 out.set_item("body", PyBytes::new(py, &body))?;
659 match sidecar {
660 Some(s) => out.set_item("sidecar", PyBytes::new(py, &s))?,
661 None => out.set_item("sidecar", py.None())?,
662 }
663 let meta = PyDict::new(py);
664 meta.set_item("s4-codec", manifest.codec.as_str())?;
668 meta.set_item("s4-original-size", manifest.original_size.to_string())?;
669 meta.set_item("s4-compressed-size", manifest.compressed_size.to_string())?;
670 meta.set_item("s4-crc32c", manifest.crc32c.to_string())?;
671 if framed {
672 meta.set_item("s4-framed", "true")?;
673 }
674 out.set_item("metadata", meta)?;
675 Ok(out)
676}
677
678#[pyfunction]
695#[pyo3(signature = (sidecar, source_compressed_size, source_etag = None))]
696fn bind_index<'py>(
697 py: Python<'py>,
698 sidecar: &Bound<'py, PyBytes>,
699 source_compressed_size: u64,
700 source_etag: Option<String>,
701) -> PyResult<Bound<'py, PyBytes>> {
702 let raw = Bytes::copy_from_slice(sidecar.as_bytes());
703 let mut idx: FrameIndex = s4_codec_rs::index::decode_index(raw).map_err(index_err_to_py)?;
704 idx.source_compressed_size = Some(source_compressed_size);
705 idx.source_etag = source_etag;
706 Ok(PyBytes::new(py, &s4_codec_rs::index::encode_index(&idx)))
707}
708
709#[pymodule]
710fn s4_codec(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
711 m.add_class::<PyCpuZstd>()?;
712 m.add_class::<PyCpuGzip>()?;
713 m.add_class::<PyCpuZstdDict>()?;
714 m.add_function(wrap_pyfunction!(gpu_available, m)?)?;
715 m.add_function(wrap_pyfunction!(read_frame, m)?)?;
719 m.add_function(wrap_pyfunction!(frame_iter, m)?)?;
720 m.add_function(wrap_pyfunction!(decode_index, m)?)?;
721 m.add_function(wrap_pyfunction!(crc32c_py, m)?)?;
722 m.add_function(wrap_pyfunction!(encode_s4_object, m)?)?;
725 m.add_function(wrap_pyfunction!(bind_index, m)?)?;
726 m.add_function(wrap_pyfunction!(pick_chunk_size, m)?)?;
727 m.add(
728 "FRAME_MAGIC",
729 PyBytes::new(py, s4_codec_rs::multipart::FRAME_MAGIC),
730 )?;
731 m.add(
732 "PADDING_MAGIC",
733 PyBytes::new(py, s4_codec_rs::multipart::PADDING_MAGIC),
734 )?;
735 m.add(
736 "FRAME_HEADER_BYTES",
737 s4_codec_rs::multipart::FRAME_HEADER_BYTES,
738 )?;
739 m.add("SIDECAR_SUFFIX", s4_codec_rs::index::SIDECAR_SUFFIX)?;
740 m.add("__version__", env!("CARGO_PKG_VERSION"))?;
741 m.add("S4Error", py.get_type::<S4Error>())?;
745 m.add("S4CrcMismatchError", py.get_type::<S4CrcMismatchError>())?;
746 m.add("S4SizeMismatchError", py.get_type::<S4SizeMismatchError>())?;
747 m.add(
748 "S4CodecMismatchError",
749 py.get_type::<S4CodecMismatchError>(),
750 )?;
751 m.add(
752 "S4UnregisteredCodecError",
753 py.get_type::<S4UnregisteredCodecError>(),
754 )?;
755 m.add(
756 "S4ManifestSizeExceedsLimitError",
757 py.get_type::<S4ManifestSizeExceedsLimitError>(),
758 )?;
759 m.add(
760 "S4ManifestSizeMismatchError",
761 py.get_type::<S4ManifestSizeMismatchError>(),
762 )?;
763 m.add("S4BackendError", py.get_type::<S4BackendError>())?;
764 m.add("S4IoError", py.get_type::<S4IoError>())?;
765 m.add("S4FrameError", py.get_type::<S4FrameError>())?;
766 m.add("S4IndexError", py.get_type::<S4IndexError>())?;
767 Ok(())
768}