Re: LP_c_ubyte buffer and length to python string?

On Sun, Jul 24, 2011 at 8:34 AM, Mads Kiilerich <[email protected]> wrote:

> Dan Stromberg wrote, On 07/23/2011 11:27 PM:
>
>> I've googled on this for hours, and found nothing that seems close to what
>> I want to do.
>>
>> I have a C function that accepts a preallocated buffer and a pointer to an
>> integer.  It then stashes data in the buffer, and passes back the buffer's
>> "in use" length in the integer.
>>
>> I'm creating the buffer with:
>>
>>      ctypes_compressed_buffer_char_**p = ctypes.create_string_buffer(**
>> maximum_size)
>>      ctypes_compressed_buffer = ctypes.cast(ctypes_compressed_**buffer_char_p,
>> ctypes.POINTER(ctypes.c_ubyte)**)
>>
>> Then I create my length with:
>>
>>      ctypes_compressed_size = ctypes.c_size_t(0)
>>      ctypes_compressed_size_pointer = ctypes.cast(ctypes.addressof(**ctypes_compressed_size),
>> ctypes.c_void_p)
>>
>> The length is actually a size_t, but passing a pointer to size_t in ctypes
>> was giving me type errors, hence the void *.
>>
>> Anyway, the function executes without reporting any errors, so I seem to
>> have my result - in a ctypes format.  Printing its type reports that its an
>> LP_c_ubyte.
>>
>> How can I convert the resulting LP_c_ubyte (array) and int to a Python 2.x
>> str and a 3.x bytes?
>>
>
> I assume you want to get the content of the chunk of memory that was
> allocated by create_string_buffer.


Yes.


> The simplest way to get that is
>  ctypes_compressed_buffer_char_p.raw[:length]
>

Doing this is at least giving me something of the correct length.

Besides that: It seems strange that you have to cast your buffer and your
> size_t.


It might have something to do with size_t being an alias.


> ctypes gives you all the power and danger of both Python and C at once, so
> just because it executes without reporting errors doesn't mean that you got
> it right.


Sure.  It's positively correlated with correct operation, but not perfectly
correlated.  Actually, years ago I wrote an article about the dangers of
languages that allow undefined memory references:
http://stromberg.dnsalias.org/~dstromberg/checking-early.html

Also, I "prototyped" in C (a touch ironic, yes) before getting into the
Python+ctypes version.  The C version is at
http://stromberg.dnsalias.org/svn/xz_mod/branches/ctypes/c-test

My unit test can now roundtrip compress+decompress, so I feel mostly good
about the module, though I of course still have some lingering questions
about memory leaks and undefined memory references.

Is it the case that ctypes variables are garbage collected, as long as the
functions you're calling don't do their own malloc'ing or other form of
memory allocation?

You might want to post a minimal self-contained code snippet that shows what
> function you are calling and how you are doing it.
>

I'm calling 3 functions:
1) One gets an upper bound on how big the compressed size of a block can be
2) The next compresses a block of data
3) The last uncompresses a block of data

I'm not sure how self-contained this is, but here's the class.  Note that I
really only care about in-memory compression for this project, as I'm only
compressing chunks of modest size, and I want to write them to disk later in
a way that skips the buffer cache using
http://stromberg.dnsalias.org/~dstromberg/odirect/ :

class Xz_ctypes:
    LZMA_OK                        = 0
    LZMA_TELL_NO_CHECK             = 1
    LZMA_CHECK_CRC32               = 1
    LZMA_NO_CHECK                  = 2
    LZMA_TELL_UNSUPPORTED_CHECK    = 2
    LZMA_UNSUPPORTED_CHECK         = 3
    LZMA_MEM_ERROR                 = 5
    LZMA_MEMLIMIT_ERROR            = 6
    LZMA_PRESET_DEFAULT            = 6
    LZMA_FORMAT_ERROR              = 7
    LZMA_OPTIONS_ERROR             = 8
    LZMA_DATA_ERROR                = 9
    LZMA_BUF_ERROR                 = 10
    LZMA_PROG_ERROR                = 11

    funcs = {}

    def __init__(self):
        pass

    @classmethod
    def declare_c_function(cls, library, name, argtypes=None, restype=None):
        '''Extract functions from liblzma'''
        try:
            func = getattr(cls.LZMA_LIB, '_%s' % name)
            cls.funcs[name] = func
        except AttributeError:
            func = getattr(cls.LZMA_LIB, name)
            cls.funcs[name] = func
        if argtypes is not None:
            func.argtypes = argtypes
        if restype is not None:
            func.restype = restype

    @classmethod
    def class_init(cls):
        cls.LZMA_LIBPATH = _find_lib('lzma')
        cls.C_LIBPATH = _find_lib('c')

        # *ix way - windows is different
        cls.LZMA_LIB = ctypes.CDLL(cls.LZMA_LIBPATH)
        cls.C_LIB = ctypes.CDLL(cls.LZMA_LIBPATH)

        cls.declare_c_function(cls.LZMA_LIB, 'lzma_stream_buffer_bound',
(ctypes.c_size_t, ), ctypes.c_size_t)
        #cls.declare_c_function(cls.C_LIB, 'malloc', (ctypes.c_size_t, ),
