From 04e532f4dea210c8bf1fec8cc48d97fe6ee6a1f9 Mon Sep 17 00:00:00 2001 From: Kimoon Han Date: Tue, 15 Apr 2025 18:14:39 +0900 Subject: [PATCH 1/3] fix: make `packed.load` and `binarywave.load` thread-safe --- igor2/binarywave.py | 34 +++++++++++++++++++--------------- igor2/packed.py | 45 +++++++++++++++++++++++++-------------------- 2 files changed, 44 insertions(+), 35 deletions(-) diff --git a/igor2/binarywave.py b/igor2/binarywave.py index a0390fc..957ff6c 100644 --- a/igor2/binarywave.py +++ b/igor2/binarywave.py @@ -793,18 +793,23 @@ def post_unpack(self, parents, data): count=0, array=True), ]) -Wave = _DynamicStructure( - name='Wave', - fields=[ - DynamicVersionField( - 'h', - 'version', - help='Version number for backwards compatibility.'), - DynamicWaveField( - Wave1, - 'wave', - help='The rest of the wave data.'), - ]) + +def setup_wave(byte_order='='): + wave = _DynamicStructure( + name='Wave', + fields=[ + DynamicVersionField( + 'h', + 'version', + help='Version number for backwards compatibility.'), + DynamicWaveField( + Wave1, + 'wave', + help='The rest of the wave data.'), + ], + byte_order=byte_order) + wave.setup() + return wave def load(filename): @@ -813,9 +818,8 @@ def load(filename): else: f = open(filename, 'rb') try: - Wave.byte_order = '=' - Wave.setup() - data = Wave.unpack_stream(f) + wave = setup_wave() + data = wave.unpack_stream(f) finally: if not hasattr(filename, 'read'): f.close() diff --git a/igor2/packed.py b/igor2/packed.py index 44b33df..3e4e665 100644 --- a/igor2/packed.py +++ b/igor2/packed.py @@ -25,17 +25,6 @@ # files, you must skip any record with a record type that is not # listed above. -PackedFileRecordHeader = _Structure( - name='PackedFileRecordHeader', - fields=[ - _Field('H', 'recordType', help='Record type plus superceded flag.'), - _Field('h', 'version', - help='Version information depends on the type of record.'), - _Field('l', 'numDataBytes', - help='Number of data bytes in the record following this' - 'record header.'), - ]) - # CR_STR = '\x15' (\r) PACKEDRECTYPE_MASK = 0x7FFF # Record type = (recordType & PACKEDREC_TYPE_MASK) @@ -43,6 +32,22 @@ # a later record in the packed file. +def setup_packed_file_record_header(byte_order='@'): + record_header = _Structure( + name='PackedFileRecordHeader', + fields=[ + _Field('H', 'recordType', help='Record type plus superceded flag.'), + _Field('h', 'version', + help='Version information depends on the type of record.'), + _Field('l', 'numDataBytes', + help='Number of data bytes in the record following this' + 'record header.'), + ], + byte_order=byte_order) + record_header.setup() + return record_header + + def load(filename, strict=True, ignore_unknown=True, initial_byte_order=None): """Load a packed experiment file. @@ -77,17 +82,17 @@ def load(filename, strict=True, ignore_unknown=True, initial_byte_order=None): initial_byte_order = '=' try: while True: - PackedFileRecordHeader.byte_order = initial_byte_order - PackedFileRecordHeader.setup() - b = bytes(f.read(PackedFileRecordHeader.size)) + header_struct = setup_packed_file_record_header( + byte_order=initial_byte_order) + b = bytes(f.read(header_struct.size)) if not b: break - if len(b) < PackedFileRecordHeader.size: + if len(b) < header_struct.size: raise ValueError( ('not enough data for the next record header ({} < {})' - ).format(len(b), PackedFileRecordHeader.size)) + ).format(len(b), header_struct.size)) logger.debug('reading a new packed experiment file record') - header = PackedFileRecordHeader.unpack_from(b) + header = header_struct.unpack_from(b) if header['version'] and not byte_order: need_to_reorder = _need_to_reorder_bytes(header['version']) byte_order = initial_byte_order = _byte_order(need_to_reorder) @@ -95,9 +100,9 @@ def load(filename, strict=True, ignore_unknown=True, initial_byte_order=None): 'get byte order from version: %s (reorder? %s)', byte_order, need_to_reorder) if need_to_reorder: - PackedFileRecordHeader.byte_order = byte_order - PackedFileRecordHeader.setup() - header = PackedFileRecordHeader.unpack_from(b) + header_struct = setup_packed_file_record_header( + byte_order=byte_order) + header = header_struct.unpack_from(b) logger.debug( 'reordered version: %s', header['version']) data = bytes(f.read(header['numDataBytes'])) From 6d8e86ff3cc571d938223d2311f2010a8ee2b815 Mon Sep 17 00:00:00 2001 From: Kimoon Han Date: Tue, 15 Apr 2025 18:14:44 +0900 Subject: [PATCH 2/3] test: add tests --- tests/test_pxp.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/test_pxp.py b/tests/test_pxp.py index d1b3bf6..3f258cc 100644 --- a/tests/test_pxp.py +++ b/tests/test_pxp.py @@ -1,3 +1,5 @@ +import threading + import numpy as np from igor2.packed import load as loadpxp @@ -152,3 +154,25 @@ def test_pxt(): 12647., 14242., 14470., 13913., 14158., 14754., 14462., 14346., 14219., 13467., 13595., 14331., 13960., 12934., 12897., 13557., 13105., 12797., 13234., 13053., 13455., 12825.], dtype='>f8')) + + +def test_thread_safe(): + + def worker(fileobj, thread_id): + expt = None + for bo in ('<', '>'): + try: + _, expt = loadpxp(fileobj, initial_byte_order=bo) + except ValueError: + pass + if expt is None: + raise ValueError(f"No experiment loaded for thread {thread_id}") + + threads = [] + for i, fname in enumerate([data_dir / 'packed-byteorder.pxt'] * 100): + t = threading.Thread(target=worker, args=(fname, i)) + threads.append(t) + t.start() + + for t in threads: + t.join() From aad6c0026bc1bdd0e58470c6c7e1b8932c5917b0 Mon Sep 17 00:00:00 2001 From: Kimoon Han Date: Tue, 15 Apr 2025 18:22:18 +0900 Subject: [PATCH 3/3] style: format code --- igor2/packed.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/igor2/packed.py b/igor2/packed.py index 3e4e665..06249e0 100644 --- a/igor2/packed.py +++ b/igor2/packed.py @@ -36,7 +36,8 @@ def setup_packed_file_record_header(byte_order='@'): record_header = _Structure( name='PackedFileRecordHeader', fields=[ - _Field('H', 'recordType', help='Record type plus superceded flag.'), + _Field('H', 'recordType', + help='Record type plus superceded flag.'), _Field('h', 'version', help='Version information depends on the type of record.'), _Field('l', 'numDataBytes',