1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
from __future__ import annotations
import contextlib
import os
import pickle
import sys
import pytest
# 3.14.0b3+, though sys.implementation.supports_isolated_interpreters is being added in b4
# Can be simplified when we drop support for the first three betas
CONCURRENT_INTERPRETERS_SUPPORT = (
sys.version_info >= (3, 14)
and (
sys.version_info != (3, 14, 0, "beta", 1)
and sys.version_info != (3, 14, 0, "beta", 2)
)
and (
sys.version_info == (3, 14, 0, "beta", 3)
or sys.implementation.supports_isolated_interpreters
)
)
def get_interpreters(*, modern: bool):
if modern and CONCURRENT_INTERPRETERS_SUPPORT:
from concurrent import interpreters
def create():
return contextlib.closing(interpreters.create())
def run_string(
interp: interpreters.Interpreter,
code: str,
*,
shared: dict[str, object] | None = None,
) -> Exception | None:
if shared:
interp.prepare_main(**shared)
try:
interp.exec(code)
return None
except interpreters.ExecutionFailed as err:
return err
return run_string, create
if sys.version_info >= (3, 12):
interpreters = pytest.importorskip(
"_interpreters" if sys.version_info >= (3, 13) else "_xxsubinterpreters"
)
@contextlib.contextmanager
def create(config: str = ""):
try:
if config:
interp = interpreters.create(config)
else:
interp = interpreters.create()
except TypeError:
pytest.skip(f"interpreters module needs to support {config} config")
try:
yield interp
finally:
interpreters.destroy(interp)
def run_string(
interp: int, code: str, shared: dict[str, object] | None = None
) -> Exception | None:
kwargs = {"shared": shared} if shared else {}
return interpreters.run_string(interp, code, **kwargs)
return run_string, create
pytest.skip("Test requires the interpreters stdlib module")
@pytest.mark.skipif(
sys.platform.startswith("emscripten"), reason="Requires loadable modules"
)
def test_independent_subinterpreters():
"""Makes sure the internals object differs across independent subinterpreters"""
sys.path.append(".")
run_string, create = get_interpreters(modern=True)
m = pytest.importorskip("mod_per_interpreter_gil")
if not m.defined_PYBIND11_HAS_SUBINTERPRETER_SUPPORT:
pytest.skip("Does not have subinterpreter support compiled in")
code = """
import mod_per_interpreter_gil as m
import pickle
with open(pipeo, 'wb') as f:
pickle.dump(m.internals_at(), f)
"""
with create() as interp1, create() as interp2:
try:
res0 = run_string(interp1, "import mod_shared_interpreter_gil")
if res0 is not None:
res0 = str(res0)
except Exception as e:
res0 = str(e)
pipei, pipeo = os.pipe()
run_string(interp1, code, shared={"pipeo": pipeo})
with open(pipei, "rb") as f:
res1 = pickle.load(f)
pipei, pipeo = os.pipe()
run_string(interp2, code, shared={"pipeo": pipeo})
with open(pipei, "rb") as f:
res2 = pickle.load(f)
assert "does not support loading in subinterpreters" in res0, (
"cannot use shared_gil in a default subinterpreter"
)
assert res1 != m.internals_at(), "internals should differ from main interpreter"
assert res2 != m.internals_at(), "internals should differ from main interpreter"
assert res1 != res2, "internals should differ between interpreters"
@pytest.mark.skipif(
sys.platform.startswith("emscripten"), reason="Requires loadable modules"
)
@pytest.mark.skipif(not CONCURRENT_INTERPRETERS_SUPPORT, reason="Requires 3.14.0b3+")
def test_independent_subinterpreters_modern():
"""Makes sure the internals object differs across independent subinterpreters. Modern (3.14+) syntax."""
sys.path.append(".")
m = pytest.importorskip("mod_per_interpreter_gil")
if not m.defined_PYBIND11_HAS_SUBINTERPRETER_SUPPORT:
pytest.skip("Does not have subinterpreter support compiled in")
from concurrent import interpreters
code = """
import mod_per_interpreter_gil as m
values.put_nowait(m.internals_at())
"""
with contextlib.closing(interpreters.create()) as interp1, contextlib.closing(
interpreters.create()
) as interp2:
with pytest.raises(
interpreters.ExecutionFailed,
match="does not support loading in subinterpreters",
):
interp1.exec("import mod_shared_interpreter_gil")
values = interpreters.create_queue()
interp1.prepare_main(values=values)
interp1.exec(code)
res1 = values.get_nowait()
interp2.prepare_main(values=values)
interp2.exec(code)
res2 = values.get_nowait()
assert res1 != m.internals_at(), "internals should differ from main interpreter"
assert res2 != m.internals_at(), "internals should differ from main interpreter"
assert res1 != res2, "internals should differ between interpreters"
@pytest.mark.skipif(
sys.platform.startswith("emscripten"), reason="Requires loadable modules"
)
def test_dependent_subinterpreters():
"""Makes sure the internals object differs across subinterpreters"""
sys.path.append(".")
run_string, create = get_interpreters(modern=False)
m = pytest.importorskip("mod_shared_interpreter_gil")
if not m.defined_PYBIND11_HAS_SUBINTERPRETER_SUPPORT:
pytest.skip("Does not have subinterpreter support compiled in")
code = """
import mod_shared_interpreter_gil as m
import pickle
with open(pipeo, 'wb') as f:
pickle.dump(m.internals_at(), f)
"""
with create("legacy") as interp1:
pipei, pipeo = os.pipe()
run_string(interp1, code, shared={"pipeo": pipeo})
with open(pipei, "rb") as f:
res1 = pickle.load(f)
assert res1 != m.internals_at(), "internals should differ from main interpreter"
|
