On 2007-05-08, Michele Simionato <[EMAIL PROTECTED]> wrote:
> On May 8, 4:53 am, [EMAIL PROTECTED] (Alex Martelli) wrote:
>> What do you expect from "timers on Linux" that you could not get with a
>> simple "sleep for the next N milliseconds"? A timer (on Linux or
>> elsewhere) can jog your process N milliseconds from now, e.g. with a
>> SIGALRM or SIGPROF, and you can set one with the setitimer syscall
>> (presumably accessible via ctypes, worst case -- I've never used it from
>> Python, yet), but how would that help you (compared to plain sleep,
>> select, poll, or whatever else best fits your need)?
>
> I hoped there was a library such thay I could register a Python
> callable (say
> a thunk) and having it called by the linux timer at time t without
> blocking
I once played with the following module to do something similar.
Maybe it is usefull to you as is, or can give you an idea on how
to proceed. I only tested it on linux.
---------------------------- alarm.py --------------------
m signal import signal, SIG_IGN, SIGALRM
from time import time
from thread import allocate_lock
from heapq import heappush, heappop
from os import kill, getpid
import errno
from select import select, error as SelectException
from ctypes import *
libc = cdll.LoadLibrary("/lib/libc.so.6")
class _timeval(Structure):
_fields_ = [("tv_sec" , c_long), ("tv_usec", c_long)]
def timeval(tm):
sec = int(tm)
usec = int(1000000 * (tm - sec))
return _timeval(sec, usec)
class itimerval(Structure):
_fields_ = [("it_interval", _timeval), ("it_value", _timeval)]
def alarm(tm):
tv = timeval(tm)
ti = timeval(0.0)
ntv = itimerval(ti, tv)
otv = itimerval(timeval(0.0), timeval(0.0))
rt = libc.setitimer(0, byref(ntv), byref(otv))
#print otv.it_value.tv_sec , otv.it_value.tv_usec
if rt:
raise ValueError
else:
return otv.it_value.tv_sec + otv.it_value.tv_usec / 1000000.0
def sleep(tm):
wakeup = time() + tm
while tm >= 0:
try:
select([],[],[],tm)
except SelectException , Err_Info:
#print dir(Err_Info)
if Err_Info[0] != errno.EINTR:
raise
tm = wakeup - time()
alarms = []
alarm_lock = allocate_lock()
def AlarmHandler(sgnr, frame):
alarm_lock.acquire()
now = time()
while alarms and alarms[0].moment <= now:
current = heappop(alarms)
if not current.canceled:
current.func(*current.args, **current.kwds)
current.executed = True
now = time()
alarm_lock.release()
if alarms:
#print alarms[0].moment - now, alarms
alarm(alarms[0].moment - now)
signal(SIGALRM, AlarmHandler)
class Alarm(object):
def __init__(self, tp, func, *args, **kwds):
alarm(0)
try:
alarm_lock.acquire()
self.canceled = False
self.executed = False
self.func = func
self.args = args
self.kwds = kwds
self.moment = tp
heappush(alarms, self)
now = time()
delta = alarms[0].moment - now
#print alarms
finally:
alarm_lock.release()
if delta <= 0:
pass
kill(getpid(), SIGALRM)
else:
alarm(delta)
def __cmp__(self, other):
return cmp(self.moment, other.moment)
def __str__(self):
return "<Alarm for %d>" % self.moment
__repr__ = __str__
def Cancel(self):
try:
alarm_lock.acquire()
if self.executed:
raise ValueError, "Cancelation was too late"
else:
self.canceled = True
except:
alarm_lock.release()
-----------------------------------------------------------------------------------------------
You use it as follows:
from alarm import Alarm
alert = Alarm(exucutemoment, function, positionalarguments, keywordarguments)
# unless alert.Cancel is called before the alert went off, the function
# with its arguments will be called at the specified time.
# If you are using threads, you are advised to do most of the work in a
# different thread and leave the main thread to only treat the alarms.
--
http://mail.python.org/mailman/listinfo/python-list
