Add a test for sharing a dictionary across threads

Author: emmatyping

Lib/test/test_zstd.py

@@ -2506,6 +2506,28 @@ def run_method(method, input_data, output_data):
         self.assertEqual(expected, actual)
 
 
+    @threading_helper.reap_threads
+    @threading_helper.requires_working_threading()
+    def test_compress_shared_dict(self):
+        num_threads = 8
+        
+        def run_method(b):
+            level = threading.get_ident() % 2
+            # sync threads to increase chance of contention on
+            # capsule storing dictionary levels
+            b.wait()
+            ZstdCompressor(level=level, zstd_dict=TRAINED_DICT.as_digested_dict)
+        threads = []
+
+        b = threading.Barrier(num_threads)
+        for i in range(num_threads):
+            thread = threading.Thread(target=run_method, args=(b,))
+
+            threads.append(thread)
+
+        with threading_helper.start_threads(threads):
+            pass
+
 
 if __name__ == "__main__":
     unittest.main()