Skip to content

Compress SChunk writes through c-blosc2's *_buffer functions (no new buffer per chunk for frames) #765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
FrancescAlted merged 1 commit into from
Nov 6, 2026
+304 −191
Merged

Compress SChunk writes through c-blosc2's *_buffer functions (no new buffer per chunk for frames) #765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
299 changes: 119 additions & 180 deletions src/blosc2/blosc2_ext.pyx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -515,7 +515,11 @@ cdef extern from "blosc2.h":
int64_t blosc2_schunk_fill_special(blosc2_schunk *schunk, int64_t nitems, int special_value,
int32_t chunksize);

int64_t blosc2_schunk_append_buffer(blosc2_schunk *schunk, void *src, int32_t nbytes)
int64_t blosc2_schunk_append_buffer(blosc2_schunk *schunk, const void *src, int32_t nbytes) nogil
int64_t blosc2_schunk_insert_buffer(blosc2_schunk *schunk, int64_t nchunk, const void *src,
int32_t nbytes) nogil
int64_t blosc2_schunk_update_buffer(blosc2_schunk *schunk, int64_t nchunk, const void *src,
int32_t nbytes) nogil
int blosc2_schunk_decompress_chunk(blosc2_schunk *schunk, int64_t nchunk, void *dest, int32_t nbytes)

