Multithreading issues with postfilters
- 主要语言
- Python
- 星标
- 211
- 派生
- 58
- 平均合并
- 1 天 17 小时
- 30 天内合并 PR
- 6
描述
We are facing issues when using multithreading in combination with postfilters (with prefilters probably too). The next simple script hangs forever (probaly due to some pthread lock waiting to be unlocked):
```
from time import time
import blosc2
import numpy as np
# Size and dtype of super-chunks
# nchunks = 20_000
# chunkshape = 50_000
nchunks = 1
chunkshape = 20_000
dtype = np.dtype(np.int32)
chunksize = chunkshape * dtype.itemsize
# Set the compression and decompression parameters
cparams = {"clevel": 1, "codec": blosc2.Codec.LZ4HC, "typesize": 4, "nthreads": 8}
dparams = {"nthreads": 8}
storage = {"cparams": cparams, "dparams": dparams}
schunk = blosc2.SChunk(chunksize=chunksize, **storage)
data = np.zeros(chunkshape, dtype=dtype)
t0 = time()
for i in range(nchunks):
schunk.append_data(data)
print(f"time append: {time() - t0:.2f}s")
print(f"cratio: {schunk.cratio:.2f}x")
# Associate a postfilter to schunk
@schunk.postfilter(np.dtype(dtype))
def py_postfilter(input, output, offset):
output[:] = input + 1
#pass
t0 = time()
sum = 0
for chunk in schunk.iterchunks(dtype):
sum += chunk.sum()
print(f"time sum (postfilter): {time() - t0:.2f}s")
print(sum)
```
The output is:
```
time append: 0.00s
cratio: 2500.00x
Process finished with exit code 143 (interrupted by signal 15: SIGTERM)
```
Above the time for sum is not printed because I needed to kill the process (`killall python`). However, if we comment out the postfilter registration, the script finishes correctly:
```
time append: 0.00s
cratio: 2500.00x
time sum (postfilter): 0.00s
0
```
Furthermore, I have been able to create a C program doing exactly the same thing and I cannot reproduce the blocking:
```
#include
#include
#include
#define KB 1024.
#define MB (1024*KB)
#define GB (1024*MB)
#define CHUNKSIZE (20 * 1000)
//#define NCHUNKS 1000
#define NCHUNKS 1
#define NTHREADS 8
int postfilter_func(blosc2_postfilter_params *postparams) {
int nelems = postparams->size / postparams->typesize;
int32_t *in = ((int32_t *)(postparams->in));
int32_t *out = ((int32_t *)(postparams->out));
for (int i = 0; i < nelems; i++) {
out[i] = in[i] + 1;
}
return 0;
}
int main(void) {
blosc2_init();
static int32_t data[CHUNKSIZE];
static int32_t data_dest[CHUNKSIZE];
int32_t isize = CHUNKSIZE * sizeof(int32_t);
int i, nchunk;
int64_t nchunks;
blosc_timestamp_t last;
printf("Blosc version info: %s (%s)\n",
BLOSC2_VERSION_STRING, BLOSC2_VERSION_DATE);
/* Create a super-chunk container */
blosc2_cparams cparams = BLOSC2_CPARAMS_DEFAULTS;
cparams.typesize = sizeof(int32_t);
cparams.compcode = BLOSC_LZ4HC;
cparams.clevel = 1;
cparams.nthreads = NTHREADS;
blosc2_dparams dparams = BLOSC2_DPARAMS_DEFAULTS;
dparams.nthreads = NTHREADS;
// Set some postfilter parameters and function
dparams.postfilter = (blosc2_postfilter_fn)postfilter_func;
// We need to zero the contents of the postparams
blosc2_postfilter_params postparams = {0};
// In this case we are not passing any additional input in tpostparams
dparams.postparams = &postparams;
blosc2_storage storage = {.cparams=&cparams, .dparams=&dparams};
blosc2_schunk* schunk = blosc2_schunk_new(&storage);
// Add some data
blosc_set_timestamp(&last);
for (nchunk = 0; nchunk < NCHUNKS; nchunk++) {
for (i = 0; i < CHUNKSIZE; i++) {
data[i] = i + nchunk * CHUNKSIZE;
}
nchunks = blosc2_schunk_append_buffer(schunk, data, isize);
assert(nchunks == nchunk + 1);
}
/* Retrieve and decompress the chunks from the super-chunks and compare values */
for (nchunk = 0; nchunk < NCHUNKS; nchunk++) {
int32_t dsize = blosc2_schunk_decompress_chunk(schunk, nchunk, data_dest, isize);
if (dsize < 0) {
printf("Decompression error in schunk. Error code: %d\n", dsize);
return dsize;
}
/* Check integrity of this chunk */
for (i = 0; i < CHUNKSIZE; i++) {
assert (data_dest[i] == 1 + i + nchunk * CHUNKSIZE);
}
}
printf("Successful roundtrip schunk <-> frame <-> fileframe !\n");
/* Free resources */
blosc2_schunk_free(schunk);
blosc2_destroy();
return 0;
}
```
This runs perfectly well (even with 1000 chunks):
```
Blosc version info: 2.4.4.dev ($Date:: 2022-10-23 #$)
Successful roundtrip schunk <-> frame <-> fileframe !
Process finished with exit code 0
```
So the issues should be in python-blosc2. I strongly suspect that we are not correctly holding the GIL during the invocation of the postfilter, or, for some reason, calling Python code from multithreaded code is not safe. Needs some investigation. One possibility is to disable multithreading completely for pre-/post-filters.
贡献指南
调研方向
首先,在将 nthreads 设置为 8 并注册 postfilter 的情况下复现提供的 Python 脚本,然后将其行为与可正常工作的 C 示例进行比较。检查 Python 的 postfilter callback 以及 threading/GIL 处理;完成的标准是迭代能够在不挂起的情况下完成并产生预期结果,包括使用 pre- 或 postfilters 时也是如此。
由索引模型根据 Issue 内容生成。
评估
- 技术栈
- numpy, python
- 领域
- backend
- Issue 类型
- 缺陷
- 难度
- 4/5
- 预计耗时
- 3-5 天
- 活跃度
- 停滞
- 描述清晰度
- 基本清楚
- 新手友好度
- 35/100