Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/to_interpreter.py: 26%

1from __future__ import annotations 

2 

3__all__ = ( 

4 "current_default_interpreter_limiter", 

5 "run_sync", 

6) 

7 

8import atexit 

9import os 

10import sys 

11from collections import deque 

12from collections.abc import Callable 

13from typing import Any, Final, TypeVar 

14 

15from . import current_time, to_thread 

16from ._core._exceptions import BrokenWorkerInterpreter 

17from ._core._synchronization import CapacityLimiter 

18from .lowlevel import RunVar 

19 

20if sys.version_info >= (3, 11): 

21 from typing import TypeVarTuple, Unpack 

22else: 

23 from typing_extensions import TypeVarTuple, Unpack 

24 

25if sys.version_info >= (3, 14): 

26 from concurrent.interpreters import ExecutionFailed, create 

27 

28 def _interp_call( 

29 func: Callable[..., Any], args: tuple[Any, ...] 

30 ) -> tuple[Any, bool]: 

31 try: 

32 retval = func(*args) 

33 except BaseException as exc: 

34 return exc, True 

35 else: 

36 return retval, False 

37 

38 class _Worker: 

39 last_used: float = 0 

40 

41 def __init__(self) -> None: 

42 self._interpreter = create() 

43 

44 def destroy(self) -> None: 

45 self._interpreter.close() 

46 

47 def call( 

48 self, 

49 func: Callable[..., T_Retval], 

50 args: tuple[Any, ...], 

51 ) -> T_Retval: 

52 try: 

53 res, is_exception = self._interpreter.call(_interp_call, func, args) 

54 except ExecutionFailed as exc: 

55 raise BrokenWorkerInterpreter(exc.excinfo) from exc 

56 

57 if is_exception: 

58 raise res 

59 

60 return res 

61elif sys.version_info >= (3, 13): 

62 import _interpqueues 

63 import _interpreters 

64 

65 UNBOUND: Final = 2 # I have no clue how this works, but it was used in the stdlib 

66 FMT_UNPICKLED: Final = 0 

67 FMT_PICKLED: Final = 1 

68 QUEUE_PICKLE_ARGS: Final = (FMT_PICKLED, UNBOUND) 

69 QUEUE_UNPICKLE_ARGS: Final = (FMT_UNPICKLED, UNBOUND) 

70 

71 _run_func = compile( 

72 """ 

73import _interpqueues 

74from _interpreters import NotShareableError 

75from pickle import loads, dumps, HIGHEST_PROTOCOL 

76 

77QUEUE_PICKLE_ARGS = (1, 2) 

78QUEUE_UNPICKLE_ARGS = (0, 2) 

79 

80item = _interpqueues.get(queue_id)[0] 

81try: 

82 func, args = loads(item) 

83 retval = func(*args) 

84except BaseException as exc: 

85 is_exception = True 

86 retval = exc 

87else: 

88 is_exception = False 

89 

90try: 

91 _interpqueues.put(queue_id, (retval, is_exception), *QUEUE_UNPICKLE_ARGS) 

92except NotShareableError: 

93 retval = dumps(retval, HIGHEST_PROTOCOL) 

94 _interpqueues.put(queue_id, (retval, is_exception), *QUEUE_PICKLE_ARGS) 

95 """, 

96 "<string>", 

97 "exec", 

98 ) 

99 

100 class _Worker: 

101 last_used: float = 0 

102 

103 def __init__(self) -> None: 

104 self._interpreter_id = _interpreters.create() 

105 self._queue_id = _interpqueues.create(1, *QUEUE_UNPICKLE_ARGS) 

106 _interpreters.set___main___attrs( 

107 self._interpreter_id, {"queue_id": self._queue_id} 

108 ) 

109 

110 def destroy(self) -> None: 

111 _interpqueues.destroy(self._queue_id) 

112 _interpreters.destroy(self._interpreter_id) 

113 

114 def call( 

115 self, 

116 func: Callable[..., T_Retval], 

117 args: tuple[Any, ...], 

118 ) -> T_Retval: 

119 import pickle 

120 

121 item = pickle.dumps((func, args), pickle.HIGHEST_PROTOCOL) 

122 _interpqueues.put(self._queue_id, item, *QUEUE_PICKLE_ARGS) 

123 exc_info = _interpreters.exec(self._interpreter_id, _run_func) 

124 if exc_info: 

125 raise BrokenWorkerInterpreter(exc_info) 

126 

127 res = _interpqueues.get(self._queue_id) 

128 (res, is_exception), fmt = res[:2] 

129 if fmt == FMT_PICKLED: 

130 res = pickle.loads(res) 

131 

132 if is_exception: 

133 raise res 

134 

135 return res 

136else: 

137 

138 class _Worker: 

139 last_used: float = 0 

140 

141 def __init__(self) -> None: 

142 raise RuntimeError("subinterpreters require at least Python 3.13") 

143 

144 def call( 

145 self, 

146 func: Callable[..., T_Retval], 

147 args: tuple[Any, ...], 

148 ) -> T_Retval: 

149 raise NotImplementedError 

150 

151 def destroy(self) -> None: 

152 pass 

153 

154 

155DEFAULT_CPU_COUNT: Final = 8 # this is just an arbitrarily selected value 

156MAX_WORKER_IDLE_TIME = ( 

157 30 # seconds a subinterpreter can be idle before becoming eligible for pruning 

158) 

159 

160T_Retval = TypeVar("T_Retval") 

161PosArgsT = TypeVarTuple("PosArgsT") 

162 

163_idle_workers = RunVar[deque[_Worker]]("_available_workers") 

164_default_interpreter_limiter = RunVar[CapacityLimiter]("_default_interpreter_limiter") 

165 

166 

167def _stop_workers(workers: deque[_Worker]) -> None: 

168 for worker in workers: 

169 worker.destroy() 

170 

171 workers.clear() 

172 

173 

174async def run_sync( 

175 func: Callable[[Unpack[PosArgsT]], T_Retval], 

176 *args: Unpack[PosArgsT], 

177 limiter: CapacityLimiter | None = None, 

178) -> T_Retval: 

179 """ 

180 Call the given function with the given arguments in a subinterpreter. 

181 

182 .. warning:: On Python 3.13, the :mod:`concurrent.interpreters` module was not yet 

183 available, so the code path for that Python version relies on an undocumented, 

184 private API. As such, it is recommended to not rely on this function for anything 

185 mission-critical on Python 3.13. 

186 

187 :param func: a callable 

188 :param args: the positional arguments for the callable 

189 :param limiter: capacity limiter to use to limit the total number of subinterpreters 

190 running (if omitted, the default limiter is used) 

191 :return: the result of the call 

192 :raises BrokenWorkerInterpreter: if there's an internal error in a subinterpreter 

193 

194 """ 

195 if limiter is None: 

196 limiter = current_default_interpreter_limiter() 

197 

198 try: 

199 idle_workers = _idle_workers.get() 

200 except LookupError: 

201 idle_workers = deque() 

202 _idle_workers.set(idle_workers) 

203 atexit.register(_stop_workers, idle_workers) 

204 

205 async with limiter: 

206 try: 

207 worker = idle_workers.pop() 

208 except IndexError: 

209 worker = _Worker() 

210 

211 try: 

212 return await to_thread.run_sync( 

213 worker.call, 

214 func, 

215 args, 

216 limiter=limiter, 

217 ) 

218 finally: 

219 # Prune workers that have been idle for too long 

220 now = current_time() 

221 while idle_workers: 

222 if now - idle_workers[0].last_used <= MAX_WORKER_IDLE_TIME: 

223 break 

224 

225 await to_thread.run_sync(idle_workers.popleft().destroy, limiter=limiter) 

226 

227 worker.last_used = current_time() 

228 idle_workers.append(worker) 

229 

230 

231def current_default_interpreter_limiter() -> CapacityLimiter: 

232 """ 

233 Return the capacity limiter used by default to limit the number of concurrently 

234 running subinterpreters. 

235 

236 Defaults to the number of CPU cores. 

237 

238 :return: a capacity limiter object 

239 

240 """ 

241 try: 

242 return _default_interpreter_limiter.get() 

243 except LookupError: 

244 limiter = CapacityLimiter(os.cpu_count() or DEFAULT_CPU_COUNT) 

245 _default_interpreter_limiter.set(limiter) 

246 return limiter