Op 2005-12-16 said:
How to send alarm to a thread?
I can set alarm in main thread, but how then send exception to another thread to wake it if it executes too long?
The short answer is you can't and you should preferably look for an
other solution. If for example your other thread is simple loop
like:
while condition:
...
You could change this to:
while not timeout and condition:
timeout is then a variable that could be set by a differnt thread.
If your code doesn't lend itself for this kind of solution you could
consider the following module I once wrote. However some caveats:
1) You need ctypes.
2) It can only interrupt a thread while it is excuting python
code. No interruptions while in C-extention code.
3) It is not extensivly tested. It was just a proof of concept
thing.
----
import os
import ctypes
from time import sleep
from random import randint
class TimeOut(Exception):
pass
class Alarm(Exception):
pass
import threading
class Xthread(threading.Thread):
def start(self):
self.__original_run = self.run
self.run = self.__run
threading.Thread.start(self)
def __run(self):
self._thrd_id = threading._get_ident()
try:
self.__original_run()
finally:
self.run = self.__original_run
def raize(self, excpt):
Nr = ctypes.pythonapi.PyThreadState_SetAsyncExc(self._thrd_id, ctypes.py_object(excpt))
#print self.id , "Exception Thrown" , Nr
while Nr > 1:
#print self.id , "Exception Canceled" , Nr
ctypes.pythonapi.PyThreadState_SetAsyncExc(self._thrd_id, None)
sleep(0.1)
Nr = ctypes.pythonapi.PyThreadState_SetAsyncExc(self._thrd_id, ctypes.py_object(excpt))
#if Nr == 0:
# print self.id , "Exception Canceled" , Nr
# ctypes.pythonapi.PyThreadState_SetAsyncExc(self._thrd_id, None)
def alarm(self, tm):
alrm = threading.Timer(tm, self.raize, (TimeOut,))
alrm.start()
return alrm
if __name__ == "__main__":
class Continue(Xthread):
def run(self):
self.id = os.getpid()
print self.id, self._thrd_id, "Begin"
i = 0
try:
for _ in xrange(randint(0,20)):
for e in xrange(4 * 100000):
i = i + e
print self.id, "Finished"
except Alarm:
print self.id, "Interupted"
lst = [Continue() for _ in xrange(10)]
for T in lst:
T.start()
try:
sleep(10)
finally:
for T in lst:
T.raize(Alarm)
