Skip to content

Commit

Permalink
Merge pull request #3787 from t20100/bslz4-buffer-input
Browse files Browse the repository at this point in the history 
`silx.opencl.codec` : Added support of pyopencl's `Buffer` and `Array` to `BitshuffleLz4.decompress`
  • Loading branch information
@kif
kif authored Apr 6, 2023
2 parents 77440ac + e925227 commit d5e322e
Show file tree
Hide file tree
Showing 2 changed files with 103 additions and 56 deletions.
72 changes: 43 additions & 29 deletions src/silx/opencl/codec/bitshuffle_lz4.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Project: Sift implementation in Python + OpenCL
# https://github.com/silx-kit/silx
#
# Copyright (C) 2022-2022 European Synchrotron Radiation Facility, Grenoble, France
# Copyright (C) 2022-2023 European Synchrotron Radiation Facility, Grenoble, France
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
Expand Down Expand Up @@ -50,9 +50,7 @@

class BitshuffleLz4(OpenclProcessing):
"""Perform the bitshuffle-lz4 decompression on the GPU
See :class:`OpenclProcessing` for optional arguments description.
:param int cmp_size:
Size of the raw stream for decompression.
It can be (slightly) larger than the array.
Expand Down Expand Up @@ -100,43 +98,59 @@ def __init__(self, cmp_size, dec_size, dtype,
self.compile_kernels([os.path.join("codec", "bitshuffle_lz4")])
self.block_size = min(self.block_size, kernel_workgroup_size(self.program, "bslz4_decompress_block"))

def decompress(self, raw, out=None, wg=None):
def decompress(self, raw, out=None, wg=None, nbytes=None):
"""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.
:return: The decompressed image as an pyopencl array.
:rtype: pyopencl.array
"""

events = []
with self.sem:
len_raw = numpy.uint64(len(raw))
if len_raw > self.cmp_size:
self.cmp_size = len_raw
logger.info("increase cmp buffer size to %s", self.cmp_size)
self.cl_mem["cmp"] = pyopencl.array.empty(self.queue, self.cmp_size, dtype=numpy.uint8)
wg = int(wg or self.block_size)

evt = pyopencl.enqueue_copy(self.queue,
self.cl_mem["cmp"].data,
raw,
is_blocking=False)
dest_size = struct.unpack(">Q", raw[:8])
self_dest_nbyte = self.dec_size * self.dec_dtype.itemsize
if dest_size<self_dest_nbyte:
num_blocks = numpy.uint32((dest_size+self.LZ4_BLOCK_SIZE-1) // self.LZ4_BLOCK_SIZE)
elif dest_size>self_dest_nbyte:
num_blocks = numpy.uint32((dest_size+self.LZ4_BLOCK_SIZE-1) // self.LZ4_BLOCK_SIZE)
self.cl_mem["dec"] = pyopencl.array.empty(self.queue,dest_size , self.dec_dtype)
self.dec_size = dest_size // self.dec_dtype.itemsize
if nbytes is not None:
assert nbytes <= raw.size
len_raw = numpy.uint64(nbytes)
elif isinstance(raw, pyopencl.Buffer):
len_raw = numpy.uint64(raw.size)
else:
len_raw = numpy.uint64(len(raw))

if isinstance(raw, pyopencl.array.Array):
cmp_buffer = raw.data
num_blocks = self.num_blocks

events.append(EventDescription("copy raw H -> D", evt))
elif isinstance(raw, pyopencl.Buffer):
cmp_buffer = raw
num_blocks = self.num_blocks
else:
if len_raw > self.cmp_size:
self.cmp_size = len_raw
logger.info("increase cmp buffer size to %s", self.cmp_size)
self.cl_mem["cmp"] = pyopencl.array.empty(self.queue, self.cmp_size, dtype=numpy.uint8)
evt = pyopencl.enqueue_copy(self.queue,
self.cl_mem["cmp"].data,
raw,
is_blocking=False)
events.append(EventDescription("copy raw H -> D", evt))
cmp_buffer = self.cl_mem["cmp"].data

dest_size = struct.unpack(">Q", raw[:8])
self_dest_nbyte = self.dec_size * self.dec_dtype.itemsize
if dest_size<self_dest_nbyte:
num_blocks = numpy.uint32((dest_size+self.LZ4_BLOCK_SIZE-1) // self.LZ4_BLOCK_SIZE)
elif dest_size>self_dest_nbyte:
num_blocks = numpy.uint32((dest_size+self.LZ4_BLOCK_SIZE-1) // self.LZ4_BLOCK_SIZE)
self.cl_mem["dec"] = pyopencl.array.empty(self.queue,dest_size , self.dec_dtype)
self.dec_size = dest_size // self.dec_dtype.itemsize
else:
num_blocks = self.num_blocks

wg = int(wg or self.block_size)

evt = self.program.lz4_unblock(self.queue, (1,), (1,),
self.cl_mem["cmp"].data,
cmp_buffer,
len_raw,
self.cl_mem["block_position"].data,
num_blocks,
Expand All @@ -148,9 +162,9 @@ def decompress(self, raw, out=None, wg=None):
else:
assert out.dtype == self.dec_dtype
assert out.size == self.dec_size

evt = self.program.bslz4_decompress_block(self.queue, (self.num_blocks*wg,), (wg,),
self.cl_mem["cmp"].data,
cmp_buffer,
out.data,
self.cl_mem["block_position"].data,
self.cl_mem["nb_blocks"].data,
Expand Down
87 changes: 60 additions & 27 deletions src/silx/opencl/codec/test/test_bitshuffle_lz4.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Project: Bitshuffle-LZ4 decompression in OpenCL
# https://github.com/silx-kit/silx
#
# Copyright (C) 2022-2022 European Synchrotron Radiation Facility,
# Copyright (C) 2022-2023 European Synchrotron Radiation Facility,
# Grenoble, France
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
Expand Down Expand Up @@ -36,22 +36,36 @@
__copyright__ = "2022 European Synchrotron Radiation Facility, Grenoble, France"
__date__ = "07/11/2022"

import logging
import struct
import numpy
import pytest

try:
import bitshuffle
except:
bitshuffle=None
except ImportError:
bitshuffle = None
from silx.opencl.common import ocl, pyopencl
from silx.opencl.codec.bitshuffle_lz4 import BitshuffleLz4
import unittest
logger = logging.getLogger(__name__)


@unittest.skipUnless(ocl and pyopencl and bitshuffle,
"PyOpenCl or bitshuffle is missing")
class TestBitshuffle(unittest.TestCase):
TESTCASES = ( # dtype, shape
("uint64", (103, 503)),
("int64", (101, 509)),
("uint32", (229, 659)),
("int32", (233, 653)),
("uint16", (743, 647)),
("int16", (751, 643)),
("uint8", (157, 1373)),
("int8", (163, 1367)),
)


@pytest.mark.skipif(
not ocl or not pyopencl or bitshuffle is None,
reason="PyOpenCl or bitshuffle is missing"
)
class TestBitshuffleLz4:
"""Test pyopencl bishuffle+LZ4 decompression"""

@staticmethod
def _create_test_data(shape, lam=100, dtype="uint32"):
Expand All @@ -62,28 +76,47 @@ def _create_test_data(shape, lam=100, dtype="uint32"):
:return: (reference image array, compressed stream)
"""
ref = numpy.random.poisson(lam, size=shape).astype(dtype)
raw = struct.pack(">Q", ref.nbytes) +b"\x00"*4+bitshuffle.compress_lz4(ref).tobytes()
raw = struct.pack(">Q", ref.nbytes) + b"\x00"*4 + bitshuffle.compress_lz4(ref).tobytes()
return ref, raw

