|
7 | 7 | import os |
8 | 8 | import unittest |
9 | 9 | import tempfile |
| 10 | +import textwrap |
10 | 11 | import threading |
11 | 12 |
|
| 13 | +from test import support |
12 | 14 | from test.support.import_helper import import_module |
| 15 | +from test.support.script_helper import assert_python_ok |
13 | 16 | from test.support import threading_helper |
14 | 17 | from test.support import _1M |
15 | 18 |
|
@@ -1548,6 +1551,42 @@ def test_len(self): |
1548 | 1551 | self.assertEqual(len(TRAINED_DICT), len(TRAINED_DICT.dict_content)) |
1549 | 1552 | self.assertIn(str(len(TRAINED_DICT)), str(TRAINED_DICT)) |
1550 | 1553 |
|
| 1554 | + @unittest.skipUnless(SUPPORT_MULTITHREADING, |
| 1555 | + "zstd build doesn't support multi-threaded compression") |
| 1556 | + @support.requires_subprocess() |
| 1557 | + def test_collect_cycle_with_busy_compressor(self): |
| 1558 | + # The dictionary must outlive the compression context when both are |
| 1559 | + # collected by the garbage collector while a worker thread is still |
| 1560 | + # using the dictionary. Whether the failure is detected depends on |
| 1561 | + # the memory allocator, so use a fresh process for each case. |
| 1562 | + code = textwrap.dedent(""" |
| 1563 | + import gc, random, sys |
| 1564 | + from compression.zstd import (CompressionParameter, |
| 1565 | + ZstdCompressor, ZstdDict) |
| 1566 | +
|
| 1567 | + class Cycle: |
| 1568 | + pass |
| 1569 | +
|
| 1570 | + job_size = 1 << 20 |
| 1571 | + content = random.randbytes(1 << 10) * (1 << 8) |
| 1572 | + zd = ZstdDict(content, is_raw=True) |
| 1573 | + options = {CompressionParameter.compression_level: 10, |
| 1574 | + CompressionParameter.nb_workers: 1, |
| 1575 | + CompressionParameter.job_size: job_size} |
| 1576 | + c = ZstdCompressor(options=options, |
| 1577 | + zstd_dict=getattr(zd, sys.argv[1])) |
| 1578 | + cycle = Cycle() |
| 1579 | + cycle.cycle = cycle |
| 1580 | + cycle.compressor = c |
| 1581 | + # Returns while the worker thread is compressing the first job. |
| 1582 | + c.compress(random.randbytes(1 << 10) * ((job_size >> 10) + 64)) |
| 1583 | + del zd, c, cycle |
| 1584 | + gc.collect() |
| 1585 | + """) |
| 1586 | + for attr in 'as_digested_dict', 'as_prefix': |
| 1587 | + with self.subTest(attr=attr): |
| 1588 | + assert_python_ok('-c', code, attr) |
| 1589 | + |
1551 | 1590 | class FileTestCase(unittest.TestCase): |
1552 | 1591 | def setUp(self): |
1553 | 1592 | self.DECOMPRESSED_42 = b'a'*42 |
|
0 commit comments