Merge pull request #135 from pycompression/allow_threads

Add GIL escape to checksums and compress/decompress

Author: rhpvorderman

CHANGELOG.rst

@@ -7,6 +7,13 @@ Changelog
 .. This document is user facing. Please word the changes in such a way
 .. that users understand how the changes affect the new version.
 
+version 1.2.0-dev
+-----------------
++ Escape GIL when calling inflate, deflate, crc32 and adler32 functions to
+  just like in CPython. This allows for utilising more CPU cores in combination
+  with the threading module. This comes with a very slight cost in efficiency
+  for strict single-threaded applications.
+
 version 1.1.0
 -----------------
 + Added tests and support for Python 3.11.

src/isal/igzip_libmodule.c

@@ -23,7 +23,8 @@ Changes compared to CPython:
 
 typedef struct {
     PyObject_HEAD
-    PyObject *unused_data;
+    PyObject *unused_data;    
+    PyThread_type_lock lock;
     PyObject *zdict;
     uint8_t *input_buffer;
     Py_ssize_t input_buffer_size;
@@ -41,6 +42,7 @@ typedef struct {
 static void
 IgzipDecompressor_dealloc(IgzipDecompressor *self)
 {
+    PyThread_free_lock(self->lock);
     PyMem_Free(self->input_buffer);
     Py_CLEAR(self->unused_data);
     Py_CLEAR(self->zdict);
@@ -99,7 +101,10 @@ decompress_buf(IgzipDecompressor *self, Py_ssize_t max_length)
             else if (obuflen == -2)
                 break;
 
+            Py_BEGIN_ALLOW_THREADS;
             err = isal_inflate(&(self->state));
+            Py_END_ALLOW_THREADS;
+
             if (err != ISAL_DECOMP_OK){
                 isal_inflate_error(err);
                 goto error;
@@ -357,18 +362,23 @@ igzip_lib_IgzipDecompressor_decompress(IgzipDecompressor *self,
 {
     static char *keywords[] = {"", "max_length", NULL};
     static char *format = "y*|n:decompress";
+
+    PyObject *result = NULL;
     Py_buffer data = {NULL, NULL};
     Py_ssize_t max_length = -1;
 
     if (!PyArg_ParseTupleAndKeywords(
             args, kwargs, format, keywords, &data, &max_length)) {
         return NULL;
     }
-    PyObject *result = NULL;
-    if (self->eof)
+    ENTER_ZLIB(self);
+    if (self->eof) {
         PyErr_SetString(PyExc_EOFError, "End of stream already reached");
-    else
+    }
+    else {
         result = decompress(self, data.buf, data.len, max_length);
+    }
+    LEAVE_ZLIB(self);
     PyBuffer_Release(&data);
     return result;
 }
@@ -416,6 +426,12 @@ igzip_lib_IgzipDecompressor__new__(PyTypeObject *type,
         Py_CLEAR(self);
         return NULL;
     }
+    self->lock = PyThread_allocate_lock();
+    if (self->lock == NULL) {
+        Py_DECREF(self);
+        PyErr_SetString(PyExc_MemoryError, "Unable to allocate lock");
+        return NULL;
+    }
     isal_inflate_init(&(self->state));
     self->state.hist_bits = hist_bits;
     self->state.crc_flag = flag;

src/isal/isal_shared.h

@@ -28,6 +28,14 @@ Changes compared to CPython:
 
 static PyObject *IsalError;
 
+#define ENTER_ZLIB(obj) do {                      \
+    if (!PyThread_acquire_lock((obj)->lock, 0)) { \
+        Py_BEGIN_ALLOW_THREADS                    \
+        PyThread_acquire_lock((obj)->lock, 1);    \
+        Py_END_ALLOW_THREADS                      \
+    } } while (0)
+#define LEAVE_ZLIB(obj) PyThread_release_lock((obj)->lock);
+
 /* Initial buffer size. */
 #define DEF_BUF_SIZE (16*1024)
 #define DEF_MAX_INITIAL_BUF_SIZE (16 * 1024 * 1024)
@@ -317,7 +325,10 @@ igzip_lib_compress_impl(Py_buffer *data,
                         "Unsufficient memory for buffer allocation");
                 goto error;
             }
+
+            Py_BEGIN_ALLOW_THREADS
             err = isal_deflate(&zst);
+            Py_END_ALLOW_THREADS
 
             if (err != COMP_OK) {
                 isal_deflate_error(err);
@@ -376,7 +387,10 @@ igzip_lib_decompress_impl(Py_buffer *data, int flag,
                 goto error;
             }
 
+            Py_BEGIN_ALLOW_THREADS
             err = isal_inflate(&zst);
+            Py_END_ALLOW_THREADS
+
             if (err != ISAL_DECOMP_OK) {
                 isal_inflate_error(err);
                 goto error;

src/isal/isal_zlibmodule.c

@@ -174,7 +174,9 @@ isal_zlib_adler32(PyObject *module, PyObject *const *args, Py_ssize_t nargs)
             return NULL;
         }
     }
+    Py_BEGIN_ALLOW_THREADS
     value = isal_adler32(value, data.buf, (uint64_t)data.len);
+    Py_END_ALLOW_THREADS
     return_value = PyLong_FromUnsignedLong(value & 0xffffffffU);
     PyBuffer_Release(&data);
     return return_value;
@@ -219,7 +221,9 @@ isal_zlib_crc32(PyObject *module, PyObject *const *args, Py_ssize_t nargs)
             return NULL;
         }
     }
+    Py_BEGIN_ALLOW_THREADS
     value = crc32_gzip_refl(value, data.buf, (uint64_t)data.len);
+    Py_END_ALLOW_THREADS
     return_value = PyLong_FromUnsignedLong(value & 0xffffffffU);
     PyBuffer_Release(&data);
     return return_value;
@@ -325,6 +329,7 @@ typedef struct
     uint8_t *level_buf;
     PyObject *zdict;
     int is_initialised;
+    PyThread_type_lock lock;
     // isal_zstream should be at the bottom as it contains buffers inside the struct.
     struct isal_zstream zst;
 } compobject;