ctypes.c_void_p)
        #cls.declare_c_function(cls.C_LIB, 'free', (ctypes.c_void_p, ))

        # lzma_easy_buffer_encode notes:
        # Argument 1, uint32_t preset:
        #    Just a uint32 - simple
        # Argument 2, lzma_check check:
        #    An enum to the C programmer - that should usually be an
unsigned integer in the C runtime.
        # Argument 3, lzma_allocator *lzma_allocator:
        #    This is really a pointer to a struct, but happily, we only need
to pass NULL to it, so we can just treat it as a void *
        # Argument 4: uint8_t *in:
        # Argument 5: size_t in_size:
        # Argument 6: uint8_t *out
        # Argument 7: size_t *out_pos
        # Argument 8: size_t out_size
        #
        # The return type is also an enum, so probably an unsigned int
        c_size_t_p = ctypes.POINTER(ctypes.c_size_t)
        cls.declare_c_function(
            cls.LZMA_LIB,
            'lzma_easy_buffer_encode',
                (
                ctypes.c_uint32,                    # preset
                ctypes.c_uint,                      # check
                ctypes.c_void_p,                    # lzma_allocator
                ctypes.POINTER(ctypes.c_uint8),     # uint8_t *in
                ctypes.c_size_t,                    # size_t in_size
                ctypes.POINTER(ctypes.c_uint8),     # uint8_t *out
                ctypes.c_void_p,                    # size_t *out_pos #
ctypes made size_t * think it was a long, so we use void_p
                ctypes.c_size_t,                    # size_t out_size
                ),
            ctypes.c_uint,
            )

        cls.declare_c_function(
            cls.LZMA_LIB,
            'lzma_stream_buffer_decode',
                (
                ctypes.POINTER(ctypes.c_uint64),    # uint64_t *memlimit
                ctypes.c_uint32,                    # uint32_t flags
                ctypes.c_void_p,                    # lzma_allocator
*allocator,
                ctypes.POINTER(ctypes.c_uint8),     # const uint8_t *in
                ctypes.c_void_p,                    # size_t *in_pos
                ctypes.c_size_t,                    # size_t in_size,
                ctypes.POINTER(ctypes.c_uint8),     # uint8_t *out
                ctypes.c_void_p,                    # size_t *out_pos
                ctypes.c_size_t,                    # size_t out_size
                ),
            ctypes.c_uint,
            )

    @classmethod
    def get_xz_error(cls, ret_xz):
        '''Decode an lzma_ret enum to an at-least-somewhat-descriptive
string'''
        if ret_xz == cls.LZMA_PRESET_DEFAULT:
            return 'LZMA_PRESET_DEFAULT'
        elif ret_xz == cls.LZMA_CHECK_CRC32:
            return 'LZMA_CHECK_CRC32'
        elif ret_xz == cls.LZMA_OK:
            return 'LZMA_OK'
        elif ret_xz == cls.LZMA_FORMAT_ERROR:
            return 'LZMA_FORMAT_ERROR'
        elif ret_xz == cls.LZMA_OPTIONS_ERROR:
            return 'LZMA_OPTIONS_ERROR'
        elif ret_xz == cls.LZMA_DATA_ERROR:
            return 'LZMA_DATA_ERROR'
        elif ret_xz == cls.LZMA_NO_CHECK:
            return 'LZMA_NO_CHECK'
        elif ret_xz == cls.LZMA_TELL_NO_CHECK:
            return 'LZMA_TELL_NO_CHECK'
        elif ret_xz == cls.LZMA_UNSUPPORTED_CHECK:
            return 'LZMA_UNSUPPORTED_CHECK'
        elif ret_xz == cls.LZMA_TELL_UNSUPPORTED_CHECK:
            return 'LZMA_TELL_UNSUPPORTED_CHECK'
        elif ret_xz == cls.LZMA_MEM_ERROR:
            return 'LZMA_MEM_ERROR'
        elif ret_xz == cls.LZMA_MEMLIMIT_ERROR:
            return 'LZMA_MEMLIMIT_ERROR: Memory usage limit was reached.
