53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""vast_c 卡死用例定位：逐个用例子进程隔离 + 10s 超时，terminate 强杀，找出挂死/异常用例。
|
||
|
||
注：ProcessPoolExecutor 无法终止正在执行的 future（shutdown 会等 worker 退出 → 主脚本卡死），
|
||
必须用 multiprocessing.Process + terminate() 强杀。
|
||
"""
|
||
import sys
|
||
import multiprocessing as _Mp
|
||
import queue as _Queue
|
||
|
||
sys.path.insert(0, r'd:\Users\TermiNexus\Desktop\TransPyC')
|
||
sys.path.insert(0, r'd:\Users\TermiNexus\Desktop\TransPyC\lib')
|
||
|
||
from _test_vast_c_diff import CASES
|
||
|
||
|
||
def _RunOne(Src, OutQ):
|
||
try:
|
||
import vast_c
|
||
T = vast_c.parse(Src)
|
||
OutQ.put(('ok', len(repr(T))))
|
||
except BaseException as E: # noqa: BLE001
|
||
OutQ.put(('err', f'{type(E).__name__}: {str(E)[:100]}'))
|
||
|
||
|
||
def main():
|
||
_Mp.freeze_support()
|
||
for I, Src in enumerate(CASES):
|
||
OutQ = _Mp.Queue()
|
||
P = _Mp.Process(target=_RunOne, args=(Src, OutQ), daemon=True)
|
||
P.start()
|
||
P.join(timeout=10)
|
||
if P.is_alive():
|
||
print(f'HANG [{I}] {Src[:60]!r}')
|
||
P.terminate()
|
||
P.join()
|
||
print('已强杀卡死进程')
|
||
return
|
||
try:
|
||
Status, Info = OutQ.get(timeout=1)
|
||
except (_Queue.Empty, Exception):
|
||
print(f'ERR [{I}] {Src[:60]!r} -> 无返回')
|
||
continue
|
||
if Status == 'ok':
|
||
print(f'OK [{I}] {Src[:60]!r}')
|
||
else:
|
||
print(f'ERR [{I}] {Src[:60]!r} -> {Info}')
|
||
print('=' * 40)
|
||
print('全部用例无卡死')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|