@@ -334,6 +339,7 @@ Comp_dealloc(compobject *self)
 {
     if (self->is_initialised && self->level_buf != NULL)
         PyMem_Free(self->level_buf);
+    PyThread_free_lock(self->lock);
     Py_XDECREF(self->zdict);
     Py_TYPE(self)->tp_free((PyObject *)self);
 }
@@ -348,6 +354,12 @@ newcompobject()
     self->is_initialised = 0;
     self->zdict = NULL;
     self->level_buf = NULL;
+    self->lock = PyThread_allocate_lock();
+    if (self->lock == NULL) {
+        Py_DECREF(self);
+        PyErr_SetString(PyExc_MemoryError, "Unable to allocate lock");
+        return NULL;
+    }
     return self;
 }
 
@@ -361,13 +373,15 @@ typedef struct
     int is_initialised;
     int method_set;
     char eof;
+    PyThread_type_lock lock;
     // inflate_state should be at the bottom as it contains buffers inside the struct.
     struct inflate_state zst;
 } decompobject;
 
 static void
 Decomp_dealloc(decompobject *self)
 {
+    PyThread_free_lock(self->lock);
     Py_XDECREF(self->unused_data);
     Py_XDECREF(self->unconsumed_tail);
     Py_XDECREF(self->zdict);
@@ -419,6 +433,12 @@ newdecompobject()
     if (self->unconsumed_tail == NULL) {
         Py_DECREF(self);
         return NULL;
+    }
+        self->lock = PyThread_allocate_lock();
+    if (self->lock == NULL) {
+        Py_DECREF(self);
+        PyErr_SetString(PyExc_MemoryError, "Unable to allocate lock");
+        return NULL;
     }
     return self;
 }
@@ -562,6 +582,8 @@ isal_zlib_Compress_compress_impl(compobject *self, Py_buffer *data)
     Py_ssize_t ibuflen, obuflen = DEF_BUF_SIZE;
     int err;
 
+    ENTER_ZLIB(self);
+
     self->zst.next_in = data->buf;
     ibuflen = data->len;
 
