Finally, it is implemented using this link. The final code is shared below :-
import time
from concurrent.futures import ThreadPoolExecutor
import ctypes
def terminate_thread(thread):
"""Terminates a python thread from another thread.
:param thread: a threading.Thread instance
"""
if not thread.isAlive():
return
exc = ctypes.py_object(SystemExit)
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
ctypes.c_long(thread.ident), exc)
if res == 0:
raise ValueError("nonexistent thread id")
elif res > 1:
# """if it returns a number greater than one, you're in trouble,
# and you should call it again with exc=NULL to revert the effect"""
ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
raise SystemError("PyThreadState_SetAsyncExc failed")
def test1(a, b, c):
time.sleep(5)
d=a+b+c
print(d)
def test2(a, b):
time.sleep(5)
d=a+b
print(d)
with ThreadPoolExecutor(max_workers=1) as executor:
try:
executor.submit(test1, 1,2,3).result(timeout=1)
except:
executor.shutdown(wait=False)
for t in executor._threads:
terminate_thread(t)
print("Pass")
with ThreadPoolExecutor(max_workers=1) as executor:
try:
executor.submit(test2, 1,2).result(timeout=8)
except:
executor.shutdown(wait=False)
for t in executor._threads:
terminate_thread(t)
print("Pass-2")
与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…