Improve of threading.Timer
Hi, I recently needed to run a function every half a second and the first option was threading.Timer, but it just tick once. Based on https://stackoverflow.com/a/48741004 I derived a class from Timer to run endlessly, but there are two things I do not like: - The function, args and kwargs attributes of Timer are just the same as the _target, _args and _kwargs from Thread. - It feels better that Timer can "tick" a given number of times. So I changed Timer to be more inline with the Thread class and make it possible to run the function more than once: class Timer(Thread): FOREVER = Ellipsis def __init__(self, interval, function, args=None, kwargs=None, name=None, *, ticks=1, daemon=None): # Keep compatibility with the old Timer, where the default # args and kwargs where mutable. if args is None: args = [] if kwargs is None: kwargs = {} Thread.__init__(self, target=function, args=args, kwargs=kwargs, name=name, daemon=daemon) self.interval = interval self.ticks = ticks self.finished = Event() def cancel(self): """Stop the timer if it hasn't finished yet.""" self.finished.set() def run(self): try: while not self.finished.wait(self.interval): if self._target and not self.finished.is_set(): self._target(*self._args, **self._kwargs) if self.ticks is self.FOREVER: continue self.ticks -= 1 if self.ticks <= 0: break finally: # Remove the reference to the function for the same # reason it is done in Thread.run del self._target # However, for backwards compatibility do not remove # _args and _kwargs as is done in Thread.run # del self._args, self._kwargs self.finished.set() def join(self, timeout=None): # If the timer was set to repeat endlessly # it will never join if it is not cancelled before. if self.ticks is self.FOREVER: self.cancel() super().join(timeout=timeout) @property def function(self): return getattr(self, '_target', None) @function.setter def function(self, value): self._target = value @property def args(self): return self._args @args.setter def args(self, value): self._args = value @property def kwargs(self): return self._kwargs @kwargs.setter def kwargs(self, value): self._kwargs = value The default option for ticks is to run once, just like the current Timer. With this version it is possible to do things like:
Other option is to leave Timer as it is and create a new class, so it is not necessary to keep the compatibility with the function, args and kwargs attributes: class Ticker(Thread): FOREVER = Ellipsis def __init__(self, interval, target=None, args=(), kwargs=None, group=None, name=None, *, ticks=1, daemon=None): super().__init__(group=group, target=target, args=args, kwargs=kwargs, name=name, daemon=daemon) self.interval = interval self.ticks = ticks self.finished = Event() def cancel(self): """Stop the timer if it hasn't finished yet.""" self.finished.set() def evaluate(self): if self._target and not self.finished.is_set(): self._target(*self._args, **self._kwargs) def run(self): try: while not self.finished.wait(self.interval): self.evaluate() if self.ticks is self.FOREVER: continue self.ticks -= 1 if self.ticks <= 0: break finally: # Remove the reference to the function for the same # reason it is done in Thread.run del self._target, self._args, self._kwargs self.finished.set() def join(self, timeout=None): # If the timer was set to repeat endlessly # it will never join if it is not cancelled before. if self.ticks is self.FOREVER: self.cancel() super().join(timeout=timeout) Do you think it is useful to try to clean up and add this code to threading or it is just something too specific to consider changing the standard library? Regards, Andrés
Event scheduler library already exists and can be "tweaked" to become periodic. - https://docs.python.org/3/library/sched.html Best Regards, Ronie On Sun, Jun 9, 2019, 2:35 AM <killerrex@gmail.com> wrote:
Thanks for the hint, although using sched it is more complex for just one background activity, the priority part solves many problems with 2 or more activities in parallel. I managed to create a task with equivalent functionality using an action that re-schedules itself each time is called... the solution has some cutting edges: - I need an object to keep the event id or otherwise it is impossible to cancel the action (the event changes each time) - scheduler.run blocks the main thread (or you need to call it constantly) so it is necessary to call it in its own Thread. but it has the great benefit of controlling the priority between two or more actions. My original Ticker class is simpler but looking at this option with sched it may be too simple. Nevertheless for a user that just want one easy command run periodically (like printing to stdout a heartbeat message) maybe is worth to have it.
Event scheduler library already exists and can be "tweaked" to become periodic. - https://docs.python.org/3/library/sched.html Best Regards, Ronie On Sun, Jun 9, 2019, 2:35 AM <killerrex@gmail.com> wrote:
Thanks for the hint, although using sched it is more complex for just one background activity, the priority part solves many problems with 2 or more activities in parallel. I managed to create a task with equivalent functionality using an action that re-schedules itself each time is called... the solution has some cutting edges: - I need an object to keep the event id or otherwise it is impossible to cancel the action (the event changes each time) - scheduler.run blocks the main thread (or you need to call it constantly) so it is necessary to call it in its own Thread. but it has the great benefit of controlling the priority between two or more actions. My original Ticker class is simpler but looking at this option with sched it may be too simple. Nevertheless for a user that just want one easy command run periodically (like printing to stdout a heartbeat message) maybe is worth to have it.
participants (3)
-
Andrés Ayala -
killerrex＠gmail.com
-
Ronie Martinez