mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 18:59:04 +08:00
Jeong, YunWon
7abf8cc247
- Use POP_TOP instead of POP_ITER for for-loop break/return cleanup - Expand duplicate_end_returns to clone final return for jump predecessors - Restrict late jump threading pass to unconditional jumps only - Skip exception blocks in inline/reorder passes - Simplify threaded_jump_instr NoInterrupt handling
78 lines
2.0 KiB
Python
78 lines
2.0 KiB
Python
import multiprocessing
|
|
import os
|
|
import threading
|
|
|
|
|
|
def import_in_thread(module_name):
|
|
outcome = {}
|
|
error = {}
|
|
|
|
def worker():
|
|
try:
|
|
module = __import__(module_name, fromlist=["*"])
|
|
outcome["name"] = module.__name__
|
|
except Exception as exc:
|
|
error["exc"] = exc
|
|
|
|
thread = threading.Thread(target=worker)
|
|
thread.start()
|
|
thread.join(timeout=5)
|
|
assert not thread.is_alive(), "thread did not finish in time"
|
|
if "exc" in error:
|
|
raise error["exc"]
|
|
|
|
assert outcome["name"] == module_name
|
|
|
|
|
|
def run_exec(code):
|
|
result = {}
|
|
error = {}
|
|
|
|
def worker():
|
|
try:
|
|
scope = {"__builtins__": __builtins__}
|
|
exec(code, scope, scope) # noqa: S102 - intentional threaded exec regression test
|
|
result["scope"] = scope
|
|
except Exception as exc:
|
|
error["exc"] = exc
|
|
|
|
thread = threading.Thread(target=worker)
|
|
thread.start()
|
|
thread.join(timeout=5)
|
|
assert not thread.is_alive(), "thread did not finish in time"
|
|
if "exc" in error:
|
|
raise error["exc"]
|
|
return result["scope"]
|
|
|
|
|
|
def child_process():
|
|
return None
|
|
|
|
|
|
def start_fork_process_after_thread():
|
|
if not hasattr(os, "fork"):
|
|
return
|
|
|
|
import_in_thread("multiprocessing.connection")
|
|
|
|
ctx = multiprocessing.get_context("fork")
|
|
process = ctx.Process(target=child_process)
|
|
process.start()
|
|
process.join(timeout=10)
|
|
assert process.exitcode == 0, process.exitcode
|
|
|
|
|
|
import_in_thread("functools")
|
|
import_in_thread("tempfile")
|
|
import_in_thread("multiprocessing.connection")
|
|
start_fork_process_after_thread()
|
|
|
|
scope = run_exec("import functools")
|
|
assert scope["functools"].__name__ == "functools"
|
|
|
|
scope = run_exec("from collections import namedtuple")
|
|
assert scope["namedtuple"].__name__ == "namedtuple"
|
|
|
|
scope = run_exec("module = __import__('multiprocessing.connection', fromlist=['*'])")
|
|
assert scope["module"].__name__ == "multiprocessing.connection"
|