minimum required memlimit value was stored to *memlimit'
        elif ret_xz == cls.LZMA_BUF_ERROR:
            return 'LZMA_BUF_ERROR: Output buffer was too small'
        elif ret_xz == cls.LZMA_PROG_ERROR:
            return 'LZMA_PROG_ERROR'
        else:
            return 'Unrecognized lzma_ret value: %d' % ret_xz

    @classmethod
    def compress(cls, input_data):
        '''Compress data into xz format using ctypes to access liblzma.so'''
        # maximum_size = lzma_stream_buffer_bound(input_buffer_size);
        length_input_data = len(input_data)
        maximum_size =
cls.funcs['lzma_stream_buffer_bound'](length_input_data)

        # This is an efficient way of creating a readonly ctypes string
        ctypes_input_data_char_p = ctypes.c_char_p(input_data)
        ctypes_input_data = ctypes.cast(ctypes_input_data_char_p,
ctypes.POINTER(ctypes.c_ubyte))

        # This is a less-fast but mutable way of creating a ctypes string
        ctypes_compressed_buffer_char_p =
ctypes.create_string_buffer(maximum_size)
        ctypes_compressed_buffer =
ctypes.cast(ctypes_compressed_buffer_char_p, ctypes.POINTER(ctypes.c_ubyte))

        ctypes_compressed_size = ctypes.c_size_t(0)
        #ctypes_compressed_size_pointer = ctypes.POINTER(ctypes.c_size_t)
        ctypes_compressed_size_pointer =
ctypes.cast(ctypes.addressof(ctypes_compressed_size), ctypes.c_void_p)

        # lzma_easy_buffer_encode
        ret_xz = cls.funcs['lzma_easy_buffer_encode'](
          cls.LZMA_PRESET_DEFAULT,
            cls.LZMA_CHECK_CRC32,
            None,
            ctypes_input_data,
            length_input_data,
            ctypes_compressed_buffer,
            ctypes_compressed_size_pointer,
            maximum_size,
            )

        if ret_xz != cls.LZMA_OK:
            raise OSError(cls.get_xz_error(ret_xz))

        resultant_length = int(ctypes_compressed_size.value)
        result = ctypes_compressed_buffer_char_p.raw[:resultant_length]

        return result

    @classmethod
    def decompress(cls, input_data, max_result_size = 2 ** 24):
        '''Uncompress data from xz format using ctypes to access
liblzma.so'''
        ctypes_memlimit = ctypes.c_uint64(max_result_size)
        ctypes_memlimit_pointer =
ctypes.cast(ctypes.addressof(ctypes_memlimit),
ctypes.POINTER(ctypes.c_uint64))

        ctypes_input_data_char_p = ctypes.c_char_p(input_data)
        ctypes_input_data = ctypes.cast(ctypes_input_data_char_p,
ctypes.POINTER(ctypes.c_ubyte))

        ctypes_in_pos = ctypes.c_size_t(0)
        ctypes_in_pos_pointer = ctypes.cast(ctypes.addressof(ctypes_in_pos),
ctypes.c_void_p)

        in_size = ctypes.c_size_t(len(input_data))

        ctypes_uncompressed_buffer_char_p =
ctypes.create_string_buffer(max_result_size)
        ctypes_uncompressed_buffer =
ctypes.cast(ctypes_uncompressed_buffer_char_p,
ctypes.POINTER(ctypes.c_ubyte))

        ctypes_out_pos = ctypes.c_size_t(0)
        ctypes_out_pos_pointer =
ctypes.cast(ctypes.addressof(ctypes_out_pos), ctypes.c_void_p)

        out_size = ctypes.c_size_t(max_result_size)

        ret_xz = cls.funcs['lzma_stream_buffer_decode'](
            ctypes_memlimit_pointer,            # uint64_t *memlimit
            cls.LZMA_TELL_NO_CHECK,             # uint32_t flags
            None,                               # lzma_allocator *allocator,
            ctypes_input_data,                  # const uint8_t *in
            ctypes_in_pos_pointer,              # size_t *in_pos
            in_size,                            # size_t in_size,
            ctypes_uncompressed_buffer,         # uint8_t *out
            ctypes_out_pos_pointer,             # size_t *out_pos
            out_size,                           # size_t out_size
            )

        if ret_xz != cls.LZMA_OK:
            raise OSError(cls.get_xz_error(ret_xz))

        resultant_length = int(ctypes_out_pos.value)
        result = ctypes_uncompressed_buffer_char_p.raw[:resultant_length]

        return result

if HAVE_CTYPES:
    XZ_CTYPES = Xz_ctypes()
    XZ_CTYPES.class_init()

------------------------------------------------------------------------------
Magic Quadrant for Content-Aware Data Loss Prevention
Research study explores the data loss prevention market. Includes in-depth
analysis on the changes within the DLP market, and the criteria used to
evaluate the strengths and weaknesses of these DLP solutions.
http://www.accelacomm.com/jaw/sfnl/114/51385063/

_______________________________________________
ctypes-users mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/ctypes-users