DorothyLocker and Tool diff

Index: PyDS/Tool.py
===================================================================
RCS file: /pyds/PyDS/PyDS/Tool.py,v
retrieving revision 1.191
diff -r1.191 Tool.py
51a52
> from PyDS.DorothyLocker import DorothyRLock, ReleaseFailure, print_minitrace
117a119,125
> def traceargs(): 
> 	import inspect
> 	frame = inspect.currentframe().f_back
> 	name = frame.f_code.co_name
> 	args = inspect.formatargvalues(*inspect.getargvalues(frame))
> 	PyDS.Tool.debugOutput("%s%s" % (name, args))
> 
131c139
< def getFromWhereCalled():
---
> def getFromWhereCalled(frames=None):
133c141
< 	frames = traceback.extract_stack()
---
> 	if frames is None: frames = traceback.extract_stack()
1137c1145
< 		self.lock = threading.RLock()
---
> 		self.lock = DorothyRLock() 
1365,1381c1373,1389
< 		if blocking:
< 			if not self.lock.acquire(blocking=0):
< 				f = traceback.extract_stack()
< 				try: fun = f[-2][2]
< 				except: fun = '?'
< 				now = time.time()
< 				if self.lockedFrom:
< 					print "%s(%s): waiting for lock <%s> at %s" % (self.name, fun, self.lockedFrom or 'unknown', time.ctime(now))
< 				else:
< 					print "%s(%s): waiting for lock at %s" % (self.name, fun, time.ctime(now))
< 				self.lock.acquire(blocking=1)
< 				then = time.time()
< 				print "%s(%s): acquired lock at %s after %d seconds" % (self.name, fun, time.ctime(then), int(then-now))
< 			if _PyDS.verbose: self.lockedFrom = getFromWhereCalled()
< 				
< 		else:
< 			self.lock.acquire(blocking)
---
> 		try: 
> 			return self.lock.acquire(blocking)
> 		except ReleaseFailure, e: 
> 			msg, frame, t = e
> 			print "%s caught; lock was acquired in frame id %d at time %s" % (
> 			      msg, id(frame), time.ctime(t))
> 			print "Trace of the initial lock acquisition:"			
> 			print_minitrace(frame)
> 			print
> 			raise # let the caller know we had a problem
> 		except: 
> 			print "_acquire: "
> 			(e, d, tb) = sys.exc_info()
> 			print 'Exception %s: %s' % (e, d)
> 			for row in traceback.extract_tb(tb):
> 				print repr(row)
> 			raise # let the caller know we had a problem
1384,1385c1392,1407
< 		self.lockedFrom = None
< 		self.lock.release()
---
> 		try: 
> 			return self.lock.release()
> 		except ReleaseFailure, e: 
> 			msg, frame, t = e
> 			print "%s caught; lock was acquired in frame id %d at time %s" % (
> 			      msg, id(frame), time.ctime(t))
> 			print "Trace of the initial lock acquisition:"
> 			print_minitrace(frame)
> 			raise # let the caller know we had a problem
> 		except: 
> 			print "_release: "
> 			(e, d, tb) = sys.exc_info()
> 			print 'Exception %s: %s' % (e, d)
> 			for row in traceback.extract_tb(tb):
> 				print repr(row)
> 			raise # let the caller know we had a problem

"""
"There's no place like home." -- Dorothy

The DorothyLocker module provides a subclass of threading.RLock that 
insists upon release() being called from the same execution frame that 
acquire()d it in the first place. RLock itself is vulnerable to missing 
a release() or tossing in an acquire() too many times within a single 
thread, and it won't show until some *other* thread tries to acquire a 
lock you think is free, and blocks. 

As a PyDS-specific feature, DorothyRLock skips functions named in IGNORE. 
This is because each tool routes calls to it's lock's acquire method 
via PyDS.Tool._acquire, and similarly treats release. By ignoring frames 
from _acquire and _release, DorothyRLock can concentrate on the frames 
actually causing the locks and releases. 
"""