def one_decompression(self, dtype, shape):
@pytest.mark.parametrize("dtype,shape", TESTCASES)
def test_decompress(self, dtype, shape):
"""
tests the byte offset decompression on GPU
Tests the byte offset decompression on GPU with various configuration
"""
ref, raw = self._create_test_data(shape=shape, dtype=dtype)
bs = BitshuffleLz4(len(raw), numpy.prod(shape), dtype=dtype)
res = bs.decompress(raw).get()
print(numpy.where(res-ref.ravel()))
self.assertEqual(numpy.all(res==ref.ravel()), True, "Checks decompression works")

def test_decompress(self):
"""
tests the byte offset decompression on GPU with various configuration
"""
self.one_decompression("uint64", (103,503))
self.one_decompression("int64", (101,509))
self.one_decompression("uint32", (229,659))
self.one_decompression("int32", (233,653))
self.one_decompression("uint16", (743,647))
self.one_decompression("int16", (751,643))
self.one_decompression("uint8", (157,1373))
self.one_decompression("int8", (163,1367))
assert numpy.array_equal(res, ref.ravel()), "Checks decompression works"

@pytest.mark.parametrize("dtype,shape", TESTCASES)
def test_decompress_from_buffer(self, dtype, shape):
"""Test reading compressed data from pyopencl Buffer"""
ref, raw = self._create_test_data(shape=shape, dtype=dtype)

bs = BitshuffleLz4(0, numpy.prod(shape), dtype=dtype)

buffer = pyopencl.Buffer(
bs.ctx,
flags=pyopencl.mem_flags.COPY_HOST_PTR | pyopencl.mem_flags.READ_ONLY,
hostbuf=raw,
)

res = bs.decompress(buffer).get()
assert numpy.array_equal(res, ref.ravel()), "Checks decompression works"

@pytest.mark.parametrize("dtype,shape", TESTCASES)
def test_decompress_from_array(self, dtype, shape):
"""Test reading compressed data from pyopencl Array"""
ref, raw = self._create_test_data(shape=shape, dtype=dtype)

bs = BitshuffleLz4(0, numpy.prod(shape), dtype=dtype)

array = pyopencl.array.to_device(
bs.queue,
numpy.frombuffer(raw, dtype=numpy.uint8),
array_queue=bs.queue,
)

res = bs.decompress(array).get()
assert numpy.array_equal(res, ref.ravel()), "Checks decompression works"

0 comments on commit d5e322e

Please sign in to comment.