lz4: unbloc compressed stream using cython code when input data are on the cpu

src/silx/meson.build

@@ -1,6 +1,7 @@
 subdir('io')
 subdir('math')
 subdir('image')
+subdir('opencl')
 
 py.install_sources([
     '__init__.py',
@@ -17,7 +18,6 @@ subdir: 'silx',  # Folder relative to site-packages to install to
 pure_subdirs = [
   'app',
   'gui',
-  'opencl',
   'resources',
   'sx',
   'test',

src/silx/opencl/codec/bitshuffle_lz4.py

@@ -34,7 +34,7 @@
 __contact__ = "jerome.kieffer@esrf.eu"
 __license__ = "MIT"
 __copyright__ = "European Synchrotron Radiation Facility, Grenoble, France"
-__date__ = "13/03/2026"
+__date__ = "05/06/2026"
 __status__ = "production"
 
 
@@ -44,6 +44,7 @@
 from ..common import ocl  # noqa F401 Initialize OpenCL
 from ..common import pyopencl, kernel_workgroup_size
 from ..processing import BufferDescription, EventDescription, OpenclProcessing
+from ._unblock_bslz4 import unblock_bslz4
 
 import logging
 
@@ -122,12 +123,20 @@ def __init__(
             kernel_workgroup_size(self.program, "bslz4_decompress_block"),
         )
 
-    def decompress(self, raw, out=None, wg=None, nbytes=None):
+    def decompress(
+        self,
+        raw,
+        out=None,
+        wg: int = None,
+        nbytes: int = None,
+        force_unblock_on_device: bool = False,
+    ):
         """This function actually performs the decompression by calling the kernels
         :param numpy.ndarray raw: The compressed data as a 1D numpy array of char or string
         :param pyopencl.array out: pyopencl array in which to place the result.
         :param wg: tuneable parameter with the workgroup size.
         :param int nbytes: (Optional) Number of bytes occupied by the chunk in raw.
+        :param bool force_unblock_on_device: set to False for allowing Cython unblocking
         :return: The decompressed image as an pyopencl array.
         :rtype: pyopencl.array
         """
@@ -142,13 +151,15 @@ def decompress(self, raw, out=None, wg=None, nbytes=None):
             else:
                 len_raw = numpy.uint64(len(raw))
 
+            unblock_on_device = True
             if isinstance(raw, pyopencl.array.Array):
                 cmp_buffer = raw.data
                 num_blocks = self.num_blocks
             elif isinstance(raw, pyopencl.Buffer):
                 cmp_buffer = raw
                 num_blocks = self.num_blocks
             else:
+                unblock_on_device = force_unblock_on_device
                 if len_raw > self.cmp_size:
                     self.cmp_size = len_raw
                     logger.info("increase cmp buffer size to %s", self.cmp_size)
@@ -180,17 +191,42 @@ def decompress(self, raw, out=None, wg=None, nbytes=None):
 
             wg = int(wg or self.block_size)
 
-            evt = self.kernels.lz4_unblock(
-                self.queue,
-                (1,),
-                (1,),
-                cmp_buffer,
-                len_raw,
-                self.cl_mem["block_position"].data,
-                num_blocks,
-                self.cl_mem["nb_blocks"].data,
-            )
-            events.append(EventDescription("LZ4 unblock", evt))
+            if unblock_on_device:
+                evt = self.kernels.lz4_unblock(
+                    self.queue,
+                    (1,),
+                    (1,),
+                    cmp_buffer,
+                    len_raw,
+                    self.cl_mem["block_position"].data,
+                    num_blocks,
+                    self.cl_mem["nb_blocks"].data,
+                )
+                events.append(EventDescription("LZ4 unblock", evt))
+            else:
+                # Perform unblock using Cython
+                block_position = unblock_bslz4(raw)
+                size = block_position.size
+                if size > self.num_blocks:
+                    self.num_blocks = size
+                    self.cl_mem["block_position"] = pyopencl.array.empty(
+                        self.queue, self.num_blocks, numpy.uint64
+                    )
+
+                evt = pyopencl.enqueue_copy(
+                    self.queue,
+                    self.cl_mem["block_position"].data,
+                    block_position,
+                    is_blocking=False,
+                )
+                events.append(EventDescription("copy block_position H -> D", evt))
+                evt = pyopencl.enqueue_copy(
+                    self.queue,
+                    self.cl_mem["nb_blocks"].data,
+                    numpy.array([size], numpy.uint32),
+                    is_blocking=False,
+                )
+                events.append(EventDescription("copy nb_blocks H -> D", evt))
 
             if out is None:
                 out = self.cl_mem["dec"]

src/silx/opencl/codec/ext/_unblock_bslz4.pyx

@@ -0,0 +1,83 @@
+#cython: embedsignature=True, language_level=3
+## This is for optimization
+# cython: boundscheck=False, wraparound=False, cdivision=True, initializedcheck=False,
+## This is for developing:
+##cython: profile=True, warn.undeclared=True, warn.unused=True, warn.unused_result=False, warn.unused_arg=True
+
+"""This module provides :func:`unblock_bslz4` which scans a dataset for different blocks.
+"""
+
+
+__authors__ = ["Jérôme Kieffer"]
+__license__ = "MIT"
+__date__ = "05/06/2026"
+
+
+from libc.stdint cimport uint8_t, uint32_t, uint64_t
+import numpy
+
+
+cdef inline uint64_t load64_at(const uint8_t[::1] src,
+                               uint64_t pos) noexcept nogil :
+    """Read a 64-bit BIG-endian integer at the given position in the stream.
+
+    :param src: byte array containing the data
+    :param pos: position in the stream
+    :return: 64-bit unsigned integer
+    """
+    cdef uint64_t result
+    result = (<uint64_t>(src[pos + 0]) << 56) | \
+             (<uint64_t>(src[pos + 1]) << 48) | \
+             (<uint64_t>(src[pos + 2]) << 40) | \
+             (<uint64_t>(src[pos + 3]) << 32) | \
+             (<uint64_t>(src[pos + 4]) << 24) | \
+             (<uint64_t>(src[pos + 5]) << 16) | \
+             (<uint64_t>(src[pos + 6]) << 8) | \
+             (<uint64_t>(src[pos + 7]))
+    return result
+
+
+cdef inline uint32_t load32_at(const uint8_t[::1] src,
+                      uint64_t pos) noexcept nogil :
+    """Read a 32-bit BIG-endian integer at the given position in the stream.
+
+    :param src: byte array containing the data
+    :param pos: position in the stream
+    :return: 32-bit unsigned integer
+    """
+    cdef uint32_t result
+    result = (<uint32_t>(src[pos + 0]) << 24) | \
+             (<uint32_t>(src[pos + 1]) << 16) | \
+             (<uint32_t>(src[pos + 2]) <<  8) | \
+             (<uint32_t>(src[pos + 3]))
+    return result
+
+
+def unblock_bslz4(bytes src):
+    """
+    Parse a compressed Bitshuffle-LZ4 stream and record the start of each LZ4-block
+
+    :param src: compressed LZ4 stream
+    :return: list of start of block in the input stream (as numpy array)
+    """
+    cdef:
+        uint32_t size = len(src)
+        uint64_t total_nbytes = load64_at(src, 0)
+        uint32_t block_nbytes = load32_at(src, 8)
+        const uint8_t[::1] buffer = src
+        uint32_t block_max = min(size//4+1, 1<<32-1)
+        uint64_t[::1] block_start = numpy.empty(block_max, dtype=numpy.uint64)
+        uint32_t block_idx = 0
+        uint64_t pos = 12
+        uint64_t end = pos + 4
+        uint64_t block_size
+
+    with nogil:
+        while ((end + 4 < size) and (block_idx < block_max)):
+            block_size = load32_at(buffer, pos)
+            block_start[block_idx] = end
+            block_idx +=1
+            pos = end + block_size
+            end = pos + 4
+
+    return numpy.asarray(block_start[:block_idx])

src/silx/opencl/codec/ext/meson.build

@@ -0,0 +1,5 @@
+py.extension_module('_unblock_bslz4', '_unblock_bslz4.pyx',
+        subdir: 'silx/opencl/codec',
+        dependencies : [py_dep],
+        install: true,
+        )

src/silx/opencl/codec/meson.build

@@ -0,0 +1,18 @@
+subdir('ext')
+
+py.install_sources([
+    '__init__.py',
+    'bitshuffle_lz4.py',
+    'byte_offset.py'
+],
+subdir: 'silx/opencl/codec',  # Folder relative to site-packages to install to
+)
+
+
+pure_subdirs = [
+  'test',
+]
+
+foreach subdir: pure_subdirs
+  install_subdir(subdir, install_dir: silx_dir / 'opencl/codec')
+endforeach

src/silx/opencl/codec/test/test_bitshuffle_lz4.py

@@ -34,7 +34,7 @@
 __contact__ = "jerome.kieffer@esrf.eu"
 __license__ = "MIT"
 __copyright__ = "2022 European Synchrotron Radiation Facility, Grenoble, France"
-__date__ = "07/11/2022"
+__date__ = "01/06/2026"
 
 import struct
 import numpy
@@ -123,3 +123,8 @@ def test_decompress_from_array(self, dtype, shape):
 
         res = bs.decompress(array).get()
         assert numpy.array_equal(res, ref.ravel()), "Checks decompression works"
+
+        # test cython accelerator:
+        ref_dev = bs.decompress(array, force_unblock_on_device=True).get()
+        res_cyt = bs.decompress(array, force_unblock_on_device=False).get()
+        assert numpy.array_equal(ref_dev, res_cyt), "Checks cython accelerator works"

src/silx/opencl/meson.build

@@ -0,0 +1,32 @@
+subdir('codec')
+
+py.install_sources([
+    '__init__.py',
+    'atomic.py',
+	'common.py',
+    'convolution.py',
+    'medfilt.py',
+    'projection.py',
+    'sinofilter.py',
+    'statistics.py',
+    'backprojection.py',
+    'conftest.py',
+    'image.py',
+    'linalg.py',
+    'processing.py',
+    'reconstruction.py',
+    'sparse.py',
+    'utils.py'
+],
+subdir: 'silx/opencl',  # Folder relative to site-packages to install to
+)
+
+
+pure_subdirs = [
+  'sift',
+  'test',
+]
+
+foreach subdir: pure_subdirs
+  install_subdir(subdir, install_dir: silx_dir / 'opencl')
+endforeach