killableThread.py

'''
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'])