Fix thread safety issues (#24)

* fix: make `packed.load` and `binarywave.load` thread-safe

* test: add tests

* style: format code

Author: kmnhan

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()

igor2/packed.py

@@ -25,24 +25,30 @@
 # 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)
 SUPERCEDED_MASK = 0x8000  # Bit is set if the record is superceded by
 # 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,27 +83,27 @@ 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)
                 logger.debug(
                     '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']))

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()