Grant Edwards <[EMAIL PROTECTED]> wrote: > On 2006-07-03, Nick Craig-Wood <[EMAIL PROTECTED]> wrote: > > Alex Martelli <[EMAIL PROTECTED]> wrote: > >> DarkBlue <[EMAIL PROTECTED]> wrote: > >> > try for 10 seconds > >> > if database.connected : > >> > do your remote thing > >> > except raise after 10 seconds > >> > abort any connection attempt > >> > do something else > >> > >> Sure, see function setdefaulttimeout in module socket. Just call it as > >> you wish before trying the DB connection and catch the resulting > >> exception. > > > > It would be nice to have language support for the general case, ie a > > general purpose timeout. > > I just use signal.alarm():
There are quite a lot of problems with alarm() 1) doesn't work on windows 2) there is only one alarm so nested alarms are tricky 3) alarm() only has 1 second resolution, however you can use the POSIX setitimer() which has better resolution 4) signals make havoc with some things (signals are basically interrupts passed to user space under unix) 5) alarm() doesn't play nice with threads All that said it does work under unix for a lot of programs. Here is a module which fixes point 2) anyway... It is a lot easier to follow than the threads example I posted earlier. There are still potential race conditions though even with something as simple as alarm()! ............................................................ """ Implement time limited function calls. Only works on platforms which support the signal module and signal.alarm() Eg from time_limit import time_limit, TimeOut def might_infinite_loop(arg): while 1: pass try: time_limit(10, might_infinite_loop, "some arg") except TimeOut: print "Oops took too long" else: print "Ran just fine" """ import signal import time import inspect class TimeOut(Exception): """Thrown on alarm""" pass class _NestedTimeOut(Exception): """Thrown on TimeOut detected for next layer""" pass def _sig_alarm(signum, frame): """Alarm handler for this module""" raise TimeOut() def time_limit(t_limit, fn, *args, **kwargs): """ Calls fn with the *args and **kwargs returning its result or raising a TimeOut exception if it doesn't complete within t seconds """ # Turn alarm off and read old value old_t_limit = signal.alarm(0) start_time = time.time() # Install new handler remembering old old_handler = signal.signal(signal.SIGALRM, _sig_alarm) # Set the timer going signal.alarm(t_limit) try: try: try: rc = fn(*args, **kwargs) except _NestedTimeOut: raise TimeOut() finally: # Disable alarm signal.alarm(0) finally: # Alarm will be disabled or will have gone off when we get here # Restore the old handler signal.signal(signal.SIGALRM, old_handler) # If there was an alarm running restore it if old_t_limit > 0: # Calculate how much of the old timeout is left elapsed_time = time.time() - start_time old_t_limit = int(round(old_t_limit - elapsed_time, 0)) if old_t_limit <= 0: # Alarm should have gone off so call old signal handler if old_handler == _sig_alarm: # If handler is our handler then... raise _NestedTimeOut() else: # Call old handler old_handler(signal.SIGALRM, inspect.currentframe()) else: # Re-enable alarm signal.alarm(old_t_limit) return rc if __name__ == "__main__": def _spin(t): """Spins for t seconds""" start = time.time() end = start + t while time.time() < end: pass def _test_time_limit(name, expecting_time_out, t_limit, fn, *args, **kwargs): """Test time_limit""" start = time.time() if expecting_time_out: print "Test",name,"should timeout" else: print "Test",name,"shouldn't timeout" try: time_limit(t_limit, fn, *args, **kwargs) except TimeOut, e: if expecting_time_out: print "Timeout generated OK" else: raise RuntimeError("Wasn't expecting TimeOut Here") else: if expecting_time_out: raise RuntimeError("Was expecting TimeOut Here") else: print "No timeout generated OK" elapsed = time.time() - start print "That took",elapsed,"seconds for timeout of",t_limit def test(): """Test code""" # no nesting _test_time_limit("simple #1", True, 5, _spin, 10) _test_time_limit("simple #2", False, 10, _spin, 5) # 1 level of nesting _test_time_limit("nested #1", True, 4, _test_time_limit, "nested #1a", True, 5, _spin, 10) _test_time_limit("nested #2", False, 6, _test_time_limit, "nested #2a", True, 5, _spin, 10) _test_time_limit("nested #3", True, 4, _test_time_limit, "nested #3a", False, 10, _spin, 5) _test_time_limit("nested #4", False, 6, _test_time_limit, "nested #4a", False, 10, _spin, 5) # 2 level of nesting _test_time_limit("nested #5", True, 3, _test_time_limit, "nested #5a", True, 4, _test_time_limit, "nested #5b", True, 5, _spin, 10) _test_time_limit("nested #6", True, 3, _test_time_limit, "nested #6a", False, 6, _test_time_limit, "nested #6b", True, 5, _spin, 10) _test_time_limit("nested #7", True, 3, _test_time_limit, "nested #7a", True, 4, _test_time_limit, "nested #7b", False, 10, _spin, 5) _test_time_limit("nested #8", True, 3, _test_time_limit, "nested #8a", False, 6, _test_time_limit, "nested #8b", False, 10, _spin, 5) _test_time_limit("nested #9", False, 7, _test_time_limit, "nested #9a", True, 4, _test_time_limit, "nested #9b", True, 5, _spin, 10) _test_time_limit("nested #10", False, 7, _test_time_limit, "nested #10a", False, 6, _test_time_limit, "nested #10b", True, 5, _spin, 10) _test_time_limit("nested #11", False, 7, _test_time_limit, "nested #11a", True, 4, _test_time_limit, "nested #11b", False, 10, _spin, 5) _test_time_limit("nested #12", False, 7, _test_time_limit, "nested #12a", False, 6, _test_time_limit, "nested #12b", False, 10, _spin, 5) print "All tests OK" test() -- Nick Craig-Wood <[EMAIL PROTECTED]> -- http://www.craig-wood.com/nick -- http://mail.python.org/mailman/listinfo/python-list