from threading import _RLock, currentThread, Thread
import time
import inspect

IGNORE = ['_acquire', '_release']

def ltrimCommon(oblist1, oblist2): 
	"Return the arguments with all common first elements removed."
	for pos in range(min(len(oblist1), len(oblist2))): 
		if oblist1[pos] is not oblist2[pos]: 
			break
	else: 
		pos = pos + 1
	return oblist1[pos:], oblist2[pos:]

def extract_minitrace(f):
	"Extract a mini-trace from a locking context."
	mycontext = context()
	items = []
	while f: 
		if f in mycontext: 
			break
		items.append((
			f.f_code.co_filename, 
			f.f_lineno, 
			f.f_code.co_name))
		f = f.f_back
	return items

def print_minitrace(f): 
	"Print a mini-trace from a locking context."
	for filename, line, name in extract_minitrace(f): 
		print 'File "%s", line %d, in %s' % (
		      filename, line, name)
		
def context(ignorecodes=[], ignorenames=[]): 
	"""Distil a calling context, ignoring certain code objects and 
	function names."""
	try: 
		stack = inspect.stack()
		codestack = []
		for frame, filename, lineno, co_name, lines, index in stack: 
			if frame.f_code is context.func_code \
			or frame.f_code in ignorecodes \
			or co_name in ignorenames: 
				continue
			codestack.append(frame)
		codestack.reverse()
		return tuple(codestack)
	finally: 
		del frame

class ReleaseFailure(Exception): 
	"Raised if DorothyRLock detects a failure to release an acquired lock."
	pass

class DorothyRLock(_RLock): 
	"""DorothyRLock insists upon release() being called from the same 
	execution frame that acquire()d it in the first place. Frames named 
	in DorothyLocker.IGNORE will be ignored.
	
	If a subframe re-acquires, that's okay, so long as it releases before 
	acquire or release is called via further up the frame stack."""
	
	def __init__(self, verbose=None): 
		"Initialise the RLock."
		_RLock.__init__(self, verbose)
		self.__threads = {}
		self.__ignores = [
			self.acquire.func_code, 
			self.release.func_code,
			self.callContext.func_code]
		
	def callContext(self): 
		"Distil our calling context, with instance-specific ignores."
		return context(self.__ignores, IGNORE)
		
	def acquire(self, blocking=1): 
		"""Acquire the lock, first checking that any previous locks have 
		been released."""
		# First, get our context: 
		me = currentThread()
		myLocks = self.__threads.setdefault(me, [])
		mycontext, myt = self.callContext(), time.time()

		# If there were any locks... 
		if myLocks: 
			# Try to figure out whether the previous lock should have been 
			# released. 
			# TODO: shouldn't we check all the locks? 
			prevcontext, prevt = myLocks[-1]
			prevframe = prevcontext[-1]
			# eliminate the common parts of the call stack
			pc, mc = ltrimCommon(prevcontext, mycontext)
			if pc:
				if mc: 
					# different parts of the call stack
					# => failure to release
					raise ReleaseFailure, ("failure to release", prevframe, prevt)
				else: 
					# mc matched, but shorter
					# => re-acquire from calling frame
					raise ReleaseFailure, ("failure to release", prevframe, prevt)
			else: 
				if mc: 
					# re-acquired from further down the call stack
					pass
				else: 
					# not pc AND not mc
					# => identical contexts
					raise ReleaseFailure, ("frame re-called acquire", prevframe, prevt)
				
		# If we got this far, it's safe to lock. 
		if blocking: 
			result = _RLock.acquire(self, blocking=0)
			if not result: 
				# Whine whilst we wait for the lock to clear. 
				whiner = LockWhiner(self, mycontext, myt, prevcontext, prevt)
				whiner.start()
				result = _RLock.acquire(self, blocking)
				whiner.stop()
		else: 
			result = _RLock.acquire(self, blocking=0)
			
		if result: 
			myLocks.append((mycontext, myt))

		return result

	def release(self): 
		"""Release the lock, first checking that we're releasing from the 
		same frame that acquired us."""
		result = _RLock.release(self) # raises AssertionError if not acquired in this thread
		context = self.callContext()
		me = currentThread()
		myLocks = self.__threads[me]
		prevcontext, prevt = myLocks[-1]
		if prevcontext == context: 
			myLocks.pop() # Permament! myLocks is bound to the 
			              # original, NOT a copy. 
			return
		raise ReleaseFailure, ("failure to release", prevcontext[-1], prevt)

