Created
September 3, 2015 23:43
-
-
Save iakovgan/058a597f6beab87c0068 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
''' | |
This module provides: | |
Thread class: Thread with terminate method. 7 | |
based on this: http://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread-in-python/15274929#15274929 | |
Exemples: see tests | |
''' | |
import threading | |
import ctypes | |
import time | |
def _async_raise(tid, excobj): | |
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(excobj)) | |
if res == 0: # pragma: no cover | |
raise ValueError("nonexistent thread id") | |
elif res > 1: # pragma: no cover | |
# """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(tid, 0) | |
raise SystemError("PyThreadState_SetAsyncExc failed") | |
class Thread(threading.Thread): | |
def terminate(self): | |
if not self.isAlive(): | |
return | |
_async_raise(self.ident, SystemExit) # python 2.6 | |
def test_no_terminate(): | |
def quick(*args): | |
return sum(args) | |
tr = Thread(target=quick) | |
tr.start() | |
time_start = time.time() | |
time.sleep(0.3) | |
assert not tr.isAlive() | |
assert (time.time() - time_start) > 0.29 | |
def test_terminate_stops_the_thread(): | |
def long(*args): | |
time.sleep(1) | |
return sum(args) | |
tr = Thread(target=long) | |
tr.start() | |
time_start = time.time() | |
time.sleep(0.1) # let it finish | |
assert tr.isAlive() | |
tr.terminate() | |
assert (time.time() - time_start) < 1.0 | |
if __name__ == '__main__': # pragma: no cover | |
import nose | |
nose.run(argv=['', __file__, '-v']) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment