Aaron Brady
3/8/2008 2:19:00 AM
On Mar 7, 4:07 pm, Steven D'Aprano <st...@REMOVE-THIS-
cybersource.com.au> wrote:
> > I want to be able to detect if [certain threads] fail with error,
>
> You can't? Why ever not?
Try this. ext can be found in 'C Function in a Python Context' on
google groops.
import ext
extA= ext.Ext()
extA[ 'araise' ]= r"""
int araise( int a, PyObject* exc ) {
int res= PyThreadState_SetAsyncExc( a, exc);
return res;
}
""", ("i","i","O")
LastCallException= type( 'LastCallException', ( Exception, ),
{ 'canstayhere': True } )
import thread
import threading
import time
partystart= threading.Event()
doorsclose= threading.Event()
def thd():
partystart.set()
try:
while 1:
print( 'running, ha ha!' )
time.sleep( .2 )
except Exception:
print( '\nclean thread exit\n' )
finally:
doorsclose.set()
partyid= thread.start_new_thread( thd, () )
partystart.wait()
print( 'waiting a second\n' )
time.sleep( 1 )
ret= extA.araise( partyid, LastCallException )
doorsclose.wait( 1 )
if not doorsclose.isSet():
print( 'Tell me about it.' )
print( 'clean exit\n' )
'''
waiting a second
running, ha ha!
running, ha ha!
running, ha ha!
running, ha ha!
running, ha ha!
clean thread exit
clean exit
'''