Add multithreaded tests (#62)

* add multithreaded tests

* add comment

* Fix typo

---------

Co-authored-by: Jonathan Ehwald <github@ehwald.info>

Author: ngoldbaum

tests/test_archive_reader.py

@@ -2,6 +2,7 @@
 import os
 import sys
 import tarfile
+from concurrent.futures import ThreadPoolExecutor
 from random import randint
 
 import psutil
@@ -558,3 +559,28 @@ def test_unpack_raises_when_archive_entries_are_not_iterable(target_path, archiv
             ArchiveUnpackingError, match="failed to iterate over archive"
         ):
             reader.unpack(target_path)
+
+
+def test_unpack_raises_when_sent_to_another_thread(
+    source_path, target_path, archive_path, write_mode, read_mode
+):
+    input_dir = source_path / "dir"
+    input_dir.mkdir()
+    input_file = input_dir / "file.txt"
+    input_file.touch()
+
+    with tarfile.open(archive_path, write_mode) as archive:
+        archive.add(input_dir)
+        archive.add(input_file)
+
+    def worker(reader, target_path):
+        reader.unpack(target_path)
+
+    # pyo3's PanicException isn't exported to Python and inherits from BaseException
+    # see https://github.com/PyO3/pyo3/issues/3918
+    with pytest.raises(
+        BaseException, match="fastar::reader::ArchiveReader is unsendable"
+    ):
+        with ArchiveReader.open(archive_path, read_mode) as reader:
+            with ThreadPoolExecutor(max_workers=1) as tpe:
+                tpe.submit(worker, reader, target_path).result()

tests/test_archive_writer.py

@@ -1,5 +1,7 @@
 import sys
 import tarfile
+import threading
+from concurrent.futures import ThreadPoolExecutor
 
 import psutil
 import pytest
@@ -1036,3 +1038,39 @@ def test_append_handles_arcnames_of_type_str(
     with tarfile.open(archive_path, read_mode) as archive:
         assert archive.getnames() == ["nested/file.txt"]
         assert archive.getmember("nested/file.txt").isfile()
+
+
+def test_multithreaded_append(
+    source_path, target_path, archive_path, write_mode, read_mode
+):
+    def worker(barrier, writer, target_path, lock, thread_index):
+        barrier.wait()
+        thread_target_path = target_path / f"thread_{thread_index}"
+        thread_target_path.touch()
+        # PyO3 ensures only one mutable borrow is allowed at a time so we use a
+        # lock. Calling append concurrently will raise "RuntimeError: already
+        # borrowed" here.
+        with lock:
+            writer.append(thread_target_path)
+
+    num_workers = 4
+    barrier = threading.Barrier(num_workers)
+    lock = threading.Lock()
+
+    with ArchiveWriter.open(archive_path, write_mode) as writer:
+        with ThreadPoolExecutor(max_workers=num_workers) as tpe:
+            try:
+                futures = []
+                for i in range(num_workers):
+                    futures.append(
+                        tpe.submit(worker, barrier, writer, target_path, lock, i)
+                    )
+            finally:
+                # avoid deadlocks if any threads failed to spawn
+                if len(futures) < num_workers:
+                    barrier.abort()
+        # join spawned threads
+        [f.result() for f in futures]
+
+    with tarfile.open(archive_path, read_mode) as archive:
+        assert sorted(archive.getnames()) == [f"thread_{i}" for i in range(num_workers)]