@@ -573,7 +595,9 @@ isal_zlib_Compress_compress_impl(compobject *self, Py_buffer *data)
             if (obuflen < 0)
                 goto error;
 
+            Py_BEGIN_ALLOW_THREADS
             err = isal_deflate(&self->zst);
+            Py_END_ALLOW_THREADS
 
             if (err != COMP_OK) {
                 isal_deflate_error(err);
@@ -591,6 +615,7 @@ isal_zlib_Compress_compress_impl(compobject *self, Py_buffer *data)
  error:
     Py_CLEAR(RetVal);
  success:
+    LEAVE_ZLIB(self);
     return RetVal;
 }
 
@@ -668,13 +693,16 @@ isal_zlib_Decompress_decompress_impl(decompobject *self, Py_buffer *data,
         }
         self->method_set = 1;
     }
-    self->zst.next_in = data->buf;
-    ibuflen = data->len;
 
     /* limit amount of data allocated to max_length */
     if (max_length && obuflen > max_length)
         obuflen = max_length;
 
+    ENTER_ZLIB(self);
+
+    self->zst.next_in = data->buf;
+    ibuflen = data->len;
+
     do {
         arrange_input_buffer(&(self->zst.avail_in), &ibuflen);
 
@@ -693,7 +721,10 @@ isal_zlib_Decompress_decompress_impl(decompobject *self, Py_buffer *data,
                 goto abort;
             }
 
+            Py_BEGIN_ALLOW_THREADS
             err = isal_inflate(&self->zst);
+            Py_END_ALLOW_THREADS
+
             if (err != ISAL_DECOMP_OK){
                 isal_inflate_error(err);
                 goto abort;
@@ -719,6 +750,7 @@ isal_zlib_Decompress_decompress_impl(decompobject *self, Py_buffer *data,
  abort:
     Py_CLEAR(RetVal);
  success:
+    LEAVE_ZLIB(self);
     return RetVal;
 }
 
@@ -732,8 +764,12 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode)
     /* Flushing with Z_NO_FLUSH is a no-op, so there's no point in
        doing any work at all; just return an empty string. */
     if (mode == Z_NO_FLUSH) {
-        return PyBytes_FromStringAndSize(NULL, 0);
-    } else if (mode == Z_FINISH) {
+        return PyBytes_FromStringAndSize(NULL, 0);        
+    } 
+    
+    ENTER_ZLIB(self);
+
+    if (mode == Z_FINISH) {
         self->zst.flush = FULL_FLUSH;
         self->zst.end_of_stream = 1;
     } else if (mode == Z_FULL_FLUSH){
@@ -756,7 +792,9 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode)
             goto error;
         }
 
+        Py_BEGIN_ALLOW_THREADS
         err = isal_deflate(&self->zst);
+        Py_END_ALLOW_THREADS
 
         if (err != COMP_OK) {
             isal_deflate_error(err);
@@ -784,6 +822,7 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode)
         Py_CLEAR(RetVal);
 
  error:
+    LEAVE_ZLIB(self);
     return RetVal;
 }
 
@@ -800,7 +839,10 @@ isal_zlib_Decompress_flush_impl(decompobject *self, Py_ssize_t length)
         return NULL;
     }
 
+    ENTER_ZLIB(self);
+
     if (PyObject_GetBuffer(self->unconsumed_tail, &data, PyBUF_SIMPLE) == -1) {
+        LEAVE_ZLIB(self);
         return NULL;
     }
 
@@ -817,7 +859,9 @@ isal_zlib_Decompress_flush_impl(decompobject *self, Py_ssize_t length)
             if (length < 0)
                 goto abort;
 
+            Py_BEGIN_ALLOW_THREADS
             err = isal_inflate(&self->zst);
+            Py_END_ALLOW_THREADS
 
             if (err != ISAL_DECOMP_OK) {
                 isal_inflate_error(err);
@@ -846,6 +890,7 @@ isal_zlib_Decompress_flush_impl(decompobject *self, Py_ssize_t length)
     Py_CLEAR(RetVal);
  success:
     PyBuffer_Release(&data);
+    LEAVE_ZLIB(self);
     return RetVal;
 }