int blosc2_schunk_get_chunk(blosc2_schunk *schunk, int64_t nchunk, uint8_t ** chunk,
Expand Down Expand Up @@ -1399,6 +1403,55 @@ cdef int _raise_callback_error() except -1:
SystemError."""
return -1 if PyErr_Occurred() != NULL else 0


cdef enum _write_op:
_WRITE_APPEND
_WRITE_INSERT
_WRITE_UPDATE


cdef inline int64_t _store_buffer_nogil(blosc2_schunk *schunk, _write_op op, int64_t nchunk,
const void *src, int32_t nbytes) noexcept nogil:
if op == _WRITE_APPEND:
return blosc2_schunk_append_buffer(schunk, src, nbytes)
if op == _WRITE_INSERT:
return blosc2_schunk_insert_buffer(schunk, nchunk, src, nbytes)
return blosc2_schunk_update_buffer(schunk, nchunk, src, nbytes)


cdef int64_t _store_buffer(blosc2_schunk *schunk, _write_op op, int64_t nchunk,
const void *src, Py_ssize_t nbytes) except? -1:
"""Compress `nbytes` at `src` with the super-chunk's context and append it (`nchunk` is
ignored then), insert it at `nchunk` or replace chunk `nchunk` with it; c-blosc2 reuses its
compression buffer for frames. Returns the number of chunks, or a negative c-blosc2 error
code."""
cdef int64_t rc
if nbytes > BLOSC2_MAX_BUFFERSIZE:
raise ValueError(f"Cannot compress more than {BLOSC2_MAX_BUFFERSIZE} bytes into one chunk")
if _can_compress_without_gil(schunk.cctx):
with nogil:
rc = _store_buffer_nogil(schunk, op, nchunk, src, <int32_t> nbytes)
else:
rc = _store_buffer_nogil(schunk, op, nchunk, src, <int32_t> nbytes)
return rc


cdef int64_t _store_data(blosc2_schunk *schunk, _write_op op, int64_t nchunk, data,
str error) except -1:
"""_store_buffer() for an object exposing the buffer protocol; raises RuntimeError(error)
if c-blosc2 fails."""
cdef Py_buffer buf
cdef int64_t rc
PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE)
try:
rc = _store_buffer(schunk, op, nchunk, buf.buf, buf.len)
finally:
PyBuffer_Release(&buf)
if rc < 0:
_raise_callback_error()
raise RuntimeError(error)
return rc

def get_blocksize():
return blosc1_get_blocksize()

Expand Down Expand Up @@ -1995,14 +2048,10 @@ cdef class SChunk:
if chunksize > INT_MAX:
raise ValueError("Maximum chunksize allowed is 2^31 - 1")
self.schunk.chunksize = chunksize
cdef const uint8_t[:] typed_view
cdef int64_t index
cdef Py_buffer buf
cdef uint8_t *buf_ptr
cdef int comp_size
cdef int32_t csize
cdef uint8_t* chunk
cdef int32_t len_chunk
cdef int64_t len_chunk
if data is not None:
PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE)
buf_ptr = <uint8_t *> buf.buf
Expand All @@ -2013,26 +2062,11 @@ cdef class SChunk:
if i == (nchunks - 1):
len_chunk = len_data - i * chunksize
index = i * chunksize
csize = <int32_t> (len_chunk + BLOSC2_MAX_OVERHEAD)
chunk = <uint8_t*> malloc(csize)
self.schunk.current_nchunk = i
comp_size = _compress_ctx(self.schunk.cctx, buf_ptr + index, len_chunk, chunk, csize)
if comp_size < 0:
free(chunk)
rc = _store_buffer(self.schunk, _WRITE_APPEND, i, buf_ptr + index, len_chunk)
if rc < 0:
PyBuffer_Release(&buf)
_raise_callback_error()
raise RuntimeError("Could not compress the data")
elif comp_size == 0:
free(chunk)
PyBuffer_Release(&buf)
raise RuntimeError("The result could not fit")
chunk = <uint8_t*> realloc(chunk, comp_size)
_check_comp_length('chunk', comp_size)
nchunks_ = blosc2_schunk_append_chunk(self.schunk, chunk, False)
if nchunks_ != (i + 1):
free(chunk)
PyBuffer_Release(&buf)
raise RuntimeError("An error occurred while appending the chunks")
raise RuntimeError("Could not compress and append the chunk")
PyBuffer_Release(&buf)

@property
Expand Down Expand Up @@ -2180,28 +2214,8 @@ cdef class SChunk:

@_handle_locked
def append_data(self, data):
cdef Py_buffer buf
PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE)
cdef int size
cdef int32_t len_chunk = <int32_t> (buf.len + BLOSC2_MAX_OVERHEAD)
cdef uint8_t* chunk = <uint8_t*> malloc(len_chunk)
self.schunk.current_nchunk = self.schunk.nchunks
size = _compress_ctx(self.schunk.cctx, buf.buf, <int32_t> buf.len, chunk, len_chunk)
PyBuffer_Release(&buf)
if size < 0:
free(chunk)
_raise_callback_error()
raise RuntimeError("Could not compress the data")
elif size == 0:
free(chunk)
raise RuntimeError("The result could not fit")
chunk = <uint8_t*> realloc(chunk, size)
_check_comp_length('chunk', size)
rc = blosc2_schunk_append_chunk(self.schunk, chunk, False)
if rc < 0:
free(chunk)
raise RuntimeError("Could not append the chunk")
return rc
return _store_data(self.schunk, _WRITE_APPEND, self.schunk.nchunks, data,
"Could not compress and append the chunk")

@_handle_locked
def fill_special(self, nitems, special_value, value):
Expand Down Expand Up @@ -2409,32 +2423,9 @@ cdef class SChunk:
return rc

@_handle_locked
def insert_data(self, nchunk, data, copy):
cdef blosc2_context *cctx
cdef Py_buffer buf
PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE)
cdef int size
cdef int32_t len_chunk = <int32_t> (buf.len + BLOSC2_MAX_OVERHEAD)
cdef uint8_t* chunk = <uint8_t*> malloc(len_chunk)
self.schunk.current_nchunk = nchunk # prefilter needs this value to be set
size = _compress_ctx(self.schunk.cctx, buf.buf, <int32_t> buf.len, chunk, len_chunk)
PyBuffer_Release(&buf)
if size <= 0:
free(chunk)
if size < 0:
_raise_callback_error()
raise RuntimeError("Could not compress the data")
raise RuntimeError("The result could not fit ")

chunk = <uint8_t*> realloc(chunk, size)
_check_comp_length('chunk', size)
rc = blosc2_schunk_insert_chunk(self.schunk, nchunk, chunk, copy)
# Without a copy, the schunk owns the chunk only if the insert succeeded
if copy or rc < 0:
free(chunk)
if rc < 0:
raise RuntimeError("Could not insert the desired chunk")
return rc
def insert_data(self, int64_t nchunk, data, copy=False):
return _store_data(self.schunk, _WRITE_INSERT, nchunk, data,
"Could not compress and insert the chunk")

@_handle_locked
def update_chunk(self, nchunk, chunk):
Expand All @@ -2456,32 +2447,9 @@ cdef class SChunk:
return None

@_handle_locked
def update_data(self, nchunk, data, copy):
cdef Py_buffer buf
PyObject_GetBuffer(data, &buf, PyBUF_SIMPLE)
cdef int size
cdef int32_t len_chunk = <int32_t> (buf.len + BLOSC2_MAX_OVERHEAD)
cdef uint8_t* chunk = <uint8_t*> malloc(len_chunk)
self.schunk.current_nchunk = nchunk # prefilter needs this value to be set
size = _compress_ctx(self.schunk.cctx, buf.buf, <int32_t> buf.len, chunk, len_chunk)

PyBuffer_Release(&buf)
if size <= 0:
free(chunk)
if size < 0:
_raise_callback_error()
raise RuntimeError("Could not compress the data")
raise RuntimeError("The result could not fit ")

chunk = <uint8_t*> realloc(chunk, size)
_check_comp_length('chunk', size)
rc = blosc2_schunk_update_chunk(self.schunk, nchunk, chunk, copy)
# Without a copy, the schunk owns the chunk only if the update succeeded
if copy or rc < 0:
free(chunk)
if rc < 0:
raise RuntimeError("Could not update the desired chunk")
return rc
def update_data(self, int64_t nchunk, data, copy=False):
return _store_data(self.schunk, _WRITE_UPDATE, nchunk, data,
"Could not compress and update the chunk")

# This is used internally for prefiltering
@_handle_locked
Expand Down Expand Up @@ -2539,93 +2507,64 @@ cdef class SChunk:
cdef int64_t nbytes_copy = min(nbytes, buf.len - buf_pos)
cdef int64_t data_start
cdef uint8_t *data
cdef uint8_t *chunk
cdef int32_t alloc_len
cdef int32_t chunk_nbytes
cdef int32_t chunksize
cdef int comp_rc
if buf.len < nbytes:
raise ValueError("Not enough data for writing the slice")

if stop > nitems:
# Increase SChunk's size
if start < nitems:
rc = blosc2_schunk_set_slice_buffer(self.schunk, start, nitems, buf.buf)
if rc < 0:
PyBuffer_Release(&buf)
_raise_callback_error()
raise RuntimeError("Error while setting the slice")
buf_pos = (nitems - start) * self.schunk.typesize
if self.schunk.nbytes % self.schunk.chunksize != 0:
# Update last chunk before appending any other
if stop * self.schunk.typesize >= self.schunk.chunksize * self.schunk.nchunks:
chunk_nbytes = self.schunk.chunksize
nbytes_copy = min(nbytes_copy, self.schunk.chunksize * self.schunk.nchunks - nitems * self.schunk.typesize)
else:
chunk_nbytes = (stop * self.schunk.typesize) % self.schunk.chunksize
data = <uint8_t *> malloc(chunk_nbytes)
rc = blosc2_schunk_decompress_chunk(self.schunk, self.schunk.nchunks - 1, data, chunk_nbytes)
if rc < 0:
free(data)
PyBuffer_Release(&buf)
_raise_callback_error()
raise RuntimeError("Error while decompressing the chunk")
data_start = self.schunk.nbytes - (self.schunk.nchunks - 1) * self.schunk.chunksize
memcpy(data + data_start, buf_ptr + buf_pos, nbytes_copy)
chunk = <uint8_t *> malloc(chunk_nbytes + BLOSC2_MAX_OVERHEAD)
self.schunk.current_nchunk = self.schunk.nchunks - 1
comp_rc = _compress_ctx(self.schunk.cctx, data, chunk_nbytes, chunk,
chunk_nbytes + BLOSC2_MAX_OVERHEAD)
free(data)
if comp_rc < 0:
free(chunk)
PyBuffer_Release(&buf)
_raise_callback_error()
raise RuntimeError("Error while compressing the data")
elif comp_rc == 0:
free(chunk)
PyBuffer_Release(&buf)
raise RuntimeError("The result could not fit")
rc = blosc2_schunk_update_chunk(self.schunk, self.schunk.nchunks - 1, chunk, True)
free(chunk)
if rc < 0:
PyBuffer_Release(&buf)
raise RuntimeError("Error while updating the chunk")
buf_pos += nbytes_copy
# Append data if needed
if buf_pos < buf.len:
nappends = int(stop * self.schunk.typesize / self.schunk.chunksize - self.schunk.nchunks)
if (stop * self.schunk.typesize) % self.schunk.chunksize != 0:
nappends += 1
for i in range(nappends):
if (self.schunk.nchunks + 1) * self.schunk.chunksize <= stop * self.schunk.typesize:
chunksize = self.schunk.chunksize
cdef int64_t rc
try:
if buf.len < nbytes:
raise ValueError("Not enough data for writing the slice")

if stop > nitems:
# Increase SChunk's size
if start < nitems:
rc = blosc2_schunk_set_slice_buffer(self.schunk, start, nitems, buf.buf)
if rc < 0:
_raise_callback_error()
raise RuntimeError("Error while setting the slice")
buf_pos = (nitems - start) * self.schunk.typesize
if self.schunk.nbytes % self.schunk.chunksize != 0:
# Update last chunk before appending any other
if stop * self.schunk.typesize >= self.schunk.chunksize * self.schunk.nchunks:
chunk_nbytes = self.schunk.chunksize
nbytes_copy = min(nbytes_copy, self.schunk.chunksize * self.schunk.nchunks - nitems * self.schunk.typesize)
else:
chunksize = (stop * self.schunk.typesize) % self.schunk.chunksize
alloc_len = <int32_t> (chunksize + BLOSC2_MAX_OVERHEAD)
chunk = <uint8_t*> malloc(alloc_len)
self.schunk.current_nchunk = self.schunk.nchunks
comp_rc = _compress_ctx(self.schunk.cctx, buf_ptr + buf_pos, chunksize, chunk, alloc_len)
if comp_rc < 0:
free(chunk)
PyBuffer_Release(&buf)
chunk_nbytes = (stop * self.schunk.typesize) % self.schunk.chunksize
data = <uint8_t *> malloc(chunk_nbytes)
rc = blosc2_schunk_decompress_chunk(self.schunk, self.schunk.nchunks - 1, data, chunk_nbytes)
if rc < 0:
free(data)
_raise_callback_error()
raise RuntimeError("Error while compressing the chunk")
elif comp_rc == 0:
free(chunk)
PyBuffer_Release(&buf)
raise RuntimeError("The result could not fit")
chunk = <uint8_t*> realloc(chunk, comp_rc)
_check_comp_length('chunk', comp_rc)
rc = blosc2_schunk_append_chunk(self.schunk, chunk, False)
raise RuntimeError("Error while decompressing the chunk")
data_start = self.schunk.nbytes - (self.schunk.nchunks - 1) * self.schunk.chunksize
memcpy(data + data_start, buf_ptr + buf_pos, nbytes_copy)
try:
rc = _store_buffer(self.schunk, _WRITE_UPDATE, self.schunk.nchunks - 1, data, chunk_nbytes)
finally:
free(data)
if rc < 0:
free(chunk)
PyBuffer_Release(&buf)
raise RuntimeError("Error while appending the chunk")
buf_pos += chunksize
else:
rc = blosc2_schunk_set_slice_buffer(self.schunk, start, stop, buf.buf)
PyBuffer_Release(&buf)
_raise_callback_error()
raise RuntimeError("Error while compressing and updating the chunk")
buf_pos += nbytes_copy
# Append data if needed
if buf_pos < buf.len:
nappends = int(stop * self.schunk.typesize / self.schunk.chunksize - self.schunk.nchunks)
if (stop * self.schunk.typesize) % self.schunk.chunksize != 0:
nappends += 1
for i in range(nappends):
if (self.schunk.nchunks + 1) * self.schunk.chunksize <= stop * self.schunk.typesize:
chunksize = self.schunk.chunksize
else:
chunksize = (stop * self.schunk.typesize) % self.schunk.chunksize
rc = _store_buffer(self.schunk, _WRITE_APPEND, self.schunk.nchunks, buf_ptr + buf_pos,
chunksize)
if rc < 0:
_raise_callback_error()
raise RuntimeError("Error while compressing and appending the chunk")
buf_pos += chunksize
else:
rc = blosc2_schunk_set_slice_buffer(self.schunk, start, stop, buf.buf)
finally:
PyBuffer_Release(&buf)
_check_rc(rc, "Error while setting the slice")

@_handle_locked
Expand Down
Loading
Loading