class LockWhiner(Thread): 
	def __init__(self, dorothy, waitcontext, waittime, prevcontext, prevtime, every=10): 
		Thread.__init__(self)
		self.dorothy = dorothy
		self.whineTime = waittime
		self.whineContext = waitcontext
		self.lockTime = prevtime
		self.lockContext = prevcontext
		self.whineEvery = every
		self.thread = currentThread()
		self.active = 1
		self.setDaemon(1) # let PyDS shut down even if we're active
	
	def stop(self): 
		self.stopTime = time.time()
		self.active = 0
		print "%s: LockWhiner stopped; lock acquired after %.2f seconds" % (
		      time.ctime(self.stopTime), 
		      self.stopTime - self.whineTime)

	def run(self): 
		try: 
			#import pprint
			#pprint.pprint(self.__dict__)
			self._run()
		except: 
			print "_LockWhiner.run: Bugger."
			(e, d, tb) = sys.exc_info()
			print 'Exception %s: %s' % (e, d)
			for row in traceback.extract_tb(tb):
				print repr(row)

	def _run(self): 
		fun = self.waitStack[-1][2]
		print "%s: LockWhiner started by %s thread %d" % (\
		      time.ctime(self.whineTime), 
		      self.tool.name, 
		      id(self.thread))
		waits = 0
		while self.active: 
			time.sleep(self.whineEvery)
			waits = waits + 1
			now = time.time()
			print "%s: LockWhiner has been waiting for %.2fs" % (
			      time.ctime(),
			      now - self.whineTime)
			if waits==1: 
				print "-"*70
				print "Just in case this is a deadlock, here's some additional "
				print "information for debugging purposes:\n"
				print "Lock obtained: %s" % (time.ctime(self.lockTime))
				print "\nLock holding stack:"
				print_minitrace(self.lockContext[-1])
				print "\nWaiting stack:" 
				print_minitrace(self.whineContext[-1])
				print "\nLock held by:",
				lockOwner = self.dorothy._RLock__owner
				if lockOwner is None: 
					print "(None.)"
				else: 
					print id(lockOwner)
				print "-"*70

if __name__ == '__main__': 
	import pprint
	import traceback

	nrl = DorothyRLock()
	def spit(): 
		pprint.pprint(nrl.callContext())
	def a(): 
		spit()
	def b(): 
		a()
	a()
	print
	b()
	def _release():
		assert nrl.callContext()[-1].f_code.co_name != '_release'
	_release()
	
	nrl.acquire()
	nrl.release()

	def fail1_top(): 
		fail1_mid1()
		fail1_mid2()
	def fail1_mid1(): 
		nrl.acquire()
	def fail1_mid2(): 
		nrl.release()

	def fail2_top(): 
		fail2_mid1()
		fail2_mid2()
	def fail2_mid1(): 
		nrl.acquire()
	def fail2_mid2(): 
		nrl.acquire()
		nrl.release()

	for test in [fail1_top, fail2_top]: 
		try: 
			print "\nTest: %s" % test.func_name
			test()
		except ReleaseFailure, e: 
			msg, frame, t = e
			print "%s caught; lock was acquired in frame id %d at time %s" % (
			      msg, id(frame), time.ctime(t))
			print_minitrace(frame)