How to periodically call a function using multiprocessing in Python?

Question

I would like to call a function that checks a value on a bus periodically using the multiprocessing library (non-blocking desired). There is a way to do it using the threading library but it doesn't use multiple processes.

I've seen example code of this kind of functionality using the threading library but I'd like to achieve the same thing using the multiprocessing library. The official documentation for multiprocessing states:

"Note multiprocessing contains no analogues of threading.active_count(), threading.enumerate(), threading.settrace(), threading.setprofile(), threading.Timer"

however, in the examples I've seen for the threading library, they use threading.Timer. Is there a similar function for multiprocessing?

import time, threading
def foo():
    print(time.ctime())
    threading.Timer(10, foo).start()

foo()

#output:
#Thu Dec 22 14:46:08 2011
#Thu Dec 22 14:46:18 2011
#Thu Dec 22 14:46:28 2011
#Thu Dec 22 14:46:38 2011

Executing periodic actions in Python

The code above is an example for used with the threading library. Also, I was wondering if that was bad practice because the threads are never terminated (.join()).

@wwii It's not a duplicate because I want to use the multiprocessing library functions to do this. I DO NOT want to use the threading library to make a periodic function call. — John, Aug 01 '19 at 19:26

score 1 · Accepted Answer · answered Aug 01 '19 at 19:45

Basically, there is no timer class in the multiprocessing module. It should be, but seems like that when they migrated from pyprocessing to multiprocessing they didn't include that part. It's explained here https://stackoverflow.com/a/25297863/6598433.

You can make a working timer for the multiprocessing library manually as dano posted in https://stackoverflow.com/a/25297758/6598433

from multiprocessing import Process, Event

class Timer(Process):
    def __init__(self, interval, function, args=[], kwargs={}):
        super(Timer, self).__init__()
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished.set()

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

Check here the full question: Why no Timer class in Python's multiprocessing module?

The best way to do a periodic call is by having a run() function that has a while loop. At the end of the while loop, you add a function that checks for a set period and for the initial time at the start of the while loop. Then the function waits for the difference between the current time and the start time if the period is greater than that. so ```if( time.time() - start_time > desired_period) { time.sleep( desired_period-(time.time()-start_time) ) }``` — John, Oct 22 '19 at 16:30

score 0 · Answer 2 · answered Jan 21 '22 at 06:18

To extend Nosvan's answer to have a truly periodic timer (well it drifts by a few milliseconds), you simply can extend the Timer class like this:

from typing import Callable


class PeriodicTimer(Timer):
    def __init__(self, interval: int, function: Callable):
        super(PeriodicTimer, self).__init__(interval, function)

    def run(self):
        while not self.finished_event.is_set():
            # function callback could set the stop event
            self.function(*self.args, **self.kwargs)
            self.finished_event.wait(self.timeout)

and may I suggest to modify the Timer class to not have mutable arguments in the constructor:

from multiprocessing import Process
from multiprocessing import Event
from typing import Callable


class Timer(Process):
    def __init__(self, timeout: int, function: Callable, args=None, kwargs=None):
        super(Timer, self).__init__()
        self.timeout = timeout
        self.function = function
        self.args = [] if args is None else args
        self.kwargs = {} if kwargs is None else kwargs
        self.finished_event = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished_event.set()

    def run(self):
        self.finished_event.wait(self.timeout)
        if not self.finished_event.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished_event.set()

Here is an exhaustive explanation of why to avoid mutable default arguments:

https://web.archive.org/web/20200221224620/http://effbot.org/zone/default-values.htm

How to periodically call a function using multiprocessing in Python?

2 Answers2