From 0a47f4918062c3f95ab3f468ceb94d910e4a846a Mon Sep 17 00:00:00 2001 From: Eirik Stavestrand Date: Sat, 4 Jul 2026 21:03:21 +0200 Subject: [PATCH] Make reading from file handle thread safe by using the `io_ctx` parameter instead of a global mutable variable. Ref: https://github.com/Roche/pyreadstat/issues/342 --- README.md | 2 +- pyreadstat/_readstat_parser.pyx | 20 ++++++++------------ tests/test_narwhalified.py | 28 ++++++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index d1fb846..b4ff416 100644 --- a/README.md +++ b/README.md @@ -969,4 +969,4 @@ pull request to ReadStat first. [alchemyst](https://github.com/alchemyst): improvements to docstrings -[bmwiedemann](https://github.com/bmwiedemann), [toddrme2178 ](https://github.com/toddrme2178), [Martin Thorsen Ranang](https://github.com/mtr): improvements to source code +[bmwiedemann](https://github.com/bmwiedemann), [toddrme2178 ](https://github.com/toddrme2178), [Martin Thorsen Ranang](https://github.com/mtr), [Eirik Stavestrand](https://github.com/eirki): improvements to source code diff --git a/pyreadstat/_readstat_parser.pyx b/pyreadstat/_readstat_parser.pyx index 802fd4c..3de0814 100644 --- a/pyreadstat/_readstat_parser.pyx +++ b/pyreadstat/_readstat_parser.pyx @@ -852,7 +852,6 @@ cdef int handle_open(const char *u8_path, void *io_ctx) except READSTAT_HANDLER_ return -1 -cdef object _file_object_ctx = None cdef int pyobject_open_handler(const char *path, void *io_ctx) noexcept: """File is already open - this is a no-op""" @@ -864,13 +863,12 @@ cdef int pyobject_close_handler(void *io_ctx) noexcept: cdef ssize_t pyobject_read_handler(void *buf, size_t nbyte, void *io_ctx) noexcept: """Bridge Python file.read() to C read operation""" - global _file_object_ctx cdef bytes data cdef ssize_t bytes_read cdef char *data_ptr - + cdef object file_obj = io_ctx + try: - file_obj = _file_object_ctx data = file_obj.read(nbyte) bytes_read = len(data) if bytes_read > 0: @@ -882,18 +880,17 @@ cdef ssize_t pyobject_read_handler(void *buf, size_t nbyte, void *io_ctx) noexce cdef readstat_off_t pyobject_seek_handler(readstat_off_t offset, readstat_io_flags_t whence, void *io_ctx) noexcept: """Bridge Python file.seek() to C seek operation""" - global _file_object_ctx cdef int py_whence - + cdef object file_obj = io_ctx + try: - file_obj = _file_object_ctx if whence == READSTAT_SEEK_SET: py_whence = 0 elif whence == READSTAT_SEEK_CUR: py_whence = 1 else: # READSTAT_SEEK_END py_whence = 2 - + file_obj.seek(offset, py_whence) return file_obj.tell() except: @@ -916,11 +913,9 @@ cdef void check_exit_status(readstat_error_t retcode) except *: cdef void run_readstat_parser(char * filename, data_container data, py_file_extension file_extension, long row_limit, long row_offset, object file_obj=None) except *: """ Runs the parsing of the file by readstat library. - + If file_obj is provided, it will be used instead of filename for I/O operations. """ - global _file_object_ctx - cdef readstat_parser_t *parser cdef readstat_error_t error cdef readstat_metadata_handler metadata_handler @@ -959,11 +954,12 @@ cdef void run_readstat_parser(char * filename, data_container data, py_file_exte # Set up custom I/O handlers for file objects if file_obj is not None: - _file_object_ctx = file_obj + io_ctx = file_obj open_handler = pyobject_open_handler close_handler = pyobject_close_handler read_handler = pyobject_read_handler seek_handler = pyobject_seek_handler + readstat_set_io_ctx(parser, io_ctx) readstat_set_open_handler(parser, open_handler) readstat_set_close_handler(parser, close_handler) readstat_set_read_handler(parser, read_handler) diff --git a/tests/test_narwhalified.py b/tests/test_narwhalified.py index 3d9ec46..ddcc7f6 100644 --- a/tests/test_narwhalified.py +++ b/tests/test_narwhalified.py @@ -16,6 +16,7 @@ # ############################################################################# from datetime import datetime, timedelta, date +from concurrent.futures import ThreadPoolExecutor import unittest import os import sys @@ -1379,6 +1380,33 @@ def test_read_sav_file_handle(self): self.assertEqual(len(df.columns), len(self.df_pandas.columns)) self.assertEqual(len(df), len(self.df_pandas)) + def test_read_sav_file_handle_threads(self): + """Test reading SAV file from file-like object in multiple threads at once (tests thread safety)""" + sav_file = os.path.join(self.basic_data_folder, "sample.sav") + with open(sav_file, "rb") as f: + file_bytes = f.read() + + def read_sav_file(path): + with open(path, "rb") as fo: + df, meta = pyreadstat.read_sav(fo) + return df, meta + + num_threads = 10 + + with tempfile.TemporaryDirectory() as tmp_dir: + paths = [os.path.join(tmp_dir, f"sample{i}.sav") for i in range(num_threads)] + for path in paths: + with open(path, "wb") as f: + f.write(file_bytes) + + with ThreadPoolExecutor(max_workers=num_threads) as executor: + results = list(executor.map(read_sav_file, paths)) + + for df, meta in results: + self.assertEqual(len(df.columns), len(self.df_pandas.columns)) + self.assertEqual(len(df), len(self.df_pandas)) + self.assertListEqual(list(df.columns), list(self.df_pandas.columns)) + def test_read_sav_bytesio(self): """Test reading SAV file from BytesIO (simulates remote/streaming data)""" sav_file = os.path.join(self.basic_data_folder, "